camel-core 0.60.0

//! Core step compilers: Log, SetBody, SetHeader, SetProperty, Bean, Script, etc.
//!
//! These are the simplest compilers — none recursively compile child steps.

use camel_api::SpanKindHint;
use std::sync::Arc;

use camel_api::{BoxProcessor, CamelError, Exchange, IdentityProcessor, Value};
use camel_language_api::LanguageError;
use camel_processor::claim_check::ClaimCheckFilter;
use camel_processor::{LogProcessor, log::DynamicLog, script_mutator::ScriptMutator, set_property};

use super::{
    CompilationContext, CompileOutcome, CompiledStep, StepCompiler, StepCompilerRegistry,
    pack_lifecycles,
};
use crate::lifecycle::adapters::step_resolution::{
    FunctionStagingMode, compile_filter_predicate, compile_key_expression,
    compile_message_id_expression, compile_sort_expression, compile_value_source, eval_meta,
    resolve_language,
};
use crate::lifecycle::application::route_definition::{BuilderStep, ValueSourceDef};

pub(crate) struct CoreCompiler;

impl StepCompiler for CoreCompiler {
    fn compile(
        &self,
        step: BuilderStep,
        step_index: usize,
        ctx: &CompilationContext,
        registry: &StepCompilerRegistry,
    ) -> Result<CompileOutcome, CamelError> {
        match step {
            // ── Pre-built processor ──
            BuilderStep::Processor(op) => Ok(CompileOutcome::Matched(CompiledStep::Process {
                kind_hint: SpanKindHint::Internal,
                processor: op.0,
                body_contract: None,
                lifecycle: None,
                label: None,
                to_uri: None,
            })),

            // ── Stop ──
            BuilderStep::Stop => Ok(CompileOutcome::Matched(CompiledStep::Stop)),

            // ── Resequence (N4: must be top-level only) ──
            BuilderStep::Resequence { .. } => Err(CamelError::RouteError(
                "resequence must be a top-level step".into(),
            )),

            // ── Delay ──
            BuilderStep::Delay { config } => {
                let svc = camel_processor::delayer::DelayerService::new(config);
                Ok(CompileOutcome::Matched(CompiledStep::Process {
                    kind_hint: SpanKindHint::Internal,
                    processor: BoxProcessor::new(svc),
                    body_contract: None,
                    lifecycle: None,
                    label: None,
                    to_uri: None,
                }))
            }

            // ── Validate ──
            BuilderStep::Validate { predicate } => {
                let meta = eval_meta(ctx, &predicate.language, "validate", step_index, None);
                let predicate_arc = compile_filter_predicate(ctx.languages, &predicate, meta)?;
                let expression_source = predicate.source.clone();
                let svc = camel_processor::ValidateService::from_predicate(
                    predicate_arc,
                    expression_source,
                );
                Ok(CompileOutcome::Matched(CompiledStep::Process {
                    kind_hint: SpanKindHint::Internal,
                    processor: BoxProcessor::new(svc),
                    body_contract: None,
                    lifecycle: None, // Validate is stateless
                    label: None,
                    to_uri: None,
                }))
            }

            // ── Idempotent Consumer (Segment-mode, ADR-0023) ──
            // CRITICAL: Implemented as OutcomePipeline (Segment), NOT Tower
            // Service<Exchange>. compose_pipeline() maps Stopped→Ok(ex); if
            // this were Process-mode, a duplicate-detected Stopped would
            // become Ok(ex) and downstream steps would re-process the dup.
            BuilderStep::IdempotentConsumer {
                repository,
                expression,
                steps,
                eager,
                remove_on_failure,
            } => {
                use crate::lifecycle::adapters::route_compiler::compose_outcome_segment;

                let repo = ctx
                    .idempotent_repositories
                    .get(&repository)
                    .ok_or_else(|| {
                        CamelError::ComponentNotFound(format!(
                            "idempotent_consumer: repository '{repository}' is not registered"
                        ))
                    })?;
                let message_id = compile_message_id_expression(
                    ctx.languages,
                    &expression,
                    eval_meta(
                        ctx,
                        &expression.language,
                        "idempotent_consumer",
                        step_index,
                        None,
                    ),
                )?;
                let (child_segments, child_lifecycles) =
                    ctx.compile_children_segments(steps, registry)?;
                let child_pipeline = compose_outcome_segment(child_segments);
                let svc = camel_processor::IdempotentConsumerSegment::new(
                    repo,
                    message_id,
                    child_pipeline,
                    eager,
                    remove_on_failure,
                );
                Ok(CompileOutcome::Matched(CompiledStep::Segment {
                    segment: camel_api::OutcomeSegment::new(Box::new(svc)),
                    body_contract: None,
                    lifecycle: pack_lifecycles(child_lifecycles),
                    label: None,
                }))
            }

            // ── Cache (Segment-mode) ──
            // Caching EIP: lookup → on-miss sub-pipeline → write-back. Same
            // Segment-mode rationale as IdempotentConsumer above.
            BuilderStep::Cache {
                repository,
                key,
                ttl,
                max_entry_bytes,
                coalesce_misses,
                on_miss,
            } => {
                use crate::lifecycle::adapters::route_compiler::compose_outcome_segment;

                let repo_name = repository.as_deref().unwrap_or("memory");
                let repo = ctx.cache_repositories.get(repo_name).ok_or_else(|| {
                    CamelError::ComponentNotFound(format!(
                        "cache: repository '{repo_name}' is not registered"
                    ))
                })?;
                let key_expr = compile_message_id_expression(
                    ctx.languages,
                    &key,
                    eval_meta(ctx, &key.language, "cache", step_index, None),
                )?;
                let ttl_dur = ttl
                    .as_ref()
                    .map(|s| humantime::parse_duration(s))
                    .transpose()
                    .map_err(|e| {
                        CamelError::RouteError(format!(
                            "cache: invalid ttl '{}': {e}",
                            ttl.as_deref().unwrap_or("<none>")
                        ))
                    })?;
                let (child_segments, child_lifecycles) =
                    ctx.compile_children_segments(on_miss, registry)?;
                let child_pipeline = compose_outcome_segment(child_segments);
                let svc = camel_processor::CacheService::new(
                    repo,
                    key_expr,
                    ttl_dur,
                    max_entry_bytes.unwrap_or(camel_api::body::DEFAULT_MATERIALIZE_LIMIT),
                    child_pipeline,
                    ctx.rt.clone(),
                )
                .with_coalesce(coalesce_misses);
                Ok(CompileOutcome::Matched(CompiledStep::Segment {
                    segment: camel_api::OutcomeSegment::new(Box::new(svc)),
                    body_contract: None,
                    lifecycle: pack_lifecycles(child_lifecycles),
                    label: None,
                }))
            }

            // ── Cache Invalidate (Segment-mode) ──
            BuilderStep::CacheInvalidate {
                repository,
                key,
                key_prefix,
            } => {
                // Exactly one of `key` or `key_prefix` is required: both-None
                // and both-Some fail closed at compile time with a Config
                // error naming the step.
                let (expr, as_prefix) = match (&key, &key_prefix) {
                    (Some(expr), None) => (expr, false),
                    (None, Some(expr)) => (expr, true),
                    _ => {
                        return Err(CamelError::Config(
                            "cache_invalidate: exactly one of 'key' or 'key_prefix' is required"
                                .into(),
                        ));
                    }
                };
                // Backstop for the canonical RegisterRoute path (the DSL
                // fast-fails the same condition): a key/key_prefix whose
                // source trims to empty is rejected with the same message.
                if expr.source.trim().is_empty() {
                    return Err(CamelError::Config(
                        "cache_invalidate: key/key_prefix must not be empty".into(),
                    ));
                }
                let meta = eval_meta(ctx, &expr.language, "cache_invalidate", step_index, None);
                let target =
                    if as_prefix {
                        camel_processor::CacheInvalidateTarget::Prefix(
                            compile_message_id_expression(ctx.languages, expr, meta)?,
                        )
                    } else {
                        camel_processor::CacheInvalidateTarget::Key(compile_message_id_expression(
                            ctx.languages,
                            expr,
                            meta,
                        )?)
                    };
                let repo_name = repository.as_deref().unwrap_or("memory");
                let repo = ctx.cache_repositories.get(repo_name).ok_or_else(|| {
                    CamelError::ComponentNotFound(format!(
                        "cache_invalidate: repository '{repo_name}' is not registered"
                    ))
                })?;
                let svc =
                    camel_processor::CacheInvalidateService::new(repo, target, ctx.rt.clone());
                Ok(CompileOutcome::Matched(CompiledStep::Segment {
                    segment: camel_api::OutcomeSegment::new(Box::new(svc)),
                    body_contract: None,
                    lifecycle: None,
                    label: None,
                }))
            }

            // ── Cache Peek Stale (Segment-mode) ──
            BuilderStep::CachePeekStale {
                repository,
                key,
                on_miss,
            } => {
                let repo_name = repository.as_deref().unwrap_or("memory");
                let repo = ctx.cache_repositories.get(repo_name).ok_or_else(|| {
                    CamelError::ComponentNotFound(format!(
                        "cache_peek_stale: repository '{repo_name}' is not registered"
                    ))
                })?;
                let key_expr = compile_message_id_expression(
                    ctx.languages,
                    &key,
                    eval_meta(ctx, &key.language, "cache_peek_stale", step_index, None),
                )?;
                let svc = camel_processor::CachePeekStaleService::new(
                    repo,
                    key_expr,
                    on_miss,
                    ctx.rt.clone(),
                );
                Ok(CompileOutcome::Matched(CompiledStep::Segment {
                    segment: camel_api::OutcomeSegment::new(Box::new(svc)),
                    body_contract: None,
                    lifecycle: None,
                    label: None,
                }))
            }

            // ── Cache Clear (Segment-mode) ──
            BuilderStep::CacheClear { repository } => {
                let repo_name = repository.as_deref().unwrap_or("memory");
                let repo = ctx.cache_repositories.get(repo_name).ok_or_else(|| {
                    CamelError::ComponentNotFound(format!(
                        "cache_clear: repository '{repo_name}' is not registered"
                    ))
                })?;
                let svc = camel_processor::CacheClearService::new(repo);
                Ok(CompileOutcome::Matched(CompiledStep::Segment {
                    segment: camel_api::OutcomeSegment::new(Box::new(svc)),
                    body_contract: None,
                    lifecycle: None,
                    label: None,
                }))
            }

            // ── Cache Stats (Segment-mode) ──
            BuilderStep::CacheStats { repository } => {
                let repo_name = repository.as_deref().unwrap_or("memory");
                let repo = ctx.cache_repositories.get(repo_name).ok_or_else(|| {
                    CamelError::ComponentNotFound(format!(
                        "cache_stats: repository '{repo_name}' is not registered"
                    ))
                })?;
                let svc = camel_processor::CacheStatsService::new(repo);
                Ok(CompileOutcome::Matched(CompiledStep::Segment {
                    segment: camel_api::OutcomeSegment::new(Box::new(svc)),
                    body_contract: None,
                    lifecycle: None,
                    label: None,
                }))
            }

            // ── Static Log ──
            BuilderStep::Log { level, message } => {
                let svc = LogProcessor::new(level, message);
                Ok(CompileOutcome::Matched(CompiledStep::Process {
                    kind_hint: SpanKindHint::Internal,
                    processor: BoxProcessor::new(svc),
                    body_contract: None,
                    lifecycle: None,
                    label: None,
                    to_uri: None,
                }))
            }

            // ── Declarative Log (language-evaluated message) ──
            BuilderStep::DeclarativeLog { level, message } => {
                let ValueSourceDef::Expression(expression) = message else {
                    // Should never happen — literal case is converted to Processor in compile.rs
                    unreachable!(
                        "DeclarativeLog with Literal should have been compiled to a Processor"
                    );
                };
                let source = compile_value_source(ctx, &expression, "log", step_index, None)?;
                let svc = DynamicLog::new(level, source);
                Ok(CompileOutcome::Matched(CompiledStep::Process {
                    kind_hint: SpanKindHint::Internal,
                    processor: BoxProcessor::new(svc),
                    body_contract: None,
                    lifecycle: None,
                    label: None,
                    to_uri: None,
                }))
            }

            // ── SetHeader (declarative) ──
            BuilderStep::DeclarativeSetHeader { key, value } => match value {
                ValueSourceDef::Literal(value) => {
                    let svc = camel_processor::SetHeader::new(IdentityProcessor, key, value);
                    Ok(CompileOutcome::Matched(CompiledStep::Process {
                        kind_hint: SpanKindHint::Internal,
                        processor: BoxProcessor::new(svc),
                        body_contract: None,
                        lifecycle: None,
                        label: None,
                        to_uri: None,
                    }))
                }
                ValueSourceDef::Expression(expression) => {
                    let source = compile_value_source(
                        ctx,
                        &expression,
                        "set_header",
                        step_index,
                        Some(key.clone()),
                    )?;
                    let svc =
                        camel_processor::DynamicSetHeader::new(IdentityProcessor, key, source);
                    Ok(CompileOutcome::Matched(CompiledStep::Process {
                        kind_hint: SpanKindHint::Internal,
                        processor: BoxProcessor::new(svc),
                        body_contract: None,
                        lifecycle: None,
                        label: None,
                        to_uri: None,
                    }))
                }
                _ => Err(CamelError::Config("unsupported value source".into())),
            },

            // ── RemoveHeader (declarative) ──
            BuilderStep::DeclarativeRemoveHeader { key } => {
                let svc = camel_processor::RemoveHeader::new(IdentityProcessor, key);
                Ok(CompileOutcome::Matched(CompiledStep::Process {
                    kind_hint: SpanKindHint::Internal,
                    processor: BoxProcessor::new(svc),
                    body_contract: None,
                    lifecycle: None,
                    label: None,
                    to_uri: None,
                }))
            }

            // ── SetHeaderIfAbsent (declarative, internal-only) ──
            BuilderStep::DeclarativeSetHeaderIfAbsent { key, value } => match value {
                ValueSourceDef::Literal(value) => {
                    let svc =
                        camel_processor::SetHeaderIfAbsent::new(IdentityProcessor, key, value);
                    Ok(CompileOutcome::Matched(CompiledStep::Process {
                        kind_hint: SpanKindHint::Internal,
                        processor: BoxProcessor::new(svc),
                        body_contract: None,
                        lifecycle: None,
                        label: None,
                        to_uri: None,
                    }))
                }
                ValueSourceDef::Expression(expression) => {
                    let source = compile_value_source(
                        ctx,
                        &expression,
                        "set_header_if_absent",
                        step_index,
                        Some(key.clone()),
                    )?;
                    let svc = camel_processor::DynamicSetHeaderIfAbsent::new(
                        IdentityProcessor,
                        key,
                        source,
                    );
                    Ok(CompileOutcome::Matched(CompiledStep::Process {
                        kind_hint: SpanKindHint::Internal,
                        processor: BoxProcessor::new(svc),
                        body_contract: None,
                        lifecycle: None,
                        label: None,
                        to_uri: None,
                    }))
                }
                _ => Err(CamelError::Config("unsupported value source".into())),
            },

            // ── SetProperty (declarative) ──
            BuilderStep::DeclarativeSetProperty { key, value_source } => match value_source {
                ValueSourceDef::Literal(value) => {
                    let svc = set_property::SetProperty::new(IdentityProcessor, key, value);
                    Ok(CompileOutcome::Matched(CompiledStep::Process {
                        kind_hint: SpanKindHint::Internal,
                        processor: BoxProcessor::new(svc),
                        body_contract: None,
                        lifecycle: None,
                        label: None,
                        to_uri: None,
                    }))
                }
                ValueSourceDef::Expression(expression) => {
                    let source = compile_value_source(
                        ctx,
                        &expression,
                        "set_property",
                        step_index,
                        Some(key.clone()),
                    )?;
                    let svc =
                        camel_processor::DynamicSetProperty::new(IdentityProcessor, key, source);
                    Ok(CompileOutcome::Matched(CompiledStep::Process {
                        kind_hint: SpanKindHint::Internal,
                        processor: BoxProcessor::new(svc),
                        body_contract: None,
                        lifecycle: None,
                        label: None,
                        to_uri: None,
                    }))
                }
                _ => Err(CamelError::Config("unsupported value source".into())),
            },

            // ── SetBody (declarative) ──
            BuilderStep::DeclarativeSetBody { value } => match value {
                ValueSourceDef::Literal(value) => {
                    // Static body: clone the literal per exchange; `SetBody`
                    // converts the evaluated Value to a Body internally.
                    let source = Arc::new(move |_: &Exchange| value.clone())
                        as Arc<dyn Fn(&Exchange) -> Value + Send + Sync>;
                    let svc = camel_processor::SetBody::new(IdentityProcessor, source);
                    Ok(CompileOutcome::Matched(CompiledStep::Process {
                        kind_hint: SpanKindHint::Internal,
                        processor: BoxProcessor::new(svc),
                        body_contract: None,
                        lifecycle: None,
                        label: None,
                        to_uri: None,
                    }))
                }
                ValueSourceDef::Expression(expression) => {
                    let source = compile_value_source(
                        ctx,
                        &expression,
                        "set_body",
                        step_index,
                        Some("body".to_string()),
                    )?;
                    let svc = camel_processor::SetBody::new(IdentityProcessor, source);
                    Ok(CompileOutcome::Matched(CompiledStep::Process {
                        kind_hint: SpanKindHint::Internal,
                        processor: BoxProcessor::new(svc),
                        body_contract: None,
                        lifecycle: None,
                        label: None,
                        to_uri: None,
                    }))
                }
                _ => Err(CamelError::Config("unsupported value source".into())),
            },

            // ── Declarative Script (requires a mutating language) ──
            BuilderStep::DeclarativeScript { expression } => {
                compile_script(ctx, &expression.language, &expression.source, step_index)
            }

            // ── Function step ──
            BuilderStep::DeclarativeFunction { mut definition } => {
                let Some(invoker) = ctx.function_invoker.clone() else {
                    return Err(CamelError::Config(
                        "function: step requires FunctionRuntimeService registered via with_lifecycle"
                            .into(),
                    ));
                };
                definition.route_id = ctx.route_id.map(|s| s.to_string());
                definition.step_index = Some(step_index);
                match ctx.staging_mode {
                    FunctionStagingMode::DirectAdd => {
                        invoker.stage_pending(definition.clone(), ctx.route_id, 0);
                    }
                    FunctionStagingMode::HotReload { generation } => {
                        invoker.stage_pending(definition.clone(), ctx.route_id, *generation);
                    }
                    FunctionStagingMode::DryCompile => {}
                }
                let step = crate::step::function_step::FunctionStep::new(invoker, definition);
                Ok(CompileOutcome::Matched(CompiledStep::Process {
                    kind_hint: SpanKindHint::Internal,
                    processor: BoxProcessor::new(step),
                    body_contract: None,
                    lifecycle: None,
                    label: None,
                    to_uri: None,
                }))
            }

            // ── Bean invocation ──
            BuilderStep::Bean { name, method } => {
                let beans = ctx
                    .beans
                    .lock()
                    .map_err(|_| CamelError::ProcessorError("beans mutex poisoned".into()))?;
                let bean = beans.get(&name).ok_or_else(|| {
                    CamelError::ProcessorError(format!("Bean not found: {}", name))
                })?;
                let processor = tower::service_fn(move |mut exchange: Exchange| {
                    let bean = Arc::clone(&bean);
                    let method = method.clone();
                    async move {
                        // R3-L6: method-allowlist check (mirrors BeanRegistry::invoke).
                        if !bean.methods().iter().any(|m| m == &method) {
                            return Err(
                                camel_bean::BeanError::MethodNotFound(method.clone()).into()
                            );
                        }
                        bean.call(&method, &mut exchange).await?;
                        Ok(exchange)
                    }
                });
                Ok(CompileOutcome::Matched(CompiledStep::Process {
                    kind_hint: SpanKindHint::Internal,
                    processor: BoxProcessor::new(processor),
                    body_contract: None,
                    lifecycle: None,
                    label: None,
                    to_uri: None,
                }))
            }

            // ── Script (requires a mutating language) ──
            BuilderStep::Script { language, script } => {
                compile_script(ctx, &language, &script, step_index)
            }

            // ── ClaimCheck ──
            BuilderStep::ClaimCheck {
                repository,
                operation,
                key,
                filter,
            } => {
                let repo = ctx
                    .claim_check_repositories
                    .get(&repository)
                    .ok_or_else(|| {
                        CamelError::ComponentNotFound(format!(
                            "claim_check: repository '{repository}' is not registered"
                        ))
                    })?;
                let op = match operation.as_str() {
                    "set" => camel_processor::claim_check::ClaimCheckOp::Set,
                    "get" => camel_processor::claim_check::ClaimCheckOp::Get,
                    "get_and_remove" => camel_processor::claim_check::ClaimCheckOp::GetAndRemove,
                    "push" => camel_processor::claim_check::ClaimCheckOp::Push,
                    "pop" => camel_processor::claim_check::ClaimCheckOp::Pop,
                    other => {
                        return Err(CamelError::RouteError(format!(
                            "claim_check: unsupported operation '{other}'"
                        )));
                    }
                };
                let key_expr = compile_key_expression(
                    ctx.languages,
                    &key,
                    eval_meta(ctx, &key.language, "claim_check", step_index, None),
                )?;
                let mut svc = camel_processor::ClaimCheckService::new(repo, op, key_expr);
                if let Some(filter_str) = filter {
                    let f = ClaimCheckFilter::parse(&filter_str).map_err(|e| {
                        CamelError::RouteError(format!(
                            "claim_check: invalid filter '{filter_str}': {e}"
                        ))
                    })?;
                    svc = svc.with_filter(f);
                }
                Ok(CompileOutcome::Matched(CompiledStep::Process {
                    kind_hint: SpanKindHint::Internal,
                    processor: BoxProcessor::new(svc),
                    body_contract: None,
                    lifecycle: None,
                    label: None,
                    to_uri: None,
                }))
            }

            // ── Sampling (Process-mode, stateless, no StepLifecycle) ──
            BuilderStep::Sampling { period } => {
                let svc = camel_processor::SamplingService::new(period);
                Ok(CompileOutcome::Matched(CompiledStep::Process {
                    kind_hint: SpanKindHint::Internal,
                    processor: BoxProcessor::new(svc),
                    body_contract: None,
                    lifecycle: None,
                    label: None,
                    to_uri: None,
                }))
            }

            // ── Sort (Process-mode, stateless) ──
            BuilderStep::Sort {
                expression,
                reverse,
            } => {
                let sort_expr = compile_sort_expression(
                    ctx.languages,
                    &expression,
                    eval_meta(ctx, &expression.language, "sort", step_index, None),
                )?;
                let svc = camel_processor::SortService::new(sort_expr, reverse);
                Ok(CompileOutcome::Matched(CompiledStep::Process {
                    kind_hint: SpanKindHint::Internal,
                    processor: BoxProcessor::new(svc),
                    body_contract: None,
                    lifecycle: None,
                    label: None,
                    to_uri: None,
                }))
            }

            // ── Everything else: not handled ──
            _ => Ok(CompileOutcome::NotHandled(step)),
        }
    }
}

