camel-core 0.38.0

use super::*;
use crate::lifecycle::application::route_definition::BuilderStep;
use crate::shared::observability::adapters::span_test_util::{finish, test_spans};
use camel_api::SpanKindHint;
use camel_api::fragment_exchange;
use camel_api::{ExceptionPolicy, OutcomePipeline, OutcomeSegment, RedeliveryPolicy};
use opentelemetry::baggage::BaggageExt;
use opentelemetry::trace::{Span, SpanId, Status, TraceContextExt, Tracer};
use opentelemetry::{Context as OtelContext, KeyValue, global};
use std::sync::atomic::AtomicUsize;

/// A pass-through step: IdentityProcessor wrapped as a CompiledStep.
fn identity_step() -> CompiledStep {
    CompiledStep::Process {
        kind_hint: SpanKindHint::Internal,
        processor: BoxProcessor::new(IdentityProcessor),
        body_contract: None,
        lifecycle: None,
        label: None,
    }
}

/// Local double whose `call` always fails with a processor error.
#[derive(Clone)]
struct ErrProcessor;

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

    fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
        Poll::Ready(Ok(()))
    }

    fn call(&mut self, _exchange: Exchange) -> Self::Future {
        Box::pin(async { Err(CamelError::ProcessorError("boom".into())) })
    }
}

#[tokio::test]
async fn traced_pipeline_opens_root_span_with_step_children() {
    let spans = test_spans().await;
    let mut pipeline = compose_traced_pipeline(
        vec![identity_step(), identity_step()],
        "rt",
        true,
        DetailLevel::Minimal,
        None,
        None,
        PipelineRuntimeCtx::compile_time(),
    );
    let exchange = Exchange::new(Message::default());
    pipeline
        .ready()
        .await
        .expect("pipeline ready")
        .call(exchange)
        .await
        .expect("pipeline call succeeds");

    let all = finish(spans);
    let root_trace_id = all
        .iter()
        .find(|s| s.name == "rt")
        .map(|s| s.span_context.trace_id())
        .expect("root span exported");
    let in_trace: Vec<_> = all
        .iter()
        .filter(|s| s.span_context.trace_id() == root_trace_id)
        .collect();
    let root = in_trace
        .iter()
        .find(|s| s.name == "rt")
        .expect("root span exported");
    assert_eq!(
        root.parent_span_id,
        SpanId::INVALID,
        "route root span must have no parent"
    );
    let trace_id = root.span_context.trace_id();
    let root_span_id = root.span_context.span_id();
    // Select step spans by parent, not by name prefix: labeled spans are
    // named `rt:{label}` (e.g. `rt:to:direct`), so a `rt:step-` prefix
    // filter would silently miss them.
    let steps: Vec<_> = all
        .iter()
        .filter(|s| s.span_context.trace_id() == trace_id && s.parent_span_id == root_span_id)
        .collect();
    assert_eq!(steps.len(), 2, "both step spans exported");
    for step in steps {
        assert_eq!(step.parent_span_id, root_span_id, "step parented by root");
        assert!(
            step.start_time > root.start_time,
            "step must start after the root span starts"
        );
        assert!(
            step.end_time < root.end_time,
            "step must end before the root span ends"
        );
    }
}

/// Task 1.3 (span-name-enrichment): `compose_traced_pipeline` threads the
/// compiled step's label into the step span name. The label comes from the
/// same `BuilderStep::To` mapping the registry stamps (registry stamping is
/// covered by `compile_step_stamps_label`; the full DSL chain by
/// camel-test's `otel_trace_tree_test`).
#[tokio::test]
async fn compose_threads_label_to_span_name() {
    let spans = test_spans().await;
    let label = BuilderStep::To("direct:y".into()).span_label();
    let mut pipeline = compose_traced_pipeline(
        vec![CompiledStep::Process {
            kind_hint: SpanKindHint::Internal,
            processor: BoxProcessor::new(IdentityProcessor),
            body_contract: None,
            lifecycle: None,
            label: label.map(Arc::from),
        }],
        "rt",
        true,
        DetailLevel::Minimal,
        None,
        None,
        PipelineRuntimeCtx::compile_time(),
    );
    let exchange = Exchange::new(Message::default());
    pipeline
        .ready()
        .await
        .expect("pipeline ready")
        .call(exchange)
        .await
        .expect("pipeline call succeeds");

    let all = finish(spans);
    let root = all
        .iter()
        .find(|s| s.name == "rt")
        .expect("root span exported");
    let step = all
        .iter()
        .find(|s| s.name == "rt:to:direct")
        .expect("labeled step span exported with name rt:to:direct");
    assert_eq!(
        step.parent_span_id,
        root.span_context.span_id(),
        "labeled step span must be parented by the route root"
    );
}

