camel_core/lifecycle/adapters/

route_controller_trait.rs

//! `RouteController` trait implementation for `DefaultRouteController`.
//!
//! Extracted from `route_controller.rs` to reduce file size. All lifecycle methods
//! (start, stop, suspend, resume, etc.) live here.

use std::sync::Arc;
use std::time::Duration;

use tokio::sync::mpsc;
use tokio_util::sync::CancellationToken;
use tower::Service;
use tracing::{error, info, warn};

use camel_api::metrics::MetricsCollector;
use camel_api::security_policy::RouteSecurityPlan;
use camel_api::{CamelError, InFlightClaim, NoOpMetrics, StepLifecycle, StepShutdownReason};
use camel_component_api::Consumer;
use camel_component_api::{ConcurrencyModel, ConsumerContext, consumer::ExchangeEnvelope};

use crate::lifecycle::adapters::consumer_management;
use crate::lifecycle::adapters::controller_component_context::ControllerComponentContext;
use crate::lifecycle::adapters::inline_dispatcher::RouteInlineDispatcher;
use crate::lifecycle::adapters::route_compiler::CANCEL_TOKEN;
use crate::lifecycle::adapters::route_controller::DefaultRouteController;
#[cfg(test)]
use crate::lifecycle::adapters::route_helpers::emit_start_route_event;
use crate::lifecycle::adapters::route_helpers::{
    DrainGuard, handle_is_running, inferred_lifecycle_label, ready_with_backoff,
    send_reply_or_b_prime,
};
use crate::lifecycle::adapters::route_registry::DEFAULT_SHUTDOWN_TIMEOUT;
use crate::lifecycle::adapters::route_runtime_state::CompiledRoute;

/// Operator acknowledgements for public exposure per bind address and the
/// per-bind exposure gate (ADR-0061).
///
/// Canonical home is `camel_auth::bind_gate` (moved in Task 2.6 so
/// transports — which may not reference `camel_core::`, see
/// `xtask lint-component-deps` — enforce the same gate; MCP's registry is
/// the first). Re-exported here so controller call sites, the CLI, and the
/// gate tests keep their historical import paths.
pub use camel_auth::bind_gate::{BindExposureAcks, enforce_bind_exposure_gate};

/// Canonical gate key + loopback classification for a listener `from` URI.
/// Only listener schemes (http/https/ws/wss/grpc) bind sockets, so only they gate;
/// `mcp:` binds live in `McpServerConfig` (gated at the McpServerRegistry level,
/// Task 2.6) and everything else (timer, direct) never binds.
pub(super) struct BindKey {
    pub(super) key: String,
    pub(super) loopback: bool,
}

pub(super) fn bind_key_from_uri(uri: &str) -> Option<BindKey> {
    let scheme = uri.split(':').next()?;
    if !matches!(scheme, "http" | "https" | "ws" | "wss" | "grpc") {
        return None;
    }
    let authority = uri.split("://").nth(1)?;
    let authority = authority.split('/').next()?;
    if authority.is_empty() {
        return None;
    }
    if let Ok(addr) = authority.parse::<std::net::SocketAddr>() {
        return Some(BindKey {
            key: addr.to_string(),
            loopback: addr.ip().is_loopback(),
        });
    }
    // Hostname authority: loopback only for `localhost` (deterministic,
    // fail-closed for every other hostname — no DNS). The host is the
    // authority minus its port; bracketed IPv6 authorities are stripped
    // to the bare host before the check.
    let host = authority
        .rsplit_once(':')
        .map(|(h, _)| h)
        .unwrap_or(authority)
        .trim_matches(['[', ']']);
    let loopback = host.eq_ignore_ascii_case("localhost");
    Some(BindKey {
        key: authority.to_string(),
        loopback,
    })
}

/// Wire the route's security context onto a freshly created consumer —
/// the start and resume paths share this delivery.
///
/// Policy-backed routes (`sp_config` + authenticator marker) receive the
/// policy with its credential sources, the named providers, and the
/// compiled plan. Without a policy, every staged server route still
/// carries a compiled plan (Task 1.2) — deliver it plan-only so consumers
/// enforce the kernel classification (e.g. strict dispatch) from day one.
/// Routes with neither get no context.
fn deliver_security_context(consumer: &mut dyn Consumer, compiled: &CompiledRoute) {
    use camel_component_api::SecurityContext;

    if let (Some(sp_config), Some(_)) = (
        compiled.security_policy.as_ref(),
        compiled.security_authenticator.as_ref(),
    ) {
        let mut sec_ctx = SecurityContext::from_arc(Arc::clone(&sp_config.policy))
            .with_credential_sources(sp_config.credential_sources.clone());
        // Inject the route's named providers so Phase-2 transports (grpc
        // 2.1, mcp 2.6, ws 2.8, http 2.9) can resolve them from the
        // SecurityContext instead of holding their own authenticator.
        if let Some(registry) = &compiled.provider_registry {
            sec_ctx = sec_ctx.with_providers(Arc::clone(registry));
        }
        // Thread the compiled plan (Task 1.8) so transports drive
        // per-route dispatch enforcement from it.
        if let Some(plan) = &compiled.security_plan {
            sec_ctx = sec_ctx.with_plan(plan.clone());
        }
        consumer.set_security_context(sec_ctx);
    } else if let Some(plan) = compiled.security_plan.clone() {
        // Plan-only delivery (Task 1.2): every staged server route
        // carries a compiled plan even without a policy declaration —
        // deliver it so consumers enforce the kernel classification
        // (e.g. strict dispatch) from day one.
        let mut sec_ctx = SecurityContext::from_plan(plan);
        if let Some(registry) = &compiled.provider_registry {
            sec_ctx = sec_ctx.with_providers(Arc::clone(registry));
        }
        consumer.set_security_context(sec_ctx);
    }
}

