Aura/crates/aura-engine/src/blueprint.rs

//! The construction layer (C9/C19/C23): a named, param-generic graph-as-data
//! (`Blueprint`) that **compiles** to the flat, type-erased instance the run loop
//! already runs (the *compilat*). The unit of reuse is the [`Composite`]: a
//! nestable sub-graph fragment exposing an output record (one port, K re-exported
//! fields; C8) and named input roles, which `compile` **inlines** into the flat
//! `(nodes, sources, edges)` the
//! unchanged [`crate::Harness::bootstrap`] consumes.
//!
//! The compilat is wired by raw index, **not by name** (C23): a composite's
//! boundary dissolves at compile time; field/role names, where kept, are
//! non-load-bearing debug symbols (as `FieldSpec.name` already is). This module
//! adds no optimisation pass (CSE/DCE, sweep-invariant hoisting are deferred,
//! C23) and no external dependency (C16).

use aura_core::{LeafFactory, Node, ParamSpec, Scalar, ScalarKind};

use crate::harness::{BootstrapError, Edge, Harness, SourceSpec, Target};

/// One re-exported field of a composite's output record: an interior
/// `(node, output-field)` surfaced at the boundary under `name`. `name` is a
/// non-load-bearing render/debug symbol (C23) — like `FieldSpec.name` and
/// `Composite.name`, it does not reach the compilat.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct OutField {
    pub node: usize,
    pub field: usize,
    pub name: String,
}

/// A blueprint item: a leaf node or a nested composite. Both present a declared
/// interface (typed inputs + one output) to the enclosing graph.
pub enum BlueprintNode {
    Leaf(LeafFactory),
    Composite(Composite),
}

/// Ergonomic lift: a param-generic leaf recipe becomes a `Leaf` blueprint item.
impl From<LeafFactory> for BlueprintNode {
    fn from(factory: LeafFactory) -> Self {
        BlueprintNode::Leaf(factory)
    }
}

/// One named input role: role `r` (by position) fans the source value into
/// `targets`. The `name` is a non-load-bearing render symbol (C23); identity is
/// the role index, which survives lowering — the name does not.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct Role {
    pub name: String,
    pub targets: Vec<Target>,
}

/// A composite-level alias relabelling one interior leaf param slot's surface
/// name in `param_space()`. `node` is the interior item index, `slot` the param
/// slot within that leaf. Pure legibility: the alias relabels in place and never
/// reorders, adds, or removes a slot (C23 — identity stays the slot).
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct ParamAlias {
    pub name: String,
    pub node: usize,
    pub slot: usize,
}

/// A reusable sub-graph fragment compiled away by inlining (C9/C23). It is **not**
/// a [`Node`]: it is never `eval`'d. It holds interior items (local indices),
/// interior edges (local indices), input roles (role `r` fans into the interior
/// targets `input_roles[r]`), and the exposed output record (each entry
/// re-exports one interior `(node, field)` under a boundary name).
pub struct Composite {
    name: String,
    nodes: Vec<BlueprintNode>,
    edges: Vec<Edge>,
    input_roles: Vec<Role>,
    params: Vec<ParamAlias>,
    output: Vec<OutField>,
}

impl Composite {
    /// Build a composite from its authored name, interior items, interior edges
    /// (local indices), input roles, and output record. The `name` is a
    /// non-load-bearing render symbol (the cluster title for #13); it does not
    /// reach the compilat (the boundary dissolves at inline, C23).
    pub fn new(
        name: impl Into<String>,
        nodes: Vec<BlueprintNode>,
        edges: Vec<Edge>,
        input_roles: Vec<Role>,
        params: Vec<ParamAlias>,
        output: Vec<OutField>,
    ) -> Self {
        Self { name: name.into(), nodes, edges, input_roles, params, output }
    }

    /// The authored render name (cluster title, #13). Non-load-bearing.
    pub fn name(&self) -> &str {
        &self.name
    }
    /// The interior blueprint items (read-only graph-as-data, C9).
    pub fn nodes(&self) -> &[BlueprintNode] {
        &self.nodes
    }
    /// The interior edges (local indices).
    pub fn edges(&self) -> &[Edge] {
        &self.edges
    }
    /// The input roles: role `r` fans into `input_roles()[r].targets` interior
    /// targets, under the boundary name `input_roles()[r].name` (C23 — name is a
    /// render symbol, identity is the role index).
    pub fn input_roles(&self) -> &[Role] {
        &self.input_roles
    }
    /// The param aliases: each relabels one interior leaf param slot's surface
    /// name in `param_space()` (pure naming overlay; identity stays the slot, C23).
    pub fn params(&self) -> &[ParamAlias] {
        &self.params
    }
    /// The exposed output record: each entry re-exports one interior
    /// `(node, output-field)` under a boundary name (C8 — one port, K columns).
    pub fn output(&self) -> &[OutField] {
        &self.output
    }
}

/// A construction-phase fault, caught before the flat compilat reaches
/// `Harness::bootstrap`.
#[derive(Debug, PartialEq, Eq)]
pub enum CompileError {
    /// An interior edge, role target, or output index is out of range.
    BadInteriorIndex,
    /// Input role `role` fans into interior slots of differing scalar kinds.
    RoleKindMismatch { role: usize },
    /// The output port names a missing interior node or output field.
    OutputPortOutOfRange,
    /// The lowered flat compilat failed `Harness::bootstrap`'s checks (kind
    /// mismatch, bad index, or directed cycle).
    Bootstrap(BootstrapError),
    /// An injected param value's scalar kind does not match the slot's declared
    /// kind. `slot` is the flat param-space index.
    ParamKindMismatch { slot: usize, expected: ScalarKind, got: ScalarKind },
    /// The injected vector's length does not equal the sum of declared params.
    ParamArity { expected: usize, got: usize },
}

/// The root graph-as-data, before compilation: blueprint items + sources + edges,
/// all addressing blueprint-level indices.
pub struct Blueprint {
    nodes: Vec<BlueprintNode>,
    sources: Vec<SourceSpec>,
    edges: Vec<Edge>,
}

impl Blueprint {
    /// Build a blueprint from its items, sources, and edges (blueprint-level
    /// indices; a target/edge endpoint may name a composite).
    pub fn new(nodes: Vec<BlueprintNode>, sources: Vec<SourceSpec>, edges: Vec<Edge>) -> Self {
        Self { nodes, sources, edges }
    }

    /// The top-level blueprint items (read-only graph-as-data, C9).
    pub fn nodes(&self) -> &[BlueprintNode] {
        &self.nodes
    }
    /// The declared sources.
    pub fn sources(&self) -> &[SourceSpec] {
        &self.sources
    }
    /// The top-level edges (blueprint-level indices).
    pub fn edges(&self) -> &[Edge] {
        &self.edges
    }

    /// The aggregated, flat, path-qualified param-space (C12): every node's declared
    /// params, concatenated in the deterministic depth-first item order `lower_items`
    /// uses, so a param's slot here matches the later flat-node order (#31 binds
    /// slot-by-slot). Read-only graph-as-data (C9); does not compile. Names are
    /// non-load-bearing: a composite's `name()` is prefixed at each level, but
    /// same-type siblings in one composite share a name — uniqueness is at the slot.
    pub fn param_space(&self) -> Vec<ParamSpec> {
        let mut out = Vec::new();
        collect_params(&self.nodes, "", &[], &mut out);
        out
    }