#[tokio::test]
async fn traced_pipeline_nested_entry_roots_under_caller() {
    let spans = test_spans().await;
    let tracer = global::tracer("camel-core-test");
    let caller = tracer.span_builder("caller").start(&tracer);
    let trace_id = caller.span_context().trace_id();
    let caller_span_id = caller.span_context().span_id();
    let mut exchange = Exchange::new(Message::default());
    exchange.otel_context = OtelContext::current_with_span(caller);

    let mut pipeline = compose_traced_pipeline(
        vec![identity_step()],
        "rt",
        true,
        DetailLevel::Minimal,
        None,
        None,
        PipelineRuntimeCtx::compile_time(),
    );
    pipeline
        .ready()
        .await
        .expect("pipeline ready")
        .call(exchange)
        .await
        .expect("pipeline call succeeds");

    let all = finish(spans);
    let root = all
        .into_iter()
        .filter(|s| s.span_context.trace_id() == trace_id)
        .find(|s| s.name == "rt")
        .expect("root span exported under caller trace");
    assert_eq!(
        root.parent_span_id, caller_span_id,
        "route root must nest under the caller span"
    );
}

#[tokio::test]
async fn traced_pipeline_restores_entry_context() {
    let _spans = test_spans().await;
    let tracer = global::tracer("camel-core-test");
    let caller = tracer.span_builder("caller").start(&tracer);
    let caller_span_id = caller.span_context().span_id();
    let mut exchange = Exchange::new(Message::default());
    exchange.otel_context =
        OtelContext::current_with_span(caller).with_baggage([KeyValue::new("outer", "1")]);

    let mut pipeline = compose_traced_pipeline(
        vec![identity_step()],
        "rt",
        true,
        DetailLevel::Minimal,
        None,
        None,
        PipelineRuntimeCtx::compile_time(),
    );
    let ex = pipeline
        .ready()
        .await
        .expect("pipeline ready")
        .call(exchange)
        .await
        .expect("pipeline call succeeds");

    assert_eq!(
        ex.otel_context.span().span_context().span_id(),
        caller_span_id,
        "returned exchange must restore the caller's active span"
    );
    assert!(
        ex.otel_context.baggage().get("outer").is_some(),
        "entry-context baggage must survive the route"
    );
}

#[tokio::test]
async fn empty_traced_route_opens_and_closes_root() {
    let spans = test_spans().await;
    let tracer = global::tracer("camel-core-test");
    let entry = tracer.span_builder("e").start(&tracer);
    let trace_id = entry.span_context().trace_id();
    let entry_span_id = entry.span_context().span_id();
    let mut exchange = Exchange::new(Message::default());
    exchange.otel_context =
        OtelContext::current_with_span(entry).with_baggage([KeyValue::new("keep", "1")]);

    let mut pipeline = compose_traced_pipeline(
        vec![],
        "empty",
        true,
        DetailLevel::Minimal,
        None,
        None,
        PipelineRuntimeCtx::compile_time(),
    );
    let ex = pipeline
        .ready()
        .await
        .expect("pipeline ready")
        .call(exchange)
        .await
        .expect("empty traced route completes");

    assert_eq!(
        ex.otel_context.span().span_context().span_id(),
        entry_span_id,
        "returned exchange must restore the entry span"
    );
    assert!(
        ex.otel_context.baggage().get("keep").is_some(),
        "entry-context baggage must survive the empty route"
    );

    let all = finish(spans);
    let in_trace: Vec<_> = all
        .into_iter()
        .filter(|s| s.span_context.trace_id() == trace_id)
        .collect();
    let root = in_trace
        .iter()
        .find(|s| s.name == "empty")
        .expect("root span exported for empty traced route");
    assert_eq!(
        root.parent_span_id, entry_span_id,
        "empty route root nests under the entry span"
    );
    assert!(
        matches!(root.status, Status::Ok),
        "empty route root status must be Ok"
    );
    assert_eq!(
        in_trace.len(),
        1,
        "empty route root must have zero child spans"
    );
}