/// Compile a script step through one shared path for both `BuilderStep::Script`
/// and `BuilderStep::DeclarativeScript`.
///
/// A `script` step requires its language to implement `MutatingExpression`.
/// A read-only language is rejected with the shared capability message; the
/// step is never silently downgraded to a `SetBody` read-only evaluation.
fn compile_script(
    ctx: &CompilationContext,
    language: &str,
    source: &str,
    step_index: usize,
) -> Result<CompileOutcome, CamelError> {
    let meta = eval_meta(ctx, language, "script", step_index, None);
    let lang = resolve_language(ctx.languages, language)?;
    match lang.create_mutating_expression(source) {
        Ok(mut_expr) => Ok(CompileOutcome::Matched(CompiledStep::Process {
            kind_hint: SpanKindHint::Internal,
            processor: BoxProcessor::new(ScriptMutator::with_meta(mut_expr, meta)),
            body_contract: None,
            lifecycle: None,
            label: None,
            to_uri: None,
        })),
        Err(LanguageError::NotSupported { .. }) => Err(CamelError::RouteError(format!(
            "script: language '{language}' does not support mutating expressions; a `script` step requires a language that implements MutatingExpression. To replace the body with a read-only expression, use a `set_body`/`transform` step with this language instead"
        ))),
        Err(e) => Err(CamelError::RouteError(format!(
            "Failed to create mutating expression for language '{language}': {e}"
        ))),
    }
}