    /// Compile the value-empty recipe under an injected param vector: build each
    /// leaf from its kind-checked slice while lowering, then rewrite edges/sources
    /// exactly as before (structure is param-invariant, C19/C23). The vector is
    /// total and positional — one value per `param_space()` slot, in slot order.
    // The flat triple is exactly `Harness::bootstrap`'s argument list; naming it
    // would be a speculative type alias this cycle (same call as the CLI's sample).
    #[allow(clippy::type_complexity)]
    pub fn compile_with_params(
        self,
        params: &[Scalar],
    ) -> Result<(Vec<Box<dyn Node>>, Vec<SourceSpec>, Vec<Edge>), CompileError> {
        let expected = self.param_space().len();
        if params.len() != expected {
            return Err(CompileError::ParamArity { expected, got: params.len() });
        }
        let mut flat_nodes: Vec<Box<dyn Node>> = Vec::new();
        let mut flat_edges: Vec<Edge> = Vec::new();
        let mut cursor = 0usize;

        // lower every top-level item (recursively inlining composites), building
        // each leaf from its kind-checked param slice as it lowers
        let lowerings =
            lower_items(self.nodes, params, &mut cursor, &mut flat_nodes, &mut flat_edges)?;

        // rewrite top-level edges through the lowerings (fan-out into composites)
        for e in &self.edges {
            for fe in rewrite_edge(e, &lowerings, &flat_nodes)? {
                flat_edges.push(fe);
            }
        }

        // rewrite sources: each target into a composite fans into its role targets
        let mut flat_sources: Vec<SourceSpec> = Vec::with_capacity(self.sources.len());
        for src in &self.sources {
            let mut targets: Vec<Target> = Vec::new();
            for t in &src.targets {
                targets.extend(resolve_target(t, &lowerings)?);
            }
            flat_sources.push(SourceSpec { kind: src.kind, targets });
        }

        Ok((flat_nodes, flat_sources, flat_edges))
    }

    /// No-param compile (a blueprint that declares no params); errors `ParamArity`
    /// if any param is declared.
    #[allow(clippy::type_complexity)]
    pub fn compile(self) -> Result<(Vec<Box<dyn Node>>, Vec<SourceSpec>, Vec<Edge>), CompileError> {
        self.compile_with_params(&[])
    }

    /// Compile under an injected vector, then hand the flat compilat to the
    /// unchanged `Harness::bootstrap`.
    pub fn bootstrap_with_params(self, params: Vec<Scalar>) -> Result<Harness, CompileError> {
        let (nodes, sources, edges) = self.compile_with_params(&params)?;
        Harness::bootstrap(nodes, sources, edges).map_err(CompileError::Bootstrap)
    }

    /// No-param bootstrap (paramless blueprint).
    pub fn bootstrap(self) -> Result<Harness, CompileError> {
        self.bootstrap_with_params(vec![])
    }
}

/// Recursive read-only walk for `Blueprint::param_space`: a leaf contributes its
/// declared params under the running path prefix; a composite pushes its `name()`
/// onto the path and recurses, passing its own param aliases down. A leaf param
/// slot matched by an `(node, slot)` alias is relabelled in place (C23 — pure
/// naming overlay; the slot stays, order is untouched). Order mirrors `lower_items`
/// (items in declared order, composites depth-first) so a param's slot matches the
/// later flat-node order.
fn collect_params(
    items: &[BlueprintNode],
    prefix: &str,
    aliases: &[ParamAlias],
    out: &mut Vec<ParamSpec>,
) {
    for (i, item) in items.iter().enumerate() {
        match item {
            BlueprintNode::Leaf(factory) => {
                for (s, p) in factory.params().iter().enumerate() {
                    // an alias for this exact (node, slot) relabels in place;
                    // otherwise the factory param name, as today.
                    let local = aliases
                        .iter()
                        .find(|a| a.node == i && a.slot == s)
                        .map(|a| a.name.as_str())
                        .unwrap_or(p.name.as_str());
                    let name = if prefix.is_empty() {
                        local.to_string()
                    } else {
                        format!("{prefix}.{local}")
                    };
                    out.push(ParamSpec { name, kind: p.kind });
                }
            }
            BlueprintNode::Composite(c) => {
                let child = if prefix.is_empty() {
                    c.name().to_string()
                } else {
                    format!("{prefix}.{}", c.name())
                };
                collect_params(c.nodes(), &child, c.params(), out);
            }
        }
    }
}

/// How one blueprint item resolved into the flat compilat. Edges and source
/// targets to/from an item are resolved through this.
enum ItemLowering {
    /// A leaf lowered to exactly one flat node at this index.
    Leaf { index: usize },
    /// A composite lowered to its interior: its output record is these flat
    /// `(node, field)` producers (one per re-exported field, declared order), and
    /// input role `r` fans into `roles[r]` (flat targets). Names dropped (C23).
    Composite { output: Vec<(usize, usize)>, roles: Vec<Vec<Target>> },
}

/// Lower a list of blueprint items into the flat node array, appending interior
/// nodes and (for composites) their interior edges. Returns one `ItemLowering` per
/// input item, in order.
fn lower_items(
    items: Vec<BlueprintNode>,
    params: &[Scalar],
    cursor: &mut usize,
    flat_nodes: &mut Vec<Box<dyn Node>>,
    flat_edges: &mut Vec<Edge>,
) -> Result<Vec<ItemLowering>, CompileError> {
    let mut lowerings = Vec::with_capacity(items.len());
    for item in items {
        match item {
            BlueprintNode::Leaf(factory) => {
                let n = factory.params().len();
                let slice = &params[*cursor..*cursor + n]; // in range: arity checked up front
                for (i, spec) in factory.params().iter().enumerate() {
                    let got = slice[i].kind();
                    if got != spec.kind {
                        return Err(CompileError::ParamKindMismatch {
                            slot: *cursor + i,
                            expected: spec.kind,
                            got,
                        });
                    }
                }
                let index = flat_nodes.len();
                flat_nodes.push(factory.build(slice));
                *cursor += n;
                lowerings.push(ItemLowering::Leaf { index });
            }
            BlueprintNode::Composite(c) => {
                lowerings.push(inline_composite(c, params, cursor, flat_nodes, flat_edges)?);
            }
        }
    }
    Ok(lowerings)
}