#[tokio::test]
async fn traced_pipeline_failed_root_records_exception() {
    let spans = test_spans().await;
    let mut pipeline = compose_traced_pipeline(
        vec![CompiledStep::Process {
            kind_hint: SpanKindHint::Internal,
            processor: BoxProcessor::new(ErrProcessor),
            body_contract: None,
            lifecycle: None,
            label: None,
        }],
        "rt",
        true,
        DetailLevel::Minimal,
        None,
        None,
        PipelineRuntimeCtx::compile_time(),
    );
    let exchange = Exchange::new(Message::default());
    let result = pipeline
        .ready()
        .await
        .expect("pipeline ready")
        .call(exchange)
        .await;
    assert!(result.is_err(), "error must propagate to the caller");

    let all = finish(spans);
    let root_trace_id = all
        .iter()
        .find(|s| s.name == "rt")
        .map(|s| s.span_context.trace_id())
        .expect("root span exported");
    let in_trace: Vec<_> = all
        .iter()
        .filter(|s| s.span_context.trace_id() == root_trace_id)
        .collect();
    let root = in_trace
        .iter()
        .find(|s| s.name == "rt")
        .expect("root span exported");
    assert!(
        matches!(root.status, Status::Error { .. }),
        "root span status must be Error"
    );
    assert_eq!(root.events.len(), 1, "exactly one event on the root span");
    let event = &root.events[0];
    assert_eq!(event.name, "exception");
    let attr = |key: &str| {
        event
            .attributes
            .iter()
            .find(|kv| kv.key.as_str() == key)
            .map(|kv| kv.value.as_str().to_string())
    };
    assert!(
        attr("exception.type").is_some_and(|v| !v.is_empty()),
        "exception.type must be non-empty"
    );
    assert!(
        attr("exception.message").is_some_and(|v| !v.is_empty()),
        "exception.message must be non-empty"
    );
}

// ── Segment step spans (trace-model-tree T1.4) ──

/// A `CompiledStep::Segment` wrapping the given segment double.
fn segment_step(segment: OutcomeSegment) -> CompiledStep {
    CompiledStep::Segment {
        segment,
        body_contract: None,
        lifecycle: None,
        label: None,
    }
}

/// Segment double that completes with the incoming exchange untouched.
#[derive(Clone)]
struct CompleteSegment;

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

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

/// Segment double that always fails with a processor error.
#[derive(Clone)]
struct FailSegment;

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

    fn run<'a>(
        &'a mut self,
        _exchange: Exchange,
    ) -> Pin<Box<dyn Future<Output = PipelineOutcome> + Send + 'a>> {
        Box::pin(async move { PipelineOutcome::Failed(CamelError::ProcessorError("boom".into())) })
    }
}

/// Segment double that fails the first attempt; on the second attempt it
/// records the active span id, runs a fragment through a locally composed
/// traced sub-pipeline (`srt-sub`), then completes.
#[derive(Clone)]
struct RetrySubrouteSegment {
    calls: Arc<AtomicUsize>,
    second_attempt_span: Arc<Mutex<Option<SpanId>>>,
}

impl OutcomePipeline for RetrySubrouteSegment {
    fn clone_box(&self) -> Box<dyn OutcomePipeline> {
        Box::new(self.clone())
    }

    fn run<'a>(
        &'a mut self,
        exchange: Exchange,
    ) -> Pin<Box<dyn Future<Output = PipelineOutcome> + Send + 'a>> {
        Box::pin(async move {
            let attempt = self.calls.fetch_add(1, Ordering::SeqCst);
            if attempt == 0 {
                return PipelineOutcome::Failed(CamelError::ProcessorError("attempt-1".into()));
            }
            *self.second_attempt_span.lock().expect("span id lock") =
                Some(exchange.otel_context.span().span_context().span_id());
            let fragment = fragment_exchange(&exchange, Body::Text("frag".into()));
            let mut sub = compose_traced_pipeline(
                vec![identity_step()],
                "srt-sub",
                true,
                DetailLevel::Minimal,
                None,
                None,
                PipelineRuntimeCtx::compile_time(),
            );
            sub.ready()
                .await
                .expect("sub-pipeline ready")
                .call(fragment)
                .await
                .expect("sub-pipeline completes");
            PipelineOutcome::Completed(exchange)
        })
    }
}

