//! The harness — the closed root graph that runs (glossary: `harness`, C20) —
//! and its deterministic run loop. A `Harness` is a bootstrapped, frozen root
//! graph: a flat node array plus an index edge table, topologically ordered,
//! driven cycle by cycle by a k-way merge of timestamped sources (C3/C4). It is
//! the flat, monomorphized sharpening of RustAst's reference-counted,
//! interior-mutable observer push graph: no reference counting, no interior
//! mutability, no per-cycle allocation (C1/C7). Each node owns its input columns
//! (the cycle-0002 shape), so `Ctx` borrows them read-only and additionally
//! carries the cycle timestamp (`ctx.now()`, C4). A node's `eval` returns a
//! borrowed record (`Option<&[Cell]>`); each out-edge forwards one field of it
//! (`Edge::from_field`) into a consumer slot, so the K fields are co-fresh (C6).
//!
//! Firing (C5/C6) gates re-evaluation. Two read clocks drive it, both stamped per
//! input slot on every push: `fresh_at` (the `cycle_id` of the last push —
//! freshness epoch, C5) and `last_ts` (the data timestamp of that push — barrier
//! token, C6). A node fires when any `Firing::Any` input is fresh this cycle, or
//! when a `Firing::Barrier` group has every member at the current timestamp; a
//! node that does not fire pushes nothing, so consumers keep seeing the held
//! value via `window[0]` (sample-and-hold falls out of the push model). The
//! barrier token is the timestamp, not the `cycle_id`: under C4 four same-time
//! sources are four cycles, so RustAst's `cycle_id`-equality barrier could never
//! fire across sources — the timestamp generalizes it faithfully.

use aura_core::{AnyColumn, Cell, Ctx, Firing, Node, NodeSchema, Scalar, ScalarKind, Timestamp};

/// Forwards one field (`from_field`) of a producer's output record into a
/// consumer's input slot. Consuming a whole record is N such edges, one per field
/// (there is no "bind whole record" mechanism).
#[derive(Clone, Copy, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct Edge {
    pub from: usize,
    pub to: usize,
    pub slot: usize,
    pub from_field: usize,
}

/// An input slot the source value is forwarded into each cycle.
#[derive(Clone, Copy, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct Target {
    pub node: usize,
    pub slot: usize,
}

/// A declared source: the scalar kind it produces and the input slots each of
/// its records is forwarded into. The multi-source generalization of cycle
/// 0003's `(source_targets, source_kind)` pair — sources are k-way-merged by
/// timestamp at ingestion (C3); there is no merge inside the graph.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct SourceSpec {
    pub kind: ScalarKind,
    pub targets: Vec<Target>,
    /// The binding key `bind_sources` / `run_bound` match a keyed supply
    /// against. `Some(name)` for a spec lowered from a bound `Role`; `None`
    /// for a hand-built raw-index spec (supplied positionally to `run`).
    /// Load-bearing for source binding only — every other flat-graph name
    /// stays a non-load-bearing debug symbol (C23, #275).
    pub role: Option<String>,
}

impl SourceSpec {
    /// A raw-index primitive spec (no role; supplied positionally to `run`).
    pub fn raw(kind: ScalarKind, targets: Vec<Target>) -> Self {
        Self { kind, targets, role: None }
    }
    /// A named spec (the role `run_bound` / `bind_sources` match by name).
    pub fn named(role: impl Into<String>, kind: ScalarKind, targets: Vec<Target>) -> Self {
        Self { kind, targets, role: Some(role.into()) }
    }
}

/// A source-binding fault caught before the run starts (`run_bound`) — the
/// by-name analogue of the raw-index `run` arity panic. A raw-index arity
/// mismatch stays a panic (an engine wiring bug); a keyed role mismatch is a
/// user data-binding error, so it is a `Result` (#275, C4/C23).
#[derive(Debug, PartialEq, Eq)]
pub enum SourceBindError {
    /// A compiled source declares a role the supply does not provide.
    MissingFeed { role: String },
    /// The supply provides a role no compiled source declares.
    ExtraFeed { role: String },
    /// The supply provides the same role name twice.
    DuplicateFeed { role: String },
    /// A compiled source carries no role, but a keyed supply was given — the
    /// keyed path requires every source to be named (a raw-index graph uses
    /// positional `run`).
    UnnamedSource { index: usize },
}

impl core::fmt::Display for SourceBindError {
    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
        match self {
            SourceBindError::MissingFeed { role } => {
                write!(f, "the graph declares source role \"{role}\" but the supply provides no such feed")
            }
            SourceBindError::ExtraFeed { role } => {
                write!(f, "the supply provides feed \"{role}\" but the graph declares no such source role")
            }
            SourceBindError::DuplicateFeed { role } => {
                write!(f, "the supply provides feed \"{role}\" more than once")
            }
            SourceBindError::UnnamedSource { index } => {
                write!(f, "compiled source {index} carries no role; a keyed supply requires every source to be named")
            }
        }
    }
}

/// Resolve a keyed source supply into the positional `Vec` `run` consumes, in
/// `specs` order — so the merge's source-index tie-break resolves on source
/// declaration order (C4), independent of the caller's supply order. Pure and
/// unit-testable; consumes `supply` (sources are `!Clone`). A `BTreeMap` index
/// makes the `DuplicateFeed` / `ExtraFeed` diagnostics deterministic (C1).
pub fn bind_sources(
    specs: &[SourceSpec],
    supply: Vec<(String, Box<dyn Source>)>,
) -> Result<Vec<Box<dyn Source>>, SourceBindError> {
    let mut by_role: std::collections::BTreeMap<String, Box<dyn Source>> =
        std::collections::BTreeMap::new();
    for (role, src) in supply {
        if by_role.insert(role.clone(), src).is_some() {
            return Err(SourceBindError::DuplicateFeed { role });
        }
    }
    let mut resolved: Vec<Box<dyn Source>> = Vec::with_capacity(specs.len());
    for (index, spec) in specs.iter().enumerate() {
        let role = spec.role.as_ref().ok_or(SourceBindError::UnnamedSource { index })?;
        let src = by_role
            .remove(role)
            .ok_or_else(|| SourceBindError::MissingFeed { role: role.clone() })?;
        resolved.push(src);
    }
    if let Some((role, _)) = by_role.into_iter().next() {
        return Err(SourceBindError::ExtraFeed { role });
    }
    Ok(resolved)
}

/// The ingestion-boundary producer the k-way merge drives (C3). A source yields
/// timestamped scalars in ascending timestamp order (the C3 ingestion
/// precondition). Object-safe: the merge holds `Vec<Box<dyn Source>>`.
pub trait Source {
    /// The timestamp of the head record, without consuming it. `None` = the
    /// source is exhausted. Cheap and side-effect-free: it reads a pre-decoded
    /// head, never advancing the underlying stream.
    fn peek(&self) -> Option<Timestamp>;

    /// Pop the head `(timestamp, scalar)` and advance. `None` = exhausted.
    /// After `next` returns `Some`, `peek` reflects the new head.
    fn next(&mut self) -> Option<(Timestamp, Scalar)>;

    /// The inclusive `(from, to)` data extent this producer will stream, known
    /// without materializing it (#71/C18): a lazy producer reports its window
    /// bounds; an eager one reports its buffer's first/last. `None` = empty /
    /// open-ended. Cursor-independent: a property of the data extent, not the
    /// read position.
    fn bounds(&self) -> Option<(Timestamp, Timestamp)>;

    /// Records this source holds resident right now — the streaming-ring probe
    /// (#95). `None` (the default) means the source does not report a residency:
    /// the eager [`VecSource`] materializes its whole stream and has no bounded
    /// ring. A lazy chunk source overrides this to return `Some(len)` of the chunk
    /// it currently borrows, so a `Box<dyn Source>` consumer can budget memory
    /// without downcasting. Mirrors [`bounds`](Source::bounds)' Option idiom:
    /// `None` = "not reported," distinct from `Some(0)` = "reports, holds none"
    /// (e.g. exhausted). Probes the aura source ring only, not whole-process RSS.
    fn resident_records(&self) -> Option<usize> {
        None
    }
}

/// A `Source` over a fully materialized stream — the cycle-0011 eager shape,
/// named. Every pre-seam call site wraps its `Vec<(Timestamp, Scalar)>` in this,
/// so run output is byte-identical (C1) and residency is unchanged.
pub struct VecSource {
    stream: Vec<(Timestamp, Scalar)>,
    cursor: usize,
}

impl VecSource {
    pub fn new(stream: Vec<(Timestamp, Scalar)>) -> Self {
        Self { stream, cursor: 0 }
    }
}

impl Source for VecSource {
    fn peek(&self) -> Option<Timestamp> {
        self.stream.get(self.cursor).map(|&(t, _)| t)
    }
    fn next(&mut self) -> Option<(Timestamp, Scalar)> {
        let item = self.stream.get(self.cursor).copied()?;
        self.cursor += 1;
        Some(item)
    }
    fn bounds(&self) -> Option<(Timestamp, Timestamp)> {
        // the eager source reads its own buffer ends (O(1) on the slice) — the
        // consumer no longer scans a Vec. Cursor-independent (ignores self.cursor).
        Some((self.stream.first()?.0, self.stream.last()?.0))
    }
}

/// The window of a run: the union extent across its sources (min `from`, max
/// `to`), each source contributing its [`Source::bounds`]. `None` when every
/// source is empty / open-ended. The producer-supplied replacement for scanning a
/// materialized `Vec` at the call site (#71/C18): byte-identical whether a source
/// is eager (`VecSource`) or lazily streamed. Uses `<`/`>` only, so it needs no
/// `Ord` on `Timestamp`.
pub fn window_of(sources: &[Box<dyn Source>]) -> Option<(Timestamp, Timestamp)> {
    let mut acc: Option<(Timestamp, Timestamp)> = None;
    for s in sources {
        if let Some((from, to)) = s.bounds() {
            acc = Some(match acc {
                None => (from, to),
                Some((lo, hi)) => (
                    if from < lo { from } else { lo },
                    if to > hi { to } else { hi },
                ),
            });
        }
    }
    acc
}

/// A tiny, fully-deterministic, dependency-free PRNG (SplitMix64). The seed
/// completely determines the sequence; no external entropy, no global state.
/// Bit-stable across toolchains and crate versions — the property C1 needs for
/// seed-as-input reproducibility (C12).
pub struct SplitMix64 {
    state: u64,
}

impl SplitMix64 {
    pub fn new(seed: u64) -> Self {
        Self { state: seed }
    }
    pub fn next_u64(&mut self) -> u64 {
        self.state = self.state.wrapping_add(0x9E37_79B9_7F4A_7C15);
        let mut z = self.state;
        z = (z ^ (z >> 30)).wrapping_mul(0xBF58_476D_1CE4_E5B9);
        z = (z ^ (z >> 27)).wrapping_mul(0x94D0_49BB_1331_11EB);
        z ^ (z >> 31)
    }
    /// A uniform `f64` in `[0, 1)` from the top 53 bits.
    pub(crate) fn next_f64(&mut self) -> f64 {
        (self.next_u64() >> 11) as f64 / ((1u64 << 53) as f64)
    }
}

/// Config for a seeded synthetic price walk — everything *except* the seed.
/// Partially applying it (`|seed| spec.source(seed)`) yields the
/// `Fn(u64) -> impl Source` contract the Monte-Carlo family (#68) consumes
/// (C12 seed-as-input). The produced stream is an ordinary ingestion-boundary
/// `Source` (C3), generated at the data-generation edge outside the engine graph.
#[derive(Clone, Copy, Debug)]
pub struct SyntheticSpec {
    /// Opening price.
    pub start: f64,
    /// Number of M1 steps.
    pub len: usize,
    /// Timestamp increment per step.
    pub step: i64,
}

impl SyntheticSpec {
    /// Produce a synthetic price stream **fully determined by `seed`** (C1/C12):
    /// a multiplicative random walk, each step a small seeded perturbation.
    /// Returns a `Source` (the C3 ingestion producer), NOT a `Vec` — so a
    /// Monte-Carlo family can re-seed without materializing N streams (#68).
    /// (First cut collects to a `VecSource` internally; the signature does not
    /// name `Vec`, so a later lazy generator is a drop-in replacement.)
    ///
    /// `use<>`: the returned source owns its data (it captures nothing from
    /// `&self` — `start`/`len`/`step` are copied into the stream before return),
    /// so under the edition-2024 capture rules it is `'static` and boxable into
    /// the `Box<dyn Source + 'static>` the k-way merge holds (C3).
    pub fn source(&self, seed: u64) -> impl Source + use<> {
        let mut rng = SplitMix64::new(seed);
        let mut price = self.start;
        let stream: Vec<(Timestamp, Scalar)> = (0..self.len)
            .map(|i| {
                let shock = (rng.next_f64() - 0.5) * 0.004; // ±0.2% per step
                price *= 1.0 + shock;
                (Timestamp(i as i64 * self.step + 1), Scalar::f64(price))
            })
            .collect();
        VecSource::new(stream)
    }
}

/// A resolved measurement tap in the flat graph: a name paired with a flat
/// producer `(node, output-field)`. The output-side twin of `SourceSpec` — its
/// `name` survives compile and is load-bearing for by-name binding (like
/// `SourceSpec.role`, #275). `compile` resolves each blueprint `Tap` into a
/// `FlatTap` through the same lowering remap that hoists edges/`OutField`
/// re-exports (interior-composite taps hoist to the root list).
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct FlatTap {
    pub name: String,
    pub node: usize,
    pub field: usize,
}

/// The flat, type-erased, index-wired output of `Composite::compile` — the target
/// `Harness::bootstrap` consumes (C23's "flat graph", now a named type). `signatures[i]`
/// is the static signature of `nodes[i]` (gathered from each primitive's builder at
/// lowering); `bootstrap` reads kinds/output from it and `nodes[i].lookbacks()` for
/// buffer depth. `sources` are the lowered bound roles, in role-declaration order.
pub struct FlatGraph {
    pub nodes: Vec<Box<dyn Node>>,
    pub signatures: Vec<NodeSchema>,
    pub sources: Vec<SourceSpec>,
    pub edges: Vec<Edge>,
    pub taps: Vec<FlatTap>,
}

/// A tap-binding fault caught before the run. The output-side twin of
/// `SourceBindError`. `UndeclaredTap` is raised by `bind_tap` itself;
/// `DuplicateBind` is raised by the **caller** (the run-mode-owning layer) when
/// it dedups a tap name across binds — the engine method binds a single tap and
/// keeps no cross-call state, so duplicate detection is the caller's, exactly as
/// `bind_sources`' `DuplicateFeed` is a supply-level, not a per-source, fault.
#[derive(Debug, PartialEq, Eq)]
pub enum TapBindError {
    /// A binding names a tap the graph does not declare (raised by `bind_tap`).
    UndeclaredTap { name: String },
    /// The same tap name was bound more than once (raised by the caller dedup).
    DuplicateBind { name: String },
}

impl core::fmt::Display for TapBindError {
    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
        match self {
            TapBindError::UndeclaredTap { name } => {
                write!(f, "bind_tap names tap \"{name}\" but the graph declares no such tap")
            }
            TapBindError::DuplicateBind { name } => {
                write!(f, "tap \"{name}\" was bound more than once")
            }
        }
    }
}

