nexus_rt/

pipeline.rs

// Builder return types are necessarily complex — each combinator returns
// PipelineBuilder<In, Out, impl FnMut(...)>. Same pattern as iterator adapters.
#![allow(clippy::type_complexity)]

//! Pre-resolved pipeline dispatch using [`Param`] steps.
//!
//! [`PipelineStart`] begins a typed composition chain where each step
//! is a named function with [`Param`] dependencies resolved at build
//! time. The result is a monomorphized closure chain where dispatch-time
//! resource access is ~3 cycles per fetch (pre-resolved [`ResourceId`](crate::ResourceId)),
//! not a HashMap lookup.
//!
//! Two dispatch tiers in nexus-rt:
//! 1. **Pipeline** — static after build, pre-resolved, the workhorse
//! 2. **Callback** — dynamic registration with per-instance context
//!
//! # Step function convention
//!
//! Params first, step input last, returns output:
//!
//! ```ignore
//! fn validate(config: Res<Config>, order: Order) -> Option<ValidOrder> { .. }
//! fn enrich(cache: Res<MarketData>, order: ValidOrder) -> EnrichedOrder { .. }
//! fn submit(mut gw: ResMut<Gateway>, order: CheckedOrder) { gw.send(order); }
//! ```
//!
//! # Combinator split
//!
//! **IntoStep-based (pre-resolved, hot path):**
//! `.then()`, `.map()`, `.and_then()`, `.catch()`
//!
//! **Closure-based (cold path, `&mut World`):**
//! `.on_none()`, `.inspect()`, `.inspect_err()`, `.filter()`, `.ok()`,
//! `.unwrap_or()`, `.unwrap_or_else()`, `.map_err()`, `.or_else()`,
//! `.cloned()`, `.dispatch()`, `.tap()`, `.guard()`, `.route()`,
//! `.switch()`
//!
//! # Combinator quick reference
//!
//! **Bare value `T`:** `.then()`, `.tap()`, `.guard()` (→ `Option<T>`),
//! `.dispatch()`, `.route()`, `.switch()`, `.tee()`, `.dedup()` (→ `Option<T>`)
//!
//! **Tuple `(A, B, ...)` (2-5 elements):** `.splat()` (→ splat builder,
//! call `.then()` with destructured args)
//!
//! **`Option<T>`:** `.map()`, `.filter()`, `.inspect()`, `.and_then()`,
//! `.on_none()`, `.ok_or()` (→ `Result`), `.unwrap_or()` (→ `T`),
//! `.cloned()` (→ `Option<T>` from `Option<&T>`)
//!
//! **`Result<T, E>`:** `.map()`, `.and_then()`, `.catch()` (→ `Option<T>`),
//! `.map_err()`, `.inspect_err()`, `.ok()` (→ `Option<T>`),
//! `.unwrap_or()` (→ `T`), `.or_else()`
//!
//! **`bool`:** `.not()`, `.and()`, `.or()`, `.xor()`
//!
//! **Terminal:** `.build()` (→ `Pipeline<In>`), `.build_batch(cap)`
//! (→ `BatchPipeline<In>`)
//!
//! # Splat — tuple destructuring
//!
//! Pipeline steps follow a single-value-in, single-value-out convention.
//! When a step returns a tuple like `(OrderId, f64)`, the next step
//! must accept the whole tuple as one argument. `.splat()` destructures
//! the tuple so the next step receives individual arguments instead:
//!
//! ```ignore
//! // Without splat — next step takes the whole tuple:
//! fn process(pair: (OrderId, f64)) -> bool { .. }
//!
//! // With splat — next step takes individual args:
//! fn process(id: OrderId, price: f64) -> bool { .. }
//!
//! PipelineStart::<Order>::new()
//!     .then(extract, reg)   // Order → (OrderId, f64)
//!     .splat()              // destructure
//!     .then(process, reg)   // (OrderId, f64) → bool
//!     .build();
//! ```
//!
//! Supported for tuples of 2-5 elements. Beyond 5, define a named
//! struct — if a combinator stage needs that many arguments, a struct
//! makes the intent clearer and the code more maintainable.

use std::marker::PhantomData;

use crate::Handler;
use crate::dag::DagArm;
use crate::handler::Param;
use crate::world::{Registry, World};

// =============================================================================
// Step — pre-resolved step with Param state
// =============================================================================

/// Internal: pre-resolved step with cached Param state.
///
/// Users don't construct this directly — it's produced by [`IntoStep`] and
/// captured inside pipeline chain closures.
#[doc(hidden)]
pub struct Step<F, Params: Param> {
    f: F,
    state: Params::State,
    #[allow(dead_code)]
    name: &'static str,
}

// =============================================================================
// StepCall — callable trait for resolved steps
// =============================================================================

/// Internal: callable trait for resolved steps.
///
/// Used as a bound on [`IntoStep::Step`]. Users don't implement this.
#[doc(hidden)]
pub trait StepCall<In, Out> {
    /// Call this step with a world reference and input value.
    fn call(&mut self, world: &mut World, input: In) -> Out;
}

// =============================================================================
// IntoStep — converts a named function into a resolved step
// =============================================================================

/// Converts a named function into a pre-resolved pipeline step.
///
/// Params first, step input last, returns output. Arity 0 (no
/// Params) supports closures. Arities 1+ require named functions
/// (same HRTB+GAT limitation as [`IntoHandler`](crate::IntoHandler)).
///
/// # Examples
///
/// ```ignore
/// // Arity 0 — closure works
/// let step = (|x: u32| x * 2).into_step(registry);
///
/// // Arity 1 — named function required
/// fn validate(config: Res<Config>, order: Order) -> Option<ValidOrder> { .. }
/// let step = validate.into_step(registry);
/// ```
pub trait IntoStep<In, Out, Params> {
    /// The concrete resolved step type.
    type Step: StepCall<In, Out>;

    /// Resolve Param state from the registry and produce a step.
    fn into_step(self, registry: &Registry) -> Self::Step;
}

// =============================================================================
// Arity 0 — fn(In) -> Out — closures work (no HRTB+GAT issues)
// =============================================================================

impl<In, Out, F: FnMut(In) -> Out + 'static> StepCall<In, Out> for Step<F, ()> {
    #[inline(always)]
    fn call(&mut self, _world: &mut World, input: In) -> Out {
        (self.f)(input)
    }
}

impl<In, Out, F: FnMut(In) -> Out + 'static> IntoStep<In, Out, ()> for F {
    type Step = Step<F, ()>;

    fn into_step(self, registry: &Registry) -> Self::Step {
        Step {
            f: self,
            state: <() as Param>::init(registry),
            name: std::any::type_name::<F>(),
        }
    }
}

// =============================================================================
// Arities 1-8 via macro — HRTB with -> Out
// =============================================================================

macro_rules! impl_into_step {
    ($($P:ident),+) => {
        impl<In, Out, F: 'static, $($P: Param + 'static),+>
            StepCall<In, Out> for Step<F, ($($P,)+)>
        where
            for<'a> &'a mut F:
                FnMut($($P,)+ In) -> Out +
                FnMut($($P::Item<'a>,)+ In) -> Out,
        {
            #[inline(always)]
            #[allow(non_snake_case)]
            fn call(&mut self, world: &mut World, input: In) -> Out {
                #[allow(clippy::too_many_arguments)]
                fn call_inner<$($P,)+ Input, Output>(
                    mut f: impl FnMut($($P,)+ Input) -> Output,
                    $($P: $P,)+
                    input: Input,
                ) -> Output {
                    f($($P,)+ input)
                }

                // SAFETY: state was produced by init() on the same registry
                // that built this world. Single-threaded sequential dispatch
                // ensures no mutable aliasing across params.
                #[cfg(debug_assertions)]
                world.clear_borrows();
                let ($($P,)+) = unsafe {
                    <($($P,)+) as Param>::fetch(world, &mut self.state)
                };
                call_inner(&mut self.f, $($P,)+ input)
            }
        }

        impl<In, Out, F: 'static, $($P: Param + 'static),+>
            IntoStep<In, Out, ($($P,)+)> for F
        where
            for<'a> &'a mut F:
                FnMut($($P,)+ In) -> Out +
                FnMut($($P::Item<'a>,)+ In) -> Out,
        {
            type Step = Step<F, ($($P,)+)>;

            fn into_step(self, registry: &Registry) -> Self::Step {
                let state = <($($P,)+) as Param>::init(registry);
                {
                    #[allow(non_snake_case)]
                    let ($($P,)+) = &state;
                    registry.check_access(&[
                        $(
                            (<$P as Param>::resource_id($P),
                             std::any::type_name::<$P>()),
                        )+
                    ]);
                }
                Step { f: self, state, name: std::any::type_name::<F>() }
            }
        }
    };
}

macro_rules! all_tuples {
    ($m:ident) => {
        $m!(P0);
        $m!(P0, P1);
        $m!(P0, P1, P2);
        $m!(P0, P1, P2, P3);
        $m!(P0, P1, P2, P3, P4);
        $m!(P0, P1, P2, P3, P4, P5);
        $m!(P0, P1, P2, P3, P4, P5, P6);
        $m!(P0, P1, P2, P3, P4, P5, P6, P7);
    };
}

all_tuples!(impl_into_step);

// =============================================================================
// SplatCall / IntoSplatStep — splat step dispatch (tuple destructuring)
// =============================================================================
//
// Splat traits mirror StepCall/IntoStep but accept multiple owned values
// instead of a single input. This lets `.splat()` destructure a tuple
// output into individual function arguments for the next step.
//
// One trait pair per arity (2-5). Past 5, use a named struct.

// -- Splat 2 ------------------------------------------------------------------

/// Internal: callable trait for resolved 2-splat steps.
#[doc(hidden)]
pub trait SplatCall2<A, B, Out> {
    fn call_splat(&mut self, world: &mut World, a: A, b: B) -> Out;
}

/// Converts a named function into a resolved 2-splat step.
#[doc(hidden)]
pub trait IntoSplatStep2<A, B, Out, Params> {
    type Step: SplatCall2<A, B, Out>;
    fn into_splat_step(self, registry: &Registry) -> Self::Step;
}

impl<A, B, Out, F: FnMut(A, B) -> Out + 'static> SplatCall2<A, B, Out> for Step<F, ()> {
    #[inline(always)]
    fn call_splat(&mut self, _world: &mut World, a: A, b: B) -> Out {
        (self.f)(a, b)
    }
}

impl<A, B, Out, F: FnMut(A, B) -> Out + 'static> IntoSplatStep2<A, B, Out, ()> for F {
    type Step = Step<F, ()>;
    fn into_splat_step(self, registry: &Registry) -> Self::Step {
        Step {
            f: self,
            state: <() as Param>::init(registry),
            name: std::any::type_name::<F>(),
        }
    }
}

macro_rules! impl_splat2_step {
    ($($P:ident),+) => {
        impl<A, B, Out, F: 'static, $($P: Param + 'static),+>
            SplatCall2<A, B, Out> for Step<F, ($($P,)+)>
        where
            for<'a> &'a mut F:
                FnMut($($P,)+ A, B) -> Out +
                FnMut($($P::Item<'a>,)+ A, B) -> Out,
        {
            #[inline(always)]
            #[allow(non_snake_case)]
            fn call_splat(&mut self, world: &mut World, a: A, b: B) -> Out {
                #[allow(clippy::too_many_arguments)]
                fn call_inner<$($P,)+ IA, IB, Output>(
                    mut f: impl FnMut($($P,)+ IA, IB) -> Output,
                    $($P: $P,)+
                    a: IA, b: IB,
                ) -> Output {
                    f($($P,)+ a, b)
                }
                #[cfg(debug_assertions)]
                world.clear_borrows();
                let ($($P,)+) = unsafe {
                    <($($P,)+) as Param>::fetch(world, &mut self.state)
                };
                call_inner(&mut self.f, $($P,)+ a, b)
            }
        }

        impl<A, B, Out, F: 'static, $($P: Param + 'static),+>
            IntoSplatStep2<A, B, Out, ($($P,)+)> for F
        where
            for<'a> &'a mut F:
                FnMut($($P,)+ A, B) -> Out +
                FnMut($($P::Item<'a>,)+ A, B) -> Out,
        {
            type Step = Step<F, ($($P,)+)>;

            fn into_splat_step(self, registry: &Registry) -> Self::Step {
                let state = <($($P,)+) as Param>::init(registry);
                {
                    #[allow(non_snake_case)]
                    let ($($P,)+) = &state;
                    registry.check_access(&[
                        $(
                            (<$P as Param>::resource_id($P),
                             std::any::type_name::<$P>()),
                        )+
                    ]);
                }
                Step { f: self, state, name: std::any::type_name::<F>() }
            }
        }
    };
}

// -- Splat 3 ------------------------------------------------------------------

/// Internal: callable trait for resolved 3-splat steps.
#[doc(hidden)]
pub trait SplatCall3<A, B, C, Out> {
    fn call_splat(&mut self, world: &mut World, a: A, b: B, c: C) -> Out;
}

/// Converts a named function into a resolved 3-splat step.
#[doc(hidden)]
pub trait IntoSplatStep3<A, B, C, Out, Params> {
    type Step: SplatCall3<A, B, C, Out>;
    fn into_splat_step(self, registry: &Registry) -> Self::Step;
}

impl<A, B, C, Out, F: FnMut(A, B, C) -> Out + 'static> SplatCall3<A, B, C, Out> for Step<F, ()> {
    #[inline(always)]
    fn call_splat(&mut self, _world: &mut World, a: A, b: B, c: C) -> Out {
        (self.f)(a, b, c)
    }
}

impl<A, B, C, Out, F: FnMut(A, B, C) -> Out + 'static> IntoSplatStep3<A, B, C, Out, ()> for F {
    type Step = Step<F, ()>;
    fn into_splat_step(self, registry: &Registry) -> Self::Step {
        Step {
            f: self,
            state: <() as Param>::init(registry),
            name: std::any::type_name::<F>(),
        }
    }
}