/// Inline one composite: recursively lower its interior items, rewrite its interior
/// edges, then resolve its output port and per-role flat targets.
fn inline_composite(
    c: Composite,
    params: &[Scalar],
    cursor: &mut usize,
    flat_nodes: &mut Vec<Box<dyn Node>>,
    flat_edges: &mut Vec<Edge>,
) -> Result<ItemLowering, CompileError> {
    // `name` is the non-load-bearing render symbol (#13); it dissolves at inline
    // (C23 — the boundary does not reach the compilat), so it is not destructured.
    // `params` here are the composite's ParamAlias overlay (renamed to avoid
    // shadowing the injected scalar `params: &[Scalar]` arg consumed by
    // `lower_items` below).
    let Composite { name: _, nodes, edges, input_roles, params: param_aliases, output } = c;
    let item_count = nodes.len();

    // an alias must name a real interior leaf param slot (C23 — names are cosmetic
    // but a dangling handle is an author error). Mirrors the output range-check.
    for a in &param_aliases {
        let ok = a.node < item_count
            && matches!(&nodes[a.node], BlueprintNode::Leaf(f) if a.slot < f.params().len());
        if !ok {
            return Err(CompileError::BadInteriorIndex);
        }
    }

    // recursively lower interior items, then rewrite interior edges through them
    let interior = lower_items(nodes, params, cursor, flat_nodes, flat_edges)?;
    for e in &edges {
        for fe in rewrite_edge(e, &interior, flat_nodes)? {
            flat_edges.push(fe);
        }
    }

    // resolve each re-exported field to a flat (node, field), in declared order
    let mut out: Vec<(usize, usize)> = Vec::with_capacity(output.len());
    for of in &output {
        if of.node >= item_count {
            return Err(CompileError::OutputPortOutOfRange);
        }
        let resolved = match &interior[of.node] {
            ItemLowering::Leaf { index } => {
                if of.field >= flat_nodes[*index].schema().output.len() {
                    return Err(CompileError::OutputPortOutOfRange);
                }
                (*index, of.field)
            }
            ItemLowering::Composite { output: nested, .. } => {
                *nested.get(of.field).ok_or(CompileError::OutputPortOutOfRange)?
            }
        };
        out.push(resolved);
    }

    // resolve each input role to flat targets (a target into a nested composite
    // fans further) and kind-check every role
    let mut roles: Vec<Vec<Target>> = Vec::with_capacity(input_roles.len());
    for (r, role) in input_roles.iter().enumerate() {
        let mut flat_targets: Vec<Target> = Vec::new();
        for t in &role.targets {
            flat_targets.extend(resolve_target(t, &interior)?);
        }
        if let Some((first, rest)) = flat_targets.split_first() {
            let k0 = slot_kind(*first, flat_nodes)?;
            for ft in rest {
                if slot_kind(*ft, flat_nodes)? != k0 {
                    return Err(CompileError::RoleKindMismatch { role: r });
                }
            }
        }
        roles.push(flat_targets);
    }

    Ok(ItemLowering::Composite { output: out, roles })
}

/// Rewrite one blueprint-level edge into flat edges. The `from` endpoint resolves
/// to a single flat producer `(node, field)`; the `to` endpoint may fan out (a
/// composite input role fans into several interior targets).
fn rewrite_edge(
    e: &Edge,
    lowerings: &[ItemLowering],
    flat_nodes: &[Box<dyn Node>],
) -> Result<Vec<Edge>, CompileError> {
    if e.from >= lowerings.len() {
        return Err(CompileError::BadInteriorIndex);
    }
    let (from_node, from_field) = match &lowerings[e.from] {
        ItemLowering::Leaf { index } => {
            if e.from_field >= flat_nodes[*index].schema().output.len() {
                return Err(CompileError::BadInteriorIndex);
            }
            (*index, e.from_field)
        }
        ItemLowering::Composite { output, .. } => {
            *output.get(e.from_field).ok_or(CompileError::BadInteriorIndex)?
        }
    };
    let targets = resolve_target(&Target { node: e.to, slot: e.slot }, lowerings)?;
    Ok(targets
        .into_iter()
        .map(|t| Edge { from: from_node, to: t.node, slot: t.slot, from_field })
        .collect())
}

/// Resolve a blueprint-level target `(node, slot)` into flat target(s). A target
/// into a leaf is itself (remapped index); a target into a composite fans into
/// that composite's input-role flat targets.
fn resolve_target(t: &Target, lowerings: &[ItemLowering]) -> Result<Vec<Target>, CompileError> {
    if t.node >= lowerings.len() {
        return Err(CompileError::BadInteriorIndex);
    }
    match &lowerings[t.node] {
        ItemLowering::Leaf { index } => Ok(vec![Target { node: *index, slot: t.slot }]),
        ItemLowering::Composite { roles, .. } => {
            let role = roles.get(t.slot).ok_or(CompileError::BadInteriorIndex)?;
            Ok(role.clone())
        }
    }
}

/// The declared scalar kind of a flat node's input slot (for role kind-checking).
fn slot_kind(t: Target, flat_nodes: &[Box<dyn Node>]) -> Result<ScalarKind, CompileError> {
    flat_nodes[t.node]
        .schema()
        .inputs
        .get(t.slot)
        .map(|spec| spec.kind)
        .ok_or(CompileError::BadInteriorIndex)
}

#[cfg(test)]
mod tests {
    use super::*;
    use aura_core::{Ctx, FieldSpec, Firing, InputSpec, NodeSchema, Timestamp};
    use aura_std::{Exposure, Recorder, SimBroker, Sma, Sub};
    use std::sync::mpsc;