impl FlatGraph {
    /// Attach a caller-built sink to a declared tap: append `sink` + `sig` to the
    /// flat node/signature arrays and an `Edge` from the tap's `(node, field)` to
    /// the sink's input slot 0. Constructs no sink itself (the engine has no
    /// `aura-std` production dep) — the caller hands in a `Recorder`. Call BEFORE
    /// `bootstrap`; the Kahn sort then orders the appended sink. Binding a tap
    /// the graph does not declare is a typed `UndeclaredTap` refusal; duplicate
    /// detection across binds is the caller's (this method keeps no cross-call
    /// state — see `TapBindError`).
    pub fn bind_tap(
        &mut self,
        tap_name: &str,
        sink: Box<dyn Node>,
        sig: NodeSchema,
    ) -> Result<(), TapBindError> {
        let tap = self
            .taps
            .iter()
            .find(|t| t.name == tap_name)
            .ok_or_else(|| TapBindError::UndeclaredTap { name: tap_name.to_string() })?;
        let (from, from_field) = (tap.node, tap.field);
        let sink_idx = self.nodes.len();
        self.nodes.push(sink);
        self.signatures.push(sig);
        self.edges.push(Edge { from, to: sink_idx, slot: 0, from_field });
        Ok(())
    }
}

/// A wiring fault caught once, at bootstrap — C7's "type check paid at wiring"
/// generalized to the whole topology.
#[derive(Debug, PartialEq, Eq)]
pub enum BootstrapError {
    /// An edge or source-target connects mismatched scalar kinds.
    KindMismatch { producer: ScalarKind, consumer: ScalarKind },
    /// A node or slot index in an edge or target is out of range.
    BadIndex,
    /// The wiring contains a directed cycle (only an explicit delay node may close
    /// a loop; that node does not exist yet).
    Cycle,
}

/// Per-input engine bookkeeping (invisible to nodes): the two read clocks of the
/// firing machinery. `fresh_at` is the `cycle_id` of the last push into this slot
/// (freshness epoch, C5: fresh this cycle iff `fresh_at == cycle_id`); `last_ts`
/// is the data timestamp of that push (barrier token, C6: a barrier group is
/// complete iff all members carry `last_ts == T`). A never-pushed slot keeps the
/// cold sentinel `Timestamp(i64::MIN)`, which no real timestamp equals.
struct SlotState {
    fresh_at: u64,
    last_ts: Timestamp,
}

struct NodeBox {
    node: Box<dyn Node>,
    inputs: Vec<AnyColumn>,
    firing: Vec<Firing>,
    slots: Vec<SlotState>,
    out_len: usize,
}

/// A bootstrapped, frozen root graph instance plus its deterministic run loop.
pub struct Harness {
    nodes: Vec<NodeBox>,
    topo: Vec<usize>,
    out_edges: Vec<Vec<Edge>>,
    sources: Vec<SourceSpec>,
}

// Manual, node-opaque `Debug`: `Box<dyn Node>` is not `Debug`, so the struct
// cannot derive it. This summary form is enough for `Result::unwrap_err` (which
// formats the `Ok` arm on a failed bootstrap-rejection assertion) without
// printing node internals.
impl core::fmt::Debug for Harness {
    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
        f.debug_struct("Harness")
            .field("nodes", &self.nodes.len())
            .field("topo", &self.topo)
            .field("out_edges", &self.out_edges)
            .field("sources", &self.sources)
            .finish()
    }
}

impl Harness {
    /// Bind a flat graph into a frozen, runnable graph. Sizes each node's input
    /// columns — KIND/firing from its carried signature, DEPTH from the built node's
    /// `lookbacks()` — lifts each input's firing policy, initializes per-slot
    /// freshness state, kind-checks every source target and edge, and topologically
    /// orders the nodes (Kahn), rejecting any directed cycle.
    pub fn bootstrap(flat: FlatGraph) -> Result<Harness, BootstrapError> {
        // `taps` is ignored here by design: a future slice (#282's `bind_tap`) will
        // materialize resolved taps into real recorder nodes/edges before bootstrap,
        // so bootstrap will still have nothing of its own to do with the list. In
        // this slice `taps` is always empty (no `Tap` resolution exists yet), so the
        // ignore is presently a no-op, not yet load-bearing.
        let FlatGraph { nodes, signatures, sources, edges, taps: _ } = flat;
        let n = nodes.len();

        // size each node's input columns: KIND/firing from the carried signature,
        // DEPTH from the built node's lookbacks() (the one param-dependent quantity)
        let mut boxes: Vec<NodeBox> = Vec::with_capacity(n);
        for (nd, sig) in nodes.into_iter().zip(signatures.iter()) {
            let depths = nd.lookbacks();
            debug_assert_eq!(depths.len(), sig.inputs.len(), "lookbacks() arity == signature inputs");
            let inputs: Vec<AnyColumn> = sig
                .inputs
                .iter()
                .zip(depths)
                .map(|(spec, depth)| AnyColumn::with_capacity(spec.kind, depth))
                .collect();
            let firing: Vec<Firing> = sig.inputs.iter().map(|spec| spec.firing).collect();
            let slots: Vec<SlotState> = sig
                .inputs
                .iter()
                .map(|_| SlotState { fresh_at: 0, last_ts: Timestamp(i64::MIN) })
                .collect();
            let out_len = sig.output.len();
            boxes.push(NodeBox { node: nd, inputs, firing, slots, out_len });
        }

        // source targets: each source's value must match each of its target slots' kind
        for src in &sources {
            for t in &src.targets {
                let s = signatures.get(t.node).ok_or(BootstrapError::BadIndex)?;
                let slot = s.inputs.get(t.slot).ok_or(BootstrapError::BadIndex)?;
                if slot.kind != src.kind {
                    return Err(BootstrapError::KindMismatch {
                        producer: src.kind,
                        consumer: slot.kind,
                    });
                }
            }
        }

        // edges: indices in range, producer output field kind == consumer slot kind
        let mut out_edges: Vec<Vec<Edge>> = vec![Vec::new(); n];
        for &e in &edges {
            let from = signatures.get(e.from).ok_or(BootstrapError::BadIndex)?;
            let to = signatures.get(e.to).ok_or(BootstrapError::BadIndex)?;
            let field = from.output.get(e.from_field).ok_or(BootstrapError::BadIndex)?;
            let slot = to.inputs.get(e.slot).ok_or(BootstrapError::BadIndex)?;
            if field.kind != slot.kind {
                return Err(BootstrapError::KindMismatch {
                    producer: field.kind,
                    consumer: slot.kind,
                });
            }
            out_edges[e.from].push(e);
        }

        // Kahn topological sort; a leftover node means a cycle
        let mut indeg = vec![0usize; n];
        for &e in &edges {
            indeg[e.to] += 1;
        }
        let mut queue: Vec<usize> = (0..n).filter(|&i| indeg[i] == 0).collect();
        let mut topo: Vec<usize> = Vec::with_capacity(n);
        let mut head = 0;
        while head < queue.len() {
            let u = queue[head];
            head += 1;
            topo.push(u);
            for e in &out_edges[u] {
                indeg[e.to] -= 1;
                if indeg[e.to] == 0 {
                    queue.push(e.to);
                }
            }
        }
        if topo.len() != n {
            return Err(BootstrapError::Cycle);
        }

        Ok(Harness {
            nodes: boxes,
            topo,
            out_edges,
            sources,
        })
    }

    /// Drive the sources, k-way-merged in timestamp order (ties by source index,
    /// C4). One stream per source, each ascending in timestamp (C3 ingestion
    /// precondition). Recording is a node-side concern: a recording node pushes
    /// its record to a destination it holds (out of graph) inside `eval`; the
    /// engine only routes in-graph edges and is oblivious to the side effect.
    /// Allocates nothing per cycle beyond the reused scratch buffer.
    pub fn run(&mut self, mut sources_in: Vec<Box<dyn Source>>) {
        assert_eq!(
            sources_in.len(),
            self.sources.len(),
            "run: one source per SourceSpec required (got {} sources for {} specs)",
            sources_in.len(),
            self.sources.len()
        );

        // disjoint field borrows so the topo walk can read topo/out_edges/sources
        // while mutating nodes
        let Harness { nodes, topo, out_edges, sources } = self;

        let mut cycle_id: u64 = 0;
        let mut scratch: Vec<Cell> = Vec::new();

        loop {
            // pick the live source head with the smallest (timestamp, source index).
            // strictly-`<` replace + source-order scan preserves C4 tie-breaking.
            let mut pick: Option<(usize, Timestamp)> = None;
            for (s, src) in sources_in.iter().enumerate() {
                if let Some(ts) = src.peek() {
                    match pick {
                        None => pick = Some((s, ts)),
                        Some((_, best)) if ts < best => pick = Some((s, ts)),
                        _ => {}
                    }
                }
            }
            let s = match pick {
                Some((s, _)) => s,
                None => break, // all sources exhausted
            };
            // `&mut *sources_in[s]`: deref the Box to `dyn Source` then re-borrow,
            // so UFCS resolves Self = dyn Source (Box<dyn Source> does not itself
            // impl Source). Fully-qualified to never collide with Iterator::next.
            let (ts, value) =
                Source::next(&mut *sources_in[s]).expect("peeked Some ⇒ next is Some");
            cycle_id += 1;

            // forward the source value into its target slots, stamping freshness
            for t in sources[s].targets.iter() {
                let nb = &mut nodes[t.node];
                nb.inputs[t.slot].push(value).expect("source kind checked at wiring");
                nb.slots[t.slot] = SlotState { fresh_at: cycle_id, last_ts: ts };
            }

            // evaluate in topological order; gate by firing; forward Some outputs
            for &nidx in topo.iter() {
                let out_len = nodes[nidx].out_len;
                let fired = {
                    let nb = &nodes[nidx];
                    fires(&nb.firing, &nb.slots, cycle_id, ts)
                };
                if !fired {
                    continue; // hold: no eval, no push
                }
                let result: Option<&[Cell]> = {
                    let nb = &mut nodes[nidx];
                    nb.node.eval(Ctx::new(&nb.inputs, ts))
                };
                if let Some(row) = result {
                    debug_assert_eq!(row.len(), out_len, "node returned a row of the wrong width");
                    scratch.clear();
                    scratch.extend_from_slice(row);
                    for e in out_edges[nidx].iter() {
                        let nb = &mut nodes[e.to];
                        nb.inputs[e.slot].push_cell(scratch[e.from_field]);
                        nb.slots[e.slot] = SlotState { fresh_at: cycle_id, last_ts: ts };
                    }
                }
            }
        }

        // end-of-stream: flush every node once, in topological order, so folding
        // sinks (C8) emit their accumulated summary. This runs AFTER the
        // deterministic event loop, so it adds no within-sim concurrency (C1).
        for &nidx in topo.iter() {
            nodes[nidx].node.finalize();
        }
    }

    /// Drive sources supplied by role name (the blueprint / CLI production path).
    /// Resolves the keyed supply against `self.sources` then runs; the run loop is
    /// `run`'s, unchanged. A mis-bind is a named `SourceBindError`, not the
    /// raw-index arity panic.
    pub fn run_bound(
        &mut self,
        supply: Vec<(String, Box<dyn Source>)>,
    ) -> Result<(), SourceBindError> {
        let resolved = bind_sources(&self.sources, supply)?;
        self.run(resolved);
        Ok(())
    }
}

/// The firing predicate (C5/C6): does this node re-evaluate this cycle? A node
/// fires when *any* of its input groups fires (OR). A `Firing::Any` input fires
/// the node when it is fresh this cycle (`fresh_at == cycle_id`). A barrier group
/// fires when >=1 member is fresh this cycle AND every member carries
/// `last_ts == ts` — the ">=1 fresh" clause is the once-per-timestamp guard (a
/// group completed in a prior cycle has no fresh member now, so it does not
/// re-fire).
fn fires(firing: &[Firing], slots: &[SlotState], cycle_id: u64, ts: Timestamp) -> bool {
    // mode A: any fire-on-any-fresh input that is fresh this cycle
    for (i, f) in firing.iter().enumerate() {
        if matches!(f, Firing::Any) && slots[i].fresh_at == cycle_id {
            return true;
        }
    }
    // mode B: each distinct barrier group fires when complete this cycle
    if let Some(max_group) = firing.iter().filter_map(group_id).max() {
        for g in 0..=max_group {
            let mut has_member = false;
            let mut any_fresh = false;
            let mut all_at_ts = true;
            for (i, f) in firing.iter().enumerate() {
                if group_id(f) == Some(g) {
                    has_member = true;
                    if slots[i].fresh_at == cycle_id {
                        any_fresh = true;
                    }
                    if slots[i].last_ts != ts {
                        all_at_ts = false;
                    }
                }
            }
            if has_member && any_fresh && all_at_ts {
                return true;
            }
        }
    }
    false
}