#[cfg(test)]
mod tests {
    use std::collections::HashMap;
    use std::sync::{Arc, Mutex};

    use camel_api::{CamelError, Exchange, LanguageExpressionDef, Message, ProducerContext, Value};
    use camel_bean::BeanRegistry;
    use camel_component_api::{
        ComponentContext, NoOpComponentContext, RuntimeObservability,
        test_support::NoopRuntimeObservability,
    };

    use camel_language_api::Language;
    use camel_processor::aggregator::SharedLanguageRegistry;

    use crate::idempotent::MemoryIdempotentRepository;
    use crate::lifecycle::adapters::step_compilers::{
        CompilationContext, CompiledStep, StepCompilerRegistry,
    };
    use crate::lifecycle::adapters::step_resolution::FunctionStagingMode;
    use crate::lifecycle::application::route_definition::BuilderStep;

    /// Compile an idempotent_consumer step through the full route compiler
    /// and assert the output is a CompiledStep::Segment with lifecycle propagated.
    #[tokio::test]
    async fn compile_idempotent_consumer_produces_segment() {
        // ── languages with simple language ──
        let languages: SharedLanguageRegistry = {
            let mut map: HashMap<String, Arc<dyn Language>> = HashMap::new();
            map.insert(
                "simple".to_string(),
                Arc::new(camel_language_simple::SimpleLanguage::new()),
            );
            Arc::new(Mutex::new(map))
        };

        // ── register memory idempotent repository ──
        let idempotent_repositories = crate::IdempotentRegistry::new();
        let repo: Arc<dyn camel_api::IdempotentRepository> =
            Arc::new(MemoryIdempotentRepository::new("memory"));
        idempotent_repositories
            .register("memory", repo)
            .expect("register repo");
        let claim_check_repositories = crate::ClaimCheckRegistry::new();
        let cache_repositories = crate::CacheRegistry::new();

        // ── registry with only CoreCompiler ──
        let mut reg = StepCompilerRegistry::new();
        reg.register(Box::new(super::CoreCompiler));

        // ── CompilationContext ──
        let pc = ProducerContext::default();
        let rt: Arc<dyn RuntimeObservability> = Arc::new(NoopRuntimeObservability);
        let beans: Arc<Mutex<BeanRegistry>> = Arc::new(Mutex::new(BeanRegistry::new()));
        let component_ctx: Arc<dyn ComponentContext> = Arc::new(NoOpComponentContext);
        let staging = FunctionStagingMode::DirectAdd;

        let 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(),
        };

