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

//! The construction layer (C9/C19/C23): a named, param-generic graph-as-data
//! ([`Composite`]) that **compiles** to the flat, type-erased instance the run loop
//! already runs (a [`crate::FlatGraph`]). The unit of reuse is the
//! [`Composite`]: a nestable sub-graph fragment exposing an output record (one port,
//! K re-exported fields; C8) and input roles (each open, or — at the root —
//! source-bound). `compile` **inlines** the nesting into the flat `FlatGraph` the
//! unchanged [`crate::Harness::bootstrap`] consumes; the root composite IS the
//! blueprint (there is no separate `Blueprint` type — a fully source-bound composite
//! is the runnable root).
//!
//! The flat graph 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::{
    FieldSpec, Firing, Node, NodeSchema, ParamSpec, PortSpec, PrimitiveBuilder, Scalar, ScalarKind,
};

use crate::harness::{BootstrapError, Edge, FlatGraph, Harness, SourceSpec, Target};
use crate::sweep::sweep;
use crate::{GridSpace, RunReport, SweepFamily};

/// 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 flat graph.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct OutField {
    pub node: usize,
    pub field: usize,
    pub name: String,
}

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

/// Ergonomic lift: a param-generic primitive recipe becomes a `Primitive` blueprint item.
impl From<PrimitiveBuilder> for BlueprintNode {
    fn from(builder: PrimitiveBuilder) -> Self {
        BlueprintNode::Primitive(builder)
    }
}

/// Ergonomic lift: a nested composite becomes a `Composite` blueprint item, so a
/// builder's `add` can accept a sub-graph the same way it accepts a primitive.
impl From<Composite> for BlueprintNode {
    fn from(c: Composite) -> Self {
        BlueprintNode::Composite(c)
    }
}

impl BlueprintNode {
    /// The node's declared signature, pre-build, uniform across both arms — a
    /// primitive returns its builder's declared schema; a composite derives it from
    /// its interior. This is "every node has a signature in the blueprint".
    pub fn signature(&self) -> NodeSchema {
        match self {
            BlueprintNode::Primitive(b) => b.schema().clone(),
            BlueprintNode::Composite(c) => derive_signature(c),
        }
    }
}

/// Derive a composite's signature from its interior (no build): one input port per
/// input role (kind = the role's interior target slot kind; firing is a non-load-
/// bearing `Any` placeholder — a composite's ports dissolve at inline, only the kind
/// is consulted by an enclosing graph's wiring check), one output field per
/// re-exported `OutField` (kind = the interior producer's field kind), and the
/// aggregated param-space.
fn derive_signature(c: &Composite) -> NodeSchema {
    let inputs = c
        .input_roles()
        .iter()
        .map(|role| {
            let kind = role
                .targets
                .first()
                .map(|t| interior_slot_kind(c.nodes(), c.edges(), t))
                .unwrap_or(ScalarKind::F64);
            PortSpec { kind, firing: Firing::Any, name: role.name.clone() }
        })
        .collect();
    let output = c
        .output()
        .iter()
        .map(|of| {
            // bounds-total: a structurally-invalid OutField (out-of-range node/field)
            // yields a placeholder kind, never a panic — the real fault is reported by
            // validate_wiring's guarded output check (OutputPortOutOfRange). signature()
            // is computed speculatively by the parent's edge/role/connectivity checks
            // before the recursion validates this composite's interior (cycle 0040).
            let kind = c
                .nodes()
                .get(of.node)
                .and_then(|n| n.signature().output.get(of.field).map(|f| f.kind))
                .unwrap_or(ScalarKind::F64);
            FieldSpec { name: leak_name(&of.name), kind }
        })
        .collect();
    let mut params = Vec::new();
    collect_params(c.nodes(), "", &mut params);
    NodeSchema { inputs, output, params }
}

/// The scalar kind of the interior input slot a composite target addresses,
/// resolving one level (a target into a nested composite reads that composite's
/// derived input-port kind).
fn interior_slot_kind(nodes: &[BlueprintNode], _edges: &[Edge], t: &Target) -> ScalarKind {
    // bounds-total: a target into a missing node/slot yields a placeholder kind, never a
    // panic — the real fault is reported by validate_wiring's guarded index checks.
    nodes
        .get(t.node)
        .and_then(|n| n.signature().inputs.get(t.slot).map(|p| p.kind))
        .unwrap_or(ScalarKind::F64)
}

/// Leak a boundary name into a `&'static str` for a derived `FieldSpec`. Acceptable
/// because `derive_signature` is a cold, pre-build, render/validation path (never the
/// hot loop), and the leaked names are bounded by the static blueprint.
fn leak_name(s: &str) -> &'static str {
    Box::leak(s.to_string().into_boxed_str())
}

/// 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>,
    /// `None` = an open interior port (wired by the enclosing graph's edges);
    /// `Some(kind)` = a bound ingestion feed of `kind` (only meaningful at the root,
    /// where it lowers to a `FlatGraph` source). C3: sources bind at ingestion only.
    pub source: Option<ScalarKind>,
}

/// 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>,
    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 flat graph (the boundary dissolves at inline, C23).
    pub fn new(
        name: impl Into<String>,
        nodes: Vec<BlueprintNode>,
        edges: Vec<Edge>,
        input_roles: Vec<Role>,
        output: Vec<OutField>,
    ) -> Self {
        Self { name: name.into(), nodes, edges, input_roles, 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 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
    }

    /// The aggregated, flat, path-qualified param-space (C12): every node's declared
    /// params, concatenated in lowering order. Each param is `<node>.<param>` (the
    /// node name = its instance name, default = lowercased type label), with the
    /// composite path prefixed at every level including the root — so a top-level
    /// leaf carries its own node segment (e.g. `sma.length`). Interior composite
    /// names prefix via the recursion in `collect_params`.
    pub fn param_space(&self) -> Vec<ParamSpec> {
        let mut out = Vec::new();
        collect_params(&self.nodes, "", &mut out);
        out
    }

    /// Compile this composite as the ROOT graph under an injected param vector:
    /// validate structurally pre-build (via `signature()`, no node built), require
    /// every root role bound, then lower (build each primitive, gather its signature,
    /// inline composites, rewrite edges, lower bound roles to flat sources).
    pub fn compile_with_params(self, params: &[Scalar]) -> Result<FlatGraph, CompileError> {
        // structural validation, all pre-build (no node constructed):
        let space = self.param_space();
        check_param_namespace_injective(&space)?;
        validate_wiring(&self.nodes, &self.edges, &self.input_roles, &self.output)?;
        for (r, role) in self.input_roles.iter().enumerate() {
            if role.source.is_none() {
                return Err(CompileError::UnboundRootRole { role: r });
            }
        }

        if params.len() != space.len() {
            return Err(CompileError::ParamArity { expected: space.len(), got: params.len() });
        }

        let mut flat_nodes: Vec<Box<dyn Node>> = Vec::new();
        let mut flat_signatures: Vec<NodeSchema> = Vec::new();
        let mut flat_edges: Vec<Edge> = Vec::new();
        let mut cursor = 0usize;

        let lowerings = lower_items(
            self.nodes,
            params,
            &mut cursor,
            &mut flat_nodes,
            &mut flat_signatures,
            &mut flat_edges,
        )?;

        for e in &self.edges {
            for fe in rewrite_edge(e, &lowerings, &flat_signatures)? {
                flat_edges.push(fe);
            }
        }

        // each bound root role lowers to a flat source, in role-declaration order
        let mut flat_sources: Vec<SourceSpec> = Vec::with_capacity(self.input_roles.len());
        for role in &self.input_roles {
            let kind = role.source.expect("root role bound (checked above)");
            let mut targets: Vec<Target> = Vec::new();
            for t in &role.targets {
                targets.extend(resolve_target(t, &lowerings)?);
            }
            flat_sources.push(SourceSpec { kind, targets });
        }

        Ok(FlatGraph { nodes: flat_nodes, signatures: flat_signatures, sources: flat_sources, edges: flat_edges })
    }

    /// No-param compile (errors `ParamArity` if any param is declared).
    pub fn compile(self) -> Result<FlatGraph, CompileError> {
        self.compile_with_params(&[])
    }

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

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

    /// Begin binding this blueprint's knobs **by name** for a single run (the
    /// fluent alternative to a positional `bootstrap_with_params` vector). The
    /// bound name is the exact `param_space()` name — `<node>.<param>` at every
    /// level, e.g. `sma_cross.fast.length` for a composite-interior knob and
    /// `exposure.scale` for a root-level knob.
    pub fn with(self, name: &str, v: impl Into<Scalar>) -> Binder {
        Binder { bp: self, bound: vec![(name.to_string(), v.into())] }
    }

    /// Begin binding this blueprint's knobs **by name** as sweep axes (the fluent
    /// authoring alternative to a positional `GridSpace`). The bound name is the
    /// exact `param_space()` name (path-qualified for a composite-interior knob,
    /// bare for a root-level knob).
    pub fn axis(self, name: &str, vals: impl IntoIterator<Item = impl Into<Scalar>>) -> SweepBinder {
        let axis = vals.into_iter().map(Into::into).collect();
        SweepBinder { bp: self, axes: vec![(name.to_string(), axis)] }
    }
}