/// ADR-0061 Task 2.9 strict-mode dispatch check (the flip deferred from
/// Task 2.2): every transport mints the typed carrier at its request
/// boundary (grpc 2.1, mcp 2.6, ws 2.8, http 2.9), so a non-Public plan
/// REQUIRES the carrier on the Exchange — absent or wrong-provider is
/// denied BEFORE the pipeline runs; the transport renders the denial in
/// its own idiom via `reply_tx`. Returns `true` when the dispatch was
/// denied (caller must `continue`).
///
/// The denial reply is a reply to a dispatched send, so it carries the same
/// abandoned-receiver hazard as the pipeline reply sites: when the producer
/// timed out and dropped the receiver, the denial would vanish silently.
/// It therefore goes through [`send_reply_or_b_prime`] (rc-e2r9 review,
/// Important 1 disposition: covered, not out-of-contract — there is no
/// probe path here where the receiver is *expected* to be gone; the
/// fire-and-forget case is the `None` branch below, which keeps its plain
/// `warn!`). A `None` channel is a no-op for the helper.
fn strict_dispatch_denies(
    dispatch_plan: &Option<camel_api::security_policy::RouteSecurityPlan>,
    exchange: &camel_api::Exchange,
    reply_tx: &mut Option<tokio::sync::oneshot::Sender<Result<camel_api::Exchange, CamelError>>>,
    metrics: &Option<Arc<dyn MetricsCollector>>,
    route_id: &str,
) -> bool {
    if let Some(plan) = dispatch_plan.as_ref()
        && let Err(denial) = camel_auth::enforce_dispatch(plan, exchange)
    {
        if let Some(tx) = reply_tx.take() {
            send_reply_or_b_prime(Some(tx), Err(denial), metrics, route_id, "dispatch:denied");
        } else {
            // log-policy: handler-owned
            warn!(
                route_id = %route_id,
                error = %denial,
                "dispatch denied: no kernel carrier on Exchange"
            );
        }
        return true;
    }
    false
}

#[cfg(test)]
mod bind_key_tests {
    use super::bind_key_from_uri;

    #[test]
    fn https_and_wss_listeners_gate() {
        assert_eq!(
            bind_key_from_uri("https://0.0.0.0:8443/api").map(|b| b.key),
            Some("0.0.0.0:8443".to_string())
        );
        assert_eq!(
            bind_key_from_uri("wss://0.0.0.0:9000").map(|b| b.key),
            Some("0.0.0.0:9000".to_string())
        );
    }

    #[test]
    fn non_listener_schemes_skip() {
        assert!(bind_key_from_uri("timer:tick?period=1s").is_none());
        assert!(bind_key_from_uri("mcp:server/tool/x").is_none());
    }

    #[test]
    fn bracketed_ipv6_hostname_check_uses_bare_host() {
        let b = bind_key_from_uri("ws://[::1]:8080/path").expect("parses"); // allow-unwrap
        assert!(b.loopback, "[::1] is loopback");
    }

    #[test]
    fn localhost_authority_with_port_is_loopback() {
        // Regression: rsplit(':') once compared the PORT segment ("8080"),
        // never matching "localhost"; the host must exclude the port.
        let b = bind_key_from_uri("http://localhost:8080/api").expect("parses"); // allow-unwrap
        assert!(b.loopback, "localhost is loopback");
        let b = bind_key_from_uri("http://myhost.example:8080").expect("parses"); // allow-unwrap
        assert!(!b.loopback, "other hostnames stay non-loopback");
    }
}

/// Best-effort, reverse-order shutdown of already-started `StepLifecycle`
/// handles when `start_route` must abort. Used both mid-start-loop (the
/// `[0..idx)` already-started prefix) and for any post-start failure path
/// (e.g. `create_route_consumer`, the aggregate spawn branch, the consumer
/// startup handshake) so the ADR-0022 SPI holds: if `start_route` returns
/// `Err`, no started handle is left running.
///
/// Mirrors `StepLifecycle::shutdown`'s best-effort contract — each error is
/// logged and swallowed so one failing shutdown cannot block rollback of the
/// remaining handles.
async fn rollback_started(route_id: &str, handles: &[Arc<dyn StepLifecycle>]) {
    for handle in handles.iter().rev() {
        if let Err(e) = handle.shutdown(StepShutdownReason::RouteStop).await {
            warn!(
                route_id = %route_id,
                step = handle.name(),
                error = %e,
                "best-effort step shutdown during start rollback failed"
            );
        }
    }
}