#[tokio::test]
async fn segment_step_opens_span_and_restores_root() {
    let spans = test_spans().await;
    let mut pipeline = compose_traced_pipeline(
        vec![
            segment_step(OutcomeSegment::new(Box::new(CompleteSegment))),
            identity_step(),
        ],
        "srt",
        true,
        DetailLevel::Minimal,
        None,
        None,
        PipelineRuntimeCtx::compile_time(),
    );
    let exchange = Exchange::new(Message::default());
    pipeline
        .ready()
        .await
        .expect("pipeline ready")
        .call(exchange)
        .await
        .expect("pipeline call succeeds");

    let all = finish(spans);
    let root_trace_id = all
        .iter()
        .find(|s| s.name == "srt")
        .map(|s| s.span_context.trace_id())
        .expect("root span exported");
    let in_trace: Vec<_> = all
        .into_iter()
        .filter(|s| s.span_context.trace_id() == root_trace_id)
        .collect();
    let root = in_trace
        .iter()
        .find(|s| s.name == "srt")
        .expect("root span exported");
    let root_span_id = root.span_context.span_id();
    let segment_span = in_trace
        .iter()
        .find(|s| s.name == "srt:step-0")
        .expect("segment attempt span exported");
    assert_eq!(
        segment_span.parent_span_id, root_span_id,
        "segment step span must be parented by the route root"
    );
    let next_step = in_trace
        .iter()
        .find(|s| s.name == "srt:step-1")
        .expect("following step span exported");
    assert_eq!(
        next_step.parent_span_id, root_span_id,
        "step after the segment must chain onto the route root, not the segment span"
    );
}

/// Task 1.3 (span-name-enrichment): segment step spans use the segment's
/// label (the registry stamps `split` for splitter segments) with the same
/// positional `step-{index}` fallback as process steps.
#[tokio::test]
async fn segment_span_uses_label() {
    let spans = test_spans().await;
    let mut pipeline = compose_traced_pipeline(
        vec![CompiledStep::Segment {
            segment: OutcomeSegment::new(Box::new(CompleteSegment)),
            body_contract: None,
            lifecycle: None,
            label: Some("split".into()),
        }],
        "srt",
        true,
        DetailLevel::Minimal,
        None,
        None,
        PipelineRuntimeCtx::compile_time(),
    );
    let exchange = Exchange::new(Message::default());
    pipeline
        .ready()
        .await
        .expect("pipeline ready")
        .call(exchange)
        .await
        .expect("pipeline call succeeds");

    let all = finish(spans);
    let root = all
        .iter()
        .find(|s| s.name == "srt")
        .expect("root span exported");
    let segment = all
        .iter()
        .find(|s| s.name == "srt:split")
        .expect("segment span exported with name srt:split");
    assert_eq!(
        segment.parent_span_id,
        root.span_context.span_id(),
        "labeled segment span must be parented by the route root"
    );
}

#[tokio::test]
async fn segment_failure_records_exception_event() {
    let spans = test_spans().await;
    let mut pipeline = compose_traced_pipeline(
        vec![segment_step(OutcomeSegment::new(Box::new(FailSegment)))],
        "srt",
        true,
        DetailLevel::Minimal,
        None,
        None,
        PipelineRuntimeCtx::compile_time(),
    );
    let result = pipeline
        .ready()
        .await
        .expect("pipeline ready")
        .call(Exchange::new(Message::default()))
        .await;
    assert!(result.is_err(), "segment failure must propagate");

    let all = finish(spans);
    let root_trace_id = all
        .iter()
        .find(|s| s.name == "srt")
        .map(|s| s.span_context.trace_id())
        .expect("root span exported");
    let in_trace: Vec<_> = all
        .into_iter()
        .filter(|s| s.span_context.trace_id() == root_trace_id)
        .collect();
    let step = in_trace
        .iter()
        .find(|s| s.name == "srt:step-0")
        .expect("segment attempt span exported");
    assert!(
        matches!(step.status, Status::Error { .. }),
        "failed attempt span status must be Error"
    );
    assert_eq!(
        step.events.len(),
        1,
        "exactly one event on the attempt span"
    );
    assert_eq!(step.events[0].name, "exception");
    let root = in_trace
        .iter()
        .find(|s| s.name == "srt")
        .expect("root span exported");
    assert!(
        matches!(root.status, Status::Error { .. }),
        "root span status must be Error"
    );
    assert_eq!(root.events.len(), 1, "root records its own exception event");
    assert_eq!(root.events[0].name, "exception");
}