/// A fault in resolving a named binding against `param_space()` — the authoring
/// layer over `bootstrap_with_params` / `sweep`. Name-qualified: each message
/// names the offending knob, not a slot index. The total error order is documented
/// on the resolution path; the first failing check wins.
#[derive(Debug, PartialEq)]
pub enum BindError {
    /// A bound name matches no `param_space()` slot's exact name.
    UnknownKnob(String),
    /// A `param_space()` slot was left unbound.
    MissingKnob(String),
    /// A bound value's kind does not equal the slot's declared kind.
    KindMismatch { knob: String, expected: ScalarKind, got: ScalarKind },
    /// The same name was bound twice.
    DuplicateBinding(String),
    /// (sweep, iteration 2) An axis was given zero values.
    EmptyAxis(String),
    /// The resolved point passed name resolution but failed downstream bootstrap.
    Compile(CompileError),
}

/// A fluent accumulator of named knob bindings for a single run, terminated by
/// [`Binder::bootstrap`]. Bindings are resolved against `param_space()` once, at
/// the terminal.
pub struct Binder {
    bp: Composite,
    bound: Vec<(String, Scalar)>,
}

impl Binder {
    /// Bind one more knob by name; raw literals lower via `Into<Scalar>`.
    pub fn with(mut self, name: &str, v: impl Into<Scalar>) -> Binder {
        self.bound.push((name.to_string(), v.into()));
        self
    }

    /// Resolve the accumulated bindings against `param_space()` and bootstrap.
    pub fn bootstrap(self) -> Result<Harness, BindError> {
        let space = self.bp.param_space();
        check_param_namespace_injective(&space).map_err(BindError::Compile)?;
        let point = resolve(&space, &self.bound)?;
        self.bp.bootstrap_with_params(point).map_err(BindError::Compile)
    }
}

/// A fluent accumulator of named sweep axes, terminated by [`SweepBinder::sweep`].
/// Axes are resolved against `param_space()` once, at the terminal; resolution is a
/// superset of `GridSpace::new`'s checks, so the grid it builds cannot fail.
pub struct SweepBinder {
    bp: Composite,
    axes: Vec<(String, Vec<Scalar>)>,
}

impl SweepBinder {
    /// Bind one more sweep axis by name; raw literals lower via `Into<Scalar>`.
    pub fn axis(mut self, name: &str, vals: impl IntoIterator<Item = impl Into<Scalar>>) -> SweepBinder {
        self.axes.push((name.to_string(), vals.into_iter().map(Into::into).collect()));
        self
    }

    /// Resolve the named axes against `param_space()` into a positional grid and
    /// run the disjoint sweep.
    pub fn sweep<F>(self, run_one: F) -> Result<SweepFamily, BindError>
    where
        F: Fn(&[Scalar]) -> RunReport + Sync,
    {
        let space = self.bp.param_space();
        check_param_namespace_injective(&space).map_err(BindError::Compile)?;
        let ordered = resolve_axes(&space, &self.axes)?;
        let grid = GridSpace::new(&space, ordered)
            .expect("named layer pre-validates arity/kind/non-empty");
        Ok(sweep(&grid, run_one))
    }
}

/// Resolve named sweep axes to a positional `Vec<Vec<Scalar>>` in `param_space()`
/// slot order. (Body filled in Step 3.)
fn resolve_axes(space: &[ParamSpec], axes: &[(String, Vec<Scalar>)]) -> Result<Vec<Vec<Scalar>>, BindError> {
    // Phase 1 — per axis in input order: claim slots by exact name (a, b, c, d).
    let mut claimed: Vec<Option<Vec<Scalar>>> = vec![None; space.len()];
    for (name, values) in axes {
        let matches: Vec<usize> = space
            .iter()
            .enumerate()
            .filter(|(_, p)| p.name == *name)
            .map(|(i, _)| i)
            .collect();
        match matches.as_slice() {
            [] => return Err(BindError::UnknownKnob(name.clone())), // a
            [idx] => {
                if values.is_empty() {
                    return Err(BindError::EmptyAxis(name.clone())); // c
                }
                if claimed[*idx].is_some() {
                    return Err(BindError::DuplicateBinding(name.clone())); // d
                }
                claimed[*idx] = Some(values.clone());
            }
            _ => unreachable!("param_space() is injective — checked before resolve"),
        }
    }
    // Phase 2 — slot walk in param_space() order: completeness (e) + per-element kind (f).
    let mut ordered = Vec::with_capacity(space.len());
    for (i, p) in space.iter().enumerate() {
        match &claimed[i] {
            None => return Err(BindError::MissingKnob(p.name.clone())), // e
            Some(values) => {
                for v in values {
                    if v.kind() != p.kind {
                        return Err(BindError::KindMismatch {
                            knob: p.name.clone(),
                            expected: p.kind,
                            got: v.kind(),
                        }); // f — first offending element in axis order
                    }
                }
                ordered.push(values.clone());
            }
        }
    }
    Ok(ordered)
}

/// Structural validation (param-value-independent): the `param_space()` name
/// projection is the by-name knob address space (C12/C19) and must be injective —
/// a duplicated path is a knob no binding can select alone. The first duplicate in
/// `param_space()` order is reported (the order is deterministic). Single source of
/// duplicate detection; called from `compile_with_params` and from both binders
/// before name resolution.
fn check_param_namespace_injective(space: &[ParamSpec]) -> Result<(), CompileError> {
    let mut seen = std::collections::HashSet::new();
    for p in space {
        if !seen.insert(p.name.as_str()) {
            return Err(CompileError::DuplicateParamPath(p.name.clone()));
        }
    }
    Ok(())
}

/// Resolve named bindings to a positional `Vec<Scalar>` in `param_space()` slot
/// order. (Body filled in Step 3.)
fn resolve(space: &[ParamSpec], bound: &[(String, Scalar)]) -> Result<Vec<Scalar>, BindError> {
    // Phase 1 — per binding in input order: claim slots by exact name (a, b, d).
    let mut claimed: Vec<Option<Scalar>> = vec![None; space.len()];
    for (name, value) in bound {
        let matches: Vec<usize> = space
            .iter()
            .enumerate()
            .filter(|(_, p)| p.name == *name)
            .map(|(i, _)| i)
            .collect();
        match matches.as_slice() {
            [] => return Err(BindError::UnknownKnob(name.clone())), // a
            [idx] => {
                if claimed[*idx].is_some() {
                    return Err(BindError::DuplicateBinding(name.clone())); // d
                }
                claimed[*idx] = Some(*value);
            }
            _ => unreachable!("param_space() is injective — checked before resolve"),
        }
    }
    // Phase 2 — slot walk in param_space() order (e, f).
    let mut point = Vec::with_capacity(space.len());
    for (i, p) in space.iter().enumerate() {
        match claimed[i] {
            None => return Err(BindError::MissingKnob(p.name.clone())), // e
            Some(v) => {
                if v.kind() != p.kind {
                    return Err(BindError::KindMismatch {
                        knob: p.name.clone(),
                        expected: p.kind,
                        got: v.kind(),
                    }); // f
                }
                point.push(v);
            }
        }
    }
    Ok(point)
}