#[async_trait::async_trait]
impl camel_api::RouteController for DefaultRouteController {
    async fn start_route(&mut self, route_id: &str) -> Result<(), CamelError> {
        // Check if route exists and can be started.
        {
            let managed = self
                .routes
                .get_mut(route_id)
                .ok_or_else(|| CamelError::RouteError(format!("Route '{}' not found", route_id)))?;

            let consumer_running = handle_is_running(&managed.consumer_handle);
            let pipeline_running = handle_is_running(&managed.pipeline_handle);
            if consumer_running && pipeline_running {
                return Ok(());
            }
            if !consumer_running && pipeline_running {
                return Err(CamelError::RouteError(format!(
                    "Route '{}' is suspended; use resume_route() to resume, or stop_route() then start_route() for full restart",
                    route_id
                )));
            }
            if consumer_running && !pipeline_running {
                return Err(CamelError::RouteError(format!(
                    "Route '{}' has inconsistent execution state; stop_route() then retry start_route()",
                    route_id
                )));
            }
        }

        info!(route_id = %route_id, "Starting route");

        // Get the resolved route info
        let (from_uri, pipeline, concurrency, dispatch_plan) = {
            let managed = self
                .routes
                .get(route_id)
                .expect("invariant: route must exist after prior existence check"); // allow-unwrap
            (
                managed.from_uri.clone(),
                Arc::clone(&managed.pipeline),
                managed.concurrency.clone(),
                managed.compiled.security_plan.clone(),
            )
        };

        // ADR-0061 per-bind exposure gate: refuse to start Public routes on
        // non-loopback binds without operator acknowledgement. Runs before
        // any lifecycle step starts, so nothing needs rolling back. All
        // sibling plans on the same bind are aggregated so the error/warn
        // names every Public route on the bind.
        if let Some(bind) = bind_key_from_uri(&from_uri) {
            let owned = self.plans_for_bind(&bind.key);
            let siblings: Vec<(&str, &RouteSecurityPlan)> =
                owned.iter().map(|(id, plan)| (id.as_str(), plan)).collect();
            enforce_bind_exposure_gate(
                &bind.key,
                bind.loopback,
                &siblings,
                self.bind_acks.acknowledged(&bind.key),
            )?;
        }

        // ADR-0022: await each stateful step's `start()` before spawning the
        // pipeline or consumer. On the Nth failure, roll back the already-
        // started steps in reverse order (best-effort) and return the original
        // start error WITHOUT spawning anything. Handles come from the compiled
        // pipeline assembly, already collected in route order at compile time.
        let lifecycle_handles: Vec<Arc<dyn StepLifecycle>> = pipeline.load().lifecycle.clone();
        for (idx, handle) in lifecycle_handles.iter().enumerate() {
            if let Err(start_err) = handle.start().await {
                warn!(
                    route_id = %route_id,
                    step = handle.name(),
                    "step start failed; rolling back already-started steps"
                );
                // Only [0..idx) have started; the Nth handle itself never did.
                rollback_started(route_id, &lifecycle_handles[0..idx]).await;
                return Err(start_err);
            }
        }

        // Clone crash notifier for consumer task
        let crash_notifier = self.crash_notifier.clone();
        let runtime_for_consumer = self.runtime.clone();

        let consumer_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.to_string()),
                self.tracer_gating.levers.components_enabled(),
            )
            .with_in_flight(Arc::clone(&self.in_flight_total)),
        );
        let consumer_rt: Arc<dyn camel_component_api::RuntimeObservability> =
            Arc::clone(&consumer_component_ctx) as Arc<_>;
        let (mut consumer, consumer_concurrency) = match consumer_management::create_route_consumer(
            consumer_rt,
            &self.registry,
            &from_uri,
            consumer_component_ctx.as_ref(),
        ) {
            Ok(v) => v,
            // ADR-0022 SPI: every started handle must be rolled back
            // before start_route returns Err, so no stateful step is left
            // running. This is the first post-start fallible step.
            Err(e) => {
                rollback_started(route_id, &lifecycle_handles).await;
                return Err(e);
            }
        };

        // Resolve effective concurrency: route override > consumer default
        let effective_concurrency = concurrency.unwrap_or(consumer_concurrency);

        // Wire security context before spawning consumer. The
        // `security_authenticator` marker stays in the guard as the
        // route's security classification; the authenticator itself no
        // longer rides the context (kernel plan + providers do). DSL
        // compile sets the marker only alongside the policy path, so the
        // marker term is redundant for DSL routes — it bites programmatic
        // ones (marker without sp_config classifies non-Public but injects
        // no context; strict dispatch fails closed downstream).
        let managed = self
            .routes
            .get_mut(route_id)
            .expect("invariant: route must exist after prior existence check"); // allow-unwrap
        deliver_security_context(consumer.as_mut(), &managed.compiled);

        // Create channel for consumer to send exchanges
        let (tx, mut rx) = mpsc::channel::<ExchangeEnvelope>(256);
        // Create child tokens for independent lifecycle control
        let consumer_cancel = managed.consumer_cancel_token.child_token();
        let pipeline_cancel = managed.pipeline_cancel_token.child_token();
        let drain_in_flight = Arc::clone(&managed.drain_in_flight);
        // Clone sender for storage (to reuse on resume)
        let tx_for_storage = tx.clone();
        // drainclaim: the consumer context mints one claim per envelope
        // sent through `send`/`send_and_wait` — counted from acceptance.
        let consumer_ctx = ConsumerContext::new(tx, consumer_cancel.clone(), route_id.to_string())
            .with_in_flight_counter(Arc::clone(&self.in_flight_total));

        // direct-inline-dispatch Task 2.2: publish the inline dispatcher
        // capability for every non-Concurrent topology. Sequential and the
        // `#[non_exhaustive]` wildcard arm (the spawn match below) are
        // Sequential-equivalent; Concurrent models keep the capability
        // `None` (channel path). Published before the consumer spawns,
        // i.e. before any `mark_ready`, per the set-once contract.
        //
        // rc-2sba publication guard: an aggregate-split route's
        // `managed.pipeline` is an identity shell (`compose_pipeline(vec![])`
        // — the real work lives in the split pre/agg/post pipelines driven
        // by the aggregate engine) and must never be exposed to inline
        // execution. The capability is skipped entirely, so producers take
        // the existing capability-unavailable channel path.
        if !matches!(effective_concurrency, ConcurrencyModel::Concurrent { .. })
            && managed.aggregate_split.is_none()
        {
            let dispatcher = RouteInlineDispatcher::new(
                route_id.to_string(),
                Arc::clone(&pipeline),
                pipeline_cancel.clone(),
                Arc::clone(&drain_in_flight),
                Arc::clone(&self.cohort),
                Some(Arc::clone(&self.in_flight_total)),
            );
            consumer_ctx.set_inline_dispatcher(Arc::new(dispatcher));
        }

        // --- Aggregator v2: check for aggregate route with timeout ---
        let split_clone = managed.aggregate_split.clone();
        if let Some(split) = split_clone {
            let result = self
                .start_aggregate_route(
                    route_id,
                    split,
                    consumer,
                    consumer_ctx,
                    rx,
                    crash_notifier,
                    runtime_for_consumer,
                    tx_for_storage,
                    pipeline_cancel,
                    drain_in_flight,
                )
                .await;
            // ADR-0022 SPI: roll back already-started handles if the aggregate
            // spawn/startup path returns Err.
            if result.is_err() {
                // rc-kh7c: cancel consumer's cancel token to stop child tasks
                // spawned by consumer.start() that observe ctx.cancelled().
                if let Some(managed) = self.routes.get_mut(route_id) {
                    managed.consumer_cancel_token.cancel();
                }
                rollback_started(route_id, &lifecycle_handles).await;
            }
            return result;
        }
        // --- End aggregator v2 branch ---

        // Clone for the startup-failure cleanup path (rc-kh7c): pipeline_cancel
        // is moved into the spawn closure below; this clone stays in scope so
        // the error handler can cancel it to force immediate pipeline exit.
        let pipeline_cancel_for_cleanup = pipeline_cancel.clone();

        // rc-e2r9: capture metrics for b′ emission at reply-drop sites.
        let metrics_for_reply_drop = self.tracer_metrics.clone();

        // rc-jxkj cohort gate: the drain loop parks each dequeued envelope
        // until the startup cohort opens the gate. Subscribed once here; the
        // spawned task owns the receiver (`wait_for` needs &mut), mirroring
        // the pipeline_cancel capture.
        let mut cohort_rx = self.cohort.subscribe();

        // Spawn pipeline task with its own cancellation token
        let pipeline_handle = match effective_concurrency {
            ConcurrencyModel::Concurrent { max } => {
                // Owned for the spawned 'static task (route_id is a borrow).
                // Review minor 2: `Arc<str>` so the per-envelope dispatch
                // tasks only bump a refcount — the String allocation happens
                // once per pipeline task, not once per envelope.
                let route_id: Arc<str> = Arc::from(route_id);
                let sem = max.map(|n| Arc::new(tokio::sync::Semaphore::new(n)));
                // rc-e2r9: metrics for b′ emission at reply-drop sites.
                let metrics_for_reply_drop = metrics_for_reply_drop.clone();
                tokio::spawn(async move {
                    loop {
                        // B2 (ADR-0044): acquire permit BEFORE dequeue.
                        // Cancel-aware: route stop is not blocked waiting for a permit.
                        let permit = match &sem {
                            Some(s) => {
                                let acquired = tokio::select! {
                                    p = Arc::clone(s).acquire_owned() => p.expect("semaphore closed"), // allow-unwrap
                                    _ = pipeline_cancel.cancelled() => return,
                                };
                                Some(acquired)
                            }
                            None => None,
                        };

                        let envelope = tokio::select! {
                            envelope = rx.recv() => match envelope {
                                Some(e) => e,
                                None => return,
                            },
                            _ = pipeline_cancel.cancelled() => return,
                        };
                        let ExchangeEnvelope {
                            exchange,
                            mut reply_tx,
                            in_flight_claim,
                        } = envelope;
                        // drainclaim: `in_flight_claim` is moved into the
                        // per-envelope task below and held across the
                        // pipeline future; any earlier exit of THIS scope
                        // (denied dispatch, cohort-gate cancel) drops it —
                        // drop = release.
                        // rc-jxkj cohort gate: park dispatch until the startup
                        // cohort completes. Level-triggered — after the first
                        // open, later envelopes pass without parking.
                        // rc-z5qz: `biased` with the gate polled FIRST —
                        // with the cohort open and the pipeline token
                        // cancelled, an unbiased select picks randomly and
                        // may drop a deliverable envelope. Gate-open wins
                        // deterministically; a closed gate still drops.
                        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.
                                return;
                            }
                        }
                        // ADR-0061 Task 2.9 strict-mode dispatch check (the
                        // flip deferred from Task 2.2): every transport now
                        // mints the typed carrier at its request boundary
                        // (grpc 2.1, mcp 2.6, ws 2.8, http 2.9), so a
                        // non-Public plan REQUIRES the carrier on the
                        // Exchange — absent or wrong-provider is denied
                        // BEFORE the pipeline runs; the transport renders
                        // the denial in its own idiom via reply_tx.
                        if strict_dispatch_denies(
                            &dispatch_plan,
                            &exchange,
                            &mut reply_tx,
                            &metrics_for_reply_drop,
                            &route_id,
                        ) {
                            continue;
                        }
                        let pipe_ref = Arc::clone(&pipeline);
                        let cancel = pipeline_cancel.clone();
                        let drain_clone = Arc::clone(&drain_in_flight);
                        // rc-e2r9: capture for b′ emission at reply-drop.
                        let inner_metrics = metrics_for_reply_drop.clone();
                        let inner_route_id = Arc::clone(&route_id);
                        tokio::spawn(async move {
                            // Permit owned by this task — released on completion (RAII).
                            let _permit = permit;
                            let _drain_guard = DrainGuard::new(drain_clone);
                            // drainclaim: the claim lives as long as this
                            // task — completion, abort, panic, or the
                            // readiness early-return below all drop it.
                            let _in_flight_claim = in_flight_claim;
                            // claimfamily (rc-hllkk): split a sibling claim
                            // onto the exchange so residency inside
                            // pipeline-embedded stash sites (resequencer
                            // buffers, aggregator buckets) stays counted
                            // after this task completes. The sibling is
                            // taken back from an in-band Ok result below —
                            // exchanges that complete inside the pipeline
                            // release at task end (drainclaim semantics);
                            // stash emissions escape with theirs.
                            let mut exchange = exchange;
                            exchange.in_flight_claim =
                                _in_flight_claim.as_ref().map(InFlightClaim::split);

                            // Load current pipeline from ArcSwap
                            let mut pipe = pipe_ref.load().processor.clone_inner();

                            // Wait for service ready with circuit breaker backoff
                            if let Err(e) = ready_with_backoff(&mut pipe, &cancel).await {
                                // rc-e2r9 review, Important 2: the real error is
                                // cloned by the helper BEFORE the send — the old
                                // code moved it into the send and reported
                                // ChannelClosed, which defeated ConsumerStopping
                                // suppression at this site.
                                send_reply_or_b_prime(
                                    reply_tx,
                                    Err(e),
                                    &inner_metrics,
                                    &inner_route_id,
                                    "concurrent:ready",
                                );
                                return;
                            }

                            // B1: scope CANCEL_TOKEN so run_steps can check
                            // cancellation between steps.
                            let mut result = CANCEL_TOKEN
                                .scope(cancel, async move { pipe.call(exchange).await })
                                .await;
                            // claimfamily: reclaim the sibling from an
                            // in-band result (input completed inside this
                            // pipeline) so release stays at task end. A
                            // stash emission already escaped with its claim;
                            // a transformed result carries None (its input's
                            // sibling dropped on consume).
                            if let Ok(ref mut ex) = result {
                                ex.in_flight_claim = None;
                            }
                            if let Some(tx) = reply_tx {
                                // Helper clones the error before the send, so a
                                // dropped receiver emits b′ with the REAL error.
                                send_reply_or_b_prime(
                                    Some(tx),
                                    result,
                                    &inner_metrics,
                                    &inner_route_id,
                                    "concurrent:pipeline",
                                );
                            } else if let Err(ref e) = result {
                                // log-policy: system-broken
                                error!("Pipeline error: {e}");
                            }
                        });
                    }
                })
            }
            // Forward-compat: an unknown future variant is treated as
            // Sequential — the safe, simplest pipeline topology. A consumer
            // that needs Concurrent semantics for a future variant must
            // override the route's `?concurrent=` setting explicitly so the
            // operator (not the wildcard) chooses the topology.
            _ => {
                // Owned for the spawned 'static task (route_id is a borrow).
                let route_id = route_id.to_string();
                // rc-e2r9: metrics for b′ emission at reply-drop sites.
                let metrics_for_reply_drop = metrics_for_reply_drop.clone();
                tokio::spawn(async move {
                    loop {
                        // Use select! to exit promptly on cancellation even when idle
                        let envelope = tokio::select! {
                            envelope = rx.recv() => match envelope {
                                Some(e) => e,
                                None => return, // Channel closed
                            },
                            _ = pipeline_cancel.cancelled() => {
                                // Cancellation requested - exit gracefully
                                return;
                            }
                        };
                        let ExchangeEnvelope {
                            exchange,
                            mut reply_tx,
                            in_flight_claim,
                        } = envelope;
                        // drainclaim: hold the claim for the whole iteration
                        // — the readiness early-return below and every
                        // `continue`/`return` exit drop it via scope exit.
                        let _in_flight_claim = in_flight_claim;
                        // rc-jxkj cohort gate: park dispatch until the startup
                        // cohort completes. Level-triggered — after the first
                        // open, later envelopes pass without parking.
                        // rc-z5qz: `biased` with the gate polled FIRST —
                        // with the cohort open and the pipeline token
                        // cancelled, an unbiased select picks randomly and
                        // may drop a deliverable envelope. Gate-open wins
                        // deterministically; a closed gate still drops.
                        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.
                                return;
                            }
                        }

                        // ADR-0061 Task 2.9 strict-mode dispatch check — see
                        // the Concurrent branch above for the full contract.
                        if strict_dispatch_denies(
                            &dispatch_plan,
                            &exchange,
                            &mut reply_tx,
                            &metrics_for_reply_drop,
                            &route_id,
                        ) {
                            continue;
                        }

                        // Load current pipeline from ArcSwap (picks up hot-reloaded pipelines)
                        let mut pipeline = pipeline.load().processor.clone_inner();

                        if let Err(e) = ready_with_backoff(&mut pipeline, &pipeline_cancel).await {
                            // rc-e2r9 review, Important 2: the real error is
                            // cloned by the helper BEFORE the send — the old
                            // code moved it into the send and reported
                            // ChannelClosed, which defeated ConsumerStopping
                            // suppression at this site.
                            send_reply_or_b_prime(
                                reply_tx,
                                Err(e),
                                &metrics_for_reply_drop,
                                &route_id,
                                "sequential:ready",
                            );
                            return;
                        }

                        // B1: scope CANCEL_TOKEN so run_steps can check cancellation
                        // between steps. Per-start task-local — child token expires
                        // when this pipeline task exits; the next start re-scopes a
                        // fresh one (avoids the lifecycle bug where a compiled-in
                        // child token stays cancelled after stop→restart).
                        let cancel = pipeline_cancel.clone();
                        let _drain_guard = DrainGuard::new(Arc::clone(&drain_in_flight));
                        // claimfamily (rc-hllkk): split a sibling claim onto
                        // the exchange so residency inside pipeline-embedded
                        // stash sites (resequencer buffers, aggregator
                        // buckets) stays counted after this iteration
                        // completes. Taken back from an in-band Ok result
                        // below — exchanges that complete inside the
                        // pipeline release at iteration end (drainclaim
                        // semantics); stash emissions escape with theirs.
                        let mut exchange = exchange;
                        exchange.in_flight_claim =
                            _in_flight_claim.as_ref().map(InFlightClaim::split);
                        let mut result = CANCEL_TOKEN
                            .scope(cancel, async move { pipeline.call(exchange).await })
                            .await;
                        // claimfamily: reclaim the sibling from an in-band
                        // result (input completed inside this pipeline) so
                        // release stays at iteration end. A stash emission
                        // already escaped with its claim; a transformed
                        // result carries None (its input's sibling dropped
                        // on consume).
                        if let Ok(ref mut ex) = result {
                            ex.in_flight_claim = None;
                        }
                        if let Some(tx) = reply_tx {
                            // Helper clones the error before the send, so a
                            // dropped receiver emits b′ with the REAL error.
                            send_reply_or_b_prime(
                                Some(tx),
                                result,
                                &metrics_for_reply_drop,
                                &route_id,
                                "sequential:pipeline",
                            );
                        } else if let Err(ref e) = result {
                            // log-policy: system-broken
                            error!("Pipeline error: {e}");
                        }
                    }
                })
            }
        };
        #[cfg(test)]
        emit_start_route_event("pipeline_spawned", route_id);

        // Start consumer after pipeline task is spawned to minimize the chance of
        // fire-and-forget events being produced before the pipeline loop is active.
        let (consumer_handle, startup_rx, watcher_inputs, outer_inputs) =
            consumer_management::spawn_consumer_task(
                route_id.to_string(),
                consumer,
                consumer_ctx,
                crash_notifier,
                runtime_for_consumer,
                false,
            );
        #[cfg(test)]
        emit_start_route_event("consumer_spawned", route_id);

        // rc-w1u9: await consumer startup handshake. For Explicit consumers
        // (HTTP, WebSocket) it propagates bind failures as proper startup
        // errors. For Immediate consumers the receiver is pre-resolved
        // (StartupReceiver::immediate) so this returns instantly — the
        // controller never yields during the Immediate handshake (rc-slvd).
        let startup_result =
            consumer_management::await_consumer_startup(startup_rx, "startup").await;
        match startup_result {
            Ok(()) => {}
            Err(e) => {
                // rc-kh7c: abort the orphaned consumer task and cancel the
                // pipeline so neither runs detached after start_route returns
                // Err. Dropping a JoinHandle detaches the task (Tokio
                // contract); abort() forces termination. The pipeline task
                // would eventually self-clean via rx-drop, but explicit
                // cancellation makes it immediate.
                consumer_handle.abort();
                pipeline_cancel_for_cleanup.cancel();
                // Cancel the consumer's cancel token so child tasks spawned
                // by consumer.start() that observe ctx.cancelled() also stop.
                consumer_cancel.cancel();
                rollback_started(route_id, &lifecycle_handles).await;
                return Err(e);
            }
        }

        // Detached failure watcher for Immediate consumers (rc-slvd).
        // The route owns the JoinHandle; the watcher owns the AbortHandle,
        // oneshot, and command_id — ownership split prevents any coupling.
        if let Some(inputs) = watcher_inputs {
            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.
        if let Some(outer) = outer_inputs {
            consumer_management::spawn_outer_task_watcher(outer);
        }

        // Store handles and update status
        let managed = self
            .routes
            .get_mut(route_id)
            .expect("invariant: route must exist after prior existence check"); // allow-unwrap
        managed.consumer_handle = Some(consumer_handle);
        managed.pipeline_handle = Some(pipeline_handle);
        managed.channel_sender = Some(tx_for_storage);

        info!(route_id = %route_id, "Route started");
        self.health_registry().mark_route_started(route_id);
        Ok(())
    }

    async fn stop_route(&mut self, route_id: &str) -> Result<(), CamelError> {
        self.stop_route_internal(route_id).await?;
        self.health_registry().mark_route_stopped(route_id);
        Ok(())
    }

    async fn restart_route(&mut self, route_id: &str) -> Result<(), CamelError> {
        self.stop_route(route_id).await?;
        tokio::time::sleep(Duration::from_millis(100)).await;
        self.start_route(route_id).await
    }

    async fn suspend_route(&mut self, route_id: &str) -> Result<(), CamelError> {
        // Check route exists and state.
        let managed = self
            .routes
            .get_mut(route_id)
            .ok_or_else(|| CamelError::RouteError(format!("Route '{}' not found", route_id)))?;

        let consumer_running = handle_is_running(&managed.consumer_handle);
        let pipeline_running = handle_is_running(&managed.pipeline_handle);

        // Can only suspend from active started state.
        if !consumer_running || !pipeline_running {
            return Err(CamelError::RouteError(format!(
                "Cannot suspend route '{}' with execution lifecycle {}",
                route_id,
                inferred_lifecycle_label(managed)
            )));
        }

        info!(route_id = %route_id, "Suspending route (consumer only, keeping pipeline)");

        // Cancel consumer token only (keep pipeline running)
        let managed = self
            .routes
            .get_mut(route_id)
            .expect("invariant: route must exist after prior existence check"); // allow-unwrap
        managed.consumer_cancel_token.cancel();

        // Take and join consumer handle
        let managed = self
            .routes
            .get_mut(route_id)
            .expect("invariant: route must exist after prior existence check"); // allow-unwrap
        let consumer_handle = managed.consumer_handle.take();

        // Wait for consumer task to complete with timeout
        let timeout_result = tokio::time::timeout(DEFAULT_SHUTDOWN_TIMEOUT, async {
            if let Some(handle) = consumer_handle {
                let _ = handle.await;
            }
        })
        .await;

        if timeout_result.is_err() {
            warn!(route_id = %route_id, "Consumer shutdown timed out during suspend");
        }

        // Get the managed route again (can't hold across await)
        let managed = self
            .routes
            .get_mut(route_id)
            .expect("invariant: route must exist after prior existence check"); // allow-unwrap

        // Create fresh cancellation token for consumer (for resume)
        managed.consumer_cancel_token = CancellationToken::new();

        info!(route_id = %route_id, "Route suspended (pipeline still running)");
        self.health_registry().mark_route_stopped(route_id);
        Ok(())
    }

    async fn resume_route(&mut self, route_id: &str) -> Result<(), CamelError> {
        // Check route exists and is Suspended-equivalent execution state.
        let managed = self
            .routes
            .get(route_id)
            .ok_or_else(|| CamelError::RouteError(format!("Route '{}' not found", route_id)))?;

        let consumer_running = handle_is_running(&managed.consumer_handle);
        let pipeline_running = handle_is_running(&managed.pipeline_handle);
        if consumer_running || !pipeline_running {
            return Err(CamelError::RouteError(format!(
                "Cannot resume route '{}' with execution lifecycle {} (expected Suspended)",
                route_id,
                inferred_lifecycle_label(managed)
            )));
        }

        // Get the stored channel sender (must exist for a suspended route)
        let sender = managed.channel_sender.clone().ok_or_else(|| {
            CamelError::RouteError("Suspended route has no channel sender".into())
        })?;

        // Get from_uri and the concurrency override for creating the new
        // consumer (the override feeds the inline-dispatcher gate below,
        // mirroring the start path's effective-model resolution).
        let from_uri = managed.from_uri.clone();
        let concurrency_override = managed.concurrency.clone();

        // ADR-0061 per-bind exposure gate on resume too (see start path).
        if let Some(bind) = bind_key_from_uri(&from_uri) {
            let owned = self.plans_for_bind(&bind.key);
            let siblings: Vec<(&str, &RouteSecurityPlan)> =
                owned.iter().map(|(id, plan)| (id.as_str(), plan)).collect();
            enforce_bind_exposure_gate(
                &bind.key,
                bind.loopback,
                &siblings,
                self.bind_acks.acknowledged(&bind.key),
            )?;
        }

        info!(route_id = %route_id, "Resuming route (spawning consumer only)");

        let consumer_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.to_string()),
                self.tracer_gating.levers.components_enabled(),
            )
            .with_in_flight(Arc::clone(&self.in_flight_total)),
        );
        let consumer_rt: Arc<dyn camel_component_api::RuntimeObservability> =
            Arc::clone(&consumer_component_ctx) as Arc<_>;
        let (mut consumer, consumer_concurrency) = consumer_management::create_route_consumer(
            consumer_rt,
            &self.registry,
            &from_uri,
            consumer_component_ctx.as_ref(),
        )?;

        // Wire security context before spawning consumer (authenticator
        // marker guard: see start path above).
        let managed = self
            .routes
            .get(route_id)
            .expect("invariant: route must exist after prior existence check"); // allow-unwrap
        deliver_security_context(consumer.as_mut(), &managed.compiled);

        // Get the managed route for mutation
        let managed = self
            .routes
            .get_mut(route_id)
            .expect("invariant: route must exist after prior existence check"); // allow-unwrap

        // Create child token for consumer lifecycle
        let consumer_cancel = managed.consumer_cancel_token.child_token();

        let crash_notifier = self.crash_notifier.clone();
        let runtime_for_consumer = self.runtime.clone();

        // Create ConsumerContext with the stored sender. drainclaim: same
        // counter installation as the start path — resumed routes count.
        let consumer_ctx =
            ConsumerContext::new(sender, consumer_cancel.clone(), route_id.to_string())
                .with_in_flight_counter(Arc::clone(&self.in_flight_total));

        // direct-inline-dispatch Task 3.3 (bd rc-y4vk): mirror the
        // start_route publication on the resume path. Without this the
        // fresh resume ConsumerContext carries no capability and the
        // resumed consumer's registry entry silently falls back to the
        // channel path. Same gate as start (non-Concurrent only, wildcard
        // = Sequential-equivalent), same handle captures (pipeline swap
        // source, pipeline_cancel child scope, shared drain counter), and
        // published before the resumed consumer spawns — a fresh ctx means
        // a fresh OnceLock, so the set-once keep-first contract cannot
        // interfere with the pre-suspend publication.
        //
        // rc-2sba publication guard (mirrors start): an aggregate-split
        // route's `managed.pipeline` is an identity shell
        // (`compose_pipeline(vec![])`) and must never be exposed to inline
        // execution — the split pre/agg/post pipelines stay driven by the
        // aggregate engine over the channel path, so no capability is
        // published here either.
        if !matches!(
            concurrency_override.unwrap_or(consumer_concurrency),
            ConcurrencyModel::Concurrent { .. }
        ) && managed.aggregate_split.is_none()
        {
            let (pipeline, pipeline_cancel, drain_in_flight) = {
                let managed = self
                    .routes
                    .get(route_id)
                    .expect("invariant: route must exist after prior existence check"); // allow-unwrap
                (
                    Arc::clone(&managed.pipeline),
                    managed.pipeline_cancel_token.child_token(),
                    Arc::clone(&managed.drain_in_flight),
                )
            };
            let dispatcher = RouteInlineDispatcher::new(
                route_id.to_string(),
                pipeline,
                pipeline_cancel,
                drain_in_flight,
                Arc::clone(&self.cohort),
                Some(Arc::clone(&self.in_flight_total)),
            );
            consumer_ctx.set_inline_dispatcher(Arc::new(dispatcher));
        }

        // Spawn consumer task
        let (consumer_handle, startup_rx, watcher_inputs, outer_inputs) =
            consumer_management::spawn_consumer_task(
                route_id.to_string(),
                consumer,
                consumer_ctx,
                crash_notifier,
                runtime_for_consumer,
                true,
            );

        // rc-w1u9: await consumer startup handshake on resume too — bind
        // failures during resume must surface as resume errors.
        // For Immediate consumers the receiver is pre-resolved (rc-slvd).
        let resume_result = consumer_management::await_consumer_startup(startup_rx, "resume").await;
        if let Err(e) = resume_result {
            // rc-kh7c cleanup parity with the start path: the consumer task
            // must not run detached after a failed resume, and child tasks
            // spawned by consumer.start() that observe ctx.cancelled() must
            // stop too.
            consumer_handle.abort();
            consumer_cancel.cancel();
            return Err(e);
        }

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

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

        // Store consumer handle and update status
        let managed = self
            .routes
            .get_mut(route_id)
            .expect("invariant: route must exist after prior existence check"); // allow-unwrap
        managed.consumer_handle = Some(consumer_handle);

        info!(route_id = %route_id, "Route resumed");
        self.health_registry().mark_route_started(route_id);
        Ok(())
    }

    async fn start_all_routes(&mut self) -> Result<(), CamelError> {
        // Only start routes where auto_startup() == true
        // Sort by startup_order() ascending before starting
        let route_ids: Vec<String> = {
            let pairs = self.routes.auto_startup_sorted();
            pairs.into_iter().map(|(id, _)| id).collect()
        };

        info!("Starting {} auto-startup routes", route_ids.len());

        // Collect errors but continue starting remaining routes
        let mut errors: Vec<String> = Vec::new();
        for route_id in route_ids {
            if let Err(e) = self.start_route(&route_id).await {
                errors.push(format!("Route '{}': {}", route_id, e));
            }
        }

        if !errors.is_empty() {
            return Err(CamelError::RouteError(format!(
                "Failed to start routes: {}",
                errors.join(", ")
            )));
        }

        info!("All auto-startup routes started");
        Ok(())
    }

    async fn stop_all_routes(&mut self) -> Result<(), CamelError> {
        // Sort by startup_order descending (reverse order)
        let route_ids: Vec<String> = {
            let pairs = self.routes.shutdown_sorted();
            pairs.into_iter().map(|(id, _)| id).collect()
        };

        info!("Stopping {} routes", route_ids.len());

        for route_id in route_ids {
            let _ = self.stop_route(&route_id).await;
        }

        info!("All routes stopped");
        Ok(())
    }
}

// ── rc-e2r9: b′ signal emission on reply-drop ──
// The shared `send_reply_or_b_prime` helper lives in
// `super::route_helpers` (single copy — the duplicated private helpers in
// this file and `route_controller.rs` drifted once already; see rc-e2r9
// review, Important 3).

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