#[tokio::test]
async fn segment_retry_attempts_each_get_span() {
    let spans = test_spans().await;
    let double = RetrySubrouteSegment {
        calls: Arc::new(AtomicUsize::new(0)),
        second_attempt_span: Arc::new(Mutex::new(None)),
    };
    let recorded_span_id = double.second_attempt_span.clone();

    let mut policy = ExceptionPolicy::new(|_| true);
    policy.retry = Some(RedeliveryPolicy::new(2).with_initial_delay(Duration::ZERO));
    let handler: Arc<dyn RouteErrorHandler> =
        Arc::new(DefaultRouteErrorHandler::new(None, vec![(policy, None)]));

    let mut pipeline = compose_traced_pipeline(
        vec![segment_step(OutcomeSegment::new(Box::new(double)))],
        "srt",
        true,
        DetailLevel::Minimal,
        None,
        Some(handler),
        PipelineRuntimeCtx::compile_time(),
    );
    pipeline
        .ready()
        .await
        .expect("pipeline ready")
        .call(Exchange::new(Message::default()))
        .await
        .expect("retry recovers on the second attempt");

    let all = finish(spans);
    let root_trace_id = all
        .iter()
        .find(|s| s.name == "srt")
        .map(|s| s.span_context.trace_id())
        .expect("root span exported");
    let in_trace: Vec<_> = all
        .into_iter()
        .filter(|s| s.span_context.trace_id() == root_trace_id)
        .collect();
    let root = in_trace
        .iter()
        .find(|s| s.name == "srt")
        .expect("root span exported");
    let root_span_id = root.span_context.span_id();

    let mut attempts: Vec<_> = in_trace.iter().filter(|s| s.name == "srt:step-0").collect();
    assert_eq!(
        attempts.len(),
        2,
        "exactly two attempt spans for the retried segment"
    );
    attempts.sort_by_key(|s| s.start_time);
    for attempt in &attempts {
        assert_eq!(
            attempt.parent_span_id, root_span_id,
            "every attempt span is parented by the route root"
        );
    }
    let second = attempts[1];
    let sub = in_trace
        .iter()
        .find(|s| s.name == "srt-sub")
        .expect("sub-route root span exported");
    assert_eq!(
        sub.parent_span_id,
        second.span_context.span_id(),
        "retry sub-route must nest under the retry attempt span"
    );
    assert_eq!(
        *recorded_span_id.lock().expect("span id lock"),
        Some(second.span_context.span_id()),
        "double's second attempt ran inside the second attempt span"
    );
}

#[tokio::test]
async fn untraced_segment_emits_no_span() {
    let spans = test_spans().await;
    let mut pipeline = compose_traced_pipeline(
        vec![segment_step(OutcomeSegment::new(Box::new(CompleteSegment)))],
        "srt",
        false,
        DetailLevel::Minimal,
        None,
        None,
        PipelineRuntimeCtx::compile_time(),
    );
    pipeline
        .ready()
        .await
        .expect("pipeline ready")
        .call(Exchange::new(Message::default()))
        .await
        .expect("pipeline call succeeds");

    let all = finish(spans);
    assert!(
        all.iter().all(|s| s.name != "srt:step-0"),
        "untraced segment step must not emit a span"
    );
}

// ── Kind hints threaded into step spans (span-kind-hint 1.3) ──
//
// These tests compile a `To` step through the REAL step-compiler registry
// (stub `ComponentContext` + `CompilationContext`, the pattern proven in
// `step_compilers/endpoints.rs`) so the returned `CompiledStep` carries the
// registry-stamped label and kind hint exactly as route compilation does.
// `compose_traced_pipeline` must thread that hint into the exported span's
// OTel kind.