/// The barrier group id of a firing policy, or `None` for mode A.
fn group_id(f: &Firing) -> Option<u8> {
    match f {
        Firing::Any => None,
        Firing::Barrier(g) => Some(*g),
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    // PortSpec / NodeSchema name the fixtures' declared signatures, brought in here
    // (production code reads them via the carried signatures, never naming the types).
    use aura_core::{FieldSpec, NodeSchema, PortSpec};
    use aura_std::{Bias, Recorder, Sma, SimBroker, Sub, When};
    use std::sync::mpsc;

    /// Build an f64 source stream from (timestamp, value) points.
    fn f64_stream(points: &[(i64, f64)]) -> Vec<(Timestamp, Scalar)> {
        points.iter().map(|&(t, v)| (Timestamp(t), Scalar::f64(v))).collect()
    }

    #[test]
    fn vec_source_reports_no_residency_by_default() {
        // The eager VecSource has no streaming ring, so it takes the Source trait
        // default: residency is "not reported" (None), distinct from Some(0). The
        // probe is reachable through `&dyn Source` without downcasting (#95).
        let v = VecSource::new(f64_stream(&[(1, 1.0), (2, 2.0)]));
        let probe: &dyn Source = &v;
        assert_eq!(probe.resident_records(), None);
    }

    #[test]
    fn seeded_source_same_seed_same_stream() {
        // Same seed -> bit-identical stream (C1/C12 determinism).
        let spec = SyntheticSpec { start: 1.0, len: 32, step: 1 };
        let drain = |seed| {
            let mut s = spec.source(seed);
            let mut out: Vec<(Timestamp, Scalar)> = Vec::new();
            while let Some(item) = Source::next(&mut s) {
                out.push(item);
            }
            out
        };
        assert_eq!(drain(7), drain(7));
    }

    #[test]
    fn seeded_source_different_seed_differs() {
        // Different seeds -> different streams.
        let spec = SyntheticSpec { start: 1.0, len: 32, step: 1 };
        let drain = |seed| {
            let mut s = spec.source(seed);
            let mut out: Vec<(Timestamp, Scalar)> = Vec::new();
            while let Some(item) = Source::next(&mut s) {
                out.push(item);
            }
            out
        };
        assert_ne!(drain(1), drain(2));
    }

    #[test]
    fn vec_source_peek_is_non_consuming_and_idempotent() {
        let mut s = VecSource::new(vec![(Timestamp(1), Scalar::f64(10.0)), (Timestamp(2), Scalar::f64(20.0))]);
        // peek twice: same head, no advance.
        assert_eq!(s.peek(), Some(Timestamp(1)));
        assert_eq!(s.peek(), Some(Timestamp(1)));
        // next pops it; peek now reflects the new head.
        assert_eq!(Source::next(&mut s), Some((Timestamp(1), Scalar::f64(10.0))));
        assert_eq!(s.peek(), Some(Timestamp(2)));
    }

    #[test]
    fn vec_source_exhaustion_yields_none() {
        let mut s = VecSource::new(vec![(Timestamp(5), Scalar::i64(7))]);
        assert_eq!(Source::next(&mut s), Some((Timestamp(5), Scalar::i64(7))));
        assert_eq!(s.peek(), None);
        assert_eq!(Source::next(&mut s), None);
    }

    #[test]
    fn vec_source_empty_peeks_none() {
        let mut s = VecSource::new(vec![]);
        assert_eq!(s.peek(), None);
        assert_eq!(Source::next(&mut s), None);
    }

    #[test]
    fn vec_source_bounds_are_buffer_ends_and_cursor_independent() {
        let mut s = VecSource::new(f64_stream(&[(10, 1.0), (20, 2.0), (30, 3.0)]));
        // bounds = (first ts, last ts) of the buffer
        assert_eq!(s.bounds(), Some((Timestamp(10), Timestamp(30))));
        // cursor-independent: advancing does not move the reported extent
        let _ = Source::next(&mut s);
        assert_eq!(s.bounds(), Some((Timestamp(10), Timestamp(30))));
        // an empty buffer has no extent
        let empty = VecSource::new(vec![]);
        assert_eq!(empty.bounds(), None);
    }

    #[test]
    fn bind_sources_resolves_by_name_into_spec_order() {
        // specs declare roles [b, a]; supply arrives [a, b] (out of order).
        let specs = vec![
            SourceSpec::named("b", ScalarKind::F64, vec![Target { node: 0, slot: 0 }]),
            SourceSpec::named("a", ScalarKind::F64, vec![Target { node: 0, slot: 1 }]),
        ];
        let supply: Vec<(String, Box<dyn Source>)> = vec![
            ("a".into(), Box::new(VecSource::new(f64_stream(&[(1, 10.0)])))),
            ("b".into(), Box::new(VecSource::new(f64_stream(&[(1, 20.0)])))),
        ];
        let resolved = bind_sources(&specs, supply).expect("both roles matched");
        // emitted in SPEC order (b then a): first source peeks the "b" stream (ts 1),
        // and popping yields 20.0 (b's value), proving b came first.
        assert_eq!(resolved.len(), 2);
        assert_eq!(resolved[0].peek(), Some(Timestamp(1)));
    }

    #[test]
    fn bind_sources_missing_feed_is_named() {
        let specs = vec![SourceSpec::named("close", ScalarKind::F64, vec![Target { node: 0, slot: 0 }])];
        let supply: Vec<(String, Box<dyn Source>)> =
            vec![("high".into(), Box::new(VecSource::new(f64_stream(&[(1, 1.0)]))))];
        // `.map(|_| ())` before `unwrap_err`: `Vec<Box<dyn Source>>` (the `Ok` arm)
        // isn't `Debug` (`Source` is intentionally not `Debug`), and `unwrap_err`
        // requires `T: Debug` for its panic message — collapsing the `Ok` payload
        // to `()` sidesteps that without touching the error path under test.
        assert_eq!(
            bind_sources(&specs, supply).map(|_| ()).unwrap_err(),
            SourceBindError::MissingFeed { role: "close".into() }
        );
    }

    #[test]
    fn bind_sources_extra_feed_is_named() {
        let specs = vec![SourceSpec::named("close", ScalarKind::F64, vec![Target { node: 0, slot: 0 }])];
        let supply: Vec<(String, Box<dyn Source>)> = vec![
            ("close".into(), Box::new(VecSource::new(f64_stream(&[(1, 1.0)])))),
            ("high".into(), Box::new(VecSource::new(f64_stream(&[(1, 2.0)])))),
        ];
        assert_eq!(
            bind_sources(&specs, supply).map(|_| ()).unwrap_err(),
            SourceBindError::ExtraFeed { role: "high".into() }
        );
    }

    #[test]
    fn bind_sources_duplicate_feed_is_named() {
        let specs = vec![SourceSpec::named("close", ScalarKind::F64, vec![Target { node: 0, slot: 0 }])];
        let supply: Vec<(String, Box<dyn Source>)> = vec![
            ("close".into(), Box::new(VecSource::new(f64_stream(&[(1, 1.0)])))),
            ("close".into(), Box::new(VecSource::new(f64_stream(&[(1, 2.0)])))),
        ];
        assert_eq!(
            bind_sources(&specs, supply).map(|_| ()).unwrap_err(),
            SourceBindError::DuplicateFeed { role: "close".into() }
        );
    }

    #[test]
    fn bind_sources_unnamed_spec_is_named() {
        let specs = vec![SourceSpec::raw(ScalarKind::F64, vec![Target { node: 0, slot: 0 }])];
        let supply: Vec<(String, Box<dyn Source>)> =
            vec![("close".into(), Box::new(VecSource::new(f64_stream(&[(1, 1.0)]))))];
        assert_eq!(
            bind_sources(&specs, supply).map(|_| ()).unwrap_err(),
            SourceBindError::UnnamedSource { index: 0 }
        );
    }

    /// The named-error prose a keyed-supply caller renders (the by-name refusal
    /// surface for decoupled-supply callers, #124/#275).
    #[test]
    fn source_bind_error_display_names_the_role() {
        assert_eq!(
            SourceBindError::MissingFeed { role: "close".into() }.to_string(),
            "the graph declares source role \"close\" but the supply provides no such feed"
        );
        assert_eq!(
            SourceBindError::DuplicateFeed { role: "high".into() }.to_string(),
            "the supply provides feed \"high\" more than once"
        );
    }

    #[test]
    fn window_of_folds_union_extent_across_sources() {
        let a: Box<dyn Source> = Box::new(VecSource::new(f64_stream(&[(10, 1.0), (40, 2.0)])));
        let b: Box<dyn Source> = Box::new(VecSource::new(f64_stream(&[(5, 1.0), (25, 2.0)])));
        // union extent: min first, max last
        assert_eq!(window_of(&[a, b]), Some((Timestamp(5), Timestamp(40))));
        // all-empty -> None
        let e1: Box<dyn Source> = Box::new(VecSource::new(vec![]));
        let e2: Box<dyn Source> = Box::new(VecSource::new(vec![]));
        assert_eq!(window_of(&[e1, e2]), None);
    }

    #[test]
    fn synthetic_source_bounds_match_analytic_span() {
        let spec = SyntheticSpec { start: 1.0, len: 32, step: 1 };
        let s = spec.source(7);
        // first ts = 1 (i=0); last ts = (len-1)*step + 1
        let last = (spec.len as i64 - 1) * spec.step + 1;
        assert_eq!(s.bounds(), Some((Timestamp(1), Timestamp(last))));
    }

    /// One f64 input port with the given firing policy (lookback 1 is the bootstrap
    /// default for these test fixtures, supplied by their `lookbacks()`).
    fn f64_port(firing: Firing) -> PortSpec {
        PortSpec { kind: ScalarKind::F64, firing, name: "in".into() }
    }

    /// Bootstrap a hand-wired graph from boxed nodes + their declared signatures.
    /// The signatures parallel `nodes` (one per node, in order); bootstrap reads
    /// kinds/firing from them and depth from each node's `lookbacks()`.
    fn boot(
        nodes: Vec<Box<dyn Node>>,
        signatures: Vec<NodeSchema>,
        sources: Vec<SourceSpec>,
        edges: Vec<Edge>,
    ) -> Result<Harness, BootstrapError> {
        Harness::bootstrap(FlatGraph { nodes, signatures, sources, edges, taps: Vec::new() })
    }

    /// The declared signature of a `Recorder` over `kinds` with the given firing.
    fn recorder_sig(kinds: &[ScalarKind], firing: Firing) -> NodeSchema {
        NodeSchema {
            inputs: kinds.iter().enumerate().map(|(i, &kind)| PortSpec { kind, firing, name: format!("col[{i}]") }).collect(),
            output: vec![],
            params: vec![],
        }
    }

    // --- firing-policy fixtures (test-local; not library nodes — C9: examples
    // for the engine's own tests, no speculative aura-std surface) ---

    /// Mode A as-of join: a 2-input f64 sum that fires whenever either input is
    /// fresh, holding the other. Warm-up returns None until both have a value.
    struct AsOfSum {
        out: [Cell; 1],
    }
    impl AsOfSum {
        fn sig() -> NodeSchema {
            NodeSchema {
                inputs: vec![f64_port(Firing::Any), f64_port(Firing::Any)],
                output: vec![FieldSpec { name: "value".into(), kind: ScalarKind::F64 }],
                params: vec![],
            }
        }
    }
    impl Node for AsOfSum {
        fn lookbacks(&self) -> Vec<usize> {
            vec![1, 1]
        }
        fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Cell]> {
            let a = ctx.f64_in(0);
            let b = ctx.f64_in(1);
            if a.is_empty() || b.is_empty() {
                return None;
            }
            self.out[0] = Cell::from_f64(a[0] + b[0]);
            Some(&self.out)
        }
    }

    /// Mode B barrier join: a 2-input f64 sum that fires only when both inputs
    /// share the current cycle timestamp (both warm by construction when it fires).
    struct BarrierSum {
        out: [Cell; 1],
    }
    impl BarrierSum {
        fn sig() -> NodeSchema {
            NodeSchema {
                inputs: vec![f64_port(Firing::Barrier(0)), f64_port(Firing::Barrier(0))],
                output: vec![FieldSpec { name: "value".into(), kind: ScalarKind::F64 }],
                params: vec![],
            }
        }
    }
    impl Node for BarrierSum {
        fn lookbacks(&self) -> Vec<usize> {
            vec![1, 1]
        }
        fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Cell]> {
            self.out[0] = Cell::from_f64(ctx.f64_in(0)[0] + ctx.f64_in(1)[0]);
            Some(&self.out)
        }
    }

    /// Test gate: `f64 -> bool`, true iff the sample is strictly above `c`.
    /// Derives a Bool gate stream **in-graph**, so the gate shares its
    /// source's cycle (separate sources at the same stamp are separate
    /// cycles — an in-graph derivation is how real gates arrive).
    struct AboveGate {
        c: f64,
        out: [Cell; 1],
    }
    impl AboveGate {
        fn sig() -> NodeSchema {
            NodeSchema {
                inputs: vec![f64_port(Firing::Any)],
                output: vec![FieldSpec { name: "value".into(), kind: ScalarKind::Bool }],
                params: vec![],
            }
        }
    }
    impl Node for AboveGate {
        fn lookbacks(&self) -> Vec<usize> {
            vec![1]
        }
        fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Cell]> {
            let v = ctx.f64_in(0);
            if v.is_empty() {
                return None;
            }
            self.out[0] = Cell::from_bool(v[0] > self.c);
            Some(&self.out)
        }
    }

    /// Mixed A+B: barrier pair (inputs 0,1 in group 0) plus an as-of input
    /// (input 2). Fires when the pair completes (holding input 2) OR when input 2
    /// ticks (holding the pair) — the OR-combine.
    struct MixedSum {
        out: [Cell; 1],
    }
    impl MixedSum {
        fn sig() -> NodeSchema {
            NodeSchema {
                inputs: vec![
                    f64_port(Firing::Barrier(0)),
                    f64_port(Firing::Barrier(0)),
                    f64_port(Firing::Any),
                ],
                output: vec![FieldSpec { name: "value".into(), kind: ScalarKind::F64 }],
                params: vec![],
            }
        }
    }
    impl Node for MixedSum {
        fn lookbacks(&self) -> Vec<usize> {
            vec![1, 1, 1]
        }
        fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Cell]> {
            let a = ctx.f64_in(0);
            let b = ctx.f64_in(1);
            let c = ctx.f64_in(2);
            if a.is_empty() || b.is_empty() || c.is_empty() {
                return None;
            }
            self.out[0] = Cell::from_f64(a[0] + b[0] + c[0]);
            Some(&self.out)
        }
    }

    /// A neutral multi-field producer: five f64 inputs bundled into one 5-field
    /// record. No trading-domain logic — it proves the K > 1 output mechanism in
    /// isolation. The five inputs are a Barrier(0) group, so the node emits one
    /// complete bar only when all five share the cycle's timestamp.
    struct Ohlcv {
        out: [Cell; 5],
    }
    impl Ohlcv {
        fn sig() -> NodeSchema {
            NodeSchema {
                inputs: vec![f64_port(Firing::Barrier(0)); 5],
                output: vec![
                    FieldSpec { name: "open".into(), kind: ScalarKind::F64 },
                    FieldSpec { name: "high".into(), kind: ScalarKind::F64 },
                    FieldSpec { name: "low".into(), kind: ScalarKind::F64 },
                    FieldSpec { name: "close".into(), kind: ScalarKind::F64 },
                    FieldSpec { name: "volume".into(), kind: ScalarKind::F64 },
                ],
                params: vec![],
            }
        }
    }
    impl Node for Ohlcv {
        fn lookbacks(&self) -> Vec<usize> {
            vec![1; 5]
        }
        fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Cell]> {
            for i in 0..5 {
                let w = ctx.f64_in(i);
                if w.is_empty() {
                    return None; // not yet warmed
                }
                self.out[i] = Cell::from_f64(w[0]);
            }
            Some(&self.out) // one 5-field record, all fields co-fresh
        }
    }

    /// A producer whose output record mixes kinds: field 0 is f64, field 1 is i64.
    /// Used only to prove the bootstrap kind check is per-field (field 0 would bind
    /// into an f64 slot; field 1 would not). Its `eval` never runs in these tests —
    /// bootstrap rejects the wiring first.
    struct TwoField {
        out: [Cell; 2],
    }
    impl TwoField {
        fn sig() -> NodeSchema {
            NodeSchema {
                inputs: vec![f64_port(Firing::Any)],
                output: vec![
                    FieldSpec { name: "f".into(), kind: ScalarKind::F64 },
                    FieldSpec { name: "i".into(), kind: ScalarKind::I64 },
                ],
                params: vec![],
            }
        }
    }
    impl Node for TwoField {
        fn lookbacks(&self) -> Vec<usize> {
            vec![1]
        }
        fn eval(&mut self, _ctx: Ctx<'_>) -> Option<&[Cell]> {
            self.out[0] = Cell::from_f64(0.0);
            self.out[1] = Cell::from_i64(0);
            Some(&self.out)
        }
    }

    /// A node that records AND forwards: it sends `(now, value)` out of the graph
    /// (sink side effect) and returns its value as a one-field output the engine
    /// forwards downstream (producer). Proves the C8 "both" role.
    struct TapForward {
        out: [Cell; 1],
        tx: mpsc::Sender<(Timestamp, Vec<Scalar>)>,
    }
    impl TapForward {
        fn sig() -> NodeSchema {
            NodeSchema {
                inputs: vec![f64_port(Firing::Any)],
                output: vec![FieldSpec { name: "value".into(), kind: ScalarKind::F64 }],
                params: vec![],
            }
        }
    }
    impl Node for TapForward {
        fn lookbacks(&self) -> Vec<usize> {
            vec![1]
        }
        fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Cell]> {
            let w = ctx.f64_in(0);
            if w.is_empty() {
                return None;
            }
            let v = w[0];
            let _ = self.tx.send((ctx.now(), vec![Scalar::f64(v)])); // sink side effect
            self.out[0] = Cell::from_f64(v);
            Some(&self.out) // producer output: engine forwards it
        }
    }

    #[test]
    fn chain_source_sma_runs() {
        // node 0 = SMA(3); source -> SMA(3).in0; node 1 = Recorder taps node 0.
        let (tx, rx) = mpsc::channel();
        let mut h = boot(
            vec![
                Box::new(Sma::new(3)),
                Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx)),
            ],
            vec![
                Sma::builder().schema().clone(),
                recorder_sig(&[ScalarKind::F64], Firing::Any),
            ],
            vec![SourceSpec::raw(ScalarKind::F64, vec![Target { node: 0, slot: 0 }])],
            vec![Edge { from: 0, to: 1, slot: 0, from_field: 0 }],
        )
        .expect("valid");
        h.run(vec![Box::new(VecSource::new(f64_stream(&[(1, 1.0), (2, 2.0), (3, 3.0), (4, 4.0), (5, 5.0)])))]);
        let got: Vec<(Timestamp, Vec<Scalar>)> = rx.try_iter().collect();
        // SMA(3) warms at cycle 3; the recorder captures only fired cycles, each
        // tagged with the cycle's timestamp (sparse — no None hold-rows).
        assert_eq!(
            got,
            vec![
                (Timestamp(3), vec![Scalar::f64(2.0)]),
                (Timestamp(4), vec![Scalar::f64(3.0)]),
                (Timestamp(5), vec![Scalar::f64(4.0)]),
            ]
        );
    }

    #[test]
    fn fan_out_join_dag_runs_deterministically() {
        // 0 = SMA(2), 1 = SMA(4), 2 = Sub; source fans into both SMAs; SMAs join
        // into Sub; node 3 = Recorder taps Sub — the 0003 baseline on the new API.
        let build = |tx| {
            boot(
                vec![
                    Box::new(Sma::new(2)),
                    Box::new(Sma::new(4)),
                    Box::new(Sub::new()),
                    Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx)),
                ],
            vec![
                Sma::builder().schema().clone(),
                Sma::builder().schema().clone(),
                Sub::builder().schema().clone(),
                recorder_sig(&[ScalarKind::F64], Firing::Any),
            ],
                vec![SourceSpec::raw(ScalarKind::F64, vec![Target { node: 0, slot: 0 }, Target { node: 1, slot: 0 }])],
                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 },
                ],
            )
            .expect("valid DAG")
        };
        let prices = f64_stream(&[(1, 10.0), (2, 12.0), (3, 14.0), (4, 16.0), (5, 18.0), (6, 20.0)]);

        let (tx, rx) = mpsc::channel();
        let mut h = build(tx);
        h.run(vec![Box::new(VecSource::new(prices.clone()))]);
        let out: Vec<(Timestamp, Vec<Scalar>)> = rx.try_iter().collect();
        // Sub fires once SMA(4) is warm (cycle 4): 15-13, 17-15, 19-17 -> 2.
        assert_eq!(
            out,
            vec![
                (Timestamp(4), vec![Scalar::f64(2.0)]),
                (Timestamp(5), vec![Scalar::f64(2.0)]),
                (Timestamp(6), vec![Scalar::f64(2.0)]),
            ]
        );

        // determinism (C1): a second identical run drains a bit-identical stream.
        let (tx2, rx2) = mpsc::channel();
        let mut h2 = build(tx2);
        h2.run(vec![Box::new(VecSource::new(prices))]);
        let out2: Vec<(Timestamp, Vec<Scalar>)> = rx2.try_iter().collect();
        assert_eq!(out2, out);
    }

    #[test]
    fn mode_a_as_of_fires_on_any_fresh_and_holds() {
        // AsOfSum @0; node 1 = Recorder taps it. source 0 ticks t=1..4; source 1
        // ticks t=2,4 (slower); both AsOfSum inputs Any.
        let build = |tx| {
            boot(
                vec![
                    Box::new(AsOfSum { out: [Cell::from_f64(0.0)] }),
                    Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx)),
                ],
            vec![
                AsOfSum::sig(),
                recorder_sig(&[ScalarKind::F64], Firing::Any),
            ],
                vec![
                    SourceSpec::raw(ScalarKind::F64, vec![Target { node: 0, slot: 0 }]),
                    SourceSpec::raw(ScalarKind::F64, vec![Target { node: 0, slot: 1 }]),
                ],
                vec![Edge { from: 0, to: 1, slot: 0, from_field: 0 }],
            )
            .expect("valid")
        };
        let s0 = f64_stream(&[(1, 10.0), (2, 20.0), (3, 30.0), (4, 40.0)]);
        let s1 = f64_stream(&[(2, 100.0), (4, 200.0)]);

        let (tx, rx) = mpsc::channel();
        let mut h = build(tx);
        h.run(vec![Box::new(VecSource::new(s0.clone())), Box::new(VecSource::new(s1.clone()))]);
        let out: Vec<(Timestamp, Vec<Scalar>)> = rx.try_iter().collect();
        // holds s1=100 across t=3 and the t=4 s0-cycle; emits on every tick once warm.
        assert_eq!(
            out,
            vec![
                (Timestamp(2), vec![Scalar::f64(120.0)]),
                (Timestamp(3), vec![Scalar::f64(130.0)]),
                (Timestamp(4), vec![Scalar::f64(140.0)]),
                (Timestamp(4), vec![Scalar::f64(240.0)]),
            ]
        );

        let (tx2, rx2) = mpsc::channel();
        let mut h2 = build(tx2);
        h2.run(vec![Box::new(VecSource::new(s0)), Box::new(VecSource::new(s1))]);
        let out2: Vec<(Timestamp, Vec<Scalar>)> = rx2.try_iter().collect();
        assert_eq!(out2, out); // deterministic
    }

    #[test]
    fn run_bound_out_of_order_matches_positional_run() {
        let build = |tx, specs: Vec<SourceSpec>| {
            boot(
                vec![
                    Box::new(AsOfSum { out: [Cell::from_f64(0.0)] }),
                    Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx)),
                ],
                vec![AsOfSum::sig(), recorder_sig(&[ScalarKind::F64], Firing::Any)],
                specs,
                vec![Edge { from: 0, to: 1, slot: 0, from_field: 0 }],
            )
            .expect("valid")
        };
        let s0 = f64_stream(&[(1, 10.0), (2, 20.0), (3, 30.0), (4, 40.0)]);
        let s1 = f64_stream(&[(2, 100.0), (4, 200.0)]);

        // positional: raw specs, sources supplied in declaration order.
        let (tx, rx) = mpsc::channel();
        let mut h = build(tx, vec![
            SourceSpec::raw(ScalarKind::F64, vec![Target { node: 0, slot: 0 }]),
            SourceSpec::raw(ScalarKind::F64, vec![Target { node: 0, slot: 1 }]),
        ]);
        h.run(vec![Box::new(VecSource::new(s0.clone())), Box::new(VecSource::new(s1.clone()))]);
        let positional: Vec<(Timestamp, Vec<Scalar>)> = rx.try_iter().collect();

        // by-name: named specs "a"@slot0 / "b"@slot1; supply given b-then-a (reversed).
        let (tx2, rx2) = mpsc::channel();
        let mut h2 = build(tx2, vec![
            SourceSpec::named("a", ScalarKind::F64, vec![Target { node: 0, slot: 0 }]),
            SourceSpec::named("b", ScalarKind::F64, vec![Target { node: 0, slot: 1 }]),
        ]);
        h2.run_bound(vec![
            ("b".into(), Box::new(VecSource::new(s1))),
            ("a".into(), Box::new(VecSource::new(s0))),
        ]).expect("both roles bind");
        let by_name: Vec<(Timestamp, Vec<Scalar>)> = rx2.try_iter().collect();

        assert!(!positional.is_empty(), "positional run drained empty");
        assert_eq!(by_name, positional, "by-name run is byte-identical to positional");
    }

    #[test]
    fn mode_b_barrier_fires_only_on_timestamp_coincidence() {
        // identical wiring to mode A, but both BarrierSum inputs are Barrier(0).
        let build = |tx| {
            boot(
                vec![
                    Box::new(BarrierSum { out: [Cell::from_f64(0.0)] }),
                    Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx)),
                ],
            vec![
                BarrierSum::sig(),
                recorder_sig(&[ScalarKind::F64], Firing::Any),
            ],
                vec![
                    SourceSpec::raw(ScalarKind::F64, vec![Target { node: 0, slot: 0 }]),
                    SourceSpec::raw(ScalarKind::F64, vec![Target { node: 0, slot: 1 }]),
                ],
                vec![Edge { from: 0, to: 1, slot: 0, from_field: 0 }],
            )
            .expect("valid")
        };
        let s0 = f64_stream(&[(1, 10.0), (2, 20.0), (3, 30.0), (4, 40.0)]);
        let s1 = f64_stream(&[(2, 100.0), (4, 200.0)]);

        let (tx, rx) = mpsc::channel();
        let mut h = build(tx);
        h.run(vec![Box::new(VecSource::new(s0.clone())), Box::new(VecSource::new(s1.clone()))]);
        let out: Vec<(Timestamp, Vec<Scalar>)> = rx.try_iter().collect();
        // records ONLY at t=2 and t=4 where both inputs share the timestamp.
        assert_eq!(
            out,
            vec![
                (Timestamp(2), vec![Scalar::f64(120.0)]),
                (Timestamp(4), vec![Scalar::f64(240.0)]),
            ]
        );

        let (tx2, rx2) = mpsc::channel();
        let mut h2 = build(tx2);
        h2.run(vec![Box::new(VecSource::new(s0)), Box::new(VecSource::new(s1))]);
        let out2: Vec<(Timestamp, Vec<Scalar>)> = rx2.try_iter().collect();
        assert_eq!(out2, out); // deterministic
    }

    #[test]
    fn within_source_diamond_rejoin_barrier_fires() {
        // One source fans out through SMA(2), SMA(4) that rejoin at a Barrier(0)
        // node; node 3 = Recorder taps the barrier. Every push in a cycle carries
        // that cycle's timestamp, so once both SMAs warm and emit in the same
        // cycle, both barrier inputs share the timestamp and the barrier fires.
        let build = |tx| {
            boot(
                vec![
                    Box::new(Sma::new(2)),
                    Box::new(Sma::new(4)),
                    Box::new(BarrierSum { out: [Cell::from_f64(0.0)] }),
                    Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx)),
                ],
            vec![
                Sma::builder().schema().clone(),
                Sma::builder().schema().clone(),
                BarrierSum::sig(),
                recorder_sig(&[ScalarKind::F64], Firing::Any),
            ],
                vec![SourceSpec::raw(ScalarKind::F64, vec![Target { node: 0, slot: 0 }, Target { node: 1, slot: 0 }])],
                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 },
                ],
            )
            .expect("valid DAG")
        };
        let prices = f64_stream(&[(1, 10.0), (2, 12.0), (3, 14.0), (4, 16.0), (5, 18.0), (6, 20.0)]);

        let (tx, rx) = mpsc::channel();
        let mut h = build(tx);
        h.run(vec![Box::new(VecSource::new(prices.clone()))]);
        let out: Vec<(Timestamp, Vec<Scalar>)> = rx.try_iter().collect();
        // SMA(4) warms at cycle 4; from then both paths emit each cycle at the same
        // timestamp, so the barrier fires: SMA(2)+SMA(4) = 15+13, 17+15, 19+17.
        assert_eq!(
            out,
            vec![
                (Timestamp(4), vec![Scalar::f64(28.0)]),
                (Timestamp(5), vec![Scalar::f64(32.0)]),
                (Timestamp(6), vec![Scalar::f64(36.0)]),
            ]
        );

        let (tx2, rx2) = mpsc::channel();
        let mut h2 = build(tx2);
        h2.run(vec![Box::new(VecSource::new(prices))]);
        let out2: Vec<(Timestamp, Vec<Scalar>)> = rx2.try_iter().collect();
        assert_eq!(out2, out);
    }

    #[test]
    fn when_gates_a_reducer_by_composition_equal_to_the_filtered_subsequence() {
        // The composition claim the clock-enable design rests on (#281):
        // SMA(When(x, gate), N) over a stream equals SMA(N) fed the gate-true
        // subsequence directly. When's quiet cycles do not advance SMA's
        // freshness, so the reducer state moves per forwarded sample, not per
        // cycle — no gated variant of any reducer exists or is needed.
        let prices = f64_stream(&[(1, 10.0), (2, 30.0), (3, 20.0), (4, 40.0), (5, 15.0), (6, 60.0)]);
        // gate = price > 25.0 -> [f, t, f, t, f, t]; gated subsequence [30, 40, 60].
        let (tx, rx) = mpsc::channel();
        let mut h = boot(
            vec![
                Box::new(AboveGate { c: 25.0, out: [Cell::from_bool(false)] }),
                Box::new(When::new()),
                Box::new(Sma::new(2)),
                Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx)),
            ],
            vec![
                AboveGate::sig(),
                When::builder().schema().clone(),
                Sma::builder().schema().clone(),
                recorder_sig(&[ScalarKind::F64], Firing::Any),
            ],
            vec![SourceSpec::raw(
                ScalarKind::F64,
                vec![Target { node: 0, slot: 0 }, Target { node: 1, slot: 0 }],
            )],
            vec![
                Edge { from: 0, to: 1, slot: 1, from_field: 0 },
                Edge { from: 1, to: 2, slot: 0, from_field: 0 },
                Edge { from: 2, to: 3, slot: 0, from_field: 0 },
            ],
        )
        .expect("valid DAG");
        h.run(vec![Box::new(VecSource::new(prices))]);
        let gated: Vec<(Timestamp, Vec<Scalar>)> = rx.try_iter().collect();

        // SMA(2) over [30, 40, 60]: warms at the 2nd forwarded sample.
        assert_eq!(
            gated,
            vec![
                (Timestamp(4), vec![Scalar::f64(35.0)]),
                (Timestamp(6), vec![Scalar::f64(50.0)]),
            ],
            "gated SMA advances once per forwarded sample"
        );

        // Reference: the same SMA(2) fed the gated subsequence directly.
        let filtered = f64_stream(&[(2, 30.0), (4, 40.0), (6, 60.0)]);
        let (tx2, rx2) = mpsc::channel();
        let mut h2 = boot(
            vec![
                Box::new(Sma::new(2)),
                Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx2)),
            ],
            vec![Sma::builder().schema().clone(), recorder_sig(&[ScalarKind::F64], Firing::Any)],
            vec![SourceSpec::raw(ScalarKind::F64, vec![Target { node: 0, slot: 0 }])],
            vec![Edge { from: 0, to: 1, slot: 0, from_field: 0 }],
        )
        .expect("valid DAG");
        h2.run(vec![Box::new(VecSource::new(filtered))]);
        let reference: Vec<(Timestamp, Vec<Scalar>)> = rx2.try_iter().collect();

        assert_eq!(gated, reference, "composition equals the filtered subsequence");
    }

    #[test]
    fn a_quiet_when_stalls_its_barrier_group_for_the_cycle() {
        // The deliberate stall pin (#281): a quiet When member is NOT at the
        // current cycle timestamp, so its Barrier(0) group does not fire that
        // cycle (the barrier contract); the group fires again on the next
        // forwarded sample.
        let prices = f64_stream(&[(1, 10.0), (2, 30.0), (3, 20.0), (4, 40.0)]);
        // gate = price > 25.0 -> [f, t, f, t]; When forwards at t=2 and t=4 only.
        let (tx, rx) = mpsc::channel();
        let mut h = boot(
            vec![
                Box::new(AboveGate { c: 25.0, out: [Cell::from_bool(false)] }),
                Box::new(When::new()),
                Box::new(BarrierSum { out: [Cell::from_f64(0.0)] }),
                Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx)),
            ],
            vec![
                AboveGate::sig(),
                When::builder().schema().clone(),
                BarrierSum::sig(),
                recorder_sig(&[ScalarKind::F64], Firing::Any),
            ],
            vec![SourceSpec::raw(
                ScalarKind::F64,
                vec![Target { node: 0, slot: 0 }, Target { node: 1, slot: 0 }, Target { node: 2, slot: 1 }],
            )],
            vec![
                Edge { from: 0, to: 1, slot: 1, from_field: 0 },
                Edge { from: 1, to: 2, slot: 0, from_field: 0 },
                Edge { from: 2, to: 3, slot: 0, from_field: 0 },
            ],
        )
        .expect("valid DAG");
        h.run(vec![Box::new(VecSource::new(prices))]);
        let out: Vec<(Timestamp, Vec<Scalar>)> = rx.try_iter().collect();
        // t=1: gate false -> When quiet -> barrier member 0 never fresh -> no fire.
        // t=2: When forwards 30; both members at t=2 -> fires 30+30=60.
        // t=3: When quiet -> member 0 stays at t=2 -> the group STALLS (no t=3 row).
        // t=4: When forwards 40; both members at t=4 -> fires 40+40=80.
        assert_eq!(
            out,
            vec![
                (Timestamp(2), vec![Scalar::f64(60.0)]),
                (Timestamp(4), vec![Scalar::f64(80.0)]),
            ]
        );
    }

    #[test]
    fn a_producer_with_no_out_edge_runs_inertly_identical_to_its_removal() {
        // CURRENT-BEHAVIOUR PIN (#282 substrate): a producer node with a
        // NON-EMPTY output and ZERO out-edges compiles, bootstraps, and runs —
        // its per-cycle Some(row) is simply discarded, disturbing nothing. This
        // is the exact shape of an *unbound declared tap*: the tap mechanism
        // will let such a producer be declared and left un-recorded in a sweep,
        // and this pins that the engine already tolerates it (no DCE needed for
        // correctness — check_ports_connected gates only input slots, and the
        // Kahn sort emits a no-out-edge node as a valid sink). The proof is
        // observable identity: the run's drained output is byte-for-byte the same
        // with the dangling producer present as with it removed.
        let prices = f64_stream(&[(1, 10.0), (2, 20.0), (3, 30.0)]);

        // WITH a dangling producer: source fans out to Sma(1) [node 0, a real
        // producer that returns Some every warm cycle, wired to NOTHING] and to
        // a Recorder [node 1] fed directly by the source.
        let (tx, rx) = mpsc::channel();
        let mut h_with = boot(
            vec![
                Box::new(Sma::new(1)),
                Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx)),
            ],
            vec![Sma::builder().schema().clone(), recorder_sig(&[ScalarKind::F64], Firing::Any)],
            vec![SourceSpec::raw(
                ScalarKind::F64,
                vec![Target { node: 0, slot: 0 }, Target { node: 1, slot: 0 }],
            )],
            vec![], // NO edge out of node 0 — the producer dangles
        )
        .expect("a no-out-edge producer is a valid graph");
        h_with.run(vec![Box::new(VecSource::new(prices.clone()))]);
        let with_dangling: Vec<(Timestamp, Vec<Scalar>)> = rx.try_iter().collect();

        // WITHOUT it: just the Recorder, fed by the same source.
        let (tx2, rx2) = mpsc::channel();
        let mut h_without = boot(
            vec![Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx2))],
            vec![recorder_sig(&[ScalarKind::F64], Firing::Any)],
            vec![SourceSpec::raw(ScalarKind::F64, vec![Target { node: 0, slot: 0 }])],
            vec![],
        )
        .expect("valid");
        h_without.run(vec![Box::new(VecSource::new(prices))]);
        let without_dangling: Vec<(Timestamp, Vec<Scalar>)> = rx2.try_iter().collect();

        assert_eq!(
            with_dangling,
            vec![
                (Timestamp(1), vec![Scalar::f64(10.0)]),
                (Timestamp(2), vec![Scalar::f64(20.0)]),
                (Timestamp(3), vec![Scalar::f64(30.0)]),
            ],
            "the drained output is the source series; the dangling producer changes nothing"
        );
        assert_eq!(
            with_dangling, without_dangling,
            "a producer wired to nothing is observably inert — identical to its removal"
        );
    }

    #[test]
    fn flat_graph_taps_are_inert_scaffolding_at_bootstrap() {
        // CURRENT-BEHAVIOUR PIN (#282, FlatTap + FlatGraph.taps): `bootstrap`
        // destructures `taps` as `taps: _` and never reads it — populating the
        // field with a `FlatTap` (even one naming a real producer) must not
        // perturb bootstrap sizing/wiring or the run's recorded output one bit.
        // This is the flat-graph-level twin of the compile-boundary pin
        // (`flat_taps_e2e.rs`): here the field carries data, and the claim is
        // that the data is inert, not merely that nothing populates it yet.
        let prices = f64_stream(&[(1, 10.0), (2, 20.0), (3, 30.0)]);

        fn graph(tx: mpsc::Sender<(Timestamp, Vec<Scalar>)>) -> FlatGraph {
            FlatGraph {
                nodes: vec![
                    Box::new(Sma::new(1)),
                    Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx)),
                ],
                signatures: vec![Sma::builder().schema().clone(), recorder_sig(&[ScalarKind::F64], Firing::Any)],
                sources: vec![SourceSpec::raw(
                    ScalarKind::F64,
                    vec![Target { node: 0, slot: 0 }, Target { node: 1, slot: 0 }],
                )],
                edges: vec![],
                taps: vec![],
            }
        }

        let (tx, rx) = mpsc::channel();
        let mut h_bare = Harness::bootstrap(graph(tx)).expect("empty taps list bootstraps");
        h_bare.run(vec![Box::new(VecSource::new(prices.clone()))]);
        let bare_out: Vec<(Timestamp, Vec<Scalar>)> = rx.try_iter().collect();

        let (tx2, rx2) = mpsc::channel();
        let mut with_taps = graph(tx2);
        with_taps.taps = vec![
            FlatTap { name: "p".into(), node: 0, field: 0 },
            FlatTap { name: "q".into(), node: 1, field: 0 },
        ];
        let mut h_tapped = Harness::bootstrap(with_taps).expect("a populated taps list bootstraps identically");
        h_tapped.run(vec![Box::new(VecSource::new(prices))]);
        let tapped_out: Vec<(Timestamp, Vec<Scalar>)> = rx2.try_iter().collect();

        assert_eq!(
            bare_out, tapped_out,
            "a populated FlatGraph.taps list must not change bootstrap/run behaviour in this slice"
        );
    }

    #[test]
    fn mixed_a_and_b_or_combine_on_one_node() {
        // MixedSum @0 (in0,in1 barrier group 0; in2 as-of); node 1 = Recorder taps it.
        let (tx, rx) = mpsc::channel();
        let mut h = boot(
            vec![
                Box::new(MixedSum { out: [Cell::from_f64(0.0)] }),
                Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx)),
            ],
            vec![
                MixedSum::sig(),
                recorder_sig(&[ScalarKind::F64], Firing::Any),
            ],
            vec![
                SourceSpec::raw(ScalarKind::F64, vec![Target { node: 0, slot: 0 }]),
                SourceSpec::raw(ScalarKind::F64, vec![Target { node: 0, slot: 1 }]),
                SourceSpec::raw(ScalarKind::F64, vec![Target { node: 0, slot: 2 }]),
            ],
            vec![Edge { from: 0, to: 1, slot: 0, from_field: 0 }],
        )
        .expect("valid");
        let s0 = f64_stream(&[(2, 20.0), (5, 50.0)]); // in0 (barrier)
        let s1 = f64_stream(&[(2, 200.0)]);           // in1 (barrier)
        let s2 = f64_stream(&[(1, 1.0), (3, 3.0)]);   // in2 (as-of)

        h.run(vec![Box::new(VecSource::new(s0)), Box::new(VecSource::new(s1)), Box::new(VecSource::new(s2))]);
        let out: Vec<(Timestamp, Vec<Scalar>)> = rx.try_iter().collect();
        // c3: barrier pair completes at t=2, holds c=1 -> 221. c4: as-of input
        // ticks at t=3, holds the pair -> 223. c1 filters; c2,c5 hold (no record).
        assert_eq!(
            out,
            vec![
                (Timestamp(2), vec![Scalar::f64(221.0)]),
                (Timestamp(3), vec![Scalar::f64(223.0)]),
            ]
        );
    }

    #[test]
    fn bootstrap_rejects_a_cycle() {
        // two SMA(1) nodes wired a -> b -> a
        let err = boot(
            vec![Box::new(Sma::new(1)), Box::new(Sma::new(1))],
            vec![
                Sma::builder().schema().clone(),
                Sma::builder().schema().clone(),
            ],
            vec![],
            vec![Edge { from: 0, to: 1, slot: 0, from_field: 0 }, Edge { from: 1, to: 0, slot: 0, from_field: 0 }],
        )
        .unwrap_err();
        assert_eq!(err, BootstrapError::Cycle);
    }

    #[test]
    fn bootstrap_rejects_a_kind_mismatch() {
        // SMA(1) declares an f64 input; an i64 source mismatches
        let err = boot(
            vec![Box::new(Sma::new(1))],
            vec![
                Sma::builder().schema().clone(),
            ],
            vec![SourceSpec::raw(ScalarKind::I64, vec![Target { node: 0, slot: 0 }])],
            vec![],
        )
        .unwrap_err();
        assert_eq!(
            err,
            BootstrapError::KindMismatch { producer: ScalarKind::I64, consumer: ScalarKind::F64 }
        );
    }

    #[test]
    fn bootstrap_rejects_a_bad_index() {
        // an edge target node (9) that does not exist -> BadIndex. (The old trigger
        // — an out-of-range observe index — is gone with `observe`; BadIndex itself
        // is unchanged, only the path that reaches it.)
        let err = boot(
            vec![Box::new(Sma::new(1)), Box::new(Sma::new(1))],
            vec![
                Sma::builder().schema().clone(),
                Sma::builder().schema().clone(),
            ],
            vec![SourceSpec::raw(ScalarKind::F64, vec![Target { node: 0, slot: 0 }])],
            vec![Edge { from: 0, to: 9, slot: 0, from_field: 0 }],
        )
        .unwrap_err();
        assert_eq!(err, BootstrapError::BadIndex);
    }

    /// Build five timestamp-aligned f64 sources feeding Ohlcv's five barrier slots.
    fn ohlcv_streams() -> Vec<Box<dyn Source>> {
        vec![
            Box::new(VecSource::new(f64_stream(&[(1, 10.0), (2, 20.0)]))), // open
            Box::new(VecSource::new(f64_stream(&[(1, 15.0), (2, 25.0)]))), // high
            Box::new(VecSource::new(f64_stream(&[(1, 8.0), (2, 19.0)]))),  // low
            Box::new(VecSource::new(f64_stream(&[(1, 12.0), (2, 22.0)]))), // close
            Box::new(VecSource::new(f64_stream(&[(1, 100.0), (2, 200.0)]))), // volume
        ]
    }

    fn ohlcv_sources() -> Vec<SourceSpec> {
        (0..5)
            .map(|slot| SourceSpec::raw(ScalarKind::F64, vec![Target { node: 0, slot }]))
            .collect()
    }

    #[test]
    fn ohlcv_bundles_five_field_record() {
        // node 0 = Ohlcv; five sources feed O/H/L/C/V; node 1 = a 5-input Recorder
        // taps all five fields via five edges. The barrier fires once all five share
        // the timestamp, so each bar is recorded once, on the fifth cycle of its ts.
        let (tx, rx) = mpsc::channel();
        let mut h = boot(
            vec![
                Box::new(Ohlcv { out: [Cell::from_f64(0.0); 5] }),
                Box::new(Recorder::new(
                    &[ScalarKind::F64, ScalarKind::F64, ScalarKind::F64, ScalarKind::F64, ScalarKind::F64],
                    Firing::Any,
                    tx,
                )),
            ],
            vec![
                Ohlcv::sig(),
                recorder_sig(&[ScalarKind::F64, ScalarKind::F64, ScalarKind::F64, ScalarKind::F64, ScalarKind::F64], Firing::Any),
            ],
            ohlcv_sources(),
            vec![
                Edge { from: 0, to: 1, slot: 0, from_field: 0 }, // open
                Edge { from: 0, to: 1, slot: 1, from_field: 1 }, // high
                Edge { from: 0, to: 1, slot: 2, from_field: 2 }, // low
                Edge { from: 0, to: 1, slot: 3, from_field: 3 }, // close
                Edge { from: 0, to: 1, slot: 4, from_field: 4 }, // volume
            ],
        )
        .expect("valid");
        h.run(ohlcv_streams());
        let out: Vec<(Timestamp, Vec<Scalar>)> = rx.try_iter().collect();
        assert_eq!(
            out,
            vec![
                (Timestamp(1), vec![
                    Scalar::f64(10.0),
                    Scalar::f64(15.0),
                    Scalar::f64(8.0),
                    Scalar::f64(12.0),
                    Scalar::f64(100.0),
                ]),
                (Timestamp(2), vec![
                    Scalar::f64(20.0),
                    Scalar::f64(25.0),
                    Scalar::f64(19.0),
                    Scalar::f64(22.0),
                    Scalar::f64(200.0),
                ]),
            ]
        );
    }

    #[test]
    fn edge_binds_single_field_high_minus_low() {
        // [Ohlcv (0), Sub (1), Recorder (2)]; Sub binds high (field 1) and low
        // (field 2) of the Ohlcv record -> high - low; the Recorder taps Sub.
        // Proves from_field routes the right columns (not field 0) and the two
        // bound fields are co-fresh (Sub's Any inputs both fire in the bar's cycle).
        let build = |tx| {
            boot(
                vec![
                    Box::new(Ohlcv { out: [Cell::from_f64(0.0); 5] }),
                    Box::new(Sub::new()),
                    Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx)),
                ],
            vec![
                Ohlcv::sig(),
                Sub::builder().schema().clone(),
                recorder_sig(&[ScalarKind::F64], Firing::Any),
            ],
                ohlcv_sources(),
                vec![
                    Edge { from: 0, to: 1, slot: 0, from_field: 1 }, // high
                    Edge { from: 0, to: 1, slot: 1, from_field: 2 }, // low
                    Edge { from: 1, to: 2, slot: 0, from_field: 0 }, // Sub -> Recorder
                ],
            )
            .expect("valid DAG")
        };
        let (tx, rx) = mpsc::channel();
        let mut h = build(tx);
        h.run(ohlcv_streams());
        let out: Vec<(Timestamp, Vec<Scalar>)> = rx.try_iter().collect();
        // bar1: 15 - 8 = 7; bar2: 25 - 19 = 6 (each on the bar's fifth cycle).
        assert_eq!(
            out,
            vec![
                (Timestamp(1), vec![Scalar::f64(7.0)]),
                (Timestamp(2), vec![Scalar::f64(6.0)]),
            ]
        );

        let (tx2, rx2) = mpsc::channel();
        let mut h2 = build(tx2);
        h2.run(ohlcv_streams());
        let out2: Vec<(Timestamp, Vec<Scalar>)> = rx2.try_iter().collect();
        assert_eq!(out2, out);
    }

    #[test]
    fn distinct_edges_read_distinct_fields() {
        // Same Ohlcv, a different consumer: Sub binds close (field 3) and open
        // (field 0) -> close - open; the Recorder taps Sub. Proves two edges on one
        // record read two different fields (3 and 0, not the high/low pair above).
        let (tx, rx) = mpsc::channel();
        let mut h = boot(
            vec![
                Box::new(Ohlcv { out: [Cell::from_f64(0.0); 5] }),
                Box::new(Sub::new()),
                Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx)),
            ],
            vec![
                Ohlcv::sig(),
                Sub::builder().schema().clone(),
                recorder_sig(&[ScalarKind::F64], Firing::Any),
            ],
            ohlcv_sources(),
            vec![
                Edge { from: 0, to: 1, slot: 0, from_field: 3 }, // close
                Edge { from: 0, to: 1, slot: 1, from_field: 0 }, // open
                Edge { from: 1, to: 2, slot: 0, from_field: 0 }, // Sub -> Recorder
            ],
        )
        .expect("valid DAG");
        h.run(ohlcv_streams());
        let out: Vec<(Timestamp, Vec<Scalar>)> = rx.try_iter().collect();
        // bar1: 12 - 10 = 2; bar2: 22 - 20 = 2.
        assert_eq!(
            out,
            vec![
                (Timestamp(1), vec![Scalar::f64(2.0)]),
                (Timestamp(2), vec![Scalar::f64(2.0)]),
            ]
        );
    }

    #[test]
    fn bootstrap_rejects_from_field_out_of_range() {
        // Sma(0) has a 1-field output (index 0 only); an edge reading field 9 is
        // out of range -> BadIndex (caught before any kind check).
        let err = boot(
            vec![Box::new(Sma::new(1)), Box::new(Sma::new(1))],
            vec![
                Sma::builder().schema().clone(),
                Sma::builder().schema().clone(),
            ],
            vec![SourceSpec::raw(ScalarKind::F64, vec![Target { node: 0, slot: 0 }])],
            vec![Edge { from: 0, to: 1, slot: 0, from_field: 9 }],
        )
        .unwrap_err();
        assert_eq!(err, BootstrapError::BadIndex);
    }

    #[test]
    fn bootstrap_rejects_per_field_kind_mismatch() {
        // TwoField(0) output: field 0 f64, field 1 i64. Binding field 1 (i64) into
        // Sma(1)'s f64 input slot is a per-field kind mismatch -> KindMismatch. (The
        // mismatch is field-specific: from_field 0 would have matched.)
        let err = boot(
            vec![Box::new(TwoField { out: [Cell::from_f64(0.0), Cell::from_i64(0)] }), Box::new(Sma::new(1))],
            vec![
                TwoField::sig(),
                Sma::builder().schema().clone(),
            ],
            vec![SourceSpec::raw(ScalarKind::F64, vec![Target { node: 0, slot: 0 }])],
            vec![Edge { from: 0, to: 1, slot: 0, from_field: 1 }],
        )
        .unwrap_err();
        assert_eq!(
            err,
            BootstrapError::KindMismatch { producer: ScalarKind::I64, consumer: ScalarKind::F64 }
        );
    }

    #[test]
    fn multi_sink_records_distinct_interior_streams() {
        // Two recorders tap SMA(2) and SMA(4) in ONE run -> one run records many
        // streams (the #2 headline). Each drained stream is individually correct.
        let (tx_fast, rx_fast) = mpsc::channel();
        let (tx_slow, rx_slow) = mpsc::channel();
        let mut h = boot(
            vec![
                Box::new(Sma::new(2)),
                Box::new(Sma::new(4)),
                Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx_fast)),
                Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx_slow)),
            ],
            vec![
                Sma::builder().schema().clone(),
                Sma::builder().schema().clone(),
                recorder_sig(&[ScalarKind::F64], Firing::Any),
                recorder_sig(&[ScalarKind::F64], Firing::Any),
            ],
            vec![SourceSpec::raw(ScalarKind::F64, vec![Target { node: 0, slot: 0 }, Target { node: 1, slot: 0 }])],
            vec![
                Edge { from: 0, to: 2, slot: 0, from_field: 0 }, // SMA(2) -> recorder fast
                Edge { from: 1, to: 3, slot: 0, from_field: 0 }, // SMA(4) -> recorder slow
            ],
        )
        .expect("valid DAG");
        h.run(vec![Box::new(VecSource::new(f64_stream(&[(1, 10.0), (2, 12.0), (3, 14.0), (4, 16.0), (5, 18.0)])))]);
        let fast: Vec<(Timestamp, Vec<Scalar>)> = rx_fast.try_iter().collect();
        let slow: Vec<(Timestamp, Vec<Scalar>)> = rx_slow.try_iter().collect();
        // SMA(2) warms at cycle 2, SMA(4) at cycle 4 — two different-rate streams.
        assert_eq!(
            fast,
            vec![
                (Timestamp(2), vec![Scalar::f64(11.0)]),
                (Timestamp(3), vec![Scalar::f64(13.0)]),
                (Timestamp(4), vec![Scalar::f64(15.0)]),
                (Timestamp(5), vec![Scalar::f64(17.0)]),
            ]
        );
        assert_eq!(
            slow,
            vec![
                (Timestamp(4), vec![Scalar::f64(13.0)]),
                (Timestamp(5), vec![Scalar::f64(15.0)]),
            ]
        );
    }

    #[test]
    fn recorder_taps_all_fields_of_a_record() {
        // A 5-input Recorder taps all five OHLCV fields via five field-wise edges
        // (0005: N edges, no whole-record bind); its recorded row is the whole bar.
        let (tx, rx) = mpsc::channel();
        let mut h = boot(
            vec![
                Box::new(Ohlcv { out: [Cell::from_f64(0.0); 5] }),
                Box::new(Recorder::new(
                    &[ScalarKind::F64, ScalarKind::F64, ScalarKind::F64, ScalarKind::F64, ScalarKind::F64],
                    Firing::Any,
                    tx,
                )),
            ],
            vec![
                Ohlcv::sig(),
                recorder_sig(&[ScalarKind::F64, ScalarKind::F64, ScalarKind::F64, ScalarKind::F64, ScalarKind::F64], Firing::Any),
            ],
            ohlcv_sources(),
            vec![
                Edge { from: 0, to: 1, slot: 0, from_field: 0 },
                Edge { from: 0, to: 1, slot: 1, from_field: 1 },
                Edge { from: 0, to: 1, slot: 2, from_field: 2 },
                Edge { from: 0, to: 1, slot: 3, from_field: 3 },
                Edge { from: 0, to: 1, slot: 4, from_field: 4 },
            ],
        )
        .expect("valid");
        h.run(vec![
            Box::new(VecSource::new(f64_stream(&[(1, 10.0)]))),
            Box::new(VecSource::new(f64_stream(&[(1, 15.0)]))),
            Box::new(VecSource::new(f64_stream(&[(1, 8.0)]))),
            Box::new(VecSource::new(f64_stream(&[(1, 12.0)]))),
            Box::new(VecSource::new(f64_stream(&[(1, 100.0)]))),
        ]);
        let out: Vec<(Timestamp, Vec<Scalar>)> = rx.try_iter().collect();
        assert_eq!(out.len(), 1);
        assert_eq!(out[0].1.len(), 5); // all five fields recorded as one row
        assert_eq!(
            out,
            vec![(Timestamp(1), vec![
                Scalar::f64(10.0),
                Scalar::f64(15.0),
                Scalar::f64(8.0),
                Scalar::f64(12.0),
                Scalar::f64(100.0),
            ])]
        );
    }

    #[test]
    fn recorder_records_mixed_scalar_kinds() {
        // A recorder with i64 + f64 + bool + timestamp inputs records a four-field
        // mixed-kind row -> recording is not f64-only. Four sources tick once each
        // at t=1,2,3,4; only on cycle 4 are all slots warm, so it records once,
        // holding the earlier-ticked values.
        let (tx, rx) = mpsc::channel();
        let mut h = boot(
            vec![Box::new(Recorder::new(
                &[ScalarKind::I64, ScalarKind::F64, ScalarKind::Bool, ScalarKind::Timestamp],
                Firing::Any,
                tx,
            ))],
            vec![
                recorder_sig(&[ScalarKind::I64, ScalarKind::F64, ScalarKind::Bool, ScalarKind::Timestamp], Firing::Any),
            ],
            vec![
                SourceSpec::raw(ScalarKind::I64, vec![Target { node: 0, slot: 0 }]),
                SourceSpec::raw(ScalarKind::F64, vec![Target { node: 0, slot: 1 }]),
                SourceSpec::raw(ScalarKind::Bool, vec![Target { node: 0, slot: 2 }]),
                SourceSpec::raw(ScalarKind::Timestamp, vec![Target { node: 0, slot: 3 }]),
            ],
            vec![],
        )
        .expect("valid");
        h.run(vec![
            Box::new(VecSource::new(vec![(Timestamp(1), Scalar::i64(7))])),
            Box::new(VecSource::new(vec![(Timestamp(2), Scalar::f64(1.5))])),
            Box::new(VecSource::new(vec![(Timestamp(3), Scalar::bool(true))])),
            Box::new(VecSource::new(vec![(Timestamp(4), Scalar::ts(Timestamp(99)))])),
        ]);
        let out: Vec<(Timestamp, Vec<Scalar>)> = rx.try_iter().collect();
        assert_eq!(
            out,
            vec![(Timestamp(4), vec![
                Scalar::i64(7),
                Scalar::f64(1.5),
                Scalar::bool(true),
                Scalar::ts(Timestamp(99)),
            ])]
        );
    }

    #[test]
    fn node_is_producer_and_sink_at_once() {
        // TapForward records its input AND forwards it downstream; a second
        // Recorder taps the forwarded output. Both channels see the same stream ->
        // one node is producer and sink at once (C8 "both").
        let (tx_tap, rx_tap) = mpsc::channel();
        let (tx_down, rx_down) = mpsc::channel();
        let mut h = boot(
            vec![
                Box::new(TapForward { out: [Cell::from_f64(0.0)], tx: tx_tap }),
                Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx_down)),
            ],
            vec![
                TapForward::sig(),
                recorder_sig(&[ScalarKind::F64], Firing::Any),
            ],
            vec![SourceSpec::raw(ScalarKind::F64, vec![Target { node: 0, slot: 0 }])],
            vec![Edge { from: 0, to: 1, slot: 0, from_field: 0 }],
        )
        .expect("valid");
        h.run(vec![Box::new(VecSource::new(f64_stream(&[(1, 10.0), (2, 20.0), (3, 30.0)])))]);
        let tapped: Vec<(Timestamp, Vec<Scalar>)> = rx_tap.try_iter().collect();
        let downstream: Vec<(Timestamp, Vec<Scalar>)> = rx_down.try_iter().collect();
        let expected = vec![
            (Timestamp(1), vec![Scalar::f64(10.0)]),
            (Timestamp(2), vec![Scalar::f64(20.0)]),
            (Timestamp(3), vec![Scalar::f64(30.0)]),
        ];
        assert_eq!(tapped, expected); // it recorded (sink side effect)
        assert_eq!(downstream, expected); // and forwarded (producer output)
    }

    #[test]
    fn recording_is_deterministic() {
        // Two fresh harnesses, two channels, identical input -> bit-identical
        // recorded streams (C1).
        let build = |tx| {
            boot(
                vec![
                    Box::new(Sma::new(3)),
                    Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx)),
                ],
            vec![
                Sma::builder().schema().clone(),
                recorder_sig(&[ScalarKind::F64], Firing::Any),
            ],
                vec![SourceSpec::raw(ScalarKind::F64, vec![Target { node: 0, slot: 0 }])],
                vec![Edge { from: 0, to: 1, slot: 0, from_field: 0 }],
            )
            .expect("valid")
        };
        let prices = f64_stream(&[(1, 1.0), (2, 2.0), (3, 3.0), (4, 4.0), (5, 5.0)]);

        let (tx_a, rx_a) = mpsc::channel();
        let mut a = build(tx_a);
        a.run(vec![Box::new(VecSource::new(prices.clone()))]);
        let run_a: Vec<(Timestamp, Vec<Scalar>)> = rx_a.try_iter().collect();

        let (tx_b, rx_b) = mpsc::channel();
        let mut b = build(tx_b);
        b.run(vec![Box::new(VecSource::new(prices))]);
        let run_b: Vec<(Timestamp, Vec<Scalar>)> = rx_b.try_iter().collect();

        assert_eq!(run_a, run_b);
        assert!(!run_a.is_empty()); // and it actually recorded something
    }

    #[test]
    fn recorder_barrier_firing_records_only_on_coincidence() {
        // A 2-input Barrier(0) recorder records only on cycles where both inputs
        // share the timestamp — the recorder's OWN firing policy gates recording.
        let (tx, rx) = mpsc::channel();
        let mut h = boot(
            vec![Box::new(Recorder::new(
                &[ScalarKind::F64, ScalarKind::F64],
                Firing::Barrier(0),
                tx,
            ))],
            vec![
                recorder_sig(&[ScalarKind::F64, ScalarKind::F64], Firing::Barrier(0)),
            ],
            vec![
                SourceSpec::raw(ScalarKind::F64, vec![Target { node: 0, slot: 0 }]),
                SourceSpec::raw(ScalarKind::F64, vec![Target { node: 0, slot: 1 }]),
            ],
            vec![],
        )
        .expect("valid");
        let s0 = f64_stream(&[(1, 10.0), (2, 20.0), (3, 30.0), (4, 40.0)]);
        let s1 = f64_stream(&[(2, 100.0), (4, 200.0)]);
        h.run(vec![Box::new(VecSource::new(s0)), Box::new(VecSource::new(s1))]);
        let out: Vec<(Timestamp, Vec<Scalar>)> = rx.try_iter().collect();
        // records ONLY at t=2 and t=4 (both inputs coincide); holds otherwise.
        assert_eq!(
            out,
            vec![
                (Timestamp(2), vec![Scalar::f64(20.0), Scalar::f64(100.0)]),
                (Timestamp(4), vec![Scalar::f64(40.0), Scalar::f64(200.0)]),
            ]
        );
    }

    #[test]
    fn recorder_any_firing_records_on_each_fresh() {
        // A 2-input Any recorder records on any-fresh once both are warm (as-of),
        // holding the stale input.
        let (tx, rx) = mpsc::channel();
        let mut h = boot(
            vec![Box::new(Recorder::new(
                &[ScalarKind::F64, ScalarKind::F64],
                Firing::Any,
                tx,
            ))],
            vec![
                recorder_sig(&[ScalarKind::F64, ScalarKind::F64], Firing::Any),
            ],
            vec![
                SourceSpec::raw(ScalarKind::F64, vec![Target { node: 0, slot: 0 }]),
                SourceSpec::raw(ScalarKind::F64, vec![Target { node: 0, slot: 1 }]),
            ],
            vec![],
        )
        .expect("valid");
        let s0 = f64_stream(&[(1, 10.0), (2, 20.0), (3, 30.0), (4, 40.0)]);
        let s1 = f64_stream(&[(2, 100.0), (4, 200.0)]);
        h.run(vec![Box::new(VecSource::new(s0)), Box::new(VecSource::new(s1))]);
        let out: Vec<(Timestamp, Vec<Scalar>)> = rx.try_iter().collect();
        // from t=2 on, records every cycle holding the stale input; two cycles fall
        // on t=4 (the s0 tick then the s1 tick).
        assert_eq!(
            out,
            vec![
                (Timestamp(2), vec![Scalar::f64(20.0), Scalar::f64(100.0)]),
                (Timestamp(3), vec![Scalar::f64(30.0), Scalar::f64(100.0)]),
                (Timestamp(4), vec![Scalar::f64(40.0), Scalar::f64(100.0)]),
                (Timestamp(4), vec![Scalar::f64(40.0), Scalar::f64(200.0)]),
            ]
        );
    }

    #[test]
    fn bootstrap_rejects_kind_mismatched_recorder_edge() {
        // TwoField output: field 0 f64, field 1 i64. Binding field 1 (i64) into a
        // Recorder's f64 input slot is a per-field kind mismatch -> KindMismatch
        // (0005's check already covers recorder edges; recording adds no new hole).
        let (tx, _rx) = mpsc::channel();
        let err = boot(
            vec![
                Box::new(TwoField { out: [Cell::from_f64(0.0), Cell::from_i64(0)] }),
                Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx)),
            ],
            vec![
                TwoField::sig(),
                recorder_sig(&[ScalarKind::F64], Firing::Any),
            ],
            vec![SourceSpec::raw(ScalarKind::F64, vec![Target { node: 0, slot: 0 }])],
            vec![Edge { from: 0, to: 1, slot: 0, from_field: 1 }],
        )
        .unwrap_err();
        assert_eq!(
            err,
            BootstrapError::KindMismatch { producer: ScalarKind::I64, consumer: ScalarKind::F64 }
        );
    }

    // --- #3 stress matrix: M-producer x N-consumer x K-sink DAGs ---
    // The closed `Harness::bootstrap` + `run` substrate must carry arbitrary
    // node-level fan-out / fan-in / depth / width deterministically (C1) and
    // compute every recorded stream correctly. Earlier cycles proved *source*
    // fan-out, single chains and firing in isolation; these close the node
    // fan-out / deep-chain / wide-layer holes #3 names. No trading domain.

    #[test]
    fn node_fan_out_identical_taps_record_identical_streams() {
        // PROPERTY: one PRODUCING node read by several consumers via distinct
        // edges from the same `from` node feeds every consumer the identical,
        // correct stream — node fan-out (not source fan-out: a single SMA(3)
        // output is forwarded down three edges to three recorders).
        let build = |t1, t2, t3| {
            boot(
                vec![
                    Box::new(Sma::new(3)),
                    Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, t1)),
                    Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, t2)),
                    Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, t3)),
                ],
            vec![
                Sma::builder().schema().clone(),
                recorder_sig(&[ScalarKind::F64], Firing::Any),
                recorder_sig(&[ScalarKind::F64], Firing::Any),
                recorder_sig(&[ScalarKind::F64], Firing::Any),
            ],
                vec![SourceSpec::raw(ScalarKind::F64, vec![Target { node: 0, slot: 0 }])],
                vec![
                    Edge { from: 0, to: 1, slot: 0, from_field: 0 },
                    Edge { from: 0, to: 2, slot: 0, from_field: 0 },
                    Edge { from: 0, to: 3, slot: 0, from_field: 0 },
                ],
            )
            .expect("valid fan-out DAG")
        };
        let prices = f64_stream(&[(1, 2.0), (2, 4.0), (3, 6.0), (4, 8.0), (5, 10.0)]);

        let (a1, ra) = mpsc::channel();
        let (b1, rb) = mpsc::channel();
        let (c1, rc) = mpsc::channel();
        let mut h = build(a1, b1, c1);
        h.run(vec![Box::new(VecSource::new(prices.clone()))]);
        let s1: Vec<(Timestamp, Vec<Scalar>)> = ra.try_iter().collect();
        let s2: Vec<(Timestamp, Vec<Scalar>)> = rb.try_iter().collect();
        let s3: Vec<(Timestamp, Vec<Scalar>)> = rc.try_iter().collect();
        // SMA(3) warms at cycle 3: mean(2,4,6)=4, mean(4,6,8)=6, mean(6,8,10)=8.
        let expected = vec![
            (Timestamp(3), vec![Scalar::f64(4.0)]),
            (Timestamp(4), vec![Scalar::f64(6.0)]),
            (Timestamp(5), vec![Scalar::f64(8.0)]),
        ];
        assert_eq!(s1, expected);
        assert_eq!(s2, expected); // every tap sees the identical shared stream
        assert_eq!(s3, expected);

        // determinism (C1): a fresh harness drains bit-identical.
        let (a2, ra2) = mpsc::channel();
        let (b2, rb2) = mpsc::channel();
        let (c2, rc2) = mpsc::channel();
        let mut h2 = build(a2, b2, c2);
        h2.run(vec![Box::new(VecSource::new(prices))]);
        assert_eq!(ra2.try_iter().collect::<Vec<_>>(), expected);
        assert_eq!(rb2.try_iter().collect::<Vec<_>>(), expected);
        assert_eq!(rc2.try_iter().collect::<Vec<_>>(), expected);
    }

    #[test]
    fn node_fan_out_divergent_consumers_each_compute_their_own() {
        // PROPERTY: one producer (SMA(2)) read by consumers that do DIFFERENT
        // things — recorded raw by one sink AND fed as input0 of a Sub whose
        // other input is a second producer (SMA(4)) — yields both the raw tap
        // and the downstream-combined stream, each independently correct.
        let build = |t_raw, t_sub| {
            boot(
                vec![
                    Box::new(Sma::new(2)),                                  // 0: shared producer
                    Box::new(Sma::new(4)),                                  // 1: second producer
                    Box::new(Sub::new()),                                   // 2: SMA(2) - SMA(4)
                    Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, t_raw)), // 3: raw tap of 0
                    Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, t_sub)), // 4: tap of Sub
                ],
            vec![
                Sma::builder().schema().clone(),
                Sma::builder().schema().clone(),
                Sub::builder().schema().clone(),
                recorder_sig(&[ScalarKind::F64], Firing::Any),
                recorder_sig(&[ScalarKind::F64], Firing::Any),
            ],
                vec![SourceSpec::raw(ScalarKind::F64, vec![Target { node: 0, slot: 0 }, Target { node: 1, slot: 0 }])],
                vec![
                    Edge { from: 0, to: 3, slot: 0, from_field: 0 }, // SMA(2) -> raw recorder
                    Edge { from: 0, to: 2, slot: 0, from_field: 0 }, // SMA(2) -> Sub.in0
                    Edge { from: 1, to: 2, slot: 1, from_field: 0 }, // SMA(4) -> Sub.in1
                    Edge { from: 2, to: 4, slot: 0, from_field: 0 }, // Sub    -> Sub recorder
                ],
            )
            .expect("valid divergent-fan-out DAG")
        };
        let prices = f64_stream(&[(1, 10.0), (2, 12.0), (3, 14.0), (4, 16.0), (5, 18.0), (6, 20.0)]);

        let (tr, rr) = mpsc::channel();
        let (ts, rs) = mpsc::channel();
        let mut h = build(tr, ts);
        h.run(vec![Box::new(VecSource::new(prices.clone()))]);
        let raw: Vec<(Timestamp, Vec<Scalar>)> = rr.try_iter().collect();
        let sub: Vec<(Timestamp, Vec<Scalar>)> = rs.try_iter().collect();
        // SMA(2): 11,13,15,17,19 at cycles 2..6 (raw tap).
        let raw_expected = vec![
            (Timestamp(2), vec![Scalar::f64(11.0)]),
            (Timestamp(3), vec![Scalar::f64(13.0)]),
            (Timestamp(4), vec![Scalar::f64(15.0)]),
            (Timestamp(5), vec![Scalar::f64(17.0)]),
            (Timestamp(6), vec![Scalar::f64(19.0)]),
        ];
        // Sub warms once SMA(4) is warm (cycle 4): 15-13, 17-15, 19-17 -> 2.
        let sub_expected = vec![
            (Timestamp(4), vec![Scalar::f64(2.0)]),
            (Timestamp(5), vec![Scalar::f64(2.0)]),
            (Timestamp(6), vec![Scalar::f64(2.0)]),
        ];
        assert_eq!(raw, raw_expected);
        assert_eq!(sub, sub_expected);

        let (tr2, rr2) = mpsc::channel();
        let (ts2, rs2) = mpsc::channel();
        let mut h2 = build(tr2, ts2);
        h2.run(vec![Box::new(VecSource::new(prices))]);
        assert_eq!(rr2.try_iter().collect::<Vec<_>>(), raw_expected); // deterministic
        assert_eq!(rs2.try_iter().collect::<Vec<_>>(), sub_expected);
    }

    #[test]
    fn node_fan_out_under_mixed_firing_each_consumer_records_per_policy() {
        // PROPERTY: one producer (SMA(2)) read simultaneously by an as-of
        // (Firing::Any) consumer and a Firing::Barrier consumer — each records
        // per ITS OWN firing policy off the SAME shared upstream value. The Any
        // tap fires on every SMA push; the BarrierSum fires only on the cycles
        // where the held SMA output and a second source coincide on a timestamp.
        let build = |t_any, t_bar| {
            boot(
                vec![
                    Box::new(Sma::new(2)),                                  // 0: shared producer (src A)
                    Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, t_any)), // 1: as-of tap of 0
                    Box::new(BarrierSum { out: [Cell::from_f64(0.0)] }),       // 2: SMA(2) + src B (barrier)
                    Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, t_bar)), // 3: tap of barrier
                ],
            vec![
                Sma::builder().schema().clone(),
                recorder_sig(&[ScalarKind::F64], Firing::Any),
                BarrierSum::sig(),
                recorder_sig(&[ScalarKind::F64], Firing::Any),
            ],
                vec![
                    SourceSpec::raw(ScalarKind::F64, vec![Target { node: 0, slot: 0 }]), // A
                    SourceSpec::raw(ScalarKind::F64, vec![Target { node: 2, slot: 1 }]), // B
                ],
                vec![
                    Edge { from: 0, to: 1, slot: 0, from_field: 0 }, // SMA(2) -> as-of recorder
                    Edge { from: 0, to: 2, slot: 0, from_field: 0 }, // SMA(2) -> barrier.in0
                    Edge { from: 2, to: 3, slot: 0, from_field: 0 }, // barrier -> recorder
                ],
            )
            .expect("valid mixed-firing fan-out DAG")
        };
        let a = f64_stream(&[(1, 10.0), (2, 12.0), (3, 14.0), (4, 16.0)]); // drives SMA(2)
        let b = f64_stream(&[(2, 100.0), (4, 200.0)]);                     // barrier.in1

        let (ta, rany) = mpsc::channel();
        let (tb, rbar) = mpsc::channel();
        let mut h = build(ta, tb);
        h.run(vec![Box::new(VecSource::new(a.clone())), Box::new(VecSource::new(b.clone()))]);
        let any: Vec<(Timestamp, Vec<Scalar>)> = rany.try_iter().collect();
        let bar: Vec<(Timestamp, Vec<Scalar>)> = rbar.try_iter().collect();
        // As-of tap records every SMA(2) fire: 11@t2, 13@t3, 15@t4.
        let any_expected = vec![
            (Timestamp(2), vec![Scalar::f64(11.0)]),
            (Timestamp(3), vec![Scalar::f64(13.0)]),
            (Timestamp(4), vec![Scalar::f64(15.0)]),
        ];
        // Barrier fires only where held SMA output and src B share a timestamp:
        // t=2 (SMA held=11 + B=100 = 111), t=4 (SMA held=15 + B=200 = 215).
        let bar_expected = vec![
            (Timestamp(2), vec![Scalar::f64(111.0)]),
            (Timestamp(4), vec![Scalar::f64(215.0)]),
        ];
        assert_eq!(any, any_expected);
        assert_eq!(bar, bar_expected);

        let (ta2, rany2) = mpsc::channel();
        let (tb2, rbar2) = mpsc::channel();
        let mut h2 = build(ta2, tb2);
        h2.run(vec![Box::new(VecSource::new(a)), Box::new(VecSource::new(b))]);
        assert_eq!(rany2.try_iter().collect::<Vec<_>>(), any_expected); // deterministic
        assert_eq!(rbar2.try_iter().collect::<Vec<_>>(), bar_expected);
    }

    #[test]
    fn deep_transform_chain_propagates_end_to_end() {
        // PROPERTY: a linear chain of several transform nodes (SMA(2) -> SMA(2)
        // -> SMA(2) -> recorder) propagates values correctly through depth, with
        // each stage's warm-up delaying the tail — closes "deep chains untested".
        let build = |tx| {
            boot(
                vec![
                    Box::new(Sma::new(2)), // 0
                    Box::new(Sma::new(2)), // 1
                    Box::new(Sma::new(2)), // 2
                    Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx)), // 3 tail
                ],
            vec![
                Sma::builder().schema().clone(),
                Sma::builder().schema().clone(),
                Sma::builder().schema().clone(),
                recorder_sig(&[ScalarKind::F64], Firing::Any),
            ],
                vec![SourceSpec::raw(ScalarKind::F64, vec![Target { node: 0, slot: 0 }])],
                vec![
                    Edge { from: 0, to: 1, slot: 0, from_field: 0 },
                    Edge { from: 1, to: 2, slot: 0, from_field: 0 },
                    Edge { from: 2, to: 3, slot: 0, from_field: 0 },
                ],
            )
            .expect("valid deep chain")
        };
        let prices = f64_stream(&[(1, 2.0), (2, 4.0), (3, 6.0), (4, 8.0), (5, 10.0)]);

        let (tx, rx) = mpsc::channel();
        let mut h = build(tx);
        h.run(vec![Box::new(VecSource::new(prices.clone()))]);
        let out: Vec<(Timestamp, Vec<Scalar>)> = rx.try_iter().collect();
        // stage1 (c2..c5): 3,5,7,9. stage2 (c3..): 4,6,8. stage3 (c4..): 5,7.
        let expected = vec![
            (Timestamp(4), vec![Scalar::f64(5.0)]),
            (Timestamp(5), vec![Scalar::f64(7.0)]),
        ];
        assert_eq!(out, expected);

        let (tx2, rx2) = mpsc::channel();
        let mut h2 = build(tx2);
        h2.run(vec![Box::new(VecSource::new(prices))]);
        assert_eq!(rx2.try_iter().collect::<Vec<_>>(), expected); // deterministic
    }

    /// A producer emitting a fixed two-field record with DISTINCT non-zero values of
    /// two different kinds (field 0 f64 = 2.5, field 1 i64 = 42). The values are
    /// chosen so each kind's bit pattern differs from its cross-kind reinterpretation
    /// — 42_i64's bits read as an f64 are a denormal ~2e-322, and 2.5_f64's bits read
    /// as an i64 are 0x4004000000000000 — so a kind-blind forward is observable.
    struct MixedKindEmit {
        out: [Cell; 2],
    }
    impl MixedKindEmit {
        fn sig() -> NodeSchema {
            NodeSchema {
                inputs: vec![f64_port(Firing::Any)],
                output: vec![
                    FieldSpec { name: "f".into(), kind: ScalarKind::F64 },
                    FieldSpec { name: "i".into(), kind: ScalarKind::I64 },
                ],
                params: vec![],
            }
        }
    }
    impl Node for MixedKindEmit {
        fn lookbacks(&self) -> Vec<usize> {
            vec![1]
        }
        fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Cell]> {
            if ctx.f64_in(0).is_empty() {
                return None;
            }
            Some(&self.out)
        }
    }

    #[test]
    fn mixed_kind_record_forwards_each_field_by_its_kind_end_to_end() {
        // PROPERTY: the inter-node forward (push_cell) reinterprets each forwarded
        // field by the DESTINATION column's kind, so a multi-kind record's i64 field
        // survives the tag-free Cell carrier intact alongside its f64 field. Every
        // other end-to-end forward test wires f64-only edges; since a Cell is one
        // untyped 8-byte word, an i64 field forwarded as if it were f64 (or vice
        // versa) would corrupt silently and still pass all f64-only tests — this
        // pins the i64 path with values whose bit patterns differ across kinds.
        let build = |tx| {
            boot(
                vec![
                    Box::new(MixedKindEmit { out: [Cell::from_f64(2.5), Cell::from_i64(42)] }),
                    Box::new(Recorder::new(&[ScalarKind::F64, ScalarKind::I64], Firing::Any, tx)),
                ],
                vec![
                    MixedKindEmit::sig(),
                    recorder_sig(&[ScalarKind::F64, ScalarKind::I64], Firing::Any),
                ],
                vec![SourceSpec::raw(ScalarKind::F64, vec![Target { node: 0, slot: 0 }])],
                vec![
                    Edge { from: 0, to: 1, slot: 0, from_field: 0 }, // f64 field -> f64 slot
                    Edge { from: 0, to: 1, slot: 1, from_field: 1 }, // i64 field -> i64 slot
                ],
            )
            .expect("kind-matched mixed-field wiring bootstraps")
        };
        let prices = f64_stream(&[(1, 5.0), (2, 7.0)]);

        let (tx, rx) = mpsc::channel();
        let mut h = build(tx);
        h.run(vec![Box::new(VecSource::new(prices.clone()))]);
        let out: Vec<(Timestamp, Vec<Scalar>)> = rx.try_iter().collect();
        // each cycle the producer fires and emits its constant record; the recorder
        // reads field 0 back as f64 2.5 and field 1 back as i64 42 — a kind-blind
        // forward would surface a denormal f64 / a giant i64 instead.
        let expected = vec![
            (Timestamp(1), vec![Scalar::f64(2.5), Scalar::i64(42)]),
            (Timestamp(2), vec![Scalar::f64(2.5), Scalar::i64(42)]),
        ];
        assert_eq!(out, expected);

        let (tx2, rx2) = mpsc::channel();
        let mut h2 = build(tx2);
        h2.run(vec![Box::new(VecSource::new(prices))]);
        assert_eq!(rx2.try_iter().collect::<Vec<_>>(), expected); // deterministic (C1)
    }

    #[test]
    fn wide_parallel_layer_multi_sink_records_each_stream() {
        // PROPERTY: one source fanned to several PARALLEL producers of different
        // params (SMA(2), SMA(3), SMA(4)), each recorded by its own sink in one
        // run, yields each parallel stream correctly — closes "wide layers
        // untested" and exercises multi-sink at width.
        let build = |t2, t3, t4| {
            boot(
                vec![
                    Box::new(Sma::new(2)), // 0
                    Box::new(Sma::new(3)), // 1
                    Box::new(Sma::new(4)), // 2
                    Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, t2)), // 3
                    Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, t3)), // 4
                    Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, t4)), // 5
                ],
            vec![
                Sma::builder().schema().clone(),
                Sma::builder().schema().clone(),
                Sma::builder().schema().clone(),
                recorder_sig(&[ScalarKind::F64], Firing::Any),
                recorder_sig(&[ScalarKind::F64], Firing::Any),
                recorder_sig(&[ScalarKind::F64], Firing::Any),
            ],
                vec![SourceSpec::raw(ScalarKind::F64, vec![
                        Target { node: 0, slot: 0 },
                        Target { node: 1, slot: 0 },
                        Target { node: 2, slot: 0 },
                    ])],
                vec![
                    Edge { from: 0, to: 3, slot: 0, from_field: 0 },
                    Edge { from: 1, to: 4, slot: 0, from_field: 0 },
                    Edge { from: 2, to: 5, slot: 0, from_field: 0 },
                ],
            )
            .expect("valid wide layer")
        };
        let prices = f64_stream(&[(1, 10.0), (2, 12.0), (3, 14.0), (4, 16.0), (5, 18.0)]);

        let (a, ra) = mpsc::channel();
        let (b, rb) = mpsc::channel();
        let (c, rc) = mpsc::channel();
        let mut h = build(a, b, c);
        h.run(vec![Box::new(VecSource::new(prices.clone()))]);
        let w2: Vec<(Timestamp, Vec<Scalar>)> = ra.try_iter().collect();
        let w3: Vec<(Timestamp, Vec<Scalar>)> = rb.try_iter().collect();
        let w4: Vec<(Timestamp, Vec<Scalar>)> = rc.try_iter().collect();
        let e2 = vec![
            (Timestamp(2), vec![Scalar::f64(11.0)]),
            (Timestamp(3), vec![Scalar::f64(13.0)]),
            (Timestamp(4), vec![Scalar::f64(15.0)]),
            (Timestamp(5), vec![Scalar::f64(17.0)]),
        ];
        let e3 = vec![
            (Timestamp(3), vec![Scalar::f64(12.0)]),
            (Timestamp(4), vec![Scalar::f64(14.0)]),
            (Timestamp(5), vec![Scalar::f64(16.0)]),
        ];
        let e4 = vec![
            (Timestamp(4), vec![Scalar::f64(13.0)]),
            (Timestamp(5), vec![Scalar::f64(15.0)]),
        ];
        assert_eq!(w2, e2);
        assert_eq!(w3, e3);
        assert_eq!(w4, e4);

        let (a2, ra2) = mpsc::channel();
        let (b2, rb2) = mpsc::channel();
        let (c2, rc2) = mpsc::channel();
        let mut h2 = build(a2, b2, c2);
        h2.run(vec![Box::new(VecSource::new(prices))]);
        assert_eq!(ra2.try_iter().collect::<Vec<_>>(), e2); // deterministic
        assert_eq!(rb2.try_iter().collect::<Vec<_>>(), e3);
        assert_eq!(rc2.try_iter().collect::<Vec<_>>(), e4);
    }

    #[test]
    fn milestone_end_to_end_mixed_dag_records_every_stream_deterministically() {
        // PROPERTY (#3 headline): a single richer DAG combining node fan-out
        // (SMA(2) -> raw recorder AND Sub) + source fan-out (source -> SMA(2),
        // SMA(4)) + fan-in (Sub) + multiple sinks of MIXED scalar kinds (two f64
        // streams + one i64 stream) records every stream correctly AND is fully
        // deterministic — the "pure compute substrate carries arbitrary
        // M-producer x N-consumer x K-sink DAGs" gate.
        let build = |t_raw, t_sub, t_i64| {
            boot(
                vec![
                    Box::new(Sma::new(2)),                                  // 0: shared f64 producer
                    Box::new(Sma::new(4)),                                  // 1: second f64 producer
                    Box::new(Sub::new()),                                   // 2: SMA(2) - SMA(4)
                    Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, t_raw)), // 3
                    Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, t_sub)), // 4
                    Box::new(Recorder::new(&[ScalarKind::I64], Firing::Any, t_i64)), // 5: i64 sink
                ],
            vec![
                Sma::builder().schema().clone(),
                Sma::builder().schema().clone(),
                Sub::builder().schema().clone(),
                recorder_sig(&[ScalarKind::F64], Firing::Any),
                recorder_sig(&[ScalarKind::F64], Firing::Any),
                recorder_sig(&[ScalarKind::I64], Firing::Any),
            ],
                vec![
                    SourceSpec::raw(ScalarKind::F64, vec![Target { node: 0, slot: 0 }, Target { node: 1, slot: 0 }]),
                    SourceSpec::raw(ScalarKind::I64, vec![Target { node: 5, slot: 0 }]),
                ],
                vec![
                    Edge { from: 0, to: 3, slot: 0, from_field: 0 }, // SMA(2) raw
                    Edge { from: 0, to: 2, slot: 0, from_field: 0 }, // SMA(2) -> Sub.in0
                    Edge { from: 1, to: 2, slot: 1, from_field: 0 }, // SMA(4) -> Sub.in1
                    Edge { from: 2, to: 4, slot: 0, from_field: 0 }, // Sub -> recorder
                ],
            )
            .expect("valid milestone DAG")
        };
        let prices =
            f64_stream(&[(1, 10.0), (2, 12.0), (3, 14.0), (4, 16.0), (5, 18.0), (6, 20.0)]);
        let counts: Vec<(Timestamp, Scalar)> = vec![
            (Timestamp(1), Scalar::i64(7)),
            (Timestamp(2), Scalar::i64(8)),
            (Timestamp(3), Scalar::i64(9)),
        ];

        let (tr, rr) = mpsc::channel();
        let (ts, rs) = mpsc::channel();
        let (ti, ri) = mpsc::channel();
        let mut h = build(tr, ts, ti);
        h.run(vec![Box::new(VecSource::new(prices.clone())), Box::new(VecSource::new(counts.clone()))]);
        let raw: Vec<(Timestamp, Vec<Scalar>)> = rr.try_iter().collect();
        let sub: Vec<(Timestamp, Vec<Scalar>)> = rs.try_iter().collect();
        let i64s: Vec<(Timestamp, Vec<Scalar>)> = ri.try_iter().collect();
        let raw_expected = vec![
            (Timestamp(2), vec![Scalar::f64(11.0)]),
            (Timestamp(3), vec![Scalar::f64(13.0)]),
            (Timestamp(4), vec![Scalar::f64(15.0)]),
            (Timestamp(5), vec![Scalar::f64(17.0)]),
            (Timestamp(6), vec![Scalar::f64(19.0)]),
        ];
        let sub_expected = vec![
            (Timestamp(4), vec![Scalar::f64(2.0)]),
            (Timestamp(5), vec![Scalar::f64(2.0)]),
            (Timestamp(6), vec![Scalar::f64(2.0)]),
        ];
        let i64_expected = vec![
            (Timestamp(1), vec![Scalar::i64(7)]),
            (Timestamp(2), vec![Scalar::i64(8)]),
            (Timestamp(3), vec![Scalar::i64(9)]),
        ];
        assert_eq!(raw, raw_expected);
        assert_eq!(sub, sub_expected);
        assert_eq!(i64s, i64_expected);

        let (tr2, rr2) = mpsc::channel();
        let (ts2, rs2) = mpsc::channel();
        let (ti2, ri2) = mpsc::channel();
        let mut h2 = build(tr2, ts2, ti2);
        h2.run(vec![Box::new(VecSource::new(prices)), Box::new(VecSource::new(counts))]);
        assert_eq!(rr2.try_iter().collect::<Vec<_>>(), raw_expected); // deterministic
        assert_eq!(rs2.try_iter().collect::<Vec<_>>(), sub_expected);
        assert_eq!(ri2.try_iter().collect::<Vec<_>>(), i64_expected);
    }

    #[test]
    fn signal_quality_loop_records_pip_equity() {
        let (tx_eq, rx_eq) = mpsc::channel();
        let mut h = boot(
            vec![
                Box::new(Sma::new(2)),                              // 0 fast
                Box::new(Sma::new(4)),                              // 1 slow
                Box::new(Sub::new()),                               // 2 raw signal
                Box::new(Bias::new(0.5)),                           // 3 bias
                Box::new(SimBroker::new(0.0001)),                   // 4 pip equity
                Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx_eq)), // 5 sink
            ],
            vec![
                Sma::builder().schema().clone(),
                Sma::builder().schema().clone(),
                Sub::builder().schema().clone(),
                Bias::builder().schema().clone(),
                SimBroker::builder(0.0001).schema().clone(),
                recorder_sig(&[ScalarKind::F64], Firing::Any),
            ],
            vec![SourceSpec::raw(ScalarKind::F64, vec![
                    Target { node: 0, slot: 0 }, // price -> SMA fast
                    Target { node: 1, slot: 0 }, // price -> SMA slow
                    Target { node: 4, slot: 1 }, // price -> broker price input
                ])],
            vec![
                Edge { from: 0, to: 2, slot: 0, from_field: 0 }, // fast     -> Sub.0
                Edge { from: 1, to: 2, slot: 1, from_field: 0 }, // slow     -> Sub.1
                Edge { from: 2, to: 3, slot: 0, from_field: 0 }, // signal   -> Bias
                Edge { from: 3, to: 4, slot: 0, from_field: 0 }, // exposure -> broker.0
                Edge { from: 4, to: 5, slot: 0, from_field: 0 }, // equity   -> recorder
            ],
        )
        .expect("valid signal-quality harness");

        h.run(vec![Box::new(VecSource::new(f64_stream(&[
            (1, 1.0000),
            (2, 1.0010),
            (3, 1.0025),
            (4, 1.0020),
            (5, 1.0040),
        ])))]);

        let equity: Vec<(Timestamp, Vec<Scalar>)> = rx_eq.try_iter().collect();
        // broker fires every cycle (price fresh each tick): five records, ts 1..=5
        assert_eq!(equity.len(), 5);
        assert_eq!(
            equity.iter().map(|(t, _)| t.0).collect::<Vec<_>>(),
            vec![1, 2, 3, 4, 5]
        );
        // flat until SMA(4) warms (cycle 4) and the held exposure meets the next
        // price move (cycle 5): the first four equities are exactly 0.
        for (_, row) in &equity[0..4] {
            assert_eq!(row, &vec![Scalar::f64(0.0)]);
        }
        // cycle 5: prev_exposure 0.00175 * (1.0040 - 1.0020) / 0.0001 = 0.035 pips
        debug_assert_eq!(equity[4].1[0].kind(), ScalarKind::F64, "equity is f64");
        let last = equity[4].1[0].as_f64();
        assert!((last - 0.035).abs() < 1e-9, "final equity = {last}, want ~0.035");
    }

    #[test]
    fn signal_quality_loop_is_deterministic() {
        let build = || {
            let (tx, rx) = mpsc::channel();
            let mut h = boot(
                vec![
                    Box::new(Sma::new(2)),
                    Box::new(Sma::new(4)),
                    Box::new(Sub::new()),
                    Box::new(Bias::new(0.5)),
                    Box::new(SimBroker::new(0.0001)),
                    Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx)),
                ],
            vec![
                Sma::builder().schema().clone(),
                Sma::builder().schema().clone(),
                Sub::builder().schema().clone(),
                Bias::builder().schema().clone(),
                SimBroker::builder(0.0001).schema().clone(),
                recorder_sig(&[ScalarKind::F64], Firing::Any),
            ],
                vec![SourceSpec::raw(ScalarKind::F64, vec![
                        Target { node: 0, slot: 0 },
                        Target { node: 1, slot: 0 },
                        Target { node: 4, slot: 1 },
                    ])],
                vec![
                    Edge { from: 0, to: 2, slot: 0, from_field: 0 },
                    Edge { from: 1, to: 2, slot: 1, from_field: 0 },
                    Edge { from: 2, to: 3, slot: 0, from_field: 0 },
                    Edge { from: 3, to: 4, slot: 0, from_field: 0 },
                    Edge { from: 4, to: 5, slot: 0, from_field: 0 },
                ],
            )
            .expect("valid harness");
            h.run(vec![Box::new(VecSource::new(f64_stream(&[
                (1, 1.0000),
                (2, 1.0010),
                (3, 1.0025),
                (4, 1.0020),
                (5, 1.0040),
            ])))]);
            rx.try_iter().collect::<Vec<(Timestamp, Vec<Scalar>)>>()
        };
        assert_eq!(build(), build());
    }

    struct FinalizeProbe {
        id: usize,
        tx: mpsc::Sender<usize>,
    }
    impl Node for FinalizeProbe {
        fn lookbacks(&self) -> Vec<usize> {
            vec![1]
        }
        fn eval(&mut self, _ctx: Ctx<'_>) -> Option<&[Cell]> {
            None
        }
        fn finalize(&mut self) {
            let _ = self.tx.send(self.id);
        }
    }

    #[test]
    fn run_finalizes_every_node_once_after_the_stream() {
        let (tx, rx) = mpsc::channel();
        // two sink probes, both fed by the single source; no inter-node edges.
        let mut h = boot(
            vec![
                Box::new(FinalizeProbe { id: 0, tx: tx.clone() }),
                Box::new(FinalizeProbe { id: 1, tx }),
            ],
            vec![
                recorder_sig(&[ScalarKind::F64], Firing::Any),
                recorder_sig(&[ScalarKind::F64], Firing::Any),
            ],
            vec![SourceSpec::raw(ScalarKind::F64, vec![Target { node: 0, slot: 0 }, Target { node: 1, slot: 0 }])],
            vec![],
        )
        .expect("valid");
        // nothing finalized before the run.
        assert!(rx.try_recv().is_err());
        h.run(vec![Box::new(VecSource::new(f64_stream(&[(1, 1.0), (2, 2.0)])))]);
        // each node finalized exactly once, after the stream drained.
        let mut fired: Vec<usize> = rx.try_iter().collect();
        fired.sort_unstable();
        assert_eq!(fired, vec![0, 1]);
    }

    #[test]
    fn bind_tap_attaches_a_caller_built_sink_and_records_the_tapped_series() {
        // A tap on a producer; bind_tap attaches a caller-built Recorder; the run
        // records exactly the producer's per-cycle output. An undeclared tap name
        // is a typed refusal.
        let prices = f64_stream(&[(1, 10.0), (2, 20.0), (3, 30.0)]);
        // graph: source -> Sma(1) [node 0, producer], declared as tap "d" on (0,0).
        let mut flat = FlatGraph {
            nodes: vec![Box::new(Sma::new(1))],
            signatures: vec![Sma::builder().schema().clone()],
            sources: vec![SourceSpec::raw(ScalarKind::F64, vec![Target { node: 0, slot: 0 }])],
            edges: vec![],
            taps: vec![FlatTap { name: "d".into(), node: 0, field: 0 }],
        };
        // undeclared name refused
        let (tx_bad, _rx_bad) = mpsc::channel();
        assert!(matches!(
            flat.bind_tap("nope", Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx_bad)),
                          recorder_sig(&[ScalarKind::F64], Firing::Any)),
            Err(TapBindError::UndeclaredTap { .. })
        ));
        // declared name binds
        let (tx, rx) = mpsc::channel();
        flat.bind_tap("d", Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx)),
                      recorder_sig(&[ScalarKind::F64], Firing::Any))
            .expect("declared tap binds");
        let mut h = Harness::bootstrap(flat).expect("bootstraps with the appended recorder");
        h.run(vec![Box::new(VecSource::new(prices))]);
        let recorded: Vec<(Timestamp, Vec<Scalar>)> = rx.try_iter().collect();
        assert_eq!(
            recorded,
            vec![
                (Timestamp(1), vec![Scalar::f64(10.0)]),
                (Timestamp(2), vec![Scalar::f64(20.0)]),
                (Timestamp(3), vec![Scalar::f64(30.0)]),
            ],
            "the bound tap records exactly the producer's per-cycle output"
        );
    }
}