/// A construction-phase fault, caught before the flat graph 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 graph 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 },
    /// Two `param_space()` slots resolved to the same path-qualified name — the
    /// by-name knob address space (C12/C19) is not injective, so no binding can
    /// select one slot without the other. Carries the duplicated path. Cure: give
    /// the colliding same-type sibling nodes distinct names with `.named(...)`.
    DuplicateParamPath(String),
    /// A root input role `role` has no bound source (`source: None`) — an open port
    /// at the root, which has no enclosing graph to wire it. Only a fully source-
    /// bound composite is runnable.
    UnboundRootRole { role: usize },
    /// An interior node's input `slot` is covered by no edge and no role target — a
    /// required port left unconnected (it would bootstrap a silent empty column).
    UnconnectedPort { node: usize, slot: usize },
    /// An interior node's input `slot` is covered by more than one edge/role target
    /// combined — a slot holds exactly one column, so >1 producer is ill-formed.
    DoubleWiredPort { node: usize, slot: usize },
}

/// Pre-build structural validation via `signature()` (no node constructed): every
/// edge's producer field and consumer slot are in range and kind-matched; every
/// output re-export and role target is in range and kind-consistent. Recurses into
/// nested composites so the checks hold at every level. This is what lets `compile`
/// reject a wiring fault before any build closure fires.
fn validate_wiring(
    nodes: &[BlueprintNode],
    edges: &[Edge],
    roles: &[Role],
    output: &[OutField],
) -> Result<(), CompileError> {
    // edges: index-range + producer/consumer kind match. The kind-mismatch variant
    // is the SAME one bootstrap returns today (Bootstrap(KindMismatch)), just raised
    // pre-build — so existing tests asserting that variant for a compiled graph stay
    // green, while the fault is now caught before any build closure fires.
    for e in edges {
        let from = nodes.get(e.from).ok_or(CompileError::BadInteriorIndex)?.signature();
        let to = nodes.get(e.to).ok_or(CompileError::BadInteriorIndex)?.signature();
        let f = from.output.get(e.from_field).ok_or(CompileError::BadInteriorIndex)?;
        let s = to.inputs.get(e.slot).ok_or(CompileError::BadInteriorIndex)?;
        if f.kind != s.kind {
            return Err(CompileError::Bootstrap(BootstrapError::KindMismatch {
                producer: f.kind,
                consumer: s.kind,
            }));
        }
    }
    // roles: every target in range, and all targets of one role share a kind
    // (RoleKindMismatch — the existing variant, today read off built schema()).
    for (r, role) in roles.iter().enumerate() {
        let mut role_kind: Option<ScalarKind> = None;
        for t in &role.targets {
            let sig = nodes.get(t.node).ok_or(CompileError::BadInteriorIndex)?.signature();
            let k = sig.inputs.get(t.slot).ok_or(CompileError::BadInteriorIndex)?.kind;
            match role_kind {
                None => role_kind = Some(k),
                Some(k0) if k0 != k => return Err(CompileError::RoleKindMismatch { role: r }),
                Some(_) => {}
            }
        }
    }
    // outputs: each re-export's field index in range
    for of in output {
        let sig = nodes.get(of.node).ok_or(CompileError::OutputPortOutOfRange)?.signature();
        if of.field >= sig.output.len() {
            return Err(CompileError::OutputPortOutOfRange);
        }
    }
    // wiring totality: every interior input slot covered by exactly one wiring act
    check_ports_connected(nodes, edges, roles)?;
    // recurse into nested composites
    for item in nodes {
        if let BlueprintNode::Composite(c) = item {
            validate_wiring(c.nodes(), c.edges(), c.input_roles(), c.output())?;
        }
    }
    Ok(())
}

/// Every interior node's every declared input slot must be covered by exactly one
/// wiring act — one interior edge OR one role target, counted uniformly. Zero = a
/// forgotten connection (would bootstrap a silent empty column); >1 = an ill-formed
/// slot (a slot holds one column). Index-based, name-free; presupposes in-range
/// edge/role indices (runs after the existing index-range checks). Mirrors the
/// single-site shape of `check_param_namespace_injective` (the wiring-side sibling).
fn check_ports_connected(
    nodes: &[BlueprintNode],
    edges: &[Edge],
    roles: &[Role],
) -> Result<(), CompileError> {
    let mut coverage: std::collections::HashMap<(usize, usize), usize> =
        std::collections::HashMap::new();
    for e in edges {
        *coverage.entry((e.to, e.slot)).or_insert(0) += 1;
    }
    for role in roles {
        for t in &role.targets {
            *coverage.entry((t.node, t.slot)).or_insert(0) += 1;
        }
    }
    for (n, item) in nodes.iter().enumerate() {
        for slot in 0..item.signature().inputs.len() {
            match coverage.get(&(n, slot)).copied().unwrap_or(0) {
                1 => {}
                0 => return Err(CompileError::UnconnectedPort { node: n, slot }),
                _ => return Err(CompileError::DoubleWiredPort { node: n, slot }),
            }
        }
    }
    Ok(())
}