use crate::lifecycle::adapters::route_controller::SharedLanguageRegistry;
use crate::lifecycle::adapters::step_compilers::{CompilationContext, build_registry};
use crate::lifecycle::adapters::step_resolution::FunctionStagingMode;
use crate::{CacheRegistry, ClaimCheckRegistry, IdempotentRegistry};
use async_trait::async_trait;
use camel_api::{BoxProcessorExt, StepLifecycle};
use camel_bean::BeanRegistry;
use camel_component_api::test_support::NoopRuntimeObservability;
use camel_component_api::{
    Component, ComponentContext, Endpoint, ProducerContext, RuntimeObservability,
};
use std::collections::HashMap;

/// Identity endpoint double: vends a pass-through producer. The stamped kind
/// depends only on the AUTHORED URI scheme (`BuilderStep::span_kind_hint`),
/// so the endpoint body is irrelevant here.
struct SchemeStubEndpoint {
    uri: String,
}

impl Endpoint for SchemeStubEndpoint {
    fn uri(&self) -> &str {
        &self.uri
    }

    fn create_consumer(
        &self,
        _rt: Arc<dyn RuntimeObservability>,
    ) -> Result<Box<dyn camel_component_api::Consumer>, CamelError> {
        Err(CamelError::EndpointCreationFailed("not a consumer".into()))
    }

    fn create_producer(
        &self,
        _rt: Arc<dyn RuntimeObservability>,
        _ctx: &ProducerContext,
    ) -> Result<BoxProcessor, CamelError> {
        Ok(BoxProcessor::from_fn(|ex| Box::pin(async move { Ok(ex) })))
    }

    fn lifecycle(&self) -> Option<Arc<dyn StepLifecycle>> {
        None
    }
}

/// Component double that vends [`SchemeStubEndpoint`] for one scheme.
struct SchemeStubComponent {
    scheme: &'static str,
}

#[async_trait]
impl Component for SchemeStubComponent {
    fn scheme(&self) -> &str {
        self.scheme
    }

    fn create_endpoint(
        &self,
        uri: &str,
        _ctx: &dyn ComponentContext,
    ) -> Result<Box<dyn Endpoint>, CamelError> {
        Ok(Box::new(SchemeStubEndpoint {
            uri: uri.to_string(),
        }))
    }
}

/// `ComponentContext` double that resolves only the stubbed scheme.
struct SchemeStubContext {
    scheme: &'static str,
}

impl ComponentContext for SchemeStubContext {
    fn resolve_component(&self, scheme: &str) -> Option<Arc<dyn Component>> {
        if scheme == self.scheme {
            Some(Arc::new(SchemeStubComponent {
                scheme: self.scheme,
            }))
        } else {
            None
        }
    }
    fn resolve_language(&self, _name: &str) -> Option<Arc<dyn camel_language_api::Language>> {
        None
    }
    fn metrics(&self) -> Arc<dyn camel_api::MetricsCollector> {
        Arc::new(NoOpMetrics)
    }
    fn platform_service(&self) -> Arc<dyn camel_api::PlatformService> {
        Arc::new(camel_api::NoopPlatformService::default())
    }
    fn register_route_health_check(
        &self,
        _route_id: &str,
        _check: Arc<dyn camel_api::AsyncHealthCheck>,
    ) {
    }
    fn unregister_route_health_check(&self, _route_id: &str) {}
}

/// Compile `BuilderStep::To(uri)` through the real step-compiler registry
/// with a stub component registered under `scheme`, so the returned step
/// carries the registry-stamped label and kind hint.
fn compile_to_via_registry(uri: &str, scheme: &'static str) -> CompiledStep {
    let pc = ProducerContext::default();
    let rt: Arc<dyn RuntimeObservability> = Arc::new(NoopRuntimeObservability);
    let languages: SharedLanguageRegistry = Arc::new(Mutex::new(HashMap::new()));
    let beans: Arc<Mutex<BeanRegistry>> = Arc::new(Mutex::new(BeanRegistry::new()));
    let component_ctx: Arc<dyn ComponentContext> = Arc::new(SchemeStubContext { scheme });
    let staging = FunctionStagingMode::DirectAdd;
    let idempotent_repositories = IdempotentRegistry::new();
    let claim_check_repositories = ClaimCheckRegistry::new();
    let cache_repositories = CacheRegistry::new();

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

    build_registry()
        .compile_step(BuilderStep::To(uri.to_string()), 0, &ctx)
        .expect("To step compiles through the real registry")
        .expect("a compiler handles the To variant")
}