macro_rules! impl_splat3_step {
    ($($P:ident),+) => {
        impl<A, B, C, Out, F: 'static, $($P: Param + 'static),+>
            SplatCall3<A, B, C, Out> for Step<F, ($($P,)+)>
        where
            for<'a> &'a mut F:
                FnMut($($P,)+ A, B, C) -> Out +
                FnMut($($P::Item<'a>,)+ A, B, C) -> Out,
        {
            #[inline(always)]
            #[allow(non_snake_case)]
            fn call_splat(&mut self, world: &mut World, a: A, b: B, c: C) -> Out {
                #[allow(clippy::too_many_arguments)]
                fn call_inner<$($P,)+ IA, IB, IC, Output>(
                    mut f: impl FnMut($($P,)+ IA, IB, IC) -> Output,
                    $($P: $P,)+
                    a: IA, b: IB, c: IC,
                ) -> Output {
                    f($($P,)+ a, b, c)
                }
                #[cfg(debug_assertions)]
                world.clear_borrows();
                let ($($P,)+) = unsafe {
                    <($($P,)+) as Param>::fetch(world, &mut self.state)
                };
                call_inner(&mut self.f, $($P,)+ a, b, c)
            }
        }

        impl<A, B, C, Out, F: 'static, $($P: Param + 'static),+>
            IntoSplatStep3<A, B, C, Out, ($($P,)+)> for F
        where
            for<'a> &'a mut F:
                FnMut($($P,)+ A, B, C) -> Out +
                FnMut($($P::Item<'a>,)+ A, B, C) -> Out,
        {
            type Step = Step<F, ($($P,)+)>;

            fn into_splat_step(self, registry: &Registry) -> Self::Step {
                let state = <($($P,)+) as Param>::init(registry);
                {
                    #[allow(non_snake_case)]
                    let ($($P,)+) = &state;
                    registry.check_access(&[
                        $(
                            (<$P as Param>::resource_id($P),
                             std::any::type_name::<$P>()),
                        )+
                    ]);
                }
                Step { f: self, state, name: std::any::type_name::<F>() }
            }
        }
    };
}

// -- Splat 4 ------------------------------------------------------------------

/// Internal: callable trait for resolved 4-splat steps.
#[doc(hidden)]
pub trait SplatCall4<A, B, C, D, Out> {
    fn call_splat(&mut self, world: &mut World, a: A, b: B, c: C, d: D) -> Out;
}

/// Converts a named function into a resolved 4-splat step.
#[doc(hidden)]
pub trait IntoSplatStep4<A, B, C, D, Out, Params> {
    type Step: SplatCall4<A, B, C, D, Out>;
    fn into_splat_step(self, registry: &Registry) -> Self::Step;
}

impl<A, B, C, D, Out, F: FnMut(A, B, C, D) -> Out + 'static> SplatCall4<A, B, C, D, Out>
    for Step<F, ()>
{
    #[inline(always)]
    fn call_splat(&mut self, _world: &mut World, a: A, b: B, c: C, d: D) -> Out {
        (self.f)(a, b, c, d)
    }
}

impl<A, B, C, D, Out, F: FnMut(A, B, C, D) -> Out + 'static> IntoSplatStep4<A, B, C, D, Out, ()>
    for F
{
    type Step = Step<F, ()>;
    fn into_splat_step(self, registry: &Registry) -> Self::Step {
        Step {
            f: self,
            state: <() as Param>::init(registry),
            name: std::any::type_name::<F>(),
        }
    }
}

macro_rules! impl_splat4_step {
    ($($P:ident),+) => {
        impl<A, B, C, D, Out, F: 'static, $($P: Param + 'static),+>
            SplatCall4<A, B, C, D, Out> for Step<F, ($($P,)+)>
        where for<'a> &'a mut F:
            FnMut($($P,)+ A, B, C, D) -> Out +
            FnMut($($P::Item<'a>,)+ A, B, C, D) -> Out,
        {
            #[inline(always)]
            #[allow(non_snake_case)]
            fn call_splat(&mut self, world: &mut World, a: A, b: B, c: C, d: D) -> Out {
                #[allow(clippy::too_many_arguments)]
                fn call_inner<$($P,)+ IA, IB, IC, ID, Output>(
                    mut f: impl FnMut($($P,)+ IA, IB, IC, ID) -> Output,
                    $($P: $P,)+ a: IA, b: IB, c: IC, d: ID,
                ) -> Output { f($($P,)+ a, b, c, d) }
                #[cfg(debug_assertions)]
                world.clear_borrows();
                let ($($P,)+) = unsafe {
                    <($($P,)+) as Param>::fetch(world, &mut self.state)
                };
                call_inner(&mut self.f, $($P,)+ a, b, c, d)
            }
        }
        impl<A, B, C, D, Out, F: 'static, $($P: Param + 'static),+>
            IntoSplatStep4<A, B, C, D, Out, ($($P,)+)> for F
        where for<'a> &'a mut F:
            FnMut($($P,)+ A, B, C, D) -> Out +
            FnMut($($P::Item<'a>,)+ A, B, C, D) -> Out,
        {
            type Step = Step<F, ($($P,)+)>;
            fn into_splat_step(self, registry: &Registry) -> Self::Step {
                let state = <($($P,)+) as Param>::init(registry);
                { #[allow(non_snake_case)] let ($($P,)+) = &state;
                  registry.check_access(&[$((<$P as Param>::resource_id($P), std::any::type_name::<$P>()),)+]); }
                Step { f: self, state, name: std::any::type_name::<F>() }
            }
        }
    };
}

// -- Splat 5 ------------------------------------------------------------------

/// Internal: callable trait for resolved 5-splat steps.
#[doc(hidden)]
pub trait SplatCall5<A, B, C, D, E, Out> {
    #[allow(clippy::many_single_char_names)]
    fn call_splat(&mut self, world: &mut World, a: A, b: B, c: C, d: D, e: E) -> Out;
}

/// Converts a named function into a resolved 5-splat step.
#[doc(hidden)]
pub trait IntoSplatStep5<A, B, C, D, E, Out, Params> {
    type Step: SplatCall5<A, B, C, D, E, Out>;
    fn into_splat_step(self, registry: &Registry) -> Self::Step;
}

impl<A, B, C, D, E, Out, F: FnMut(A, B, C, D, E) -> Out + 'static> SplatCall5<A, B, C, D, E, Out>
    for Step<F, ()>
{
    #[inline(always)]
    #[allow(clippy::many_single_char_names)]
    fn call_splat(&mut self, _world: &mut World, a: A, b: B, c: C, d: D, e: E) -> Out {
        (self.f)(a, b, c, d, e)
    }
}

impl<A, B, C, D, E, Out, F: FnMut(A, B, C, D, E) -> Out + 'static>
    IntoSplatStep5<A, B, C, D, E, Out, ()> for F
{
    type Step = Step<F, ()>;
    fn into_splat_step(self, registry: &Registry) -> Self::Step {
        Step {
            f: self,
            state: <() as Param>::init(registry),
            name: std::any::type_name::<F>(),
        }
    }
}

macro_rules! impl_splat5_step {
    ($($P:ident),+) => {
        impl<A, B, C, D, E, Out, F: 'static, $($P: Param + 'static),+>
            SplatCall5<A, B, C, D, E, Out> for Step<F, ($($P,)+)>
        where for<'a> &'a mut F:
            FnMut($($P,)+ A, B, C, D, E) -> Out +
            FnMut($($P::Item<'a>,)+ A, B, C, D, E) -> Out,
        {
            #[inline(always)]
            #[allow(non_snake_case, clippy::many_single_char_names)]
            fn call_splat(&mut self, world: &mut World, a: A, b: B, c: C, d: D, e: E) -> Out {
                #[allow(clippy::too_many_arguments)]
                fn call_inner<$($P,)+ IA, IB, IC, ID, IE, Output>(
                    mut f: impl FnMut($($P,)+ IA, IB, IC, ID, IE) -> Output,
                    $($P: $P,)+ a: IA, b: IB, c: IC, d: ID, e: IE,
                ) -> Output { f($($P,)+ a, b, c, d, e) }
                #[cfg(debug_assertions)]
                world.clear_borrows();
                let ($($P,)+) = unsafe {
                    <($($P,)+) as Param>::fetch(world, &mut self.state)
                };
                call_inner(&mut self.f, $($P,)+ a, b, c, d, e)
            }
        }
        impl<A, B, C, D, E, Out, F: 'static, $($P: Param + 'static),+>
            IntoSplatStep5<A, B, C, D, E, Out, ($($P,)+)> for F
        where for<'a> &'a mut F:
            FnMut($($P,)+ A, B, C, D, E) -> Out +
            FnMut($($P::Item<'a>,)+ A, B, C, D, E) -> Out,
        {
            type Step = Step<F, ($($P,)+)>;
            fn into_splat_step(self, registry: &Registry) -> Self::Step {
                let state = <($($P,)+) as Param>::init(registry);
                { #[allow(non_snake_case)] let ($($P,)+) = &state;
                  registry.check_access(&[$((<$P as Param>::resource_id($P), std::any::type_name::<$P>()),)+]); }
                Step { f: self, state, name: std::any::type_name::<F>() }
            }
        }
    };
}

all_tuples!(impl_splat2_step);
all_tuples!(impl_splat3_step);
all_tuples!(impl_splat4_step);
all_tuples!(impl_splat5_step);

// =============================================================================
// PipelineStart — entry point
// =============================================================================

/// Entry point for building a pre-resolved step pipeline.
///
/// `In` is the pipeline input type. Call [`.then()`](Self::then) to add
/// the first step — a named function whose [`Param`] dependencies
/// are resolved from the registry at build time.
///
/// # Examples
///
/// ```
/// use nexus_rt::{WorldBuilder, Res, ResMut, PipelineStart, Handler};
///
/// let mut wb = WorldBuilder::new();
/// wb.register::<u64>(10);
/// wb.register::<String>(String::new());
/// let mut world = wb.build();
///
/// fn double(factor: Res<u64>, x: u32) -> u64 {
///     (*factor) * x as u64
/// }
/// fn store(mut out: ResMut<String>, val: u64) {
///     *out = val.to_string();
/// }
///
/// let r = world.registry_mut();
/// let mut pipeline = PipelineStart::<u32>::new()
///     .then(double, r)
///     .then(store, r)
///     .build();
///
/// pipeline.run(&mut world, 5);
/// assert_eq!(world.resource::<String>().as_str(), "50");
/// ```
pub struct PipelineStart<In>(PhantomData<fn(In)>);

impl<In> PipelineStart<In> {
    /// Create a new step pipeline entry point.
    pub fn new() -> Self {
        Self(PhantomData)
    }

    /// Add the first step. Params resolved from the registry.
    pub fn then<Out, Params, S: IntoStep<In, Out, Params>>(
        self,
        f: S,
        registry: &Registry,
    ) -> PipelineBuilder<In, Out, impl FnMut(&mut World, In) -> Out + use<In, Out, Params, S>> {
        let mut resolved = f.into_step(registry);
        PipelineBuilder {
            chain: move |world: &mut World, input: In| resolved.call(world, input),
            _marker: PhantomData,
        }
    }

    /// Closure-based first step for N-ary conditional routing.
    ///
    /// Same as [`PipelineBuilder::switch`] but usable as the first step.
    /// See that method for full documentation.
    pub fn switch<Out>(
        self,
        mut f: impl FnMut(&mut World, In) -> Out,
    ) -> PipelineBuilder<In, Out, impl FnMut(&mut World, In) -> Out> {
        PipelineBuilder {
            chain: move |world: &mut World, input: In| f(world, input),
            _marker: PhantomData,
        }
    }
}

impl<In> Default for PipelineStart<In> {
    fn default() -> Self {
        Self::new()
    }
}

// =============================================================================
// PipelineBuilder — typestate builder
// =============================================================================

/// Builder that composes pre-resolved pipeline steps via closure nesting.
///
/// `In` is the pipeline's input type (fixed). `Out` is the current output.
/// `Chain` is the concrete composed closure type (opaque, never named by users).
///
/// Each combinator consumes `self`, captures the previous chain in a new
/// closure, and returns a new `PipelineBuilder`. The compiler
/// monomorphizes the entire chain — zero virtual dispatch through steps.
///
/// IntoStep-based methods (`.then()`, `.map()`, `.and_then()`, `.catch()`)
/// take `&Registry` to resolve Param state at build time. Closure-based
/// methods don't need the registry.
pub struct PipelineBuilder<In, Out, Chain> {
    chain: Chain,
    _marker: PhantomData<fn(In) -> Out>,
}

// =============================================================================
// Core — any Out
// =============================================================================

impl<In, Out, Chain> PipelineBuilder<In, Out, Chain>
where
    Chain: FnMut(&mut World, In) -> Out,
{
    /// Add a step. Params resolved from the registry.
    pub fn then<NewOut, Params, S: IntoStep<Out, NewOut, Params>>(
        self,
        f: S,
        registry: &Registry,
    ) -> PipelineBuilder<
        In,
        NewOut,
        impl FnMut(&mut World, In) -> NewOut + use<In, Out, NewOut, Params, Chain, S>,
    > {
        let mut chain = self.chain;
        let mut resolved = f.into_step(registry);
        PipelineBuilder {
            chain: move |world: &mut World, input: In| {
                let out = chain(world, input);
                resolved.call(world, out)
            },
            _marker: PhantomData,
        }
    }

    /// Run the pipeline directly. No boxing, no `'static` on `In`.
    pub fn run(&mut self, world: &mut World, input: In) -> Out {
        (self.chain)(world, input)
    }

    /// Dispatch pipeline output to a [`Handler<Out>`].
    ///
    /// Connects a pipeline's output to any handler — [`HandlerFn`](crate::HandlerFn),
    /// [`Callback`](crate::Callback), [`Pipeline`], or a combinator like
    /// [`fan_out!`](crate::fan_out).
    pub fn dispatch<H: Handler<Out>>(
        self,
        mut handler: H,
    ) -> PipelineBuilder<In, (), impl FnMut(&mut World, In) + use<In, Out, Chain, H>> {
        let mut chain = self.chain;
        PipelineBuilder {
            chain: move |world: &mut World, input: In| {
                let out = chain(world, input);
                handler.run(world, out);
            },
            _marker: PhantomData,
        }
    }

    /// Conditionally wrap the output in `Option`. `Some(val)` if
    /// the predicate returns true, `None` otherwise.
    ///
    /// Enters Option-combinator land — follow with `.map()`,
    /// `.and_then()`, `.filter()`, `.unwrap_or()`, etc.
    pub fn guard(
        self,
        mut f: impl FnMut(&mut World, &Out) -> bool,
    ) -> PipelineBuilder<In, Option<Out>, impl FnMut(&mut World, In) -> Option<Out>> {
        let mut chain = self.chain;
        PipelineBuilder {
            chain: move |world: &mut World, input: In| {
                let val = chain(world, input);
                if f(world, &val) { Some(val) } else { None }
            },
            _marker: PhantomData,
        }
    }

    /// Observe the current value without consuming or changing it.
    ///
    /// The closure receives `&mut World` and `&Out`. The value passes
    /// through unchanged. Useful for logging, metrics, or debugging
    /// mid-chain.
    pub fn tap(
        self,
        mut f: impl FnMut(&mut World, &Out),
    ) -> PipelineBuilder<In, Out, impl FnMut(&mut World, In) -> Out> {
        let mut chain = self.chain;
        PipelineBuilder {
            chain: move |world: &mut World, input: In| {
                let val = chain(world, input);
                f(world, &val);
                val
            },
            _marker: PhantomData,
        }
    }

    /// Binary conditional routing. Evaluates the predicate on the
    /// current value, then moves it into exactly one of two arms.
    ///
    /// Both arms must produce the same output type. Build each arm as
    /// a sub-pipeline from [`PipelineStart`]. For N-ary routing, nest
    /// `route` calls in the false arm.
    ///
    /// ```ignore
    /// let large = PipelineStart::new().then(large_check, reg).then(submit, reg);
    /// let small = PipelineStart::new().then(submit, reg);
    ///
    /// PipelineStart::<Order>::new()
    ///     .then(validate, reg)
    ///     .route(|_, order| order.size > 1000, large, small)
    ///     .build();
    /// ```
    pub fn route<NewOut, C0, C1, P>(
        self,
        mut pred: P,
        on_true: PipelineBuilder<Out, NewOut, C0>,
        on_false: PipelineBuilder<Out, NewOut, C1>,
    ) -> PipelineBuilder<
        In,
        NewOut,
        impl FnMut(&mut World, In) -> NewOut + use<In, Out, NewOut, Chain, C0, C1, P>,
    >
    where
        P: FnMut(&mut World, &Out) -> bool,
        C0: FnMut(&mut World, Out) -> NewOut,
        C1: FnMut(&mut World, Out) -> NewOut,
    {
        let mut chain = self.chain;
        let mut c0 = on_true.chain;
        let mut c1 = on_false.chain;
        PipelineBuilder {
            chain: move |world: &mut World, input: In| {
                let val = chain(world, input);
                if pred(world, &val) {
                    c0(world, val)
                } else {
                    c1(world, val)
                }
            },
            _marker: PhantomData,
        }
    }

    /// Fork off a multi-step side-effect chain. The arm borrows
    /// `&Out`, runs to completion (producing `()`), and the
    /// original value passes through unchanged.
    ///
    /// Multi-step version of [`tap`](Self::tap) — the arm has the
    /// full DAG combinator API with Param resolution. Build with
    /// [`DagArmStart::new()`](crate::dag::DagArmStart::new).
    pub fn tee<C>(
        self,
        side: DagArm<Out, (), C>,
    ) -> PipelineBuilder<In, Out, impl FnMut(&mut World, In) -> Out>
    where
        C: FnMut(&mut World, &Out),
    {
        let mut chain = self.chain;
        let mut side_chain = side.chain;
        PipelineBuilder {
            chain: move |world: &mut World, input: In| {
                let val = chain(world, input);
                side_chain(world, &val);
                val
            },
            _marker: PhantomData,
        }
    }

    /// Closure-based step for N-ary conditional routing.
    ///
    /// The closure receives `&mut World` and the current value `Out`
    /// **by value** (pipeline ownership semantics), returning a new
    /// value of type `NewOut`. Use with [`resolve_step`] to pre-resolve
    /// per-arm steps with independent [`Param`] sets:
    ///
    /// ```ignore
    /// let mut arm_new    = resolve_step(handle_new, reg);
    /// let mut arm_cancel = resolve_step(handle_cancel, reg);
    ///
    /// pipeline.switch(move |world, order: Order| match order.kind {
    ///     OrderKind::New    => arm_new(world, order),
    ///     OrderKind::Cancel => arm_cancel(world, order),
    /// })
    /// ```
    ///
    /// For simple cases where all arms share the same params, a named
    /// function with a `match` body passed to [`.then()`](Self::then)
    /// is sufficient.
    pub fn switch<NewOut>(
        self,
        mut f: impl FnMut(&mut World, Out) -> NewOut,
    ) -> PipelineBuilder<In, NewOut, impl FnMut(&mut World, In) -> NewOut> {
        let mut chain = self.chain;
        PipelineBuilder {
            chain: move |world: &mut World, input: In| {
                let val = chain(world, input);
                f(world, val)
            },
            _marker: PhantomData,
        }
    }
}

// =============================================================================
// Splat — tuple destructuring into individual function arguments
// =============================================================================
//
// `.splat()` transitions from a tuple output to a builder whose `.then()`
// accepts a function taking the tuple elements as individual arguments.
// After `.splat().then(f, reg)`, the user is back on PipelineBuilder.
//
// Builder types are `#[doc(hidden)]` — users only see `.splat().then()`.

// -- Splat builder types ------------------------------------------------------

macro_rules! define_splat_builders {
    (
        $N:literal,
        start: $SplatStart:ident,
        mid: $SplatBuilder:ident,
        into_trait: $IntoSplatStep:ident,
        call_trait: $SplatCall:ident,
        ($($T:ident),+),
        ($($idx:tt),+)
    ) => {
        /// Splat builder at pipeline start position.
        #[doc(hidden)]
        pub struct $SplatStart<$($T),+>(PhantomData<fn(($($T,)+))>);

        impl<$($T),+> $SplatStart<$($T),+> {
            /// Add a step that receives the tuple elements as individual arguments.
            pub fn then<Out, Params, S>(
                self,
                f: S,
                registry: &Registry,
            ) -> PipelineBuilder<
                ($($T,)+),
                Out,
                impl FnMut(&mut World, ($($T,)+)) -> Out
                    + use<$($T,)+ Out, Params, S>,
            >
            where
                S: $IntoSplatStep<$($T,)+ Out, Params>,
            {
                let mut resolved = f.into_splat_step(registry);
                PipelineBuilder {
                    chain: move |world: &mut World, input: ($($T,)+)| {
                        resolved.call_splat(world, $(input.$idx),+)
                    },
                    _marker: PhantomData,
                }
            }
        }

        impl<$($T),+> PipelineStart<($($T,)+)> {
            /// Destructure the tuple input into individual function arguments.
            pub fn splat(self) -> $SplatStart<$($T),+> {
                $SplatStart(PhantomData)
            }
        }

        /// Splat builder at mid-chain position.
        #[doc(hidden)]
        pub struct $SplatBuilder<In, $($T,)+ Chain> {
            chain: Chain,
            _marker: PhantomData<fn(In) -> ($($T,)+)>,
        }

        impl<In, $($T,)+ Chain> $SplatBuilder<In, $($T,)+ Chain>
        where
            Chain: FnMut(&mut World, In) -> ($($T,)+),
        {
            /// Add a step that receives the tuple elements as individual arguments.
            pub fn then<Out, Params, S>(
                self,
                f: S,
                registry: &Registry,
            ) -> PipelineBuilder<
                In,
                Out,
                impl FnMut(&mut World, In) -> Out
                    + use<In, $($T,)+ Out, Params, Chain, S>,
            >
            where
                S: $IntoSplatStep<$($T,)+ Out, Params>,
            {
                let mut chain = self.chain;
                let mut resolved = f.into_splat_step(registry);
                PipelineBuilder {
                    chain: move |world: &mut World, input: In| {
                        let tuple = chain(world, input);
                        resolved.call_splat(world, $(tuple.$idx),+)
                    },
                    _marker: PhantomData,
                }
            }
        }

        impl<In, $($T,)+ Chain> PipelineBuilder<In, ($($T,)+), Chain>
        where
            Chain: FnMut(&mut World, In) -> ($($T,)+),
        {
            /// Destructure the tuple output into individual function arguments.
            pub fn splat(self) -> $SplatBuilder<In, $($T,)+ Chain> {
                $SplatBuilder {
                    chain: self.chain,
                    _marker: PhantomData,
                }
            }
        }
    };
}

define_splat_builders!(2,
    start: SplatStart2,
    mid: SplatBuilder2,
    into_trait: IntoSplatStep2,
    call_trait: SplatCall2,
    (A, B),
    (0, 1)
);

define_splat_builders!(3,
    start: SplatStart3,
    mid: SplatBuilder3,
    into_trait: IntoSplatStep3,
    call_trait: SplatCall3,
    (A, B, C),
    (0, 1, 2)
);

define_splat_builders!(4,
    start: SplatStart4,
    mid: SplatBuilder4,
    into_trait: IntoSplatStep4,
    call_trait: SplatCall4,
    (A, B, C, D),
    (0, 1, 2, 3)
);

define_splat_builders!(5,
    start: SplatStart5,
    mid: SplatBuilder5,
    into_trait: IntoSplatStep5,
    call_trait: SplatCall5,
    (A, B, C, D, E),
    (0, 1, 2, 3, 4)
);

// =============================================================================
// Dedup — suppress unchanged values
// =============================================================================

impl<In, Out: PartialEq + Clone, Chain> PipelineBuilder<In, Out, Chain>
where
    Chain: FnMut(&mut World, In) -> Out,
{
    /// Suppress consecutive unchanged values. Returns `Some(val)`
    /// when the value differs from the previous invocation, `None`
    /// when unchanged. First invocation always returns `Some`.
    ///
    /// Requires `PartialEq + Clone` — the previous value is stored
    /// internally for comparison.
    pub fn dedup(
        self,
    ) -> PipelineBuilder<In, Option<Out>, impl FnMut(&mut World, In) -> Option<Out>> {
        let mut chain = self.chain;
        let mut prev: Option<Out> = None;
        PipelineBuilder {
            chain: move |world: &mut World, input: In| {
                let val = chain(world, input);
                if prev.as_ref() == Some(&val) {
                    None
                } else {
                    prev = Some(val.clone());
                    Some(val)
                }
            },
            _marker: PhantomData,
        }
    }
}

// =============================================================================
// Bool combinators
// =============================================================================

impl<In, Chain> PipelineBuilder<In, bool, Chain>
where
    Chain: FnMut(&mut World, In) -> bool,
{
    /// Invert a boolean value.
    #[allow(clippy::should_implement_trait)]
    pub fn not(self) -> PipelineBuilder<In, bool, impl FnMut(&mut World, In) -> bool> {
        let mut chain = self.chain;
        PipelineBuilder {
            chain: move |world: &mut World, input: In| !chain(world, input),
            _marker: PhantomData,
        }
    }

    /// Short-circuit AND with a second boolean from World state.
    ///
    /// If the chain produces `false`, the closure is not called.
    pub fn and(
        self,
        mut f: impl FnMut(&mut World) -> bool,
    ) -> PipelineBuilder<In, bool, impl FnMut(&mut World, In) -> bool> {
        let mut chain = self.chain;
        PipelineBuilder {
            chain: move |world: &mut World, input: In| chain(world, input) && f(world),
            _marker: PhantomData,
        }
    }

    /// Short-circuit OR with a second boolean from World state.
    ///
    /// If the chain produces `true`, the closure is not called.
    pub fn or(
        self,
        mut f: impl FnMut(&mut World) -> bool,
    ) -> PipelineBuilder<In, bool, impl FnMut(&mut World, In) -> bool> {
        let mut chain = self.chain;
        PipelineBuilder {
            chain: move |world: &mut World, input: In| chain(world, input) || f(world),
            _marker: PhantomData,
        }
    }

    /// XOR with a second boolean from World state.
    ///
    /// Both sides are always evaluated.
    pub fn xor(
        self,
        mut f: impl FnMut(&mut World) -> bool,
    ) -> PipelineBuilder<In, bool, impl FnMut(&mut World, In) -> bool> {
        let mut chain = self.chain;
        PipelineBuilder {
            chain: move |world: &mut World, input: In| chain(world, input) ^ f(world),
            _marker: PhantomData,
        }
    }
}

// =============================================================================
// Clone helpers — &T → T transitions
// =============================================================================

impl<'a, In, T: Clone, Chain> PipelineBuilder<In, &'a T, Chain>
where
    Chain: FnMut(&mut World, In) -> &'a T,
{
    /// Clone a borrowed output to produce an owned value.
    ///
    /// Transitions the pipeline from `&T` to `T`. Uses UFCS
    /// (`T::clone(val)`) — `val.clone()` on a `&&T` resolves to
    /// `<&T as Clone>::clone` and returns `&T`, not `T`.
    pub fn cloned(self) -> PipelineBuilder<In, T, impl FnMut(&mut World, In) -> T> {
        let mut chain = self.chain;
        PipelineBuilder {
            chain: move |world: &mut World, input: In| T::clone(chain(world, input)),
            _marker: PhantomData,
        }
    }
}

impl<'a, In, T: Clone, Chain> PipelineBuilder<In, Option<&'a T>, Chain>
where
    Chain: FnMut(&mut World, In) -> Option<&'a T>,
{
    /// Clone inner borrowed value. `Option<&T>` → `Option<T>`.
    pub fn cloned(self) -> PipelineBuilder<In, Option<T>, impl FnMut(&mut World, In) -> Option<T>> {
        let mut chain = self.chain;
        PipelineBuilder {
            chain: move |world: &mut World, input: In| chain(world, input).cloned(),
            _marker: PhantomData,
        }
    }
}

impl<'a, In, T: Clone, E, Chain> PipelineBuilder<In, Result<&'a T, E>, Chain>
where
    Chain: FnMut(&mut World, In) -> Result<&'a T, E>,
{
    /// Clone inner borrowed Ok value. `Result<&T, E>` → `Result<T, E>`.
    pub fn cloned(
        self,
    ) -> PipelineBuilder<In, Result<T, E>, impl FnMut(&mut World, In) -> Result<T, E>> {
        let mut chain = self.chain;
        PipelineBuilder {
            chain: move |world: &mut World, input: In| chain(world, input).cloned(),
            _marker: PhantomData,
        }
    }
}

// =============================================================================
// Option helpers — PipelineBuilder<In, Option<T>, Chain>
// =============================================================================

impl<In, T, Chain> PipelineBuilder<In, Option<T>, Chain>
where
    Chain: FnMut(&mut World, In) -> Option<T>,
{
    // -- IntoStep-based (hot path) -------------------------------------------

    /// Transform the inner value. Step not called on None.
    pub fn map<U, Params, S: IntoStep<T, U, Params>>(
        self,
        f: S,
        registry: &Registry,
    ) -> PipelineBuilder<
        In,
        Option<U>,
        impl FnMut(&mut World, In) -> Option<U> + use<In, T, U, Params, Chain, S>,
    > {
        let mut chain = self.chain;
        let mut resolved = f.into_step(registry);
        PipelineBuilder {
            chain: move |world: &mut World, input: In| {
                chain(world, input).map(|val| resolved.call(world, val))
            },
            _marker: PhantomData,
        }
    }

    /// Short-circuits on None. std: `Option::and_then`
    pub fn and_then<U, Params, S: IntoStep<T, Option<U>, Params>>(
        self,
        f: S,
        registry: &Registry,
    ) -> PipelineBuilder<
        In,
        Option<U>,
        impl FnMut(&mut World, In) -> Option<U> + use<In, T, U, Params, Chain, S>,
    > {
        let mut chain = self.chain;
        let mut resolved = f.into_step(registry);
        PipelineBuilder {
            chain: move |world: &mut World, input: In| {
                chain(world, input).and_then(|val| resolved.call(world, val))
            },
            _marker: PhantomData,
        }
    }

    // -- Closure-based (cold path, &mut World) --------------------------------

    /// Side effect on None. Complement to [`inspect`](Self::inspect).
    pub fn on_none(
        self,
        mut f: impl FnMut(&mut World),
    ) -> PipelineBuilder<In, Option<T>, impl FnMut(&mut World, In) -> Option<T>> {
        let mut chain = self.chain;
        PipelineBuilder {
            chain: move |world: &mut World, input: In| {
                let result = chain(world, input);
                if result.is_none() {
                    f(world);
                }
                result
            },
            _marker: PhantomData,
        }
    }

    /// Keep value if predicate holds. std: `Option::filter`
    pub fn filter(
        self,
        mut f: impl FnMut(&mut World, &T) -> bool,
    ) -> PipelineBuilder<In, Option<T>, impl FnMut(&mut World, In) -> Option<T>> {
        let mut chain = self.chain;
        PipelineBuilder {
            chain: move |world: &mut World, input: In| {
                chain(world, input).filter(|val| f(world, val))
            },
            _marker: PhantomData,
        }
    }

    /// Side effect on Some value. std: `Option::inspect`
    pub fn inspect(
        self,
        mut f: impl FnMut(&mut World, &T),
    ) -> PipelineBuilder<In, Option<T>, impl FnMut(&mut World, In) -> Option<T>> {
        let mut chain = self.chain;
        PipelineBuilder {
            chain: move |world: &mut World, input: In| {
                chain(world, input).inspect(|val| f(world, val))
            },
            _marker: PhantomData,
        }
    }

    /// None becomes Err(err). std: `Option::ok_or`
    ///
    /// `Clone` required because the pipeline may run many times —
    /// the error value is cloned on each `None` invocation.
    pub fn ok_or<E: Clone>(
        self,
        err: E,
    ) -> PipelineBuilder<In, Result<T, E>, impl FnMut(&mut World, In) -> Result<T, E>> {
        let mut chain = self.chain;
        PipelineBuilder {
            chain: move |world: &mut World, input: In| {
                chain(world, input).ok_or_else(|| err.clone())
            },
            _marker: PhantomData,
        }
    }

    /// None becomes Err(f()). std: `Option::ok_or_else`
    pub fn ok_or_else<E>(
        self,
        mut f: impl FnMut(&mut World) -> E,
    ) -> PipelineBuilder<In, Result<T, E>, impl FnMut(&mut World, In) -> Result<T, E>> {
        let mut chain = self.chain;
        PipelineBuilder {
            chain: move |world: &mut World, input: In| chain(world, input).ok_or_else(|| f(world)),
            _marker: PhantomData,
        }
    }

    /// Exit Option — None becomes the default value.
    ///
    /// `Clone` required because the pipeline may run many times —
    /// the default is cloned on each `None` invocation (unlike
    /// std's `unwrap_or` which consumes the value once).
    pub fn unwrap_or(self, default: T) -> PipelineBuilder<In, T, impl FnMut(&mut World, In) -> T>
    where
        T: Clone,
    {
        let mut chain = self.chain;
        PipelineBuilder {
            chain: move |world: &mut World, input: In| {
                chain(world, input).unwrap_or_else(|| default.clone())
            },
            _marker: PhantomData,
        }
    }

    /// Exit Option — None becomes `f()`.
    pub fn unwrap_or_else(
        self,
        mut f: impl FnMut(&mut World) -> T,
    ) -> PipelineBuilder<In, T, impl FnMut(&mut World, In) -> T> {
        let mut chain = self.chain;
        PipelineBuilder {
            chain: move |world: &mut World, input: In| {
                chain(world, input).unwrap_or_else(|| f(world))
            },
            _marker: PhantomData,
        }
    }
}

// =============================================================================
// Result helpers — PipelineBuilder<In, Result<T, E>, Chain>
// =============================================================================

impl<In, T, E, Chain> PipelineBuilder<In, Result<T, E>, Chain>
where
    Chain: FnMut(&mut World, In) -> Result<T, E>,
{
    // -- IntoStep-based (hot path) -------------------------------------------

    /// Transform the Ok value. Step not called on Err.
    pub fn map<U, Params, S: IntoStep<T, U, Params>>(
        self,
        f: S,
        registry: &Registry,
    ) -> PipelineBuilder<
        In,
        Result<U, E>,
        impl FnMut(&mut World, In) -> Result<U, E> + use<In, T, E, U, Params, Chain, S>,
    > {
        let mut chain = self.chain;
        let mut resolved = f.into_step(registry);
        PipelineBuilder {
            chain: move |world: &mut World, input: In| {
                chain(world, input).map(|val| resolved.call(world, val))
            },
            _marker: PhantomData,
        }
    }

    /// Short-circuits on Err. std: `Result::and_then`
    pub fn and_then<U, Params, S: IntoStep<T, Result<U, E>, Params>>(
        self,
        f: S,
        registry: &Registry,
    ) -> PipelineBuilder<
        In,
        Result<U, E>,
        impl FnMut(&mut World, In) -> Result<U, E> + use<In, T, E, U, Params, Chain, S>,
    > {
        let mut chain = self.chain;
        let mut resolved = f.into_step(registry);
        PipelineBuilder {
            chain: move |world: &mut World, input: In| {
                chain(world, input).and_then(|val| resolved.call(world, val))
            },
            _marker: PhantomData,
        }
    }

    /// Handle error and transition to Option.
    ///
    /// `Ok(val)` becomes `Some(val)` — handler not called.
    /// `Err(err)` calls the handler, then produces `None`.
    pub fn catch<Params, S: IntoStep<E, (), Params>>(
        self,
        f: S,
        registry: &Registry,
    ) -> PipelineBuilder<
        In,
        Option<T>,
        impl FnMut(&mut World, In) -> Option<T> + use<In, T, E, Params, Chain, S>,
    > {
        let mut chain = self.chain;
        let mut resolved = f.into_step(registry);
        PipelineBuilder {
            chain: move |world: &mut World, input: In| match chain(world, input) {
                Ok(val) => Some(val),
                Err(err) => {
                    resolved.call(world, err);
                    None
                }
            },
            _marker: PhantomData,
        }
    }

    // -- Closure-based (cold path, &mut World) --------------------------------

    /// Transform the error. std: `Result::map_err`
    pub fn map_err<E2>(
        self,
        mut f: impl FnMut(&mut World, E) -> E2,
    ) -> PipelineBuilder<In, Result<T, E2>, impl FnMut(&mut World, In) -> Result<T, E2>> {
        let mut chain = self.chain;
        PipelineBuilder {
            chain: move |world: &mut World, input: In| {
                chain(world, input).map_err(|err| f(world, err))
            },
            _marker: PhantomData,
        }
    }

    /// Recover from Err. std: `Result::or_else`
    pub fn or_else<E2>(
        self,
        mut f: impl FnMut(&mut World, E) -> Result<T, E2>,
    ) -> PipelineBuilder<In, Result<T, E2>, impl FnMut(&mut World, In) -> Result<T, E2>> {
        let mut chain = self.chain;
        PipelineBuilder {
            chain: move |world: &mut World, input: In| {
                chain(world, input).or_else(|err| f(world, err))
            },
            _marker: PhantomData,
        }
    }

    /// Side effect on Ok. std: `Result::inspect`
    pub fn inspect(
        self,
        mut f: impl FnMut(&mut World, &T),
    ) -> PipelineBuilder<In, Result<T, E>, impl FnMut(&mut World, In) -> Result<T, E>> {
        let mut chain = self.chain;
        PipelineBuilder {
            chain: move |world: &mut World, input: In| {
                chain(world, input).inspect(|val| f(world, val))
            },
            _marker: PhantomData,
        }
    }

    /// Side effect on Err. std: `Result::inspect_err`
    pub fn inspect_err(
        self,
        mut f: impl FnMut(&mut World, &E),
    ) -> PipelineBuilder<In, Result<T, E>, impl FnMut(&mut World, In) -> Result<T, E>> {
        let mut chain = self.chain;
        PipelineBuilder {
            chain: move |world: &mut World, input: In| {
                chain(world, input).inspect_err(|err| f(world, err))
            },
            _marker: PhantomData,
        }
    }

    /// Discard error, enter Option land. std: `Result::ok`
    pub fn ok(self) -> PipelineBuilder<In, Option<T>, impl FnMut(&mut World, In) -> Option<T>> {
        let mut chain = self.chain;
        PipelineBuilder {
            chain: move |world: &mut World, input: In| chain(world, input).ok(),
            _marker: PhantomData,
        }
    }

    /// Exit Result — Err becomes the default value.
    ///
    /// `Clone` required because the pipeline may run many times —
    /// the default is cloned on each `Err` invocation (unlike
    /// std's `unwrap_or` which consumes the value once).
    pub fn unwrap_or(self, default: T) -> PipelineBuilder<In, T, impl FnMut(&mut World, In) -> T>
    where
        T: Clone,
    {
        let mut chain = self.chain;
        PipelineBuilder {
            chain: move |world: &mut World, input: In| {
                chain(world, input).unwrap_or_else(|_| default.clone())
            },
            _marker: PhantomData,
        }
    }

    /// Exit Result — Err becomes `f(err)`.
    pub fn unwrap_or_else(
        self,
        mut f: impl FnMut(&mut World, E) -> T,
    ) -> PipelineBuilder<In, T, impl FnMut(&mut World, In) -> T> {
        let mut chain = self.chain;
        PipelineBuilder {
            chain: move |world: &mut World, input: In| match chain(world, input) {
                Ok(val) => val,
                Err(err) => f(world, err),
            },
            _marker: PhantomData,
        }
    }
}

// =============================================================================
// PipelineOutput — marker trait for build()
// =============================================================================

/// Marker trait restricting [`PipelineBuilder::build`] to pipelines
/// that produce `()`.
///
/// If your pipeline produces a value, add a final `.then()` that
/// writes it somewhere (e.g. `ResMut<T>`).
#[diagnostic::on_unimplemented(
    message = "`build()` requires the step pipeline output to be `()`",
    label = "this pipeline produces `{Self}`, not `()`",
    note = "add a final `.then()` that consumes the output"
)]
pub trait PipelineOutput {}
impl PipelineOutput for () {}
impl PipelineOutput for Option<()> {}

// =============================================================================
// build — when Out = ()
// =============================================================================

impl<In: 'static, Chain> PipelineBuilder<In, (), Chain>
where
    Chain: FnMut(&mut World, In) + 'static,
{
    /// Finalize the pipeline into a [`Pipeline`].
    ///
    /// The returned pipeline is a concrete, monomorphized type — no boxing,
    /// no virtual dispatch. Call `.run()` directly for zero-cost execution,
    /// or wrap in `Box<dyn Handler<In>>` when type erasure is needed.
    ///
    /// Only available when the pipeline ends with `()`. If your chain
    /// produces a value, add a final `.then()` that consumes the output.
    pub fn build(self) -> Pipeline<In, Chain> {
        Pipeline {
            chain: self.chain,
            _marker: PhantomData,
        }
    }
}

// =============================================================================
// build_batch — when Out: PipelineOutput (() or Option<()>)
// =============================================================================

impl<In: 'static, Out: PipelineOutput, Chain> PipelineBuilder<In, Out, Chain>
where
    Chain: FnMut(&mut World, In) -> Out + 'static,
{
    /// Finalize into a [`BatchPipeline`] with a pre-allocated input buffer.
    ///
    /// Same pipeline chain as [`build`](PipelineBuilder::build), but the
    /// pipeline owns an input buffer that drivers fill between dispatch
    /// cycles. Each call to [`BatchPipeline::run`] drains the buffer,
    /// running every item through the chain independently.
    ///
    /// Available when the pipeline ends with `()` or `Option<()>` (e.g.
    /// after `.catch()` or `.filter()`). Pipelines producing values need
    /// a final `.then()` that consumes the output.
    ///
    /// `capacity` is the initial allocation — the buffer can grow if needed,
    /// but sizing it for the expected batch size avoids reallocation.
    pub fn build_batch(self, capacity: usize) -> BatchPipeline<In, Chain> {
        BatchPipeline {
            input: Vec::with_capacity(capacity),
            chain: self.chain,
        }
    }
}

// =============================================================================
// Pipeline<In, F> — built pipeline
// =============================================================================

/// Built step pipeline implementing [`Handler<In>`](crate::Handler).
///
/// Created by [`PipelineBuilder::build`]. The entire pipeline chain is
/// monomorphized at compile time — no boxing, no virtual dispatch.
/// Call `.run()` directly for zero-cost execution, or wrap in
/// `Box<dyn Handler<In>>` when you need type erasure (single box).
///
/// Implements [`Handler<In>`](crate::Handler), so it can be stored in
/// driver handler collections alongside [`Callback`](crate::Callback)
/// and [`HandlerFn`](crate::HandlerFn). For batch processing, see
/// [`BatchPipeline`].
pub struct Pipeline<In, F> {
    chain: F,
    _marker: PhantomData<fn(In)>,
}

impl<In: 'static, F: FnMut(&mut World, In) + Send + 'static> crate::Handler<In>
    for Pipeline<In, F>
{
    fn run(&mut self, world: &mut World, event: In) {
        (self.chain)(world, event);
    }
}

// =============================================================================
// BatchPipeline<In, F> — pipeline with owned input buffer
// =============================================================================

/// Batch pipeline that owns a pre-allocated input buffer.
///
/// Created by [`PipelineBuilder::build_batch`]. Each item flows through
/// the full pipeline chain independently — the same per-item `Option`
/// and `Result` flow control as [`Pipeline`]. Errors are handled inline
/// (via `.catch()`, `.unwrap_or()`, etc.) and the batch continues to
/// the next item. No intermediate buffers between steps.
///
/// # Examples
///
/// ```
/// use nexus_rt::{WorldBuilder, Res, ResMut, PipelineStart};
///
/// let mut wb = WorldBuilder::new();
/// wb.register::<u64>(0);
/// let mut world = wb.build();
///
/// fn accumulate(mut sum: ResMut<u64>, x: u32) {
///     *sum += x as u64;
/// }
///
/// let r = world.registry_mut();
/// let mut batch = PipelineStart::<u32>::new()
///     .then(accumulate, r)
///     .build_batch(64);
///
/// batch.input_mut().extend_from_slice(&[1, 2, 3, 4, 5]);
/// batch.run(&mut world);
///
/// assert_eq!(*world.resource::<u64>(), 15);
/// assert!(batch.input().is_empty());
/// ```
pub struct BatchPipeline<In, F> {
    input: Vec<In>,
    chain: F,
}

impl<In, Out: PipelineOutput, F: FnMut(&mut World, In) -> Out> BatchPipeline<In, F> {
    /// Mutable access to the input buffer. Drivers fill this between
    /// dispatch cycles.
    pub fn input_mut(&mut self) -> &mut Vec<In> {
        &mut self.input
    }

    /// Read-only access to the input buffer.
    pub fn input(&self) -> &[In] {
        &self.input
    }

    /// Drain the input buffer, running each item through the pipeline.
    ///
    /// Each item gets independent `Option`/`Result` flow control — an
    /// error on one item does not affect subsequent items. After `run()`,
    /// the input buffer is empty but retains its allocation.
    pub fn run(&mut self, world: &mut World) {
        for item in self.input.drain(..) {
            let _ = (self.chain)(world, item);
        }
    }
}

// =============================================================================
// resolve_step — pre-resolve a step for manual dispatch (owned input)
// =============================================================================

/// Resolve a step for use in manual dispatch (e.g. inside a
/// [`.switch()`](PipelineBuilder::switch) closure).
///
/// Returns a closure with pre-resolved [`Param`] state — the same
/// build-time resolution that `.then()` performs, but as a standalone
/// value the caller can invoke from any context.
///
/// This is the pipeline (owned-input) counterpart of
/// [`dag::resolve_arm`](crate::dag::resolve_arm) (reference-input).
///
/// # Examples
///
/// ```ignore
/// let mut arm0 = resolve_step(handle_new, reg);
/// let mut arm1 = resolve_step(handle_cancel, reg);
///
/// pipeline.switch(move |world, order: Order| match order.kind {
///     OrderKind::New    => arm0(world, order),
///     OrderKind::Cancel => arm1(world, order),
/// })
/// ```
pub fn resolve_step<In, Out, Params, S>(
    f: S,
    registry: &Registry,
) -> impl FnMut(&mut World, In) -> Out + use<In, Out, Params, S>
where
    In: 'static,
    Out: 'static,
    S: IntoStep<In, Out, Params>,
{
    let mut resolved = f.into_step(registry);
    move |world: &mut World, input: In| resolved.call(world, input)
}

// =============================================================================
// Tests
// =============================================================================

#[cfg(test)]
mod tests {
    use super::*;
    use crate::{Handler, IntoHandler, Local, Res, ResMut, WorldBuilder, fan_out};

    // =========================================================================
    // Core dispatch
    // =========================================================================

    #[test]
    fn step_pure_transform() {
        let mut world = WorldBuilder::new().build();
        let r = world.registry_mut();
        let mut p = PipelineStart::<u32>::new().then(|x: u32| x as u64 * 2, r);
        assert_eq!(p.run(&mut world, 5), 10u64);
    }

    #[test]
    fn step_one_res() {
        let mut wb = WorldBuilder::new();
        wb.register::<u64>(10);
        let mut world = wb.build();

        fn multiply(factor: Res<u64>, x: u32) -> u64 {
            *factor * x as u64
        }

        let r = world.registry_mut();
        let mut p = PipelineStart::<u32>::new().then(multiply, r);
        assert_eq!(p.run(&mut world, 5), 50);
    }

    #[test]
    fn step_one_res_mut() {
        let mut wb = WorldBuilder::new();
        wb.register::<u64>(0);
        let mut world = wb.build();

        fn accumulate(mut total: ResMut<u64>, x: u32) {
            *total += x as u64;
        }

        let r = world.registry_mut();
        let mut p = PipelineStart::<u32>::new().then(accumulate, r);
        p.run(&mut world, 10);
        p.run(&mut world, 5);
        assert_eq!(*world.resource::<u64>(), 15);
    }

    #[test]
    fn step_two_params() {
        let mut wb = WorldBuilder::new();
        wb.register::<u64>(10);
        wb.register::<bool>(true);
        let mut world = wb.build();

        fn conditional(factor: Res<u64>, flag: Res<bool>, x: u32) -> u64 {
            if *flag { *factor * x as u64 } else { 0 }
        }

        let r = world.registry_mut();
        let mut p = PipelineStart::<u32>::new().then(conditional, r);
        assert_eq!(p.run(&mut world, 5), 50);
    }

    #[test]
    fn step_chain_two() {
        let mut wb = WorldBuilder::new();
        wb.register::<u64>(2);
        let mut world = wb.build();

        fn double(factor: Res<u64>, x: u32) -> u64 {
            *factor * x as u64
        }

        let r = world.registry_mut();
        let mut p = PipelineStart::<u32>::new()
            .then(double, r)
            .then(|val: u64| val + 1, r);
        assert_eq!(p.run(&mut world, 5), 11); // 2*5 + 1
    }

    // =========================================================================
    // Option combinators
    // =========================================================================

    #[test]
    fn option_map_on_some() {
        let mut wb = WorldBuilder::new();
        wb.register::<u64>(10);
        let mut world = wb.build();

        fn add_factor(factor: Res<u64>, x: u32) -> u64 {
            *factor + x as u64
        }

        let r = world.registry_mut();
        let mut p = PipelineStart::<u32>::new()
            .then(|x: u32| -> Option<u32> { Some(x) }, r)
            .map(add_factor, r);
        assert_eq!(p.run(&mut world, 5), Some(15));
    }

    #[test]
    fn option_map_skips_none() {
        let mut wb = WorldBuilder::new();
        wb.register::<bool>(false);
        let mut world = wb.build();

        fn mark(mut flag: ResMut<bool>, _x: u32) -> u32 {
            *flag = true;
            0
        }

        let r = world.registry_mut();
        let mut p = PipelineStart::<u32>::new()
            .then(|_x: u32| -> Option<u32> { None }, r)
            .map(mark, r);
        assert_eq!(p.run(&mut world, 5), None);
        assert!(!*world.resource::<bool>());
    }

    #[test]
    fn option_and_then_chains() {
        let mut wb = WorldBuilder::new();
        wb.register::<u64>(10);
        let mut world = wb.build();

        fn check(min: Res<u64>, x: u32) -> Option<u64> {
            let val = x as u64;
            if val > *min { Some(val) } else { None }
        }

        let r = world.registry_mut();
        let mut p = PipelineStart::<u32>::new()
            .then(|x: u32| Some(x), r)
            .and_then(check, r);
        assert_eq!(p.run(&mut world, 20), Some(20));
    }

    #[test]
    fn option_and_then_short_circuits() {
        let mut wb = WorldBuilder::new();
        wb.register::<u64>(10);
        let mut world = wb.build();

        fn check(min: Res<u64>, x: u32) -> Option<u64> {
            let val = x as u64;
            if val > *min { Some(val) } else { None }
        }

        let r = world.registry_mut();
        let mut p = PipelineStart::<u32>::new()
            .then(|x: u32| Some(x), r)
            .and_then(check, r);
        assert_eq!(p.run(&mut world, 5), None);
    }

    #[test]
    fn option_on_none_fires() {
        let mut wb = WorldBuilder::new();
        wb.register::<bool>(false);
        let mut world = wb.build();

        let r = world.registry_mut();
        let mut p = PipelineStart::<u32>::new()
            .then(|_x: u32| -> Option<u32> { None }, r)
            .on_none(|w| {
                *w.resource_mut::<bool>() = true;
            });
        p.run(&mut world, 0);
        assert!(*world.resource::<bool>());
    }

    #[test]
    fn option_filter_keeps() {
        let mut world = WorldBuilder::new().build();
        let r = world.registry_mut();
        let mut p = PipelineStart::<u32>::new()
            .then(|x: u32| Some(x), r)
            .filter(|_w, x| *x > 3);
        assert_eq!(p.run(&mut world, 5), Some(5));
    }

    #[test]
    fn option_filter_drops() {
        let mut world = WorldBuilder::new().build();
        let r = world.registry_mut();
        let mut p = PipelineStart::<u32>::new()
            .then(|x: u32| Some(x), r)
            .filter(|_w, x| *x > 10);
        assert_eq!(p.run(&mut world, 5), None);
    }

    // =========================================================================
    // Result combinators
    // =========================================================================

    #[test]
    fn result_map_on_ok() {
        let mut wb = WorldBuilder::new();
        wb.register::<u64>(10);
        let mut world = wb.build();

        fn add_factor(factor: Res<u64>, x: u32) -> u64 {
            *factor + x as u64
        }

        let r = world.registry_mut();
        let mut p = PipelineStart::<u32>::new()
            .then(|x: u32| -> Result<u32, String> { Ok(x) }, r)
            .map(add_factor, r);
        assert_eq!(p.run(&mut world, 5), Ok(15));
    }

    #[test]
    fn result_map_skips_err() {
        let mut wb = WorldBuilder::new();
        wb.register::<bool>(false);
        let mut world = wb.build();

        fn mark(mut flag: ResMut<bool>, _x: u32) -> u32 {
            *flag = true;
            0
        }

        let r = world.registry_mut();
        let mut p = PipelineStart::<u32>::new()
            .then(|_x: u32| -> Result<u32, String> { Err("fail".into()) }, r)
            .map(mark, r);
        assert!(p.run(&mut world, 5).is_err());
        assert!(!*world.resource::<bool>());
    }

    #[test]
    fn result_catch_handles_error() {
        let mut wb = WorldBuilder::new();
        wb.register::<String>(String::new());
        let mut world = wb.build();

        fn log_error(mut log: ResMut<String>, err: String) {
            *log = err;
        }

        let r = world.registry_mut();
        let mut p = PipelineStart::<u32>::new()
            .then(|_x: u32| -> Result<u32, String> { Err("caught".into()) }, r)
            .catch(log_error, r);
        assert_eq!(p.run(&mut world, 0), None);
        assert_eq!(world.resource::<String>().as_str(), "caught");
    }

    #[test]
    fn result_catch_passes_ok() {
        let mut wb = WorldBuilder::new();
        wb.register::<String>(String::new());
        let mut world = wb.build();

        fn log_error(mut log: ResMut<String>, err: String) {
            *log = err;
        }

        let r = world.registry_mut();
        let mut p = PipelineStart::<u32>::new()
            .then(|x: u32| -> Result<u32, String> { Ok(x) }, r)
            .catch(log_error, r);
        assert_eq!(p.run(&mut world, 5), Some(5));
        assert!(world.resource::<String>().is_empty());
    }

    // =========================================================================
    // Build + Handler
    // =========================================================================

    #[test]
    fn build_produces_handler() {
        let mut wb = WorldBuilder::new();
        wb.register::<u64>(0);
        let mut world = wb.build();

        fn accumulate(mut total: ResMut<u64>, x: u32) {
            *total += x as u64;
        }

        let r = world.registry_mut();
        let mut pipeline = PipelineStart::<u32>::new().then(accumulate, r).build();

        pipeline.run(&mut world, 10);
        pipeline.run(&mut world, 5);
        assert_eq!(*world.resource::<u64>(), 15);
    }

    #[test]
    fn run_returns_output() {
        let mut wb = WorldBuilder::new();
        wb.register::<u64>(3);
        let mut world = wb.build();

        fn multiply(factor: Res<u64>, x: u32) -> u64 {
            *factor * x as u64
        }

        let r = world.registry_mut();
        let mut p = PipelineStart::<u32>::new().then(multiply, r);
        let result: u64 = p.run(&mut world, 7);
        assert_eq!(result, 21);
    }

    // =========================================================================
    // Safety
    // =========================================================================

    #[test]
    #[should_panic(expected = "not registered")]
    fn panics_on_missing_resource() {
        let mut world = WorldBuilder::new().build();

        fn needs_u64(_val: Res<u64>, _x: u32) -> u32 {
            0
        }

        let r = world.registry_mut();
        let _p = PipelineStart::<u32>::new().then(needs_u64, r);
    }

    // =========================================================================
    // Access conflict detection
    // =========================================================================

    #[test]
    #[should_panic(expected = "conflicting access")]
    fn step_duplicate_access_panics() {
        let mut wb = WorldBuilder::new();
        wb.register::<u64>(0);
        let mut world = wb.build();

        fn bad(a: Res<u64>, b: ResMut<u64>, _x: u32) -> u32 {
            let _ = (*a, &*b);
            0
        }

        let r = world.registry_mut();
        let _p = PipelineStart::<u32>::new().then(bad, r);
    }

    // =========================================================================
    // Integration
    // =========================================================================

    #[test]
    fn local_in_step() {
        let mut wb = WorldBuilder::new();
        wb.register::<u64>(0);
        let mut world = wb.build();

        fn count(mut count: Local<u64>, mut total: ResMut<u64>, _x: u32) {
            *count += 1;
            *total = *count;
        }

        let r = world.registry_mut();
        let mut p = PipelineStart::<u32>::new().then(count, r);
        p.run(&mut world, 0);
        p.run(&mut world, 0);
        p.run(&mut world, 0);
        assert_eq!(*world.resource::<u64>(), 3);
    }

    // =========================================================================
    // Option combinators (extended)
    // =========================================================================

    #[test]
    fn option_unwrap_or_some() {
        let mut world = WorldBuilder::new().build();
        let r = world.registry_mut();
        let mut p = PipelineStart::<u32>::new()
            .then(|x: u32| -> Option<u32> { Some(x) }, r)
            .unwrap_or(99);
        assert_eq!(p.run(&mut world, 5), 5);
    }

    #[test]
    fn option_unwrap_or_none() {
        let mut world = WorldBuilder::new().build();
        let r = world.registry_mut();
        let mut p = PipelineStart::<u32>::new()
            .then(|_x: u32| -> Option<u32> { None }, r)
            .unwrap_or(99);
        assert_eq!(p.run(&mut world, 5), 99);
    }

    #[test]
    fn option_unwrap_or_else() {
        let mut world = WorldBuilder::new().build();
        let r = world.registry_mut();
        let mut p = PipelineStart::<u32>::new()
            .then(|_x: u32| -> Option<u32> { None }, r)
            .unwrap_or_else(|_w| 42);
        assert_eq!(p.run(&mut world, 0), 42);
    }

    #[test]
    fn option_ok_or() {
        let mut world = WorldBuilder::new().build();
        let r = world.registry_mut();
        let mut p = PipelineStart::<u32>::new()
            .then(|_x: u32| -> Option<u32> { None }, r)
            .ok_or("missing");
        assert_eq!(p.run(&mut world, 0), Err("missing"));
    }

    #[test]
    fn option_ok_or_some() {
        let mut world = WorldBuilder::new().build();
        let r = world.registry_mut();
        let mut p = PipelineStart::<u32>::new()
            .then(|x: u32| -> Option<u32> { Some(x) }, r)
            .ok_or("missing");
        assert_eq!(p.run(&mut world, 7), Ok(7));
    }

    #[test]
    fn option_ok_or_else() {
        let mut world = WorldBuilder::new().build();
        let r = world.registry_mut();
        let mut p = PipelineStart::<u32>::new()
            .then(|_x: u32| -> Option<u32> { None }, r)
            .ok_or_else(|_w| "computed");
        assert_eq!(p.run(&mut world, 0), Err("computed"));
    }

    #[test]
    fn option_inspect_passes_through() {
        let mut wb = WorldBuilder::new();
        wb.register::<u64>(0);
        let mut world = wb.build();
        let r = world.registry_mut();
        let mut p = PipelineStart::<u32>::new()
            .then(|x: u32| -> Option<u32> { Some(x) }, r)
            .inspect(|_w, _val| {});
        // inspect should pass through the value unchanged.
        assert_eq!(p.run(&mut world, 10), Some(10));
    }

    // =========================================================================
    // Result combinators (extended)
    // =========================================================================

    #[test]
    fn result_map_err() {
        let mut world = WorldBuilder::new().build();
        let r = world.registry_mut();
        let mut p = PipelineStart::<u32>::new()
            .then(|_x: u32| -> Result<u32, i32> { Err(-1) }, r)
            .map_err(|_w, e| e.to_string());
        assert_eq!(p.run(&mut world, 0), Err("-1".to_string()));
    }

    #[test]
    fn result_map_err_ok_passthrough() {
        let mut world = WorldBuilder::new().build();
        let r = world.registry_mut();
        let mut p = PipelineStart::<u32>::new()
            .then(|x: u32| -> Result<u32, i32> { Ok(x) }, r)
            .map_err(|_w, e| e.to_string());
        assert_eq!(p.run(&mut world, 5), Ok(5));
    }

    #[test]
    fn result_or_else() {
        let mut world = WorldBuilder::new().build();
        let r = world.registry_mut();
        let mut p = PipelineStart::<u32>::new()
            .then(|_x: u32| -> Result<u32, &str> { Err("fail") }, r)
            .or_else(|_w, _e| Ok::<u32, &str>(42));
        assert_eq!(p.run(&mut world, 0), Ok(42));
    }

    #[test]
    fn result_inspect_passes_through() {
        let mut world = WorldBuilder::new().build();
        let r = world.registry_mut();
        let mut p = PipelineStart::<u32>::new()
            .then(|x: u32| -> Result<u32, &str> { Ok(x) }, r)
            .inspect(|_w, _val| {});
        // inspect should pass through Ok unchanged.
        assert_eq!(p.run(&mut world, 7), Ok(7));
    }

    #[test]
    fn result_inspect_err_passes_through() {
        let mut world = WorldBuilder::new().build();
        let r = world.registry_mut();
        let mut p = PipelineStart::<u32>::new()
            .then(|_x: u32| -> Result<u32, &str> { Err("bad") }, r)
            .inspect_err(|_w, _e| {});
        // inspect_err should pass through Err unchanged.
        assert_eq!(p.run(&mut world, 0), Err("bad"));
    }

    #[test]
    fn result_ok_converts() {
        let mut world = WorldBuilder::new().build();
        let r = world.registry_mut();
        let mut p = PipelineStart::<u32>::new()
            .then(|x: u32| -> Result<u32, &str> { Ok(x) }, r)
            .ok();
        assert_eq!(p.run(&mut world, 5), Some(5));
    }

    #[test]
    fn result_ok_drops_err() {
        let mut world = WorldBuilder::new().build();
        let r = world.registry_mut();
        let mut p = PipelineStart::<u32>::new()
            .then(|_x: u32| -> Result<u32, &str> { Err("gone") }, r)
            .ok();
        assert_eq!(p.run(&mut world, 0), None);
    }

    #[test]
    fn result_unwrap_or() {
        let mut world = WorldBuilder::new().build();
        let r = world.registry_mut();
        let mut p = PipelineStart::<u32>::new()
            .then(|_x: u32| -> Result<u32, &str> { Err("x") }, r)
            .unwrap_or(99);
        assert_eq!(p.run(&mut world, 0), 99);
    }

    #[test]
    fn result_unwrap_or_else() {
        let mut world = WorldBuilder::new().build();
        let r = world.registry_mut();
        let mut p = PipelineStart::<u32>::new()
            .then(|_x: u32| -> Result<u32, i32> { Err(-5) }, r)
            .unwrap_or_else(|_w, e| e.unsigned_abs());
        assert_eq!(p.run(&mut world, 0), 5);
    }

    // =========================================================================
    // Batch pipeline
    // =========================================================================

    #[test]
    fn batch_accumulates() {
        let mut wb = WorldBuilder::new();
        wb.register::<u64>(0);
        let mut world = wb.build();

        fn accumulate(mut sum: ResMut<u64>, x: u32) {
            *sum += x as u64;
        }

        let r = world.registry_mut();
        let mut batch = PipelineStart::<u32>::new()
            .then(accumulate, r)
            .build_batch(16);

        batch.input_mut().extend_from_slice(&[1, 2, 3, 4, 5]);
        batch.run(&mut world);

        assert_eq!(*world.resource::<u64>(), 15);
        assert!(batch.input().is_empty());
    }

    #[test]
    fn batch_retains_allocation() {
        let mut world = WorldBuilder::new().build();
        let r = world.registry_mut();
        let mut batch = PipelineStart::<u32>::new()
            .then(|_x: u32| {}, r)
            .build_batch(64);

        batch.input_mut().extend_from_slice(&[1, 2, 3]);
        batch.run(&mut world);

        assert!(batch.input().is_empty());
        assert!(batch.input_mut().capacity() >= 64);
    }

    #[test]
    fn batch_empty_is_noop() {
        let mut wb = WorldBuilder::new();
        wb.register::<u64>(0);
        let mut world = wb.build();

        fn accumulate(mut sum: ResMut<u64>, x: u32) {
            *sum += x as u64;
        }

        let r = world.registry_mut();
        let mut batch = PipelineStart::<u32>::new()
            .then(accumulate, r)
            .build_batch(16);

        batch.run(&mut world);
        assert_eq!(*world.resource::<u64>(), 0);
    }

    #[test]
    fn batch_catch_continues_on_error() {
        let mut wb = WorldBuilder::new();
        wb.register::<u64>(0);
        wb.register::<u32>(0);
        let mut world = wb.build();

        fn validate(x: u32) -> Result<u32, &'static str> {
            if x > 0 { Ok(x) } else { Err("zero") }
        }

        fn count_errors(mut errs: ResMut<u32>, _err: &'static str) {
            *errs += 1;
        }

        fn accumulate(mut sum: ResMut<u64>, x: u32) {
            *sum += x as u64;
        }

        let r = world.registry_mut();
        let mut batch = PipelineStart::<u32>::new()
            .then(validate, r)
            .catch(count_errors, r)
            .map(accumulate, r)
            .build_batch(16);

        // Items: 1, 0 (error), 2, 0 (error), 3
        batch.input_mut().extend_from_slice(&[1, 0, 2, 0, 3]);
        batch.run(&mut world);

        assert_eq!(*world.resource::<u64>(), 6); // 1 + 2 + 3
        assert_eq!(*world.resource::<u32>(), 2); // 2 errors
    }

    #[test]
    fn batch_filter_skips_items() {
        let mut wb = WorldBuilder::new();
        wb.register::<u64>(0);
        let mut world = wb.build();

        fn accumulate(mut sum: ResMut<u64>, x: u32) {
            *sum += x as u64;
        }

        let r = world.registry_mut();
        let mut batch = PipelineStart::<u32>::new()
            .then(
                |x: u32| -> Option<u32> { if x > 2 { Some(x) } else { None } },
                r,
            )
            .map(accumulate, r)
            .build_batch(16);

        batch.input_mut().extend_from_slice(&[1, 2, 3, 4, 5]);
        batch.run(&mut world);

        assert_eq!(*world.resource::<u64>(), 12); // 3 + 4 + 5
    }

    #[test]
    fn batch_multiple_runs_accumulate() {
        let mut wb = WorldBuilder::new();
        wb.register::<u64>(0);
        let mut world = wb.build();

        fn accumulate(mut sum: ResMut<u64>, x: u32) {
            *sum += x as u64;
        }

        let r = world.registry_mut();
        let mut batch = PipelineStart::<u32>::new()
            .then(accumulate, r)
            .build_batch(16);

        batch.input_mut().extend_from_slice(&[1, 2, 3]);
        batch.run(&mut world);
        assert_eq!(*world.resource::<u64>(), 6);

        batch.input_mut().extend_from_slice(&[4, 5]);
        batch.run(&mut world);
        assert_eq!(*world.resource::<u64>(), 15);
    }

    #[test]
    fn batch_with_world_access() {
        let mut wb = WorldBuilder::new();
        wb.register::<u64>(10); // multiplier
        wb.register::<Vec<u64>>(Vec::new());
        let mut world = wb.build();

        fn multiply_and_collect(factor: Res<u64>, mut out: ResMut<Vec<u64>>, x: u32) {
            out.push(x as u64 * *factor);
        }

        let r = world.registry_mut();
        let mut batch = PipelineStart::<u32>::new()
            .then(multiply_and_collect, r)
            .build_batch(16);

        batch.input_mut().extend_from_slice(&[1, 2, 3]);
        batch.run(&mut world);

        assert_eq!(world.resource::<Vec<u64>>().as_slice(), &[10, 20, 30]);
    }

    // =========================================================================
    // Cloned combinator
    // =========================================================================

    // Named functions for proper lifetime elision (&'a u32 → &'a u32).
    // Closures get two independent lifetimes and fail to compile.
    fn ref_identity(x: &u32) -> &u32 {
        x
    }
    fn ref_wrap_some(x: &u32) -> Option<&u32> {
        Some(x)
    }
    fn ref_wrap_none(_x: &u32) -> Option<&u32> {
        None
    }
    fn ref_wrap_ok(x: &u32) -> Result<&u32, String> {
        Ok(x)
    }
    fn ref_wrap_err(_x: &u32) -> Result<&u32, String> {
        Err("fail".into())
    }

    #[test]
    fn cloned_bare() {
        let mut world = WorldBuilder::new().build();
        // val before p — val must outlive the pipeline's In = &u32
        let val = 42u32;
        let r = world.registry_mut();
        let mut p = PipelineStart::<&u32>::new().then(ref_identity, r).cloned();
        assert_eq!(p.run(&mut world, &val), 42u32);
    }

    #[test]
    fn cloned_option_some() {
        let mut world = WorldBuilder::new().build();
        let val = 42u32;
        let r = world.registry_mut();
        let mut p = PipelineStart::<&u32>::new().then(ref_wrap_some, r).cloned();
        assert_eq!(p.run(&mut world, &val), Some(42u32));
    }

    #[test]
    fn cloned_option_none() {
        let mut world = WorldBuilder::new().build();
        let val = 42u32;
        let r = world.registry_mut();
        let mut p = PipelineStart::<&u32>::new().then(ref_wrap_none, r).cloned();
        assert_eq!(p.run(&mut world, &val), None);
    }

    #[test]
    fn cloned_result_ok() {
        let mut world = WorldBuilder::new().build();
        let val = 42u32;
        let r = world.registry_mut();
        let mut p = PipelineStart::<&u32>::new().then(ref_wrap_ok, r).cloned();
        assert_eq!(p.run(&mut world, &val), Ok(42u32));
    }

    #[test]
    fn cloned_result_err() {
        let mut world = WorldBuilder::new().build();
        let val = 42u32;
        let r = world.registry_mut();
        let mut p = PipelineStart::<&u32>::new().then(ref_wrap_err, r).cloned();
        assert_eq!(p.run(&mut world, &val), Err("fail".into()));
    }

    // =========================================================================
    // Dispatch combinator
    // =========================================================================

    #[test]
    fn dispatch_to_handler() {
        let mut wb = WorldBuilder::new();
        wb.register::<u64>(0);
        let mut world = wb.build();

        fn store(mut out: ResMut<u64>, val: u32) {
            *out = val as u64;
        }

        let r = world.registry_mut();
        let handler = PipelineStart::<u32>::new().then(store, r).build();

        let mut p = PipelineStart::<u32>::new()
            .then(|x: u32| x * 2, r)
            .dispatch(handler)
            .build();

        p.run(&mut world, 5);
        assert_eq!(*world.resource::<u64>(), 10);
    }

    #[test]
    fn dispatch_to_fanout() {
        let mut wb = WorldBuilder::new();
        wb.register::<u64>(0);
        wb.register::<i64>(0);
        let mut world = wb.build();

        fn write_u64(mut sink: ResMut<u64>, event: &u32) {
            *sink += *event as u64;
        }
        fn write_i64(mut sink: ResMut<i64>, event: &u32) {
            *sink += *event as i64;
        }

        let h1 = write_u64.into_handler(world.registry());
        let h2 = write_i64.into_handler(world.registry());
        let fan = fan_out!(h1, h2);

        let r = world.registry_mut();
        let mut p = PipelineStart::<u32>::new()
            .then(|x: u32| x * 2, r)
            .dispatch(fan)
            .build();

        p.run(&mut world, 5);
        assert_eq!(*world.resource::<u64>(), 10);
        assert_eq!(*world.resource::<i64>(), 10);
    }

    #[test]
    fn dispatch_to_broadcast() {
        let mut wb = WorldBuilder::new();
        wb.register::<u64>(0);
        let mut world = wb.build();

        fn write_u64(mut sink: ResMut<u64>, event: &u32) {
            *sink += *event as u64;
        }

        let mut broadcast = crate::Broadcast::<u32>::new();
        broadcast.add(write_u64.into_handler(world.registry()));
        broadcast.add(write_u64.into_handler(world.registry()));

        let r = world.registry_mut();
        let mut p = PipelineStart::<u32>::new()
            .then(|x: u32| x + 1, r)
            .dispatch(broadcast)
            .build();

        p.run(&mut world, 4);
        assert_eq!(*world.resource::<u64>(), 10); // 5 + 5
    }

    #[test]
    fn dispatch_build_produces_handler() {
        let mut wb = WorldBuilder::new();
        wb.register::<u64>(0);
        let mut world = wb.build();

        fn store(mut out: ResMut<u64>, val: u32) {
            *out = val as u64;
        }

        let r = world.registry_mut();
        let inner = PipelineStart::<u32>::new().then(store, r).build();

        let mut pipeline: Box<dyn Handler<u32>> = Box::new(
            PipelineStart::<u32>::new()
                .then(|x: u32| x + 1, r)
                .dispatch(inner)
                .build(),
        );

        pipeline.run(&mut world, 9);
        assert_eq!(*world.resource::<u64>(), 10);
    }

    // -- Guard combinator --

    #[test]
    fn pipeline_guard_keeps() {
        fn sink(mut out: ResMut<u64>, val: Option<u64>) {
            *out = val.unwrap_or(0);
        }
        let mut wb = WorldBuilder::new();
        wb.register::<u64>(0);
        let mut world = wb.build();
        let reg = world.registry();

        let mut p = PipelineStart::<u32>::new()
            .then(|x: u32| x as u64, reg)
            .guard(|_w, v| *v > 3)
            .then(sink, reg);

        p.run(&mut world, 5u32);
        assert_eq!(*world.resource::<u64>(), 5);
    }

    #[test]
    fn pipeline_guard_drops() {
        fn sink(mut out: ResMut<u64>, val: Option<u64>) {
            *out = val.unwrap_or(999);
        }
        let mut wb = WorldBuilder::new();
        wb.register::<u64>(0);
        let mut world = wb.build();
        let reg = world.registry();

        let mut p = PipelineStart::<u32>::new()
            .then(|x: u32| x as u64, reg)
            .guard(|_w, v| *v > 10)
            .then(sink, reg);

        p.run(&mut world, 5u32);
        assert_eq!(*world.resource::<u64>(), 999);
    }

    // -- Tap combinator --

    #[test]
    fn pipeline_tap_observes_without_changing() {
        fn sink(mut out: ResMut<u64>, val: u64) {
            *out = val;
        }
        let mut wb = WorldBuilder::new();
        wb.register::<u64>(0);
        wb.register::<bool>(false);
        let mut world = wb.build();
        let reg = world.registry();

        let mut p = PipelineStart::<u32>::new()
            .then(|x: u32| x as u64 * 2, reg)
            .tap(|w, val| {
                *w.resource_mut::<bool>() = *val == 10;
            })
            .then(sink, reg);

        p.run(&mut world, 5u32);
        assert_eq!(*world.resource::<u64>(), 10); // value passed through
        assert!(*world.resource::<bool>()); // tap fired
    }

    // -- Route combinator --

    #[test]
    fn pipeline_route_true_arm() {
        fn sink(mut out: ResMut<u64>, val: u64) {
            *out = val;
        }
        let mut wb = WorldBuilder::new();
        wb.register::<u64>(0);
        let mut world = wb.build();
        let reg = world.registry();

        let arm_t = PipelineStart::new().then(|x: u64| x * 2, reg);
        let arm_f = PipelineStart::new().then(|x: u64| x * 3, reg);

        let mut p = PipelineStart::<u32>::new()
            .then(|x: u32| x as u64, reg)
            .route(|_w, v| *v > 3, arm_t, arm_f)
            .then(sink, reg);

        p.run(&mut world, 5u32); // 5 > 3 → true arm → double → 10
        assert_eq!(*world.resource::<u64>(), 10);
    }

    #[test]
    fn pipeline_route_false_arm() {
        fn sink(mut out: ResMut<u64>, val: u64) {
            *out = val;
        }
        let mut wb = WorldBuilder::new();
        wb.register::<u64>(0);
        let mut world = wb.build();
        let reg = world.registry();

        let arm_t = PipelineStart::new().then(|x: u64| x * 2, reg);
        let arm_f = PipelineStart::new().then(|x: u64| x * 3, reg);

        let mut p = PipelineStart::<u32>::new()
            .then(|x: u32| x as u64, reg)
            .route(|_w, v| *v > 10, arm_t, arm_f)
            .then(sink, reg);

        p.run(&mut world, 5u32); // 5 <= 10 → false arm → triple → 15
        assert_eq!(*world.resource::<u64>(), 15);
    }

    #[test]
    fn pipeline_route_nested() {
        fn sink(mut out: ResMut<u64>, val: u64) {
            *out = val;
        }
        let mut wb = WorldBuilder::new();
        wb.register::<u64>(0);
        let mut world = wb.build();
        let reg = world.registry();

        // N-ary via nesting: <5 → +100, 5..10 → +200, >=10 → +300
        let inner_t = PipelineStart::new().then(|x: u64| x + 200, reg);
        let inner_f = PipelineStart::new().then(|x: u64| x + 300, reg);
        let outer_t = PipelineStart::new().then(|x: u64| x + 100, reg);
        let outer_f =
            PipelineStart::new()
                .then(|x: u64| x, reg)
                .route(|_w, v| *v < 10, inner_t, inner_f);

        let mut p = PipelineStart::<u32>::new()
            .then(|x: u32| x as u64, reg)
            .route(|_w, v| *v < 5, outer_t, outer_f)
            .then(sink, reg);

        p.run(&mut world, 3u32); // 3 < 5 → +100 → 103
        assert_eq!(*world.resource::<u64>(), 103);

        p.run(&mut world, 7u32); // 7 >= 5, 7 < 10 → +200 → 207
        assert_eq!(*world.resource::<u64>(), 207);

        p.run(&mut world, 15u32); // 15 >= 5, 15 >= 10 → +300 → 315
        assert_eq!(*world.resource::<u64>(), 315);
    }

    // -- Tee combinator --

    #[test]
    fn pipeline_tee_side_effect_chain() {
        use crate::dag::DagArmStart;

        fn log_step(mut counter: ResMut<u32>, _val: &u64) {
            *counter += 1;
        }
        fn sink(mut out: ResMut<u64>, val: u64) {
            *out = val;
        }
        let mut wb = WorldBuilder::new();
        wb.register::<u64>(0);
        wb.register::<u32>(0);
        let mut world = wb.build();
        let reg = world.registry();

        let side = DagArmStart::new().then(log_step, reg);

        let mut p = PipelineStart::<u32>::new()
            .then(|x: u32| x as u64 * 2, reg)
            .tee(side)
            .then(sink, reg);

        p.run(&mut world, 5u32);
        assert_eq!(*world.resource::<u64>(), 10); // value passed through
        assert_eq!(*world.resource::<u32>(), 1); // side-effect fired

        p.run(&mut world, 7u32);
        assert_eq!(*world.resource::<u64>(), 14);
        assert_eq!(*world.resource::<u32>(), 2);
    }

    // -- Dedup combinator --

    #[test]
    fn pipeline_dedup_suppresses_unchanged() {
        fn sink(mut out: ResMut<u32>, val: Option<u64>) {
            if val.is_some() {
                *out += 1;
            }
        }
        let mut wb = WorldBuilder::new();
        wb.register::<u32>(0);
        let mut world = wb.build();
        let reg = world.registry();

        let mut p = PipelineStart::<u32>::new()
            .then(|x: u32| x as u64 / 2, reg)
            .dedup()
            .then(sink, reg);

        p.run(&mut world, 4u32); // 2 — first, Some
        assert_eq!(*world.resource::<u32>(), 1);

        p.run(&mut world, 5u32); // 2 — same, None
        assert_eq!(*world.resource::<u32>(), 1);

        p.run(&mut world, 6u32); // 3 — changed, Some
        assert_eq!(*world.resource::<u32>(), 2);
    }

    // -- Bool combinators --

    #[test]
    fn pipeline_not() {
        fn sink(mut out: ResMut<bool>, val: bool) {
            *out = val;
        }
        let mut wb = WorldBuilder::new();
        wb.register::<bool>(false);
        let mut world = wb.build();
        let reg = world.registry();

        let mut p = PipelineStart::<u32>::new()
            .then(|x: u32| x > 5, reg)
            .not()
            .then(sink, reg);

        p.run(&mut world, 3u32); // 3 > 5 = false, not = true
        assert!(*world.resource::<bool>());

        p.run(&mut world, 10u32); // 10 > 5 = true, not = false
        assert!(!*world.resource::<bool>());
    }

    #[test]
    fn pipeline_and() {
        fn sink(mut out: ResMut<bool>, val: bool) {
            *out = val;
        }
        let mut wb = WorldBuilder::new();
        wb.register::<bool>(true);
        let mut world = wb.build();
        let reg = world.registry();

        let mut p = PipelineStart::<u32>::new()
            .then(|x: u32| x > 5, reg)
            .and(|w| *w.resource::<bool>())
            .then(sink, reg);

        p.run(&mut world, 10u32); // true && true = true
        assert!(*world.resource::<bool>());

        *world.resource_mut::<bool>() = false;
        p.run(&mut world, 10u32); // true && false = false
        assert!(!*world.resource::<bool>());
    }

    #[test]
    fn pipeline_or() {
        fn sink(mut out: ResMut<bool>, val: bool) {
            *out = val;
        }
        let mut wb = WorldBuilder::new();
        wb.register::<bool>(false);
        let mut world = wb.build();
        let reg = world.registry();

        let mut p = PipelineStart::<u32>::new()
            .then(|x: u32| x > 5, reg)
            .or(|w| *w.resource::<bool>())
            .then(sink, reg);

        p.run(&mut world, 3u32); // false || false = false
        assert!(!*world.resource::<bool>());

        *world.resource_mut::<bool>() = true;
        p.run(&mut world, 3u32); // false || true = true
        assert!(*world.resource::<bool>());
    }

    #[test]
    fn pipeline_xor() {
        fn sink(mut out: ResMut<bool>, val: bool) {
            *out = val;
        }
        let mut wb = WorldBuilder::new();
        wb.register::<bool>(true);
        let mut world = wb.build();
        let reg = world.registry();

        let mut p = PipelineStart::<u32>::new()
            .then(|x: u32| x > 5, reg)
            .xor(|w| *w.resource::<bool>())
            .then(sink, reg);

        p.run(&mut world, 10u32); // true ^ true = false
        assert!(!*world.resource::<bool>());
    }

    // =========================================================================
    // Splat — tuple destructuring
    // =========================================================================

    #[test]
    fn splat2_closure_on_start() {
        let mut world = WorldBuilder::new().build();
        let r = world.registry_mut();
        let mut p = PipelineStart::<(u32, u64)>::new()
            .splat()
            .then(|a: u32, b: u64| a as u64 + b, r);
        assert_eq!(p.run(&mut world, (3, 7)), 10);
    }

    #[test]
    fn splat2_named_fn_with_param() {
        let mut wb = WorldBuilder::new();
        wb.register::<u64>(100);
        let mut world = wb.build();

        fn process(base: Res<u64>, a: u32, b: u32) -> u64 {
            *base + a as u64 + b as u64
        }

        let r = world.registry_mut();
        let mut p = PipelineStart::<(u32, u32)>::new().splat().then(process, r);
        assert_eq!(p.run(&mut world, (3, 7)), 110);
    }

    #[test]
    fn splat2_mid_chain() {
        let mut world = WorldBuilder::new().build();
        let r = world.registry_mut();
        let mut p = PipelineStart::<u32>::new()
            .then(|x: u32| (x, x * 2), r)
            .splat()
            .then(|a: u32, b: u32| a as u64 + b as u64, r);
        assert_eq!(p.run(&mut world, 5), 15); // 5 + 10
    }

    #[test]
    fn splat3_closure_on_start() {
        let mut world = WorldBuilder::new().build();
        let r = world.registry_mut();
        let mut p = PipelineStart::<(u32, u32, u32)>::new()
            .splat()
            .then(|a: u32, b: u32, c: u32| a + b + c, r);
        assert_eq!(p.run(&mut world, (1, 2, 3)), 6);
    }

    #[test]
    fn splat3_named_fn_with_param() {
        let mut wb = WorldBuilder::new();
        wb.register::<u64>(10);
        let mut world = wb.build();

        fn process(factor: Res<u64>, a: u32, b: u32, c: u32) -> u64 {
            *factor * (a + b + c) as u64
        }

        let r = world.registry_mut();
        let mut p = PipelineStart::<(u32, u32, u32)>::new()
            .splat()
            .then(process, r);
        assert_eq!(p.run(&mut world, (1, 2, 3)), 60);
    }

    #[test]
    fn splat4_mid_chain() {
        let mut world = WorldBuilder::new().build();
        let r = world.registry_mut();
        let mut p = PipelineStart::<u32>::new()
            .then(|x: u32| (x, x + 1, x + 2, x + 3), r)
            .splat()
            .then(|a: u32, b: u32, c: u32, d: u32| (a + b + c + d) as u64, r);
        assert_eq!(p.run(&mut world, 10), 46); // 10+11+12+13
    }

    #[test]
    fn splat5_closure_on_start() {
        let mut world = WorldBuilder::new().build();
        let r = world.registry_mut();
        let mut p = PipelineStart::<(u8, u8, u8, u8, u8)>::new().splat().then(
            |a: u8, b: u8, c: u8, d: u8, e: u8| {
                (a as u64) + (b as u64) + (c as u64) + (d as u64) + (e as u64)
            },
            r,
        );
        assert_eq!(p.run(&mut world, (1, 2, 3, 4, 5)), 15);
    }

    #[test]
    fn splat_build_into_handler() {
        let mut wb = WorldBuilder::new();
        wb.register::<u64>(0);
        let mut world = wb.build();

        fn store(mut out: ResMut<u64>, a: u32, b: u32) {
            *out = a as u64 + b as u64;
        }

        let r = world.registry_mut();
        let mut pipeline = PipelineStart::<(u32, u32)>::new()
            .splat()
            .then(store, r)
            .build();

        pipeline.run(&mut world, (3, 7));
        assert_eq!(*world.resource::<u64>(), 10);
    }

    #[test]
    fn splat_build_batch() {
        let mut wb = WorldBuilder::new();
        wb.register::<u64>(0);
        let mut world = wb.build();

        fn accumulate(mut sum: ResMut<u64>, a: u32, b: u32) {
            *sum += a as u64 + b as u64;
        }

        let r = world.registry_mut();
        let mut batch = PipelineStart::<(u32, u32)>::new()
            .splat()
            .then(accumulate, r)
            .build_batch(8);

        batch
            .input_mut()
            .extend_from_slice(&[(1, 2), (3, 4), (5, 6)]);
        batch.run(&mut world);
        assert_eq!(*world.resource::<u64>(), 21); // 3+7+11
    }

    #[test]
    #[should_panic]
    fn splat_access_conflict_detected() {
        let mut wb = WorldBuilder::new();
        wb.register::<u64>(0);
        let mut world = wb.build();

        fn bad(a: ResMut<u64>, _b: ResMut<u64>, _x: u32, _y: u32) {
            let _ = a;
        }

        let r = world.registry_mut();
        // Should panic on duplicate ResMut<u64>
        let _ = PipelineStart::<(u32, u32)>::new().splat().then(bad, r);
    }

    // -- Switch combinator --

    #[test]
    fn pipeline_switch_basic() {
        fn double(x: u32) -> u64 {
            x as u64 * 2
        }
        fn sink(mut out: ResMut<u64>, val: u64) {
            *out = val;
        }

        let mut wb = WorldBuilder::new();
        wb.register::<u64>(0);
        let mut world = wb.build();
        let reg = world.registry();

        let mut pipeline = PipelineStart::<u32>::new()
            .then(double, reg)
            .switch(|_world, val| if val > 10 { val * 100 } else { val + 1 })
            .then(sink, reg)
            .build();

        pipeline.run(&mut world, 10u32); // 20 > 10 → 2000
        assert_eq!(*world.resource::<u64>(), 2000);

        pipeline.run(&mut world, 3u32); // 6 <= 10 → 7
        assert_eq!(*world.resource::<u64>(), 7);
    }

    #[test]
    fn pipeline_switch_3_way() {
        fn sink(mut out: ResMut<u64>, val: u64) {
            *out = val;
        }

        let mut wb = WorldBuilder::new();
        wb.register::<u64>(0);
        let mut world = wb.build();
        let reg = world.registry();

        let mut pipeline = PipelineStart::<u32>::new()
            .switch(|_world, val| match val % 3 {
                0 => val as u64 + 100,
                1 => val as u64 + 200,
                _ => val as u64 + 300,
            })
            .then(sink, reg)
            .build();

        pipeline.run(&mut world, 6u32); // 6 % 3 == 0 → 106
        assert_eq!(*world.resource::<u64>(), 106);

        pipeline.run(&mut world, 7u32); // 7 % 3 == 1 → 207
        assert_eq!(*world.resource::<u64>(), 207);

        pipeline.run(&mut world, 8u32); // 8 % 3 == 2 → 308
        assert_eq!(*world.resource::<u64>(), 308);
    }

    #[test]
    fn pipeline_switch_with_resolve_step() {
        fn add_offset(offset: Res<i64>, val: u32) -> u64 {
            (*offset + val as i64) as u64
        }
        fn plain_double(val: u32) -> u64 {
            val as u64 * 2
        }
        fn sink(mut out: ResMut<u64>, val: u64) {
            *out = val;
        }

        let mut wb = WorldBuilder::new();
        wb.register::<u64>(0);
        wb.register::<i64>(100);
        let mut world = wb.build();
        let reg = world.registry();

        let mut arm_offset = resolve_step(add_offset, reg);
        let mut arm_double = resolve_step(plain_double, reg);

        let mut pipeline = PipelineStart::<u32>::new()
            .switch(move |world, val| {
                if val > 10 {
                    arm_offset(world, val)
                } else {
                    arm_double(world, val)
                }
            })
            .then(sink, reg)
            .build();

        pipeline.run(&mut world, 20u32); // > 10 → offset → 100 + 20 = 120
        assert_eq!(*world.resource::<u64>(), 120);

        pipeline.run(&mut world, 5u32); // <= 10 → double → 10
        assert_eq!(*world.resource::<u64>(), 10);
    }

    #[test]
    fn batch_pipeline_switch() {
        fn sink(mut out: ResMut<u64>, val: u64) {
            *out += val;
        }

        let mut wb = WorldBuilder::new();
        wb.register::<u64>(0);
        let mut world = wb.build();
        let reg = world.registry();

        let mut batch = PipelineStart::<u32>::new()
            .switch(|_world, val| {
                if val % 2 == 0 {
                    val as u64 * 10
                } else {
                    val as u64
                }
            })
            .then(sink, reg)
            .build_batch(8);

        batch.input_mut().extend([1, 2, 3, 4]);
        batch.run(&mut world);

        // 1 → 1, 2 → 20, 3 → 3, 4 → 40 = 64
        assert_eq!(*world.resource::<u64>(), 64);
    }
}