        // ── IdempotentConsumer step ──
        let step = BuilderStep::IdempotentConsumer {
            repository: "memory".into(),
            expression: LanguageExpressionDef {
                language: "simple".into(),
                source: "${header.id}".into(),
            },
            steps: vec![BuilderStep::Log {
                level: camel_processor::LogLevel::Info,
                message: "duplicate check passed".into(),
            }],
            eager: false,
            remove_on_failure: true,
        };

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

        assert!(
            matches!(compiled, CompiledStep::Segment { .. }),
            "Expected CompiledStep::Segment, got {compiled:?}"
        );
    }

    #[test]
    fn nested_resequence_rejected_at_compile_children() {
        use camel_api::CamelError;

        // N4: any Resequence reaching the compiler registry (i.e., not intercepted
        // at the top-level by the route controller) must be rejected.
        let mut reg = StepCompilerRegistry::new();
        reg.register(Box::new(super::CoreCompiler));

        let step = BuilderStep::Resequence {
            policy_config: camel_api::ResequencePolicyConfig::default(),
        };

        let result = reg.compile_step(step, 0, &dummy_context());
        let err = result.expect_err("compiler should produce an error for nested resequence");
        assert!(
            matches!(err, CamelError::RouteError(_)),
            "should be RouteError, got {err:?}"
        );
        assert!(
            err.to_string().contains("top-level"),
            "error should mention 'top-level', got: {err}"
        );
    }

    /// A bean whose `call` accepts any method, but whose `methods()` allowlist
    /// is restricted. Used to verify the compiler-level allowlist check fires
    /// before `bean.call()` is reached.
    struct PermissiveBean;

    #[async_trait::async_trait]
    impl camel_bean::BeanProcessor for PermissiveBean {
        async fn call(
            &self,
            _method: &str,
            _exchange: &mut camel_api::Exchange,
        ) -> Result<(), CamelError> {
            // Always succeeds — the allowlist check must catch unauthorized
            // methods before we get here.
            Ok(())
        }

        fn methods(&self) -> Vec<String> {
            vec!["handle".to_string()]
        }
    }

    /// R3-L6: bean step compiler must reject methods not in the allowlist
    /// before invoking `bean.call()`.
    #[tokio::test]
    async fn bean_step_rejects_method_not_in_allowlist() {
        use tower::ServiceExt;

        // Register a bean with allowlist ["handle"].
        let beans = BeanRegistry::new();
        beans.register("test_bean", PermissiveBean).unwrap();
        let beans: Arc<Mutex<BeanRegistry>> = Arc::new(Mutex::new(beans));

        // Build a CompilationContext referencing the bean registry.
        let producer_ctx: &'static ProducerContext =
            Box::leak(Box::new(ProducerContext::default()));
        let staging: &'static FunctionStagingMode =
            Box::leak(Box::new(FunctionStagingMode::DirectAdd));
        let languages: &'static SharedLanguageRegistry =
            Box::leak(Box::new(Arc::new(Mutex::new(HashMap::new()))));
        let beans_ref: &'static Arc<Mutex<BeanRegistry>> = Box::leak(Box::new(beans));
        let idempotent: &'static crate::IdempotentRegistry =
            Box::leak(Box::new(crate::IdempotentRegistry::new()));
        let claim_check: &'static crate::ClaimCheckRegistry =
            Box::leak(Box::new(crate::ClaimCheckRegistry::new()));
        let cache: &'static crate::CacheRegistry = Box::leak(Box::new(crate::CacheRegistry::new()));

        let ctx = CompilationContext {
            producer_ctx,
            rt: Arc::new(NoopRuntimeObservability),
            languages,
            beans: beans_ref,
            function_invoker: None,
            component_ctx: Arc::new(NoOpComponentContext),
            route_id: None,
            staging_mode: staging,
            idempotent_repositories: idempotent,
            claim_check_repositories: claim_check,
            cache_repositories: cache,
            intercept: crate::intercept::InterceptRules::default(),
        };

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

        // Compile a Bean step calling an unauthorized method.
        let step = BuilderStep::Bean {
            name: "test_bean".into(),
            method: "unauthorized_method".into(),
        };
        let result = reg.compile_step(step, 0, &ctx);
        let compiled = result
            .expect("compilation should succeed")
            .expect("should match");

        // Extract the processor and invoke it.
        let processor = match compiled {
            CompiledStep::Process { processor, .. } => processor,
            other => panic!("expected CompiledStep::Process, got {other:?}"),
        };

        let exchange = Exchange::new(Message::default());
        let result = processor.oneshot(exchange).await;

        assert!(
            result.is_err(),
            "bean step should reject method not in allowlist"
        );
        let err = result.unwrap_err();
        let msg = err.to_string();
        assert!(
            msg.contains("unauthorized_method") && msg.contains("not found"),
            "error should mention method name and 'not found', got: {msg}"
        );
    }

    /// End-to-end: compile a DeclarativeRemoveHeader step and run it against an
    /// exchange whose input headers contain the target key.
    #[tokio::test]
    async fn test_remove_header_end_to_end() {
        use tower::ServiceExt;

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

        let step = BuilderStep::DeclarativeRemoveHeader {
            key: "CamelHttpPath".into(),
        };

        let compiled = reg
            .compile_step(step, 0, &dummy_context())
            .expect("compilation should succeed")
            .expect("should match");

        let processor = match compiled {
            CompiledStep::Process { processor, .. } => processor,
            other => panic!("expected CompiledStep::Process, got {other:?}"),
        };

        let mut exchange = Exchange::new(Message::default());
        exchange.input.set_header("CamelHttpPath", "some-path");

        let result = processor.oneshot(exchange).await.unwrap();
        assert!(
            !result.input.headers.contains_key("CamelHttpPath"),
            "CamelHttpPath should be removed from input headers"
        );
    }

    /// RemoveHeader only mutates `input.headers` — a same-named key on
    /// `output.headers` must survive.
    #[tokio::test]
    async fn test_remove_header_input_only_preserves_output() {
        use tower::ServiceExt;

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

        let step = BuilderStep::DeclarativeRemoveHeader {
            key: "X-Shared".into(),
        };

        let compiled = reg
            .compile_step(step, 0, &dummy_context())
            .expect("compilation should succeed")
            .expect("should match");

        let processor = match compiled {
            CompiledStep::Process { processor, .. } => processor,
            other => panic!("expected CompiledStep::Process, got {other:?}"),
        };

        let mut output = Message::default();
        output.set_header("X-Shared", "output-value");

        let mut exchange = Exchange::new(Message::default());
        exchange.input.set_header("X-Shared", "input-value");
        exchange.output = Some(output);

        let result = processor.oneshot(exchange).await.unwrap();

        assert!(
            !result.input.headers.contains_key("X-Shared"),
            "X-Shared should be removed from input headers"
        );
        let output = result
            .output
            .as_ref()
            .expect("output message should be preserved");
        assert!(
            output.headers.contains_key("X-Shared"),
            "X-Shared should remain on output headers"
        );
    }

    /// A cache_clear step naming an unregistered repository must fail
    /// compilation with `ComponentNotFound` naming both the step and repo.
    #[test]
    fn cache_clear_unknown_repository_fails_compile() {
        let mut reg = StepCompilerRegistry::new();
        reg.register(Box::new(super::CoreCompiler));

        let step = BuilderStep::CacheClear {
            repository: Some("nope".into()),
        };

        let result = reg.compile_step(step, 0, &dummy_context());
        let err = result.expect_err("compilation should fail for unknown repository");
        match err {
            CamelError::ComponentNotFound(msg) => {
                assert!(
                    msg.contains("cache_clear"),
                    "error must name cache_clear: {msg}"
                );
                assert!(
                    msg.contains("nope"),
                    "error must name the repository: {msg}"
                );
            }
            other => panic!("expected ComponentNotFound, got {other:?}"),
        }
    }

    /// Compile a single cache step through the `CoreCompiler` against a
    /// context with the `simple` language and the `memory` cache repository
    /// registered. Returns the registry's `Result<Option<CompiledStep>, _>`.
    fn compile_cache_step(step: BuilderStep) -> Result<Option<CompiledStep>, CamelError> {
        // ── languages with simple language ──
        let languages: SharedLanguageRegistry = {
            let mut map: HashMap<String, Arc<dyn Language>> = HashMap::new();
            map.insert(
                "simple".to_string(),
                Arc::new(camel_language_simple::SimpleLanguage::new()),
            );
            Arc::new(Mutex::new(map))
        };

        // ── register memory cache repository ──
        let cache_repositories = crate::CacheRegistry::new();
        let repo: Arc<dyn camel_api::CacheRepository> =
            Arc::new(crate::cache::MemoryCacheRepository::new("memory", 100));
        cache_repositories
            .register("memory", repo)
            .expect("register cache repo");
        let idempotent_repositories = crate::IdempotentRegistry::new();
        let claim_check_repositories = crate::ClaimCheckRegistry::new();

        // ── registry with only CoreCompiler ──
        let mut reg = StepCompilerRegistry::new();
        reg.register(Box::new(super::CoreCompiler));

        // ── CompilationContext ──
        let pc = ProducerContext::default();
        let rt: Arc<dyn RuntimeObservability> = Arc::new(NoopRuntimeObservability);
        let beans: Arc<Mutex<BeanRegistry>> = Arc::new(Mutex::new(BeanRegistry::new()));
        let component_ctx: Arc<dyn ComponentContext> = Arc::new(NoOpComponentContext);
        let staging = FunctionStagingMode::DirectAdd;

        let 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(),
        };

        reg.compile_step(step, 0, &ctx)
    }

    /// A `cache_invalidate` step with both `key` and `key_prefix` set, or with
    /// neither set, must be rejected at compile time with a `Config` error
    /// naming the conflict.
    #[test]
    fn cache_invalidate_both_key_and_prefix_rejected() {
        for step in [
            BuilderStep::CacheInvalidate {
                repository: None,
                key: Some(LanguageExpressionDef {
                    language: "simple".into(),
                    source: "k".into(),
                }),
                key_prefix: Some(LanguageExpressionDef {
                    language: "simple".into(),
                    source: "ns:".into(),
                }),
            },
            BuilderStep::CacheInvalidate {
                repository: None,
                key: None,
                key_prefix: None,
            },
        ] {
            let err =
                compile_cache_step(step).expect_err("key/key_prefix conflict must be rejected");
            match err {
                CamelError::Config(msg) => {
                    assert!(
                        msg.contains("cache_invalidate"),
                        "error must name the invalidate step: {msg}"
                    );
                    assert!(
                        msg.contains("exactly one of"),
                        "error must require exactly one of key/key_prefix: {msg}"
                    );
                }
                other => panic!("expected CamelError::Config, got {other:?}"),
            }
        }
    }

    /// A `cache_invalidate` step with only `key_prefix` must compile to a
    /// `CompiledStep::Segment` (the prefix target reaches the service).
    #[test]
    fn cache_invalidate_prefix_compiles_to_target() {
        let step = BuilderStep::CacheInvalidate {
            repository: None,
            key: None,
            key_prefix: Some(LanguageExpressionDef {
                language: "simple".into(),
                source: "ns:".into(),
            }),
        };

        let compiled = compile_cache_step(step)
            .expect("compilation should succeed")
            .expect("should match");
        assert!(
            matches!(compiled, CompiledStep::Segment { .. }),
            "expected CompiledStep::Segment, got {compiled:?}"
        );
    }

    /// A `cache_invalidate` step whose key source trims to empty must be
    /// rejected at compile time with a `Config` error naming the empty guard.
    #[test]
    fn cache_invalidate_empty() {
        let step = BuilderStep::CacheInvalidate {
            repository: None,
            key: Some(LanguageExpressionDef {
                language: "simple".into(),
                source: "".into(),
            }),
            key_prefix: None,
        };

        let err = compile_cache_step(step).expect_err("empty key must be rejected");
        match err {
            CamelError::Config(msg) => {
                assert!(
                    msg.contains("must not be empty"),
                    "error must name the empty guard: {msg}"
                );
            }
            other => panic!("expected CamelError::Config, got {other:?}"),
        }
    }

    /// A `cache` step with `coalesce_misses: true` must compile to a
    /// `CompiledStep::Segment`. Compile-level pin only; threading proven by
    /// 2.7 route-level integration test.
    #[test]
    fn cache_coalesce_misses_compiles() {
        let step = BuilderStep::Cache {
            repository: None,
            key: LanguageExpressionDef {
                language: "simple".into(),
                source: "k".into(),
            },
            ttl: None,
            max_entry_bytes: None,
            coalesce_misses: true,
            on_miss: vec![BuilderStep::Log {
                level: camel_processor::LogLevel::Info,
                message: "miss".into(),
            }],
        };

        let compiled = compile_cache_step(step)
            .expect("compilation should succeed")
            .expect("should match");
        assert!(
            matches!(compiled, CompiledStep::Segment { .. }),
            "expected CompiledStep::Segment, got {compiled:?}"
        );
    }

    // ---- script compiler (Task 2.1) test support ----

    /// Build a `CompilationContext` with the given language registry.
    fn script_ctx(langs: Vec<(&str, Arc<dyn Language>)>) -> CompilationContext<'static> {
        use crate::lifecycle::adapters::route_controller::SharedLanguageRegistry;

        let mut map: HashMap<String, Arc<dyn Language>> = HashMap::new();
        for (name, lang) in langs {
            map.insert(name.to_string(), lang);
        }
        let languages: &'static SharedLanguageRegistry =
            Box::leak(Box::new(Arc::new(Mutex::new(map))));
        let producer_ctx: &'static ProducerContext =
            Box::leak(Box::new(ProducerContext::default()));
        let staging: &'static FunctionStagingMode =
            Box::leak(Box::new(FunctionStagingMode::DirectAdd));
        let beans: &'static Arc<Mutex<BeanRegistry>> =
            Box::leak(Box::new(Arc::new(Mutex::new(BeanRegistry::new()))));
        let idempotent: &'static crate::IdempotentRegistry =
            Box::leak(Box::new(crate::IdempotentRegistry::new()));
        let claim_check: &'static crate::ClaimCheckRegistry =
            Box::leak(Box::new(crate::ClaimCheckRegistry::new()));
        let cache: &'static crate::CacheRegistry = Box::leak(Box::new(crate::CacheRegistry::new()));

        CompilationContext {
            producer_ctx,
            rt: Arc::new(NoopRuntimeObservability),
            languages,
            beans,
            function_invoker: None,
            component_ctx: Arc::new(NoOpComponentContext),
            route_id: None,
            staging_mode: staging,
            idempotent_repositories: idempotent,
            claim_check_repositories: claim_check,
            cache_repositories: cache,
            intercept: crate::intercept::InterceptRules::default(),
        }
    }

    fn simple_language() -> Arc<dyn Language> {
        Arc::new(camel_language_simple::SimpleLanguage::new())
    }

    fn compile_script_step(
        reg: &StepCompilerRegistry,
        step: BuilderStep,
        ctx: &CompilationContext<'_>,
    ) -> Result<CompiledStep, CamelError> {
        reg.compile_step(step, 0, ctx)?
            .ok_or_else(|| CamelError::Config("step was not handled by CoreCompiler".into()))
    }

    /// A test language whose `create_mutating_expression` fails before a
    /// processor exists (`EvalError("boom")`).
    struct CreationFailLanguage;

    impl Language for CreationFailLanguage {
        fn name(&self) -> &'static str {
            "creationfail"
        }

        fn create_expression(
            &self,
            _script: &str,
        ) -> Result<Box<dyn camel_language_api::Expression>, camel_language_api::LanguageError>
        {
            Ok(Box::new(ScriptStubExpression))
        }

        fn create_predicate(
            &self,
            _script: &str,
        ) -> Result<Box<dyn camel_language_api::Predicate>, camel_language_api::LanguageError>
        {
            Ok(Box::new(ScriptStubPredicate))
        }

        fn create_mutating_expression(
            &self,
            _script: &str,
        ) -> Result<
            Box<dyn camel_language_api::MutatingExpression>,
            camel_language_api::LanguageError,
        > {
            Err(camel_language_api::LanguageError::EvalError("boom".into()))
        }
    }

    /// A test language whose mutating expression is created successfully but
    /// fails when evaluated (`EvalError("boom")`).
    struct EvalFailLanguage;

    impl Language for EvalFailLanguage {
        fn name(&self) -> &'static str {
            "evalfail"
        }

        fn create_expression(
            &self,
            _script: &str,
        ) -> Result<Box<dyn camel_language_api::Expression>, camel_language_api::LanguageError>
        {
            Ok(Box::new(ScriptStubExpression))
        }

        fn create_predicate(
            &self,
            _script: &str,
        ) -> Result<Box<dyn camel_language_api::Predicate>, camel_language_api::LanguageError>
        {
            Ok(Box::new(ScriptStubPredicate))
        }

        fn create_mutating_expression(
            &self,
            _script: &str,
        ) -> Result<
            Box<dyn camel_language_api::MutatingExpression>,
            camel_language_api::LanguageError,
        > {
            Ok(Box::new(EvalFailExpr))
        }
    }

    struct EvalFailExpr;

    #[async_trait::async_trait]
    impl camel_language_api::MutatingExpression for EvalFailExpr {
        async fn evaluate(
            &self,
            _exchange: &mut Exchange,
        ) -> Result<Value, camel_language_api::LanguageError> {
            Err(camel_language_api::LanguageError::EvalError("boom".into()))
        }
    }

    struct ScriptStubExpression;

    #[async_trait::async_trait]
    impl camel_language_api::Expression for ScriptStubExpression {
        async fn evaluate(
            &self,
            _exchange: &Exchange,
        ) -> Result<Value, camel_language_api::LanguageError> {
            Ok(Value::Null)
        }
    }

    struct ScriptStubPredicate;

    #[async_trait::async_trait]
    impl camel_language_api::Predicate for ScriptStubPredicate {
        async fn matches(
            &self,
            _exchange: &Exchange,
        ) -> Result<bool, camel_language_api::LanguageError> {
            Ok(false)
        }
    }

    /// Read-only language: both builder variants must fail before execution
    /// with the shared capability message, never silently replace the body.
    #[test]
    fn compile_script_rejects_non_mutating_language() {
        let ctx = script_ctx(vec![("simple", simple_language())]);
        let mut reg = StepCompilerRegistry::new();
        reg.register(Box::new(super::CoreCompiler));

        let err = reg
            .compile_step(
                BuilderStep::DeclarativeScript {
                    expression: LanguageExpressionDef {
                        language: "simple".into(),
                        source: "${body}".into(),
                    },
                },
                0,
                &ctx,
            )
            .expect_err("a read-only language must be rejected at compile time");

        const EXPECTED: &str = "script: language 'simple' does not support mutating expressions; a `script` step requires a language that implements MutatingExpression. To replace the body with a read-only expression, use a `set_body`/`transform` step with this language instead";
        match err {
            CamelError::RouteError(msg) => assert_eq!(msg, EXPECTED),
            other => panic!("expected RouteError, got {other:?}"),
        }
    }

    /// `Script` and `DeclarativeScript` share one compiler, so equivalent
    /// read-only inputs produce byte-identical errors.
    #[test]
    fn compile_script_both_variants_byte_identical() {
        let ctx = script_ctx(vec![("simple", simple_language())]);
        let mut reg = StepCompilerRegistry::new();
        reg.register(Box::new(super::CoreCompiler));

        let err_script = reg
            .compile_step(
                BuilderStep::Script {
                    language: "simple".into(),
                    script: "${body}".into(),
                },
                0,
                &ctx,
            )
            .expect_err("Script with a read-only language must fail");
        let err_declarative = reg
            .compile_step(
                BuilderStep::DeclarativeScript {
                    expression: LanguageExpressionDef {
                        language: "simple".into(),
                        source: "${body}".into(),
                    },
                },
                0,
                &ctx,
            )
            .expect_err("DeclarativeScript with a read-only language must fail");

        assert_eq!(err_script.to_string(), err_declarative.to_string());
    }

    /// A custom mutating language compiles to a processor that really mutates
    /// the exchange.
    #[tokio::test]
    async fn compile_script_custom_mutating_language_executes_mutation() {
        use crate::lifecycle::adapters::step_resolution::MutatingTestLanguage;
        use tower::ServiceExt;

        let ctx = script_ctx(vec![(
            "testmut",
            Arc::new(MutatingTestLanguage) as Arc<dyn Language>,
        )]);
        let mut reg = StepCompilerRegistry::new();
        reg.register(Box::new(super::CoreCompiler));

        let compiled = compile_script_step(
            &reg,
            BuilderStep::DeclarativeScript {
                expression: LanguageExpressionDef {
                    language: "testmut".into(),
                    source: "body = 1;".into(),
                },
            },
            &ctx,
        )
        .expect("mutating language must compile");

        let processor = match compiled {
            CompiledStep::Process { processor, .. } => processor,
            other => panic!("expected CompiledStep::Process, got {other:?}"),
        };

        let result = processor
            .oneshot(Exchange::default())
            .await
            .expect("mutating script must execute");
        assert_eq!(
            result.property("testmut"),
            Some(&Value::String("ran".into()))
        );
    }

    /// A runtime evaluation failure propagates as `ExpressionFailed` carrying
    /// the language and `script` verb from the compile-time metadata.
    #[tokio::test]
    async fn compile_script_mutating_eval_error_propagates_with_metadata() {
        use tower::ServiceExt;

        let ctx = script_ctx(vec![(
            "evalfail",
            Arc::new(EvalFailLanguage) as Arc<dyn Language>,
        )]);
        let mut reg = StepCompilerRegistry::new();
        reg.register(Box::new(super::CoreCompiler));

        let compiled = compile_script_step(
            &reg,
            BuilderStep::DeclarativeScript {
                expression: LanguageExpressionDef {
                    language: "evalfail".into(),
                    source: "x".into(),
                },
            },
            &ctx,
        )
        .expect("creation succeeds; failure happens at evaluation");

        let processor = match compiled {
            CompiledStep::Process { processor, .. } => processor,
            other => panic!("expected CompiledStep::Process, got {other:?}"),
        };

        let result = processor.oneshot(Exchange::default()).await;
        match result {
            Err(CamelError::ExpressionFailed { language, verb, .. }) => {
                assert_eq!(language, "evalfail");
                assert_eq!(verb, "script");
            }
            other => panic!("expected ExpressionFailed, got {other:?}"),
        }
    }

    /// A creation failure keeps its existing message and never builds a
    /// processor.
    #[test]
    fn compile_script_creation_error_unchanged() {
        let ctx = script_ctx(vec![(
            "creationfail",
            Arc::new(CreationFailLanguage) as Arc<dyn Language>,
        )]);
        let mut reg = StepCompilerRegistry::new();
        reg.register(Box::new(super::CoreCompiler));

        let err = reg
            .compile_step(
                BuilderStep::DeclarativeScript {
                    expression: LanguageExpressionDef {
                        language: "creationfail".into(),
                        source: "x".into(),
                    },
                },
                0,
                &ctx,
            )
            .expect_err("creation failure must fail compilation");

        let msg = err.to_string();
        assert!(
            msg.contains("Failed to create mutating expression for language 'creationfail'"),
            "unexpected error: {msg}"
        );
        assert!(msg.contains("boom"), "unexpected error: {msg}");
    }

    /// An unknown language keeps the existing resolution error.
    #[test]
    fn compile_script_unknown_language_error_unchanged() {
        let ctx = script_ctx(vec![]);
        let mut reg = StepCompilerRegistry::new();
        reg.register(Box::new(super::CoreCompiler));

        let err = reg
            .compile_step(
                BuilderStep::Script {
                    language: "nope".into(),
                    script: "x".into(),
                },
                0,
                &ctx,
            )
            .expect_err("unknown language must fail resolution");

        assert!(
            err.to_string().contains("not registered"),
            "unexpected error: {err}"
        );
    }

    /// Build a minimal `CompilationContext` for unit tests.
    fn dummy_context() -> CompilationContext<'static> {
        use std::collections::HashMap;
        use std::sync::Mutex;

        use camel_bean::BeanRegistry;
        use camel_component_api::{NoOpComponentContext, test_support::NoopRuntimeObservability};

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

        // Leak on-stack data to satisfy 'static lifetime (test-only safe).
        // SAFETY: these leaks are bounded — used once in a short-lived test.
        let producer_ctx: &'static ProducerContext =
            Box::leak(Box::new(ProducerContext::default()));
        let staging: &'static FunctionStagingMode =
            Box::leak(Box::new(FunctionStagingMode::DirectAdd));
        let languages: &'static SharedLanguageRegistry =
            Box::leak(Box::new(Arc::new(Mutex::new(HashMap::new()))));
        let beans: &'static Arc<Mutex<BeanRegistry>> =
            Box::leak(Box::new(Arc::new(Mutex::new(BeanRegistry::new()))));
        let idempotent: &'static crate::IdempotentRegistry =
            Box::leak(Box::new(crate::IdempotentRegistry::new()));
        let claim_check: &'static crate::ClaimCheckRegistry =
            Box::leak(Box::new(crate::ClaimCheckRegistry::new()));
        let cache: &'static crate::CacheRegistry = Box::leak(Box::new(crate::CacheRegistry::new()));

        CompilationContext {
            producer_ctx,
            rt: Arc::new(NoopRuntimeObservability),
            languages,
            beans,
            function_invoker: None,
            component_ctx: Arc::new(NoOpComponentContext),
            route_id: None,
            staging_mode: staging,
            idempotent_repositories: idempotent,
            claim_check_repositories: claim_check,
            cache_repositories: cache,
            intercept: crate::intercept::InterceptRules::default(),
        }
    }
}