/// Task 1.3 (span-kind-hint): a `To("http://…")` step compiled through the
/// real registry is stamped `Client`; `compose_traced_pipeline` threads that
/// hint into the step span's OTel kind.
#[tokio::test]
async fn compose_threads_kind_client_http() {
    let spans = test_spans().await;
    let step = compile_to_via_registry("http://stub/x", "http");
    let mut pipeline = compose_traced_pipeline(
        vec![step],
        "rt",
        true,
        DetailLevel::Minimal,
        None,
        None,
        PipelineRuntimeCtx::compile_time(),
    );
    let exchange = Exchange::new(Message::default());
    pipeline
        .ready()
        .await
        .expect("pipeline ready")
        .call(exchange)
        .await
        .expect("pipeline call succeeds");

    let all = finish(spans);
    let step_span = all
        .iter()
        .find(|s| s.name == "rt:to:http")
        .expect("step span exported with name rt:to:http");
    assert_eq!(
        step_span.span_kind,
        SpanKind::Client,
        "http To step span must be Client"
    );
}

/// Task 1.3 (span-kind-hint): a `To("kafka:…")` step compiled through the
/// real registry is stamped `Producer`; `compose_traced_pipeline` threads
/// that hint into the step span's OTel kind.
#[tokio::test]
async fn compose_threads_kind_producer_kafka() {
    let spans = test_spans().await;
    let step = compile_to_via_registry("kafka:stub-topic", "kafka");
    let mut pipeline = compose_traced_pipeline(
        vec![step],
        "rt",
        true,
        DetailLevel::Minimal,
        None,
        None,
        PipelineRuntimeCtx::compile_time(),
    );
    let exchange = Exchange::new(Message::default());
    pipeline
        .ready()
        .await
        .expect("pipeline ready")
        .call(exchange)
        .await
        .expect("pipeline call succeeds");

    let all = finish(spans);
    let step_span = all
        .iter()
        .find(|s| s.name == "rt:to:kafka")
        .expect("step span exported with name rt:to:kafka");
    assert_eq!(
        step_span.span_kind,
        SpanKind::Producer,
        "kafka To step span must be Producer"
    );
}

/// Minimal local double: counts `increment_exchanges` calls.
struct ExchangesCounter(std::sync::Mutex<u32>);

impl camel_api::MetricsCollector for ExchangesCounter {
    fn increment_exchanges(&self, _route_id: &str) {
        *self.0.lock().expect("counter lock") += 1; // allow-unwrap
    }
    fn increment_errors(&self, _route_id: &str, _error_type: &str) {}
    fn record_exchange_duration(&self, _route_id: &str, _seconds: std::time::Duration) {}
    fn set_queue_depth(&self, _queue: &str, _depth: usize) {}
    fn record_circuit_breaker_change(&self, _route: &str, _from: &str, _to: &str) {}
}

/// Task 2.1 follow-up (inter-phase review): the spans-off composition
/// branch (`!gating.spans_enabled` → plain SequentialPipeline over the
/// metrics-emitting wrappers) is executed here, not just compiled —
/// metrics flow, no spans are created.
#[tokio::test]
async fn spans_off_composition_records_metrics_without_spans() {
    let spans = test_spans().await;
    let collector = Arc::new(ExchangesCounter(std::sync::Mutex::new(0)));
    let mut pipeline = compose_traced_pipeline(
        vec![identity_step(), identity_step()],
        "rt",
        crate::lifecycle::adapters::route_compiler::TracerPipelineGating {
            pipeline_enabled: true,
            spans_enabled: false,
            levers: crate::shared::observability::domain::MetricsLeversConfig::default(),
        },
        DetailLevel::Minimal,
        Some(Arc::clone(&collector) as Arc<dyn camel_api::MetricsCollector>),
        None,
        PipelineRuntimeCtx::compile_time(),
    );
    pipeline
        .ready()
        .await
        .expect("pipeline ready")
        .call(Exchange::new(Message::default()))
        .await
        .expect("pipeline call succeeds");

    let all = finish(spans);
    assert!(
        all.is_empty(),
        "spans-off composition must export no spans, got {all:?}"
    );
    assert_eq!(
        *collector.0.lock().expect("counter lock"),
        2,
        "exchange metrics must still flow through the per-step wrappers \
         (one per step, matching the traced pipeline's wrapper semantics)"
    );
}