/// Recursive read-only walk for `Blueprint::param_space`: a leaf contributes its
/// declared params under `<prefix>.<node-name>.<param>` (the node name is the
/// instance name, default = the lowercased type label); a composite pushes its
/// `name()` onto the path and recurses. 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, out: &mut Vec<ParamSpec>) {
    for item in items {
        match item {
            BlueprintNode::Primitive(b) => {
                let node = if prefix.is_empty() {
                    b.node_name()
                } else {
                    format!("{prefix}.{}", b.node_name())
                };
                for p in b.params() {
                    out.push(ParamSpec { name: format!("{node}.{}", p.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, out);
            }
        }
    }
}

/// How one blueprint item resolved into the flat graph. 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_signatures: &mut Vec<NodeSchema>,
    flat_edges: &mut Vec<Edge>,
) -> Result<Vec<ItemLowering>, CompileError> {
    let mut lowerings = Vec::with_capacity(items.len());
    for item in items {
        match item {
            BlueprintNode::Primitive(builder) => {
                let n = builder.params().len();
                let slice = &params[*cursor..*cursor + n]; // in range: arity checked up front
                for (i, spec) in builder.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_signatures.push(builder.schema().clone());
                flat_nodes.push(builder.build(slice));
                *cursor += n;
                lowerings.push(ItemLowering::Leaf { index });
            }
            BlueprintNode::Composite(c) => {
                lowerings.push(inline_composite(
                    c,
                    params,
                    cursor,
                    flat_nodes,
                    flat_signatures,
                    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_signatures: &mut Vec<NodeSchema>,
    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 flat graph), so it is not destructured.
    // Node names join the same non-load-bearing debug-symbol class: they qualify the
    // param-space path at construction but are dropped at lowering — the injected
    // scalar `params: &[Scalar]` arg drives `lower_items` below; the flat graph stays
    // wired by raw index.
    let Composite { name: _, nodes, edges, input_roles, output } = c;
    let item_count = nodes.len();

    // recursively lower interior items, then rewrite interior edges through them
    let interior = lower_items(nodes, params, cursor, flat_nodes, flat_signatures, flat_edges)?;
    for e in &edges {
        for fe in rewrite_edge(e, &interior, flat_signatures)? {
            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_signatures[*index].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_signatures)?;
            for ft in rest {
                if slot_kind(*ft, flat_signatures)? != 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_signatures: &[NodeSchema],
) -> 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_signatures[*index].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_signatures: &[NodeSchema]) -> Result<ScalarKind, CompileError> {
    flat_signatures[t.node]
        .inputs
        .get(t.slot)
        .map(|spec| spec.kind)
        .ok_or(CompileError::BadInteriorIndex)
}

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

    #[test]
    fn resolve_axes_named_equals_positional() {
        // named axes resolve to the positional Vec<Vec<Scalar>> in slot order,
        // order-independent on the binding side
        let space = vec![
            ParamSpec { name: "sma_cross.fast".into(), kind: ScalarKind::I64 },
            ParamSpec { name: "sma_cross.slow".into(), kind: ScalarKind::I64 },
            ParamSpec { name: "scale".into(), kind: ScalarKind::F64 },
        ];
        let axes = vec![
            ("scale".to_string(), vec![Scalar::F64(0.5)]),
            ("sma_cross.fast".to_string(), vec![Scalar::I64(2), Scalar::I64(3)]),
            ("sma_cross.slow".to_string(), vec![Scalar::I64(4), Scalar::I64(5)]),
        ];
        assert_eq!(
            resolve_axes(&space, &axes),
            Ok(vec![
                vec![Scalar::I64(2), Scalar::I64(3)],
                vec![Scalar::I64(4), Scalar::I64(5)],
                vec![Scalar::F64(0.5)],
            ]),
        );
    }

    #[test]
    fn resolve_axes_empty_axis() {
        let space = vec![ParamSpec { name: "a".into(), kind: ScalarKind::I64 }];
        assert_eq!(
            resolve_axes(&space, &[("a".to_string(), vec![])]),
            Err(BindError::EmptyAxis("a".to_string())),
        );
    }

    #[test]
    fn resolve_axes_missing_knob() {
        let space = vec![
            ParamSpec { name: "a".into(), kind: ScalarKind::I64 },
            ParamSpec { name: "b".into(), kind: ScalarKind::I64 },
        ];
        assert_eq!(
            resolve_axes(&space, &[("a".to_string(), vec![Scalar::I64(1)])]),
            Err(BindError::MissingKnob("b".to_string())),
        );
    }

    #[test]
    fn resolve_axes_per_element_kind_mismatch() {
        // a MIXED-kind axis: the second element mismatches; per-element check must
        // catch it (a first-element-only check would pass it through to a panic).
        let space = vec![ParamSpec { name: "scale".into(), kind: ScalarKind::F64 }];
        let axes = vec![("scale".to_string(), vec![Scalar::F64(0.5), Scalar::I64(1)])];
        assert_eq!(
            resolve_axes(&space, &axes),
            Err(BindError::KindMismatch {
                knob: "scale".to_string(),
                expected: ScalarKind::F64,
                got: ScalarKind::I64,
            }),
        );
    }

    #[test]
    fn named_sweep_rejects_wrong_kind_axis_without_panic() {
        // builder path: an F64 axis bound to an I64 slot is rejected as a clean
        // BindError, never reaching GridSpace::new's .expect() — the run closure
        // must not be invoked.
        let (bp, _eq, _ex) = composite_sma_cross_harness();
        let result = bp
            .axis("sma_cross.fast.length", [2.0, 3.0]) // F64 values for the I64 slot
            .axis("sma_cross.slow.length", [4])
            .axis("exposure.scale", [0.5])
            .sweep(|_: &[Scalar]| -> RunReport { panic!("axis pre-validation must reject before running") });
        assert_eq!(
            result,
            Err(BindError::KindMismatch {
                knob: "sma_cross.fast.length".to_string(),
                expected: ScalarKind::I64,
                got: ScalarKind::F64,
            }),
        );
    }

    #[test]
    fn named_axes_grid_parity_with_positional() {
        // named axes enumerate the SAME GridSpace points (same order) as the
        // positional grid over the sample param-space.
        let space = composite_sma_cross_harness().0.param_space();
        let named = resolve_axes(
            &space,
            &[
                ("sma_cross.fast.length".to_string(), vec![Scalar::I64(2), Scalar::I64(3)]),
                ("sma_cross.slow.length".to_string(), vec![Scalar::I64(4), Scalar::I64(5)]),
                ("exposure.scale".to_string(), vec![Scalar::F64(0.5)]),
            ],
        )
        .expect("named axes resolve");
        let positional = vec![
            vec![Scalar::I64(2), Scalar::I64(3)],
            vec![Scalar::I64(4), Scalar::I64(5)],
            vec![Scalar::F64(0.5)],
        ];
        let named_pts = GridSpace::new(&space, named).expect("named grid").points();
        let pos_pts = GridSpace::new(&space, positional).expect("positional grid").points();
        assert_eq!(named_pts, pos_pts, "named and positional grids must enumerate identically");
    }

    #[test]
    fn resolve_named_equals_positional_vector() {
        // order-independent: shuffled bindings resolve to slot order
        let space = vec![
            ParamSpec { name: "sma_cross.fast".into(), kind: ScalarKind::I64 },
            ParamSpec { name: "sma_cross.slow".into(), kind: ScalarKind::I64 },
            ParamSpec { name: "scale".into(), kind: ScalarKind::F64 },
        ];
        let bound = vec![
            ("scale".to_string(), Scalar::F64(0.5)),
            ("sma_cross.fast".to_string(), Scalar::I64(2)),
            ("sma_cross.slow".to_string(), Scalar::I64(4)),
        ];
        assert_eq!(
            resolve(&space, &bound),
            Ok(vec![Scalar::I64(2), Scalar::I64(4), Scalar::F64(0.5)]),
        );
    }

    #[test]
    fn resolve_unknown_knob() {
        let space = vec![ParamSpec { name: "scale".into(), kind: ScalarKind::F64 }];
        assert_eq!(
            resolve(&space, &[("nope".to_string(), Scalar::F64(0.5))]),
            Err(BindError::UnknownKnob("nope".to_string())),
        );
    }

    #[test]
    fn resolve_missing_knob() {
        let space = vec![
            ParamSpec { name: "a".into(), kind: ScalarKind::I64 },
            ParamSpec { name: "b".into(), kind: ScalarKind::I64 },
        ];
        assert_eq!(
            resolve(&space, &[("a".to_string(), Scalar::I64(1))]),
            Err(BindError::MissingKnob("b".to_string())),
        );
    }

    #[test]
    fn resolve_kind_mismatch() {
        let space = vec![ParamSpec { name: "scale".into(), kind: ScalarKind::F64 }];
        assert_eq!(
            resolve(&space, &[("scale".to_string(), Scalar::I64(2))]),
            Err(BindError::KindMismatch {
                knob: "scale".to_string(),
                expected: ScalarKind::F64,
                got: ScalarKind::I64,
            }),
        );
    }

    #[test]
    fn resolve_duplicate_binding() {
        let space = vec![ParamSpec { name: "a".into(), kind: ScalarKind::I64 }];
        assert_eq!(
            resolve(
                &space,
                &[("a".to_string(), Scalar::I64(1)), ("a".to_string(), Scalar::I64(2))],
            ),
            Err(BindError::DuplicateBinding("a".to_string())),
        );
    }

    #[test]
    fn resolve_precedence_unknown_before_kind_mismatch() {
        // Phase-1 (unknown) wins over Phase-2 (kind mismatch)
        let space = vec![ParamSpec { name: "scale".into(), kind: ScalarKind::F64 }];
        let bound = vec![
            ("typo".to_string(), Scalar::I64(9)),
            ("scale".to_string(), Scalar::I64(2)),
        ];
        assert_eq!(resolve(&space, &bound), Err(BindError::UnknownKnob("typo".to_string())));
    }

    #[test]
    fn resolve_precedence_unknown_before_duplicate() {
        // intra-binding: check a (unknown) precedes check d (duplicate)
        let space = vec![ParamSpec { name: "a".into(), kind: ScalarKind::I64 }];
        let bound = vec![
            ("typo".to_string(), Scalar::I64(1)),
            ("typo".to_string(), Scalar::I64(2)),
        ];
        assert_eq!(resolve(&space, &bound), Err(BindError::UnknownKnob("typo".to_string())));
    }

    #[test]
    fn named_binder_runs_bit_identical_to_positional() {
        // C1 equivalence: the named builder bootstraps to a run bit-identical to
        // the positional vector, over the sample composite harness.
        let (bp, comp_eq, comp_ex) = composite_sma_cross_harness();
        let mut named = bp
            .with("sma_cross.fast.length", 2)
            .with("sma_cross.slow.length", 4)
            .with("exposure.scale", 0.5)
            .bootstrap()
            .expect("named binding resolves and bootstraps");
        named.run(vec![synthetic_prices()]);
        let named_eq = comp_eq.try_iter().collect::<Vec<_>>();
        let named_ex = comp_ex.try_iter().collect::<Vec<_>>();

        let (bp2, pos_eq, pos_ex) = composite_sma_cross_harness();
        let mut positional = bp2
            .bootstrap_with_params(vec![Scalar::I64(2), Scalar::I64(4), Scalar::F64(0.5)])
            .expect("positional bootstrap");
        positional.run(vec![synthetic_prices()]);
        let pos_eq_v = pos_eq.try_iter().collect::<Vec<_>>();
        let pos_ex_v = pos_ex.try_iter().collect::<Vec<_>>();

        assert!(!named_eq.is_empty(), "named run drained empty");
        assert_eq!(named_eq, pos_eq_v, "equity stream: named must equal positional");
        assert_eq!(named_ex, pos_ex_v, "exposure stream: named must equal positional");
    }

    /// One f64 input port with `Firing::Any` (the common case for these fixtures).
    fn f64_any() -> PortSpec {
        PortSpec { kind: ScalarKind::F64, firing: Firing::Any, name: "in".into() }
    }
    /// A one-f64-field output record under name `v`.
    fn out_v() -> Vec<FieldSpec> {
        vec![FieldSpec { name: "v", kind: ScalarKind::F64 }]
    }
    /// A bound root role of f64 kind, fanning into `targets`.
    fn root_role(name: &str, targets: Vec<Target>) -> Role {
        Role { name: name.into(), targets, source: Some(ScalarKind::F64) }
    }

    /// 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 lookbacks(&self) -> Vec<usize> {
            vec![1, 1]
        }
        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 lookbacks(&self) -> Vec<usize> {
            vec![1]
        }
        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 lookbacks(&self) -> Vec<usize> {
            vec![1]
        }
        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 lookbacks(&self) -> Vec<usize> {
            vec![1]
        }
        fn eval(&mut self, _ctx: Ctx<'_>) -> Option<&[Scalar]> {
            None
        }
    }

    fn pass1() -> BlueprintNode {
        PrimitiveBuilder::new(
            "Pass1",
            NodeSchema { inputs: vec![f64_any()], output: out_v(), params: vec![] },
            |_| Box::new(Pass1 { out: [Scalar::F64(0.0)] }),
        )
        .into()
    }
    fn join2() -> BlueprintNode {
        PrimitiveBuilder::new(
            "Join2",
            NodeSchema { inputs: vec![f64_any(), f64_any()], output: out_v(), params: vec![] },
            |_| Box::new(Join2 { out: [Scalar::F64(0.0)] }),
        )
        .into()
    }
    fn sink_f64() -> BlueprintNode {
        PrimitiveBuilder::new(
            "SinkF64",
            NodeSchema { inputs: vec![f64_any()], output: vec![], params: vec![] },
            |_| Box::new(SinkF64),
        )
        .into()
    }
    fn sink_i64() -> BlueprintNode {
        PrimitiveBuilder::new(
            "SinkI64",
            NodeSchema {
                inputs: vec![PortSpec { kind: ScalarKind::I64, firing: Firing::Any, name: "in".into() }],
                output: vec![],
                params: vec![],
            },
            |_| Box::new(SinkI64),
        )
        .into()
    }

    /// 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 }], source: None, }],
            vec![OutField { node: 2, field: 0, name: "out".into() }],
        )
    }

    /// The SMA-cross signal as a composite under the boundary name `sma_cross`,
    /// with its two SMA legs given `name` (or left default when `named` is false).
    /// Nested under a root so the composite name qualifies the param-space path and
    /// the fan-in check (which inspects nested composites) reaches the Sub fan-in.
    fn sma_cross_under_root(named: bool) -> Composite {
        use aura_std::{Sma, Sub};
        let (a, b) = if named {
            (Sma::builder().named("fast"), Sma::builder().named("slow"))
        } else {
            (Sma::builder(), Sma::builder())
        };
        let cross = Composite::new(
            "sma_cross",
            vec![a.into(), b.into(), Sub::builder().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 }],
                source: None,
            }],
            vec![OutField { node: 2, field: 0, name: "cross".into() }],
        );
        Composite::new(
            "root",
            vec![BlueprintNode::Composite(cross)],
            vec![],
            vec![Role {
                name: "src".into(),
                targets: vec![Target { node: 0, slot: 0 }],
                source: Some(ScalarKind::F64),
            }],
            vec![],
        )
    }

    #[test]
    fn named_siblings_path_qualify_with_node_segment() {
        // two SMAs named fast/slow under composite "sma_cross" -> the node segment
        // qualifies each leaf's param path under the composite name.
        let bp = sma_cross_under_root(true);
        let names: Vec<String> = bp.param_space().into_iter().map(|p| p.name).collect();
        assert_eq!(names, ["sma_cross.fast.length", "sma_cross.slow.length"]);
    }

    #[test]
    fn unnamed_single_primitive_uses_lowercased_type_label_segment() {
        use aura_std::Sma;
        let bp = Composite::new("root", vec![Sma::builder().into()], vec![], vec![], vec![]);
        assert_eq!(bp.param_space()[0].name, "sma.length");
    }

    #[test]
    fn unnamed_same_type_param_bearing_fan_in_is_rejected() {
        // both SMAs default to "sma" -> collide -> fan-in indistinguishable
        let bp = sma_cross_under_root(false);
        assert_eq!(
            bp.compile().err(),
            Some(CompileError::DuplicateParamPath("sma_cross.sma.length".to_string()))
        );
    }

    #[test]
    fn named_param_bearing_fan_in_bootstraps() {
        // distinct node names -> fan-in distinguishable -> compiles (the two SMA
        // length params are supplied so the only thing under test is the fan-in)
        let bp = sma_cross_under_root(true);
        assert!(bp.compile_with_params(&[Scalar::I64(2), Scalar::I64(4)]).is_ok());
    }

    #[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 = Composite::new(
            "root",
            vec![BlueprintNode::Composite(fan_composite()), sink_f64()],
            vec![Edge { from: 0, to: 1, slot: 0, from_field: 0 }],
            vec![
                Role { name: "src".into(), targets: vec![Target { node: 0, slot: 0 }], source: Some(ScalarKind::F64) },
            ],
            vec![], // output
        );
        let flat = bp.compile().expect("valid composite");
        let (nodes, sources, edges) = (flat.nodes, flat.sources, flat.edges);

        // 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 }], source: None },
                Role { name: "price2".into(), targets: vec![Target { node: 1, slot: 0 }], source: None },
            ],
            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 = Composite::new(
            "root",
            vec![BlueprintNode::Composite(c), sink_f64(), sink_f64()],
            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
            ],
            vec![
                Role { name: "src".into(), targets: vec![Target { node: 0, slot: 0 }, Target { node: 0, slot: 1 }], source: Some(ScalarKind::F64) },
            ],
            vec![], // output
        );
        let flat = bp.compile().expect("valid multi-output composite");
        let (nodes, sources, edges) = (flat.nodes, flat.sources, flat.edges);
        // 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 }], source: None }],
            vec![OutField { node: 0, field: 0, name: "out".into() }],
        );
        let bp = Composite::new(
            "root",
            vec![BlueprintNode::Composite(outer), sink_f64()],
            vec![Edge { from: 0, to: 1, slot: 0, from_field: 0 }],
            vec![
                Role { name: "src".into(), targets: vec![Target { node: 0, slot: 0 }], source: Some(ScalarKind::F64) },
            ],
            vec![], // output
        );
        let flat = bp.compile().expect("valid nested composite");
        let (nodes, sources, edges) = (flat.nodes, flat.sources, flat.edges);

        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 }], source: None },
                Role { name: "price2".into(), targets: vec![Target { node: 1, slot: 0 }], source: None },
            ],
            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 }], source: None }, // outer role 0 -> inner role 0
                Role { name: "price2".into(), targets: vec![Target { node: 0, slot: 1 }], source: None }, // outer role 1 -> inner role 1
            ],
            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 = Composite::new(
            "root",
            vec![BlueprintNode::Composite(outer), sink_f64(), sink_f64()],
            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
            ],
            vec![
                Role { name: "src".into(), targets: vec![Target { node: 0, slot: 0 }, Target { node: 0, slot: 1 }], source: Some(ScalarKind::F64) },
            ],
            vec![], // output
        );
        let flat = bp.compile().expect("valid nested multi-output");
        let (nodes, _sources, edges) = (flat.nodes, flat.sources, flat.edges);
        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 }], source: None }],
            vec![OutField { node: 0, field: 0, name: "out".into() }],
        );
        let bp = Composite::new(
            "root",
            vec![BlueprintNode::Composite(c)],
            vec![],
            vec![Role { name: "src".into(), targets: vec![Target { node: 0, slot: 0 }], source: Some(ScalarKind::F64) }],
            vec![], // output
        );
        // 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 indistinguishable_fan_in_rejected() {
        // two default-named Sma (both "sma", each a `length` param) on role price
        // into a Sub: node-name signatures collide and a param is present -> fault.
        let c = Composite::new(
            "ambig",
            vec![Sma::builder().into(), Sma::builder().into(), Sub::builder().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 }], source: None, }],
            vec![OutField { node: 2, field: 0, name: "x".into() }],
        );
        let bp = Composite::new(
            "root",
            vec![BlueprintNode::Composite(c)],
            vec![],
            vec![
                Role { name: "src".into(), targets: vec![Target { node: 0, slot: 0 }], source: Some(ScalarKind::F64) },
            ],
            vec![], // output
        );
        assert_eq!(
            bp.compile().err(),
            Some(CompileError::DuplicateParamPath("ambig.sma.length".to_string()))
        );
    }

    #[test]
    fn interchangeable_fan_in_allowed() {
        // fan_composite: two param-less Pass into a Join, equal signatures but no
        // unaliased param -> interchangeable -> Ok.
        let bp = Composite::new(
            "root",
            vec![BlueprintNode::Composite(fan_composite()), sink_f64()],
            vec![Edge { from: 0, to: 1, slot: 0, from_field: 0 }],
            vec![
                Role { name: "src".into(), targets: vec![Target { node: 0, slot: 0 }], source: Some(ScalarKind::F64) },
            ],
            vec![], // output
        );
        assert!(bp.compile().is_ok(), "param-less interchangeable fan-in must compile");
    }

    #[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 }], source: None, }],
            vec![OutField { node: 0, field: 0, name: "out".into() }],
        );
        let bp = Composite::new(
            "root",
            vec![BlueprintNode::Composite(c)],
            vec![],
            vec![Role { name: "src".into(), targets: vec![Target { node: 0, slot: 0 }], source: Some(ScalarKind::F64) }],
            vec![], // output
        );
        // 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 }], source: None }],
            vec![OutField { node: 0, field: 5, name: "out".into() }],
        );
        let bp = Composite::new(
            "root",
            vec![BlueprintNode::Composite(c)],
            vec![],
            vec![Role { name: "src".into(), targets: vec![Target { node: 0, slot: 0 }], source: Some(ScalarKind::F64) }],
            vec![], // output
        );
        // 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 unconnected_interior_slot_rejected() {
        // pass1 is fed by a role; sink_f64's one input is left unwired -> rejected.
        let bp = Composite::new(
            "root",
            vec![pass1(), sink_f64()],
            vec![],
            vec![Role { name: "src".into(), targets: vec![Target { node: 0, slot: 0 }], source: Some(ScalarKind::F64) }],
            vec![],
        );
        assert_eq!(
            bp.compile().err(),
            Some(CompileError::UnconnectedPort { node: 1, slot: 0 })
        );
    }

    #[test]
    fn double_wired_slot_rejected_edge_and_role() {
        // sink_f64's one input is targeted by BOTH an edge (from pass1) and a role.
        let bp = Composite::new(
            "root",
            vec![pass1(), sink_f64()],
            vec![Edge { from: 0, to: 1, slot: 0, from_field: 0 }],
            vec![
                Role { name: "src".into(), targets: vec![Target { node: 0, slot: 0 }], source: Some(ScalarKind::F64) },
                Role { name: "extra".into(), targets: vec![Target { node: 1, slot: 0 }], source: Some(ScalarKind::F64) },
            ],
            vec![],
        );
        assert_eq!(
            bp.compile().err(),
            Some(CompileError::DoubleWiredPort { node: 1, slot: 0 })
        );
    }

    #[test]
    fn double_wired_slot_rejected_two_edges() {
        // two producers' edges land on sink_f64's single input slot.
        let bp = Composite::new(
            "root",
            vec![pass1(), pass1(), sink_f64()],
            vec![
                Edge { from: 0, to: 2, slot: 0, from_field: 0 },
                Edge { from: 1, to: 2, slot: 0, from_field: 0 },
            ],
            vec![
                Role { name: "a".into(), targets: vec![Target { node: 0, slot: 0 }], source: Some(ScalarKind::F64) },
                Role { name: "b".into(), targets: vec![Target { node: 1, slot: 0 }], source: Some(ScalarKind::F64) },
            ],
            vec![],
        );
        assert_eq!(
            bp.compile().err(),
            Some(CompileError::DoubleWiredPort { node: 2, slot: 0 })
        );
    }

    #[test]
    fn unconnected_slot_in_nested_composite_rejected() {
        // inner composite c: pass1 (fed by c's role) + sink_f64 (interior slot
        // unwired). The root covers c's one input, so the fault surfaces only via
        // the recursion into c -> UnconnectedPort at c's interior index (1, 0).
        let c = Composite::new(
            "c",
            vec![pass1(), sink_f64()],
            vec![],
            vec![Role { name: "in".into(), targets: vec![Target { node: 0, slot: 0 }], source: None }],
            vec![OutField { node: 0, field: 0, name: "out".into() }],
        );
        let bp = Composite::new(
            "root",
            vec![BlueprintNode::Composite(c)],
            vec![],
            vec![Role { name: "src".into(), targets: vec![Target { node: 0, slot: 0 }], source: Some(ScalarKind::F64) }],
            vec![],
        );
        assert_eq!(
            bp.compile().err(),
            Some(CompileError::UnconnectedPort { node: 1, slot: 0 })
        );
    }

    #[test]
    fn open_role_provider_is_not_flagged_unconnected() {
        // a composite whose OPEN input role (source: None) feeds its interior slot,
        // used as a nested node with that role covered by the enclosing root, must
        // compile — the open role is a provider, not an unwired consumer.
        let c = Composite::new(
            "c",
            vec![pass1()],
            vec![],
            vec![Role { name: "in".into(), targets: vec![Target { node: 0, slot: 0 }], source: None }],
            vec![OutField { node: 0, field: 0, name: "out".into() }],
        );
        let bp = Composite::new(
            "root",
            vec![BlueprintNode::Composite(c)],
            vec![],
            vec![Role { name: "src".into(), targets: vec![Target { node: 0, slot: 0 }], source: Some(ScalarKind::F64) }],
            vec![],
        );
        assert!(bp.compile().is_ok(), "open role as provider must not be mis-flagged");
    }

    #[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 }], source: None }],
            vec![OutField { node: 0, field: 0, name: "a".into() }],
        );
        let bp = Composite::new(
            "root",
            vec![BlueprintNode::Composite(c), sink_f64()],
            vec![Edge { from: 0, to: 1, slot: 0, from_field: 1 }],
            vec![
                Role { name: "src".into(), targets: vec![Target { node: 0, slot: 0 }], source: Some(ScalarKind::F64) },
            ],
            vec![], // output
        );
        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 = Composite::new(
            "root",
            vec![pass1(), sink_i64()],
            vec![Edge { from: 0, to: 1, slot: 0, from_field: 0 }],
            vec![],
            vec![], // output
        );
        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:?}"),
        }
    }

    /// 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 f64_recorder_sig = || NodeSchema {
            inputs: vec![f64_any()],
            output: vec![],
            params: vec![],
        };
        let h = Harness::bootstrap(FlatGraph {
            nodes: 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)),
            ],
            signatures: vec![
                Sma::builder().schema().clone(),
                Sma::builder().schema().clone(),
                Sub::builder().schema().clone(),
                Exposure::builder().schema().clone(),
                SimBroker::builder(0.0001).schema().clone(),
                f64_recorder_sig(),
                f64_recorder_sig(),
            ],
            sources: vec![SourceSpec {
                kind: ScalarKind::F64,
                targets: vec![
                    Target { node: 0, slot: 0 },
                    Target { node: 1, slot: 0 },
                    Target { node: 4, slot: 1 },
                ],
            }],
            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 },
                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)
    }

    #[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::builder().named("fast").into(), Sma::builder().named("slow").into()],
            vec![],
            vec![
                Role { name: "price".into(), targets: vec![Target { node: 0, slot: 0 }], source: None },
                Role { name: "price2".into(), targets: vec![Target { node: 1, slot: 0 }], source: None },
            ],
            vec![
                OutField { node: 0, field: 0, name: "a".into() },
                OutField { node: 1, field: 0, name: "b".into() },
            ],
        );
        let bp = Composite::new(
            "root",
            vec![
                BlueprintNode::Composite(c),
                Recorder::builder(vec![ScalarKind::F64], Firing::Any, tx_a).into(),
                Recorder::builder(vec![ScalarKind::F64], Firing::Any, tx_b).into(),
            ],
            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
            ],
            vec![
                Role { name: "src".into(), targets: vec![Target { node: 0, slot: 0 }, Target { node: 0, slot: 1 }], source: Some(ScalarKind::F64) },
            ],
            vec![], // output
        );
        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 = bp.compile_with_params(&[Scalar::I64(2), Scalar::I64(4), Scalar::F64(0.5)]).expect("harness compiles");
        let from_flat: Vec<ParamSpec> =
            flat.signatures.iter().flat_map(|s| s.params.clone()).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_flat.len(), "param count must match the flat graph");
        assert_eq!(
            space.iter().map(|p| p.kind).collect::<Vec<_>>(),
            from_flat.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.fast.length", "sma_cross.slow.length", "exposure.scale"],
        );
        assert_eq!(
            space.iter().map(|p| p.kind).collect::<Vec<_>>(),
            [ScalarKind::I64, ScalarKind::I64, ScalarKind::F64],
        );
    }

    #[test]
    fn param_space_is_flat_path_qualified_and_slot_disambiguated() {
        use aura_std::{LinComb, Sma, Sub};
        // inner composite "fast_slow": two named SMAs (fast/slow) + a Sub
        let fast_slow = Composite::new(
            "fast_slow",
            vec![
                Sma::builder().named("fast").into(),
                Sma::builder().named("slow").into(),
                Sub::builder().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 }], source: None, }],
            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::builder(2).into()],
            vec![],
            vec![Role { name: "price".into(), targets: vec![Target { node: 0, slot: 0 }], source: None }],
            vec![OutField { node: 0, field: 0, name: "out".into() }],
        );
        let bp = Composite::new(
            "root",
            vec![BlueprintNode::Composite(strategy)],
            vec![],
            vec![],
            vec![], // output
        );

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

    #[test]
    fn param_space_reflects_only_open_knobs() {
        use aura_std::{Exposure, Sma};
        // "sma2_entry": Sma "bias" with length bound to 2 (a structural constant)
        // plus Exposure "exp" whose `scale` stays open. The bound knob must be
        // absent from param_space; only the open one remains.
        let strat = Composite::new(
            "sma2_entry",
            vec![
                Sma::builder().named("bias").bind("length", Scalar::I64(2)).into(),
                Exposure::builder().named("exp").into(),
            ],
            vec![], // edges — irrelevant to param_space()
            vec![], // input_roles
            vec![], // output
        );
        let space = strat.param_space();
        let names: Vec<&str> = space.iter().map(|p| p.name.as_str()).collect();
        // collect_params runs with an empty prefix, so a top-level leaf is qualified
        // by its OWN node segment, not the root composite's name → "exp.scale".
        // `bias.length` is GONE (bound), not present-but-fixed.
        assert_eq!(names, ["exp.scale"]);
    }

    /// 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::builder().named("fast").into(),
                Sma::builder().named("slow").into(),
                Sub::builder().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 }], source: None, }],
            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::builder(2).into()],
            // fan fast_slow's output into both LinComb terms so every interior slot
            // is wired (the totality check, cycle 0040); param order is unaffected.
            vec![
                Edge { from: 0, to: 1, slot: 0, from_field: 0 },
                Edge { from: 0, to: 1, slot: 1, from_field: 0 },
            ],
            vec![Role { name: "price".into(), targets: vec![Target { node: 0, slot: 0 }], source: None }],
            vec![OutField { node: 0, field: 0, name: "out".into() }],
        );
        let bp = Composite::new(
            "root",
            vec![BlueprintNode::Composite(strategy)],
            vec![],
            vec![Role { name: "price".into(), targets: vec![Target { node: 0, slot: 0 }], source: Some(ScalarKind::F64) }],
            vec![], // output
        );

        // 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 = bp.compile_with_params(&[Scalar::I64(2), Scalar::I64(4), Scalar::F64(1.0), Scalar::F64(-1.0)]).expect("nested composite compiles");
        let from_flat: Vec<ParamSpec> =
            flat.signatures.iter().flat_map(|s| s.params.clone()).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_flat.len(), "param count must match the flat graph");
        assert_eq!(
            space.iter().map(|p| p.kind).collect::<Vec<_>>(),
            from_flat.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_carry_the_node_segment() {
        use aura_std::Sma;
        let bp = Composite::new("root", vec![Sma::builder().into()], vec![], vec![], vec![]);
        let space = bp.param_space();
        assert_eq!(space.len(), 1);
        // the root node now carries its own node-name segment (default "sma")
        assert_eq!(space[0].name, "sma.length");
    }

    #[test]
    fn param_space_is_deterministic() {
        use aura_std::{LinComb, Sma};
        let bp = Composite::new(
            "root",
            vec![Sma::builder().into(), LinComb::builder(2).into()],
            vec![],
            vec![],
            vec![], // output
        );
        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 =
            Composite::new(
            "root",
            vec![Sub::builder().into(), Add::builder().into()],
            vec![],
            vec![],
            vec![], // output
        );
        assert!(only_paramless.param_space().is_empty());
        let empty = Composite::new(
            "root",
            vec![],
            vec![],
            vec![],
            vec![], // output
        );
        assert!(empty.param_space().is_empty());
    }

    /// A macd-like composite: one f64 input role `price`, three f64 outputs
    /// (macd/signal/histogram), three params (fast/slow/signal lengths). Mirrors the
    /// CLI `macd` composite's typed multi-output boundary, used to exercise
    /// `derive_signature` on a composite.
    fn macd_fixture() -> Composite {
        Composite::new(
            "macd",
            vec![
                Ema::builder().named("fast").into(), // 0 fast EMA
                Ema::builder().named("slow").into(),  // 1 slow EMA
                Sub::builder().into(),                // 2 macd = fast - slow
                Ema::builder().named("signal").into(), // 3 signal EMA of macd
                Sub::builder().into(),                // 4 histogram = macd - signal
            ],
            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: 2, to: 4, slot: 0, from_field: 0 },
                Edge { from: 3, to: 4, slot: 1, from_field: 0 },
            ],
            vec![root_role("price", vec![Target { node: 0, slot: 0 }, Target { node: 1, slot: 0 }])],
            vec![
                OutField { node: 2, field: 0, name: "macd".into() },
                OutField { node: 3, field: 0, name: "signal".into() },
                OutField { node: 4, field: 0, name: "histogram".into() },
            ],
        )
    }

    #[test]
    fn primitive_signature_equals_builder_schema() {
        // a primitive's pre-build signature IS its builder's declared schema
        let b = Sma::builder();
        let node = BlueprintNode::Primitive(Sma::builder());
        assert_eq!(node.signature(), b.schema().clone());
    }

    #[test]
    fn composite_signature_is_derived_from_interior() {
        // macd composite: 1 f64 input role; output macd/signal/histogram (all f64)
        let node = BlueprintNode::Composite(macd_fixture());
        let sig = node.signature();
        assert_eq!(sig.inputs.len(), 1);
        assert_eq!(sig.inputs[0].kind, ScalarKind::F64);
        assert_eq!(sig.output.iter().map(|f| f.kind).collect::<Vec<_>>(), vec![ScalarKind::F64; 3]);
        assert_eq!(sig.params.len(), 3); // fast, slow, signal
    }

    #[test]
    fn compile_rejects_kind_mismatch_without_building() {
        // a builder whose build closure PANICS if called — proves validation is pre-build
        let exploding = PrimitiveBuilder::new(
            "Boom",
            NodeSchema {
                inputs: vec![PortSpec { kind: ScalarKind::I64, firing: Firing::Any, name: "in".into() }],
                output: vec![FieldSpec { name: "v", kind: ScalarKind::I64 }],
                params: vec![],
            },
            |_| panic!("build must not run when validation fails pre-build"),
        );
        // an f64 producer wired into Boom's i64 slot
        let root = Composite::new(
            "root",
            vec![Sma::builder().into(), exploding.into()],
            vec![Edge { from: 0, to: 1, slot: 0, from_field: 0 }], // f64 -> i64 slot
            vec![Role { name: "price".into(), targets: vec![Target { node: 0, slot: 0 }], source: Some(ScalarKind::F64) }],
            vec![],
        );
        let err = root.compile_with_params(&[Scalar::I64(3)]);
        // kind fault caught pre-build (no panic) — same variant bootstrap would give
        assert!(matches!(
            err,
            Err(CompileError::Bootstrap(BootstrapError::KindMismatch { .. }))
        ));
    }

    #[test]
    fn unbound_root_role_is_rejected() {
        let root = Composite::new(
            "root",
            vec![Sma::builder().into()],
            vec![],
            vec![Role { name: "price".into(), targets: vec![Target { node: 0, slot: 0 }], source: None }],
            vec![],
        );
        // the Ok arm holds a FlatGraph (not Debug), so assert via the Err arm.
        assert_eq!(
            root.compile_with_params(&[Scalar::I64(3)]).err(),
            Some(CompileError::UnboundRootRole { role: 0 })
        );
    }

    #[test]
    fn lookbacks_arity_matches_signature_inputs() {
        use aura_std::{Add, Sma};
        // every std node: one lookback per declared input
        assert_eq!(Sma::new(3).lookbacks(), vec![3]);
        assert_eq!(Sma::new(3).lookbacks().len(), Sma::builder().schema().inputs.len());
        assert_eq!(Add::new().lookbacks().len(), Add::builder().schema().inputs.len());
    }

    #[test]
    fn non_fan_in_duplicate_path_is_rejected() {
        use aura_std::Sma;
        // two unnamed SMAs ("sma" each), each feeding its OWN sink — no shared
        // fan-in node, so the old fan-in check never fired; but param_space() has
        // the path "dup.sma.length" twice.
        let dup = Composite::new(
            "dup",
            vec![
                Sma::builder().into(), // node 0 -> dup.sma.length
                Sma::builder().into(), // node 1 -> dup.sma.length (duplicate)
                sink_f64(),            // node 2
                sink_f64(),            // node 3
            ],
            vec![
                Edge { from: 0, to: 2, slot: 0, from_field: 0 },
                Edge { from: 1, to: 3, slot: 0, from_field: 0 },
            ],
            vec![Role {
                name: "price".into(),
                targets: vec![Target { node: 0, slot: 0 }, Target { node: 1, slot: 0 }],
                source: None,
            }],
            vec![],
        );
        let bp = Composite::new(
            "root",
            vec![BlueprintNode::Composite(dup)],
            vec![],
            vec![Role {
                name: "src".into(),
                targets: vec![Target { node: 0, slot: 0 }],
                source: Some(ScalarKind::F64),
            }],
            vec![],
        );
        // two params declared (one per SMA); supply both so the only failure under
        // test is the duplicate path (green today, DuplicateParamPath after).
        assert_eq!(
            bp.compile_with_params(&[Scalar::I64(2), Scalar::I64(3)]).err(),
            Some(CompileError::DuplicateParamPath("dup.sma.length".to_string()))
        );
    }

    #[test]
    fn asymmetric_node_name_collision_compiles() {
        use aura_std::{Sma, Sub};
        // inner composite `asym`: a param-bearing Sma (default "sma") + a paramless
        // Pass1 forced to "sma", both on role price fanning into a Sub. Equal
        // node-name signatures + one param -> rejected today (IndistinguishableFanIn);
        // param_space() is the single injective entry ["asym.sma.length"] (the
        // paramless leg contributes no path) -> admitted after the change. Pass1 is
        // built inline (the pass1() helper returns an already-.into()'d node, so it
        // cannot take .named).
        let paramless_sma = PrimitiveBuilder::new(
            "Pass1",
            NodeSchema { inputs: vec![f64_any()], output: out_v(), params: vec![] },
            |_| Box::new(Pass1 { out: [Scalar::F64(0.0)] }),
        )
        .named("sma");
        let asym = Composite::new(
            "asym",
            vec![Sma::builder().into(), paramless_sma.into(), Sub::builder().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 }],
                source: None,
            }],
            vec![OutField { node: 2, field: 0, name: "out".into() }],
        );
        let bp = Composite::new(
            "root",
            vec![BlueprintNode::Composite(asym)],
            vec![],
            vec![Role {
                name: "src".into(),
                targets: vec![Target { node: 0, slot: 0 }],
                source: Some(ScalarKind::F64),
            }],
            vec![],
        );
        assert_eq!(
            bp.param_space().into_iter().map(|p| p.name).collect::<Vec<_>>(),
            ["asym.sma.length"]
        );
        assert!(bp.compile_with_params(&[Scalar::I64(2)]).is_ok());
    }

    #[test]
    fn by_name_bootstrap_of_unnamed_cross_reports_duplicate_path() {
        // the canonical by-name flow on an unnamed cross: the binder runs the
        // injectivity check before name resolution, so the author sees the
        // structural DuplicateParamPath (carrying the .named(...) cure) instead of
        // AmbiguousKnob.
        let bp = sma_cross_under_root(false);
        assert_eq!(
            bp.with("sma_cross.sma.length", 2).bootstrap().err(),
            Some(BindError::Compile(CompileError::DuplicateParamPath(
                "sma_cross.sma.length".to_string()
            )))
        );
    }

    /// E2E (cycle 0032): the whole collision → cure → run arc on the canonical
    /// by-name flow. The same un-named SMA cross that `bootstrap` rejects as
    /// `DuplicateParamPath` becomes runnable the moment its two legs are `.named()`
    /// apart — the cure the error itself prescribes. Naming makes `param_space()`
    /// injective, the by-name `.with(...).bootstrap()` terminal resolves the two
    /// now-distinct paths, and the harness runs to a populated, non-degenerate
    /// exposure trace. A regression that made injectivity gate the run (rather than
    /// only the duplicate) — or that broke the by-name resolution of the cured
    /// space — would surface here as a bootstrap error or an empty trace, not just a
    /// bad `compile()` Err.
    #[test]
    fn named_cross_resolves_by_name_and_runs_to_a_trace() {
        // root { sma_cross[ fast/slow SMA -> Sub ] -> Exposure -> SimBroker -> rec },
        // plus an exposure tap, mirroring composite_sma_cross_harness but driven
        // through the by-name binder so the injective cured space is exercised
        // end-to-end (resolve + bootstrap + run), not just at compile().
        let prices = synthetic_prices();
        let (tx_ex, rx_ex) = mpsc::channel();
        let cross = Composite::new(
            "sma_cross",
            vec![
                Sma::builder().named("fast").into(),
                Sma::builder().named("slow").into(),
                Sub::builder().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 }],
                source: None,
            }],
            vec![OutField { node: 2, field: 0, name: "out".into() }],
        );
        let bp = Composite::new(
            "root",
            vec![
                BlueprintNode::Composite(cross),
                Exposure::builder().into(),
                Recorder::builder(vec![ScalarKind::F64], Firing::Any, tx_ex).into(),
            ],
            vec![
                Edge { from: 0, to: 1, slot: 0, from_field: 0 }, // cross out -> Exposure
                Edge { from: 1, to: 2, slot: 0, from_field: 0 }, // exposure -> recorder
            ],
            vec![Role {
                name: "src".into(),
                targets: vec![Target { node: 0, slot: 0 }],
                source: Some(ScalarKind::F64),
            }],
            vec![],
        );
        // the un-named twin of this blueprint is rejected (see
        // by_name_bootstrap_of_unnamed_cross_reports_duplicate_path); the named one
        // resolves its two distinct paths by name and bootstraps.
        let mut h = bp
            .with("sma_cross.fast.length", 2)
            .with("sma_cross.slow.length", 4)
            .with("exposure.scale", 0.5)
            .bootstrap()
            .expect("named (injective) cross resolves by name and bootstraps");
        h.run(vec![prices]);
        let ex = rx_ex.try_iter().collect::<Vec<_>>();
        assert!(!ex.is_empty(), "the cured, by-name-bound cross must run to a populated exposure trace");
    }

    #[test]
    fn composite_lifts_into_blueprint_node_via_from() {
        let c = Composite::new(
            "leaf",
            vec![Sub::builder().into()],
            vec![],
            vec![],
            vec![OutField { node: 0, field: 0, name: "o".into() }],
        );
        let bn: BlueprintNode = c.into();
        assert!(matches!(bn, BlueprintNode::Composite(_)));
    }
}