    /// A 2-input f64 node, one f64 output. Test-local fixture (C9: examples for the
    /// engine's own tests, no speculative `aura-std` surface).
    struct Join2 {
        out: [Scalar; 1],
    }
    impl Node for Join2 {
        fn schema(&self) -> NodeSchema {
            NodeSchema {
                inputs: vec![
                    InputSpec { kind: ScalarKind::F64, lookback: 1, firing: Firing::Any },
                    InputSpec { kind: ScalarKind::F64, lookback: 1, firing: Firing::Any },
                ],
                output: vec![FieldSpec { name: "v", kind: ScalarKind::F64 }],
                params: vec![],
            }
        }
        fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Scalar]> {
            let a = ctx.f64_in(0);
            let b = ctx.f64_in(1);
            if a.is_empty() || b.is_empty() {
                return None;
            }
            self.out[0] = Scalar::F64(a[0] + b[0]);
            Some(&self.out)
        }
    }

    /// A 1-input f64 node, one f64 output. Test-local fixture.
    struct Pass1 {
        out: [Scalar; 1],
    }
    impl Node for Pass1 {
        fn schema(&self) -> NodeSchema {
            NodeSchema {
                inputs: vec![InputSpec { kind: ScalarKind::F64, lookback: 1, firing: Firing::Any }],
                output: vec![FieldSpec { name: "v", kind: ScalarKind::F64 }],
                params: vec![],
            }
        }
        fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Scalar]> {
            let w = ctx.f64_in(0);
            if w.is_empty() {
                return None;
            }
            self.out[0] = Scalar::F64(w[0]);
            Some(&self.out)
        }
    }

    /// A pure consumer with one f64 input and no output (sink role, C8).
    struct SinkF64;
    impl Node for SinkF64 {
        fn schema(&self) -> NodeSchema {
            NodeSchema {
                inputs: vec![InputSpec { kind: ScalarKind::F64, lookback: 1, firing: Firing::Any }],
                output: vec![],
                params: vec![],
            }
        }
        fn eval(&mut self, _ctx: Ctx<'_>) -> Option<&[Scalar]> {
            None
        }
    }

    /// A pure consumer with one i64 input and no output. Used to provoke a role /
    /// edge kind mismatch (its slot is i64 where an f64 is fanned in).
    struct SinkI64;
    impl Node for SinkI64 {
        fn schema(&self) -> NodeSchema {
            NodeSchema {
                inputs: vec![InputSpec { kind: ScalarKind::I64, lookback: 1, firing: Firing::Any }],
                output: vec![],
                params: vec![],
            }
        }
        fn eval(&mut self, _ctx: Ctx<'_>) -> Option<&[Scalar]> {
            None
        }
    }

    fn pass1() -> BlueprintNode {
        BlueprintNode::Leaf(LeafFactory::new("Pass1", vec![], |_| {
            Box::new(Pass1 { out: [Scalar::F64(0.0)] })
        }))
    }
    fn join2() -> BlueprintNode {
        BlueprintNode::Leaf(LeafFactory::new("Join2", vec![], |_| {
            Box::new(Join2 { out: [Scalar::F64(0.0)] })
        }))
    }
    fn sink_f64() -> BlueprintNode {
        BlueprintNode::Leaf(LeafFactory::new("SinkF64", vec![], |_| Box::new(SinkF64)))
    }
    fn sink_i64() -> BlueprintNode {
        BlueprintNode::Leaf(LeafFactory::new("SinkI64", vec![], |_| Box::new(SinkI64)))
    }

    /// A composite: two Pass1 leaves feeding a Join2, role 0 fanning the source
    /// into BOTH Pass1 slots, output = the Join2 field 0. The generic analogue of
    /// the SMA-cross shape.
    fn fan_composite() -> Composite {
        Composite::new(
            "fan",
            vec![pass1(), pass1(), join2()],
            vec![
                Edge { from: 0, to: 2, slot: 0, from_field: 0 },
                Edge { from: 1, to: 2, slot: 1, from_field: 0 },
            ],
            vec![Role {
                name: "price".into(),
                targets: vec![Target { node: 0, slot: 0 }, Target { node: 1, slot: 0 }],
            }],
            vec![],
            vec![OutField { node: 2, field: 0, name: "out".into() }],
        )
    }

    #[test]
    fn single_composite_inlines_with_offset_fan_and_output() {
        // composite as item 0; a source into its role 0; an edge out of it to a sink.
        let bp = Blueprint::new(
            vec![BlueprintNode::Composite(fan_composite()), sink_f64()],
            vec![SourceSpec { kind: ScalarKind::F64, targets: vec![Target { node: 0, slot: 0 }] }],
            vec![Edge { from: 0, to: 1, slot: 0, from_field: 0 }],
        );
        let (nodes, sources, edges) = bp.compile().expect("valid composite");

        // 3 interior nodes (Pass1, Pass1, Join2) at flat 0..2, then SinkF64 at 3
        assert_eq!(nodes.len(), 4);
        // interior edges rewritten at offset 0, then the output edge resolves the
        // composite's OutField (interior node 2, field 0) to the sink (flat node 3)
        assert_eq!(
            edges,
            vec![
                Edge { from: 0, to: 2, slot: 0, from_field: 0 },
                Edge { from: 1, to: 2, slot: 1, from_field: 0 },
                Edge { from: 2, to: 3, slot: 0, from_field: 0 },
            ]
        );
        // the source target into role 0 fanned into BOTH Pass1 slots
        assert_eq!(sources.len(), 1);
        assert_eq!(
            sources[0].targets,
            vec![Target { node: 0, slot: 0 }, Target { node: 1, slot: 0 }]
        );
    }

    #[test]
    fn composite_reexports_two_fields_to_distinct_consumers() {
        // composite: two independent Pass1 leaves; role 0 -> leaf 0, role 1 -> leaf 1;
        // output record re-exports leaf 0 as "a", leaf 1 as "b".
        let c = Composite::new(
            "two_out",
            vec![pass1(), pass1()],
            vec![],
            vec![
                Role { name: "price".into(), targets: vec![Target { node: 0, slot: 0 }] },
                Role { name: "price2".into(), targets: vec![Target { node: 1, slot: 0 }] },
            ],
            vec![],
            vec![
                OutField { node: 0, field: 0, name: "a".into() },
                OutField { node: 1, field: 0, name: "b".into() },
            ],
        );
        // composite is item 0; two sinks (items 1, 2) read its two output fields by
        // from_field; one source fans into both roles.
        let bp = Blueprint::new(
            vec![BlueprintNode::Composite(c), sink_f64(), sink_f64()],
            vec![SourceSpec {
                kind: ScalarKind::F64,
                targets: vec![Target { node: 0, slot: 0 }, Target { node: 0, slot: 1 }],
            }],
            vec![
                Edge { from: 0, to: 1, slot: 0, from_field: 0 }, // field "a" -> sink 1
                Edge { from: 0, to: 2, slot: 0, from_field: 1 }, // field "b" -> sink 2
            ],
        );
        let (nodes, sources, edges) = bp.compile().expect("valid multi-output composite");
        // flat layout: Pass1(0), Pass1(1), SinkF64(2), SinkF64(3)
        assert_eq!(nodes.len(), 4);
        // from_field 0 resolves to leaf 0, from_field 1 to leaf 1 — distinct producers
        assert_eq!(
            edges,
            vec![
                Edge { from: 0, to: 2, slot: 0, from_field: 0 },
                Edge { from: 1, to: 3, slot: 0, from_field: 0 },
            ]
        );
        // the source fanned into both interior leaves
        assert_eq!(
            sources[0].targets,
            vec![Target { node: 0, slot: 0 }, Target { node: 1, slot: 0 }]
        );
    }

    #[test]
    fn nested_composite_inlines() {
        // outer composite wraps the inner fan_composite as its only interior item,
        // re-exposing the inner's role 0 (outer role 0 -> inner role 0) and the
        // inner's output. A source into the outer role 0 must fan to BOTH inner
        // Pass1 slots; the inner Join2 lands at flat index 2.
        let inner = fan_composite();
        let outer = Composite::new(
            "outer",
            vec![BlueprintNode::Composite(inner)],
            vec![],
            vec![Role { name: "price".into(), targets: vec![Target { node: 0, slot: 0 }] }],
            vec![],
            vec![OutField { node: 0, field: 0, name: "out".into() }],
        );
        let bp = Blueprint::new(
            vec![BlueprintNode::Composite(outer), sink_f64()],
            vec![SourceSpec { kind: ScalarKind::F64, targets: vec![Target { node: 0, slot: 0 }] }],
            vec![Edge { from: 0, to: 1, slot: 0, from_field: 0 }],
        );
        let (nodes, sources, edges) = bp.compile().expect("valid nested composite");

        assert_eq!(nodes.len(), 4); // Pass1, Pass1, Join2, SinkF64
        assert_eq!(
            sources[0].targets,
            vec![Target { node: 0, slot: 0 }, Target { node: 1, slot: 0 }]
        );
        // inner interior edges + the output edge from the inner Join2 (flat 2) to sink
        assert_eq!(
            edges,
            vec![
                Edge { from: 0, to: 2, slot: 0, from_field: 0 },
                Edge { from: 1, to: 2, slot: 1, from_field: 0 },
                Edge { from: 2, to: 3, slot: 0, from_field: 0 },
            ]
        );
    }

    #[test]
    fn outer_reexports_two_fields_of_inner_composite() {
        // inner re-exports two leaves as "a","b"; outer re-exposes both inner roles
        // and re-exports inner field 0 and field 1 (the latter exercises the nested arm).
        let inner = Composite::new(
            "inner_two",
            vec![pass1(), pass1()],
            vec![],
            vec![
                Role { name: "price".into(), targets: vec![Target { node: 0, slot: 0 }] },
                Role { name: "price2".into(), targets: vec![Target { node: 1, slot: 0 }] },
            ],
            vec![],
            vec![
                OutField { node: 0, field: 0, name: "a".into() },
                OutField { node: 1, field: 0, name: "b".into() },
            ],
        );
        let outer = Composite::new(
            "outer_two",
            vec![BlueprintNode::Composite(inner)],
            vec![],
            vec![
                Role { name: "price".into(), targets: vec![Target { node: 0, slot: 0 }] }, // outer role 0 -> inner role 0
                Role { name: "price2".into(), targets: vec![Target { node: 0, slot: 1 }] }, // outer role 1 -> inner role 1
            ],
            vec![],
            vec![
                OutField { node: 0, field: 0, name: "x".into() }, // inner field 0
                OutField { node: 0, field: 1, name: "y".into() }, // inner field 1 (nested arm)
            ],
        );
        let bp = Blueprint::new(
            vec![BlueprintNode::Composite(outer), sink_f64(), sink_f64()],
            vec![SourceSpec {
                kind: ScalarKind::F64,
                targets: vec![Target { node: 0, slot: 0 }, Target { node: 0, slot: 1 }],
            }],
            vec![
                Edge { from: 0, to: 1, slot: 0, from_field: 0 }, // outer field x -> sink 1
                Edge { from: 0, to: 2, slot: 0, from_field: 1 }, // outer field y -> sink 2
            ],
        );
        let (nodes, _sources, edges) = bp.compile().expect("valid nested multi-output");
        assert_eq!(nodes.len(), 4); // Pass1, Pass1, SinkF64, SinkF64
        assert_eq!(
            edges,
            vec![
                Edge { from: 0, to: 2, slot: 0, from_field: 0 },
                Edge { from: 1, to: 3, slot: 0, from_field: 0 },
            ]
        );
    }

    #[test]
    fn bad_interior_index_rejected() {
        // interior edge references interior node 9, which does not exist
        let c = Composite::new(
            "c",
            vec![pass1()],
            vec![Edge { from: 0, to: 9, slot: 0, from_field: 0 }],
            vec![Role { name: "price".into(), targets: vec![Target { node: 0, slot: 0 }] }],
            vec![],
            vec![OutField { node: 0, field: 0, name: "out".into() }],
        );
        let bp = Blueprint::new(vec![BlueprintNode::Composite(c)], vec![], vec![]);
        // the Ok arm holds Box<dyn Node> (not Debug), so assert via the Err arm.
        assert_eq!(bp.compile().err(), Some(CompileError::BadInteriorIndex));
    }

    #[test]
    fn role_kind_mismatch_rejected() {
        // role 0 fans into a Pass1 f64 slot AND a SinkI64 i64 slot -> mismatch
        let c = Composite::new(
            "c",
            vec![pass1(), sink_i64()],
            vec![],
            vec![Role {
                name: "price".into(),
                targets: vec![Target { node: 0, slot: 0 }, Target { node: 1, slot: 0 }],
            }],
            vec![],
            vec![OutField { node: 0, field: 0, name: "out".into() }],
        );
        let bp = Blueprint::new(vec![BlueprintNode::Composite(c)], vec![], vec![]);
        // the Ok arm holds Box<dyn Node> (not Debug), so assert via the Err arm.
        assert_eq!(bp.compile().err(), Some(CompileError::RoleKindMismatch { role: 0 }));
    }

    #[test]
    fn output_port_out_of_range_rejected() {
        // output names field 5 of a node whose output has one field
        let c = Composite::new(
            "c",
            vec![pass1()],
            vec![],
            vec![Role { name: "price".into(), targets: vec![Target { node: 0, slot: 0 }] }],
            vec![],
            vec![OutField { node: 0, field: 5, name: "out".into() }],
        );
        let bp = Blueprint::new(vec![BlueprintNode::Composite(c)], vec![], vec![]);
        // the Ok arm holds Box<dyn Node> (not Debug), so assert via the Err arm.
        assert_eq!(bp.compile().err(), Some(CompileError::OutputPortOutOfRange));
    }

    #[test]
    fn consume_of_missing_output_field_is_rejected() {
        // a single-field composite; a consumer reads from_field 1 (past the 1-field
        // record) -> the rewrite_edge range-check rejects it.
        let c = Composite::new(
            "c",
            vec![pass1()],
            vec![],
            vec![Role { name: "price".into(), targets: vec![Target { node: 0, slot: 0 }] }],
            vec![],
            vec![OutField { node: 0, field: 0, name: "a".into() }],
        );
        let bp = Blueprint::new(
            vec![BlueprintNode::Composite(c), sink_f64()],
            vec![SourceSpec { kind: ScalarKind::F64, targets: vec![Target { node: 0, slot: 0 }] }],
            vec![Edge { from: 0, to: 1, slot: 0, from_field: 1 }], // only field 0 exists
        );
        assert_eq!(bp.compile().err(), Some(CompileError::BadInteriorIndex));
    }

    #[test]
    fn bootstrap_error_is_wrapped() {
        // a top-level kind mismatch: a Pass1 f64 output wired into a SinkI64 i64
        // input. compile() lowers it faithfully; bootstrap's kind-check rejects it.
        let bp = Blueprint::new(
            vec![pass1(), sink_i64()],
            vec![],
            vec![Edge { from: 0, to: 1, slot: 0, from_field: 0 }],
        );
        match bp.bootstrap().unwrap_err() {
            CompileError::Bootstrap(BootstrapError::KindMismatch { producer, consumer }) => {
                assert_eq!(producer, ScalarKind::F64);
                assert_eq!(consumer, ScalarKind::I64);
            }
            other => panic!("expected Bootstrap(KindMismatch), got {other:?}"),
        }
    }

    /// The built-in synthetic price stream (a local copy of the CLI sample's
    /// stream): rises through t=4 then reverses, so the trace is non-degenerate.
    fn synthetic_prices() -> Vec<(Timestamp, Scalar)> {
        [
            (1_i64, 1.0000_f64),
            (2, 1.0010),
            (3, 1.0030),
            (4, 1.0060),
            (5, 1.0040),
            (6, 1.0010),
            (7, 0.9990),
        ]
        .iter()
        .map(|&(t, p)| (Timestamp(t), Scalar::F64(p)))
        .collect()
    }

    /// Today's flat, hand-wired SMA-cross signal-quality harness (the
    /// `sample_harness` wiring from `aura-cli`), with two recording sinks.
    #[allow(clippy::type_complexity)]
    fn hand_wired_sma_cross_harness() -> (
        Harness,
        mpsc::Receiver<(Timestamp, Vec<Scalar>)>,
        mpsc::Receiver<(Timestamp, Vec<Scalar>)>,
    ) {
        let (tx_eq, rx_eq) = mpsc::channel();
        let (tx_ex, rx_ex) = mpsc::channel();
        let h = Harness::bootstrap(
            vec![
                Box::new(Sma::new(2)),
                Box::new(Sma::new(4)),
                Box::new(Sub::new()),
                Box::new(Exposure::new(0.5)),
                Box::new(SimBroker::new(0.0001)),
                Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx_eq)),
                Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx_ex)),
            ],
            vec![SourceSpec {
                kind: ScalarKind::F64,
                targets: vec![
                    Target { node: 0, slot: 0 },
                    Target { node: 1, slot: 0 },
                    Target { node: 4, slot: 1 },
                ],
            }],
            vec![
                Edge { from: 0, to: 2, slot: 0, from_field: 0 },
                Edge { from: 1, to: 2, slot: 1, from_field: 0 },
                Edge { from: 2, to: 3, slot: 0, from_field: 0 },
                Edge { from: 3, to: 4, slot: 0, from_field: 0 },
                Edge { from: 4, to: 5, slot: 0, from_field: 0 },
                Edge { from: 3, to: 6, slot: 0, from_field: 0 },
            ],
        )
        .expect("valid hand-wired DAG");
        (h, rx_eq, rx_ex)
    }

    /// The SMA-cross signal as a reusable composite: one input role (price), one
    /// output (the fast-minus-slow spread). Interior wired with raw local indices.
    /// Value-empty: the two SMA lengths are injected at compile, not baked here.
    fn sma_cross() -> Composite {
        Composite::new(
            "sma_cross",
            vec![Sma::factory().into(), Sma::factory().into(), Sub::factory().into()],
            vec![
                Edge { from: 0, to: 2, slot: 0, from_field: 0 },
                Edge { from: 1, to: 2, slot: 1, from_field: 0 },
            ],
            vec![Role {
                name: "price".into(),
                targets: vec![Target { node: 0, slot: 0 }, Target { node: 1, slot: 0 }],
            }],
            vec![],
            vec![OutField { node: 2, field: 0, name: "out".into() }],
        )
    }

    /// The same signal-quality harness authored as a composite blueprint.
    #[allow(clippy::type_complexity)]
    fn composite_sma_cross_harness() -> (
        Blueprint,
        mpsc::Receiver<(Timestamp, Vec<Scalar>)>,
        mpsc::Receiver<(Timestamp, Vec<Scalar>)>,
    ) {
        let (tx_eq, rx_eq) = mpsc::channel();
        let (tx_ex, rx_ex) = mpsc::channel();
        let bp = Blueprint::new(
            vec![
                BlueprintNode::Composite(sma_cross()),
                Exposure::factory().into(),
                SimBroker::factory(0.0001).into(),
                Recorder::factory(vec![ScalarKind::F64], Firing::Any, tx_eq).into(),
                Recorder::factory(vec![ScalarKind::F64], Firing::Any, tx_ex).into(),
            ],
            vec![SourceSpec {
                kind: ScalarKind::F64,
                targets: vec![
                    Target { node: 0, slot: 0 }, // price -> sma_cross role 0
                    Target { node: 2, slot: 1 }, // price -> SimBroker price slot
                ],
            }],
            vec![
                Edge { from: 0, to: 1, slot: 0, from_field: 0 }, // composite out -> Exposure
                Edge { from: 1, to: 2, slot: 0, from_field: 0 }, // exposure -> broker slot 0
                Edge { from: 2, to: 3, slot: 0, from_field: 0 }, // equity -> sink
                Edge { from: 1, to: 4, slot: 0, from_field: 0 }, // exposure -> sink
            ],
        );
        (bp, rx_eq, rx_ex)
    }

    #[test]
    fn composite_sma_cross_runs_bit_identical_to_hand_wired() {
        let prices = synthetic_prices();

        // (a) today's flat, hand-wired graph
        let (mut flat, flat_eq, flat_ex) = hand_wired_sma_cross_harness();
        flat.run(vec![prices.clone()]);

        // (b) the same graph authored as a composite blueprint, compiled
        let (bp, comp_eq, comp_ex) = composite_sma_cross_harness();
        let mut composed = bp
            .bootstrap_with_params(vec![Scalar::I64(2), Scalar::I64(4), Scalar::F64(0.5)])
            .expect("composite blueprint compiles");
        composed.run(vec![prices]);

        let flat_eq_v = flat_eq.try_iter().collect::<Vec<_>>();
        let flat_ex_v = flat_ex.try_iter().collect::<Vec<_>>();
        let comp_eq_v = comp_eq.try_iter().collect::<Vec<_>>();
        let comp_ex_v = comp_ex.try_iter().collect::<Vec<_>>();

        // both recording sinks captured the same equity + exposure traces, bit-for-bit
        assert_eq!(flat_eq_v, comp_eq_v, "equity traces differ");
        assert_eq!(flat_ex_v, comp_ex_v, "exposure traces differ");
        // and the trace is populated (non-degenerate), so the equality is meaningful
        assert!(!comp_eq_v.is_empty(), "equity trace must be populated");
        assert!(!comp_ex_v.is_empty(), "exposure trace must be populated");
    }

    /// E2E (cycle 0018): a composite's multi-field output record is selected
    /// field-wise downstream all the way through `bootstrap + run` — two consumers
    /// reading distinct `from_field`s off one multi-output composite record the two
    /// distinct interior producers' streams, deterministically. The Task-2 unit
    /// tests stop at `compile()` (edge resolution); this one runs the harness, so a
    /// regression that resolved both taps to the same producer (or dropped a field)
    /// would surface as identical recorded traces here, not just a bad edge table.
    #[test]
    fn multi_output_composite_taps_distinct_fields_through_a_run() {
        let prices = synthetic_prices();
        let (tx_a, rx_a) = mpsc::channel();
        let (tx_b, rx_b) = mpsc::channel();
        // composite: two SMAs of different lengths, each its own input role; the
        // output record re-exports SMA-fast as "a" (field 0) and SMA-slow as "b"
        // (field 1). One source fans into both roles; two recorders tap the two
        // fields by from_field.
        let c = Composite::new(
            "two_sma",
            vec![Sma::factory().into(), Sma::factory().into()],
            vec![],
            vec![
                Role { name: "price".into(), targets: vec![Target { node: 0, slot: 0 }] },
                Role { name: "price2".into(), targets: vec![Target { node: 1, slot: 0 }] },
            ],
            vec![],
            vec![
                OutField { node: 0, field: 0, name: "a".into() },
                OutField { node: 1, field: 0, name: "b".into() },
            ],
        );
        let bp = Blueprint::new(
            vec![
                BlueprintNode::Composite(c),
                Recorder::factory(vec![ScalarKind::F64], Firing::Any, tx_a).into(),
                Recorder::factory(vec![ScalarKind::F64], Firing::Any, tx_b).into(),
            ],
            vec![SourceSpec {
                kind: ScalarKind::F64,
                targets: vec![Target { node: 0, slot: 0 }, Target { node: 0, slot: 1 }],
            }],
            vec![
                Edge { from: 0, to: 1, slot: 0, from_field: 0 }, // field "a" (SMA-2) -> recorder a
                Edge { from: 0, to: 2, slot: 0, from_field: 1 }, // field "b" (SMA-4) -> recorder b
            ],
        );
        let mut h = bp
            .bootstrap_with_params(vec![Scalar::I64(2), Scalar::I64(4)])
            .expect("multi-output composite bootstraps");
        h.run(vec![prices]);

        let a = rx_a.try_iter().collect::<Vec<_>>();
        let b = rx_b.try_iter().collect::<Vec<_>>();
        // both fields recorded something (the equality below is meaningful only if
        // populated) and the two taps captured different streams — distinct fast vs
        // slow SMA, so the two from_field selections resolve to distinct producers.
        assert!(!a.is_empty() && !b.is_empty(), "both field taps must be populated");
        assert_ne!(a, b, "the two from_field taps must record distinct interior streams");
    }

    #[test]
    fn injecting_a_different_vector_changes_the_run() {
        let prices = synthetic_prices();
        let (bp, eq, _ex) = composite_sma_cross_harness();
        let mut a = bp.bootstrap_with_params(vec![Scalar::I64(2), Scalar::I64(4), Scalar::F64(0.5)])
            .expect("compiles");
        a.run(vec![prices.clone()]);
        let a_eq = eq.try_iter().collect::<Vec<_>>();

        let (bp2, eq2, _ex2) = composite_sma_cross_harness();
        let mut b = bp2.bootstrap_with_params(vec![Scalar::I64(5), Scalar::I64(20), Scalar::F64(1.0)])
            .expect("compiles");
        b.run(vec![prices]);
        let b_eq = eq2.try_iter().collect::<Vec<_>>();

        assert!(!a_eq.is_empty() && !b_eq.is_empty(), "both traces populated");
        assert_ne!(a_eq, b_eq, "a different vector must yield a different run");
    }

    #[test]
    fn wrong_kind_is_a_param_kind_mismatch() {
        let (bp, _eq, _ex) = composite_sma_cross_harness();
        // slot 0 is I64 (an SMA length); inject F64 there
        let err = bp.bootstrap_with_params(vec![Scalar::F64(2.0), Scalar::I64(4), Scalar::F64(0.5)])
            .unwrap_err();
        assert!(matches!(err, CompileError::ParamKindMismatch { slot: 0, .. }));
    }

    #[test]
    fn wrong_arity_is_a_param_arity_error() {
        let (short, _e1, _x1) = composite_sma_cross_harness();
        assert!(matches!(
            short.bootstrap_with_params(vec![Scalar::I64(2)]).unwrap_err(),
            CompileError::ParamArity { expected: 3, got: 1 }
        ));
        let (long, _e2, _x2) = composite_sma_cross_harness();
        assert!(matches!(
            long.bootstrap_with_params(
                vec![Scalar::I64(2), Scalar::I64(4), Scalar::F64(0.5), Scalar::F64(0.0)]
            ).unwrap_err(),
            CompileError::ParamArity { expected: 3, got: 4 }
        ));
    }

    #[test]
    fn same_vector_bootstraps_identically() {
        let prices = synthetic_prices();
        let (bp, eq, _ex) = composite_sma_cross_harness();
        let mut a = bp.bootstrap_with_params(vec![Scalar::I64(3), Scalar::I64(9), Scalar::F64(0.7)])
            .expect("compiles");
        a.run(vec![prices.clone()]);
        let (bp2, eq2, _ex2) = composite_sma_cross_harness();
        let mut b = bp2.bootstrap_with_params(vec![Scalar::I64(3), Scalar::I64(9), Scalar::F64(0.7)])
            .expect("compiles");
        b.run(vec![prices]);
        assert_eq!(eq.try_iter().collect::<Vec<_>>(), eq2.try_iter().collect::<Vec<_>>());
    }

    /// E2E (cycle 0015): the C23/#31 cross-cutting invariant — `param_space()` is a
    /// parallel projection of the *same* traversal `compile` inlines, so a param's
    /// slot in the aggregated space lines up, in order and kind, with the declared
    /// params of the compiled flat nodes (the premise #31's slot-by-slot binding
    /// rests on). Driven on the realistic SMA-cross harness, not a synthetic graph,
    /// and on the blueprint *as compiled* — so a future inliner reorder that
    /// silently desynced the two projections would fail here, not just the isolated
    /// `param_space` order tests.
    #[test]
    fn param_space_mirrors_compiled_flat_node_param_order() {
        let (bp, _rx_eq, _rx_ex) = composite_sma_cross_harness();

        // the aggregated, path-qualified projection
        let space = bp.param_space();

        // the same blueprint, actually compiled to its flat node array; each flat
        // node's own declared params, concatenated in flat-node order
        let (flat_nodes, _sources, _edges) = bp
            .compile_with_params(&[Scalar::I64(2), Scalar::I64(4), Scalar::F64(0.5)])
            .expect("harness compiles");
        let from_compilat: Vec<ParamSpec> =
            flat_nodes.iter().flat_map(|n| n.schema().params).collect();

        // same count, same per-slot kind, same order — the projection mirrors the
        // compilation (names differ: param_space path-qualifies, the raw node does
        // not, so compare on the load-bearing axis, kind-by-slot)
        assert_eq!(space.len(), from_compilat.len(), "param count must match the compilat");
        assert_eq!(
            space.iter().map(|p| p.kind).collect::<Vec<_>>(),
            from_compilat.iter().map(|p| p.kind).collect::<Vec<_>>(),
            "per-slot param kinds must line up with the compiled flat-node order",
        );
        // the realistic harness's concrete space: two SMA lengths (I64) + Exposure
        // scale (F64); Sub/SimBroker/Recorder declare none
        assert_eq!(
            space.iter().map(|p| p.name.as_str()).collect::<Vec<_>>(),
            ["sma_cross.length", "sma_cross.length", "scale"],
        );
        assert_eq!(
            space.iter().map(|p| p.kind).collect::<Vec<_>>(),
            [ScalarKind::I64, ScalarKind::I64, ScalarKind::F64],
        );
    }

    #[test]
    fn param_alias_relabels_param_space_name_in_place() {
        // two Sma leaves (each one `length` param) under a composite that aliases
        // slot 0 of node 0 -> "shortLen" and slot 0 of node 1 -> "longLen".
        let c = Composite::new(
            "cross",
            vec![Sma::factory().into(), Sma::factory().into(), Sub::factory().into()],
            vec![
                Edge { from: 0, to: 2, slot: 0, from_field: 0 },
                Edge { from: 1, to: 2, slot: 1, from_field: 0 },
            ],
            vec![Role {
                name: "price".into(),
                targets: vec![Target { node: 0, slot: 0 }, Target { node: 1, slot: 0 }],
            }],
            vec![
                ParamAlias { name: "shortLen".into(), node: 0, slot: 0 },
                ParamAlias { name: "longLen".into(), node: 1, slot: 0 },
            ],
            vec![OutField { node: 2, field: 0, name: "out".into() }],
        );
        let bp = Blueprint::new(vec![BlueprintNode::Composite(c)], vec![], vec![]);
        let names: Vec<String> = bp.param_space().into_iter().map(|p| p.name).collect();
        // aliased in place: names are the aliases, NOT two duplicate "cross.length".
        assert_eq!(names, vec!["cross.shortLen".to_string(), "cross.longLen".to_string()]);
    }

    #[test]
    fn out_of_range_param_alias_rejected() {
        // alias names node 9 (no such interior item) -> caught at compile.
        let c = Composite::new(
            "cross",
            vec![Sma::factory().into(), Sma::factory().into(), Sub::factory().into()],
            vec![
                Edge { from: 0, to: 2, slot: 0, from_field: 0 },
                Edge { from: 1, to: 2, slot: 1, from_field: 0 },
            ],
            vec![Role {
                name: "price".into(),
                targets: vec![Target { node: 0, slot: 0 }, Target { node: 1, slot: 0 }],
            }],
            vec![ParamAlias { name: "bogus".into(), node: 9, slot: 0 }],
            vec![OutField { node: 2, field: 0, name: "out".into() }],
        );
        let bp = Blueprint::new(
            vec![BlueprintNode::Composite(c)],
            vec![SourceSpec { kind: ScalarKind::F64, targets: vec![] }],
            vec![],
        );
        // two Sma leaves => two i64 length slots; supply a matching vector so the
        // ONLY error is the bad alias, not arity. (The Ok arm holds Box<dyn Node>,
        // not Debug, so assert via the Err arm — as the other reject tests do.)
        assert_eq!(
            bp.compile_with_params(&[Scalar::I64(2), Scalar::I64(4)]).err(),
            Some(CompileError::BadInteriorIndex),
        );
    }

    #[test]
    fn unaliased_params_keep_factory_names() {
        // no aliases => param_space identical to the pre-#41 path-qualified names.
        let c = Composite::new(
            "cross",
            vec![Sma::factory().into(), Sma::factory().into(), Sub::factory().into()],
            vec![
                Edge { from: 0, to: 2, slot: 0, from_field: 0 },
                Edge { from: 1, to: 2, slot: 1, from_field: 0 },
            ],
            vec![Role {
                name: "price".into(),
                targets: vec![Target { node: 0, slot: 0 }, Target { node: 1, slot: 0 }],
            }],
            vec![],
            vec![OutField { node: 2, field: 0, name: "out".into() }],
        );
        let bp = Blueprint::new(vec![BlueprintNode::Composite(c)], vec![], vec![]);
        let names: Vec<String> = bp.param_space().into_iter().map(|p| p.name).collect();
        assert_eq!(names, vec!["cross.length".to_string(), "cross.length".to_string()]);
    }

    #[test]
    fn partial_aliasing_relabels_only_the_named_slot() {
        // alias node 0 only; node 1 keeps its factory name; order intact.
        let c = Composite::new(
            "cross",
            vec![Sma::factory().into(), Sma::factory().into(), Sub::factory().into()],
            vec![
                Edge { from: 0, to: 2, slot: 0, from_field: 0 },
                Edge { from: 1, to: 2, slot: 1, from_field: 0 },
            ],
            vec![Role {
                name: "price".into(),
                targets: vec![Target { node: 0, slot: 0 }, Target { node: 1, slot: 0 }],
            }],
            vec![ParamAlias { name: "shortLen".into(), node: 0, slot: 0 }],
            vec![OutField { node: 2, field: 0, name: "out".into() }],
        );
        let bp = Blueprint::new(vec![BlueprintNode::Composite(c)], vec![], vec![]);
        let names: Vec<String> = bp.param_space().into_iter().map(|p| p.name).collect();
        assert_eq!(names, vec!["cross.shortLen".to_string(), "cross.length".to_string()]);
    }

    #[test]
    fn param_space_is_flat_path_qualified_and_slot_disambiguated() {
        use aura_std::{LinComb, Sma, Sub};
        // inner composite "fast_slow": two SMAs (same type → same param name) + a Sub
        let fast_slow = Composite::new(
            "fast_slow",
            vec![Sma::factory().into(), Sma::factory().into(), Sub::factory().into()],
            vec![
                Edge { from: 0, to: 2, slot: 0, from_field: 0 },
                Edge { from: 1, to: 2, slot: 1, from_field: 0 },
            ],
            vec![Role {
                name: "price".into(),
                targets: vec![Target { node: 0, slot: 0 }, Target { node: 1, slot: 0 }],
            }],
            vec![],
            vec![OutField { node: 2, field: 0, name: "out".into() }],
        );
        // outer composite "strategy": the inner composite + a LinComb([1,-1])
        let strategy = Composite::new(
            "strategy",
            vec![BlueprintNode::Composite(fast_slow), LinComb::factory(2).into()],
            vec![],
            vec![Role { name: "price".into(), targets: vec![Target { node: 0, slot: 0 }] }],
            vec![],
            vec![OutField { node: 0, field: 0, name: "out".into() }],
        );
        let bp = Blueprint::new(vec![BlueprintNode::Composite(strategy)], vec![], vec![]);

        let space = bp.param_space();
        let names: Vec<&str> = space.iter().map(|p| p.name.as_str()).collect();
        assert_eq!(
            names,
            [
                "strategy.fast_slow.length", // slot 0 — Sma(2)
                "strategy.fast_slow.length", // slot 1 — Sma(4): same name, distinct slot
                "strategy.weights[0]",       // slot 2 — LinComb weight 0
                "strategy.weights[1]",       // slot 3 — LinComb weight 1
            ]
        );
        assert_eq!(space[0].kind, ScalarKind::I64);
        assert_eq!(space[2].kind, ScalarKind::F64);
    }

    /// E2E (issue #34): the C23/#31 mirror invariant *under composite nesting*.
    /// `param_space()` (via `collect_params`) duplicates `lower_items`' depth-
    /// first traversal rather than sharing it, so the two orders must stay in
    /// lockstep. The single-level mirror test
    /// (`param_space_mirrors_compiled_flat_node_param_order`) never compiles a
    /// composite whose interior holds *another* composite; the nested
    /// `param_space` order test never compiles. This closes that gap: it
    /// compiles a `strategy → { fast_slow → [Sma, Sma, Sub], LinComb }` nest and
    /// asserts the aggregated space lines up, kind-by-slot, with the compiled
    /// flat-node param order — so a future inliner reorder that desynced the two
    /// projections *only under nesting* would fail here, not slip through.
    #[test]
    fn param_space_mirrors_compiled_flat_node_param_order_under_nesting() {
        use aura_std::{LinComb, Sma, Sub};
        // inner composite "fast_slow": two SMAs + a Sub (same nest as the
        // isolated path-qualification test above)
        let fast_slow = Composite::new(
            "fast_slow",
            vec![Sma::factory().into(), Sma::factory().into(), Sub::factory().into()],
            vec![
                Edge { from: 0, to: 2, slot: 0, from_field: 0 },
                Edge { from: 1, to: 2, slot: 1, from_field: 0 },
            ],
            vec![Role {
                name: "price".into(),
                targets: vec![Target { node: 0, slot: 0 }, Target { node: 1, slot: 0 }],
            }],
            vec![],
            vec![OutField { node: 2, field: 0, name: "out".into() }],
        );
        // outer composite "strategy": the inner composite + a LinComb([1,-1])
        let strategy = Composite::new(
            "strategy",
            vec![BlueprintNode::Composite(fast_slow), LinComb::factory(2).into()],
            vec![],
            vec![Role { name: "price".into(), targets: vec![Target { node: 0, slot: 0 }] }],
            vec![],
            vec![OutField { node: 0, field: 0, name: "out".into() }],
        );
        let bp = Blueprint::new(vec![BlueprintNode::Composite(strategy)], vec![], vec![]);

        // the aggregated, path-qualified projection (borrows; take it first since
        // compile() consumes self — same ordering as the single-level mirror test)
        let space = bp.param_space();

        // the same blueprint, compiled to its flat node array; each flat node's
        // own declared params, concatenated in flat-node order
        let (flat_nodes, _sources, _edges) = bp
            .compile_with_params(&[Scalar::I64(2), Scalar::I64(4), Scalar::F64(1.0), Scalar::F64(-1.0)])
            .expect("nested composite compiles");
        let from_compilat: Vec<ParamSpec> =
            flat_nodes.iter().flat_map(|n| n.schema().params).collect();

        // same count, same per-slot kind, same order — the nested projection
        // mirrors the compilation (names differ: param_space path-qualifies, the
        // raw node does not, so compare on the load-bearing axis, kind-by-slot)
        assert_eq!(space.len(), from_compilat.len(), "param count must match the compilat");
        assert_eq!(
            space.iter().map(|p| p.kind).collect::<Vec<_>>(),
            from_compilat.iter().map(|p| p.kind).collect::<Vec<_>>(),
            "per-slot param kinds must line up with the compiled flat-node order, under nesting",
        );
        // pin the concrete shape: two Sma lengths (I64), Sub none, two LinComb
        // weights (F64)
        assert_eq!(
            space.iter().map(|p| p.kind).collect::<Vec<_>>(),
            [ScalarKind::I64, ScalarKind::I64, ScalarKind::F64, ScalarKind::F64],
        );
    }

    #[test]
    fn top_level_leaf_params_are_unqualified() {
        use aura_std::Sma;
        let bp = Blueprint::new(vec![Sma::factory().into()], vec![], vec![]);
        let space = bp.param_space();
        assert_eq!(space.len(), 1);
        assert_eq!(space[0].name, "length"); // no path prefix at the top level
    }

    #[test]
    fn param_space_is_deterministic() {
        use aura_std::{LinComb, Sma};
        let bp = Blueprint::new(
            vec![Sma::factory().into(), LinComb::factory(2).into()],
            vec![],
            vec![],
        );
        assert_eq!(bp.param_space(), bp.param_space()); // pure structural function (C1)
    }

    #[test]
    fn param_space_empty_for_paramless_and_empty_blueprints() {
        use aura_std::{Add, Sub};
        let only_paramless =
            Blueprint::new(vec![Sub::factory().into(), Add::factory().into()], vec![], vec![]);
        assert!(only_paramless.param_space().is_empty());
        let empty = Blueprint::new(vec![], vec![], vec![]);
        assert!(empty.param_space().is_empty());
    }
}