feat(aura): node signature lives in the blueprint; collapse Blueprint into Composite

Consolidate the node data structure so every node's signature (NodeSchema:
inputs/output/params) is declared once and exists in the blueprint pre-build,
and dissolve the special "root graph" type. Behaviour-preserving (C1).

Signature vs sizing
- NodeSchema is now the static signature only: InputSpec -> PortSpec{kind,firing},
  with lookback removed. The signature is fully static per blueprint (input
  kinds/firing, output fields, params); LinComb's variable arity is a builder arg,
  not an injected param.
- The one param-dependent quantity, an input's buffer lookback (e.g. Sma's window =
  its injected length), moves out of the signature to Node::lookbacks() -> Vec<usize>,
  read only by bootstrap for sizing. Node::schema() is removed.
- LeafFactory -> PrimitiveBuilder, which carries the full NodeSchema. The built node
  no longer re-declares it: closes the params-declared-twice drift (#36, the 8
  per-node factory_params_match_built_node_schema lockstep tests are deleted — their
  subject is now structurally impossible) and a value-empty recipe exposes its full
  I/O interface pre-build (#43).

Root is just a bound composite
- struct Blueprint is deleted; its compile/bootstrap/param_space methods move onto
  Composite. Role gains source: Option<ScalarKind> (None = open interior port,
  Some = bound ingestion feed). A composite is runnable iff every root role is bound;
  the "main graph" is no longer a category, only the fully-source-bound composite.
  New error CompileError::UnboundRootRole for an open root role.
- BlueprintNode::signature() answers uniformly for both arms: Primitive returns the
  builder's declared schema, Composite derives it from the interior (role kinds in,
  OutField kinds out, aggregated params), pre-build, no build.

compile -> FlatGraph -> bootstrap
- compile validates structure pre-build via signature() (validate_wiring: range +
  kind, returning the same variants as before, so an edge kind fault is now caught
  before any build closure fires) and emits FlatGraph{nodes,signatures,sources,edges}.
- bootstrap consumes the FlatGraph: kinds/firing/output from the carried signatures,
  buffer depth from node.lookbacks(). SourceSpec survives as the flat descriptor.

Renames: BlueprintNode::Leaf -> Primitive, LeafFactory -> PrimitiveBuilder.

Render (aura-cli/src/graph.rs) is migrated compile-only: it takes &Composite, maps
bound roles to the same source-entry shape, so both render goldens reproduce
byte-identical output (no re-capture needed). Render-fidelity tuning is the next cycle.

Verification (orchestrator-run, not agent-reported): cargo build --workspace green;
cargo test --workspace 150 passed / 0 failed; cargo clippy --workspace --all-targets
-D warnings clean. All pinned determinism/run-output tests pass with values unchanged;
no behavioural assertion was altered to go green. 5 new tests assert the signature is
pre-build and uniform, that compile rejects a kind mismatch without building (via a
panicking builder), UnboundRootRole, and lookbacks()/signature arity agreement.

Deferred to cycle-close audit (per plan): docs/design/INDEX.md and some aura-std
module docs still name the old Node::schema()/LeafFactory/BlueprintNode::Leaf/
Blueprint::param_space contracts; prose reconciliation is the architect's at audit.

closes #43 #36

crates/aura-cli/src/graph.rs

@@ -15,10 +15,10 @@
 
 use ascii_dag::graph::{Graph, RenderMode};
 use ascii_dag::render::colors::Palette;
-use aura_core::{LeafFactory, Node, ScalarKind};
+use aura_core::{Node, PrimitiveBuilder, ScalarKind};
 use aura_engine::{
-    aliases_on, signature_of, Blueprint, BlueprintNode, Composite, Edge, OutField, ParamAlias,
-    SourceSpec, Target,
+    aliases_on, signature_of, BlueprintNode, Composite, Edge, OutField, ParamAlias, SourceSpec,
+    Target,
 };
 
 /// Where a leaf's param **names** come from in the shared leaf-label path — the one
@@ -58,18 +58,24 @@ pub enum Color {
 /// Blueprint view: the authored structure (#38). A flat main graph wires the
 /// harness with each composite shown as a single opaque node; a `where:` section
 /// defines each distinct composite type once. Flat layout only (no subgraphs).
-pub fn render_blueprint(bp: &Blueprint, color: Color) -> String {
-    // the main graph is the shared graph core (`render_graph`) over the blueprint's
-    // top-level (nodes, edges): leaves enriched exactly as `where:` interior leaves
-    // are (param names folded in, a `<field> →` prefix for any input fed by a
-    // multi-output producer — the headline: macd's `histogram` driving Exposure),
-    // composites opaque. The deltas vs a composite definition: param names come from
-    // the factory (no alias overlay at the root); the entries are sources, not roles;
-    // there is no output record (terminals are sinks) and no title.
+pub fn render_blueprint(bp: &Composite, color: Color) -> String {
+    // the main graph is the shared graph core (`render_graph`) over the root
+    // composite's top-level (nodes, edges): leaves enriched exactly as `where:`
+    // interior leaves are (param names folded in, a `<field> →` prefix for any input
+    // fed by a multi-output producer — the headline: macd's `histogram` driving
+    // Exposure), composites opaque. The deltas vs a composite definition: param names
+    // come from the builder (no alias overlay at the root); the entries are the
+    // root's source-bound roles, not interior roles; there is no output record
+    // (terminals are sinks) and no title.
     let entries: Vec<Entry> = bp
-        .sources()
+        .input_roles()
         .iter()
-        .map(|src| Entry { name: format!("source:{:?}", src.kind), targets: &src.targets })
+        .filter_map(|role| {
+            role.source.map(|kind| Entry {
+                name: format!("source:{kind:?}"),
+                targets: &role.targets,
+            })
+        })
         .collect();
     let main = render_graph(
         bp.nodes(),
@@ -94,7 +100,7 @@ pub fn render_blueprint(bp: &Blueprint, color: Color) -> String {
 /// The shared graph core both views render through (#48): a flat ascii-dag of the
 /// computation DAG plus external-entry markers. `nodes`/`edges` are a borrowed
 /// (blueprint-root or composite-interior) graph slice — never an owned
-/// `Composite` (`LeafFactory` is not `Clone`). Each leaf gets the unified
+/// `Composite` (`PrimitiveBuilder` is not `Clone`). Each leaf gets the unified
 /// [`leaf_label`]; each composite is opaque (`[name]`); a producing node carrying an
 /// `OutField` re-export gets the `name := ` binding prefix folded on (empty list →
 /// no binding, the blueprint case); every [`Entry`] is drawn as a marker wired to its
@@ -115,7 +121,7 @@ fn render_graph(
     let mut labels: Vec<String> = Vec::with_capacity(nodes.len());
     for (i, item) in nodes.iter().enumerate() {
         let base = match item {
-            BlueprintNode::Leaf(factory) => {
+            BlueprintNode::Primitive(factory) => {
                 leaf_label(nodes, edges, entries, i, factory, &param_names, stub_ctx)
             }
             BlueprintNode::Composite(c) => c.name().to_string(),
@@ -166,7 +172,7 @@ fn render_flat(labels: &[String], edges: &[(usize, usize)], color: Color) -> Str
 /// `name()` (the authoring type identity — same name implies same structure).
 /// Recurses into a composite's interior on first sight so nested composites get
 /// their own definition; a later same-name occurrence is skipped (deduped).
-fn collect_distinct_composites(bp: &Blueprint) -> Vec<&Composite> {
+fn collect_distinct_composites(bp: &Composite) -> Vec<&Composite> {
     fn walk<'a>(items: &'a [BlueprintNode], seen: &mut Vec<&'a str>, out: &mut Vec<&'a Composite>) {
         for item in items {
             if let BlueprintNode::Composite(c) = item
@@ -209,7 +215,7 @@ fn signature(c: &Composite) -> String {
                 .nodes()
                 .get(a.node)
                 .and_then(|n| match n {
-                    BlueprintNode::Leaf(f) => f.params().get(a.slot).map(|p| kind_str(p.kind)),
+                    BlueprintNode::Primitive(f) => f.params().get(a.slot).map(|p| kind_str(p.kind)),
                     BlueprintNode::Composite(_) => None,
                 })
                 .unwrap_or("?");
@@ -247,7 +253,7 @@ fn leaf_label(
     edges: &[Edge],
     entries: &[Entry],
     index: usize,
-    factory: &LeafFactory,
+    factory: &PrimitiveBuilder,
     param_names: &ParamNames,
     stub_ctx: Option<&Composite>,
 ) -> String {
@@ -334,7 +340,7 @@ fn multi_output_field_name(nodes: &[BlueprintNode], from: usize, from_field: usi
                 None
             }
         }
-        BlueprintNode::Leaf(_) => (from_field > 0).then(|| from_field.to_string()),
+        BlueprintNode::Primitive(_) => (from_field > 0).then(|| from_field.to_string()),
     }
 }
 
@@ -396,7 +402,7 @@ fn slot_source(c: &Composite, index: usize, slot: usize) -> (String, usize, bool
 /// rendered identifier never goes below.
 fn signature_base_len(c: &Composite, node: usize) -> usize {
     match &c.nodes()[node] {
-        BlueprintNode::Leaf(_) => 1 + aliases_on(c.params(), node).count(),
+        BlueprintNode::Primitive(_) => 1 + aliases_on(c.params(), node).count(),
         BlueprintNode::Composite(_) => 1,
     }
 }

crates/aura-cli/src/main.rs

@@ -10,8 +10,8 @@ mod graph;
 
 use aura_core::{Firing, Scalar, ScalarKind, Timestamp};
 use aura_engine::{
-    f64_field, summarize, Blueprint, BlueprintNode, Composite, Edge, Harness, OutField,
-    ParamAlias, Role, RunManifest, RunReport, SourceSpec, Target,
+    f64_field, summarize, BlueprintNode, Composite, Edge, FlatGraph, Harness, OutField, ParamAlias,
+    Role, RunManifest, RunReport, SourceSpec, Target,
 };
 use aura_std::{Ema, Exposure, Recorder, SimBroker, Sma, Sub};
 use std::io::IsTerminal;
@@ -50,8 +50,13 @@ fn sample_harness() -> (
 ) {
     let (tx_eq, rx_eq) = mpsc::channel();
     let (tx_ex, rx_ex) = mpsc::channel();
-    let h = Harness::bootstrap(
-        vec![
+    let f64_recorder_sig = || aura_engine::NodeSchema {
+        inputs: vec![aura_engine::PortSpec { kind: ScalarKind::F64, firing: Firing::Any }],
+        output: vec![],
+        params: vec![],
+    };
+    let h = Harness::bootstrap(FlatGraph {
+        nodes: vec![
             Box::new(Sma::new(2)),                                          // 0 fast SMA
             Box::new(Sma::new(4)),                                          // 1 slow SMA
             Box::new(Sub::new()),                                           // 2 spread
@@ -60,7 +65,16 @@ fn sample_harness() -> (
             Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx_eq)), // 5 equity sink
             Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx_ex)), // 6 exposure sink
         ],
-        vec![SourceSpec {
+        signatures: vec![
+            Sma::builder().schema().clone(),
+            Sma::builder().schema().clone(),
+            Sub::builder().schema().clone(),
+            Exposure::builder().schema().clone(),
+            SimBroker::builder(0.0001).schema().clone(),
+            f64_recorder_sig(),
+            f64_recorder_sig(),
+        ],
+        sources: vec![SourceSpec {
             kind: ScalarKind::F64,
             targets: vec![
                 Target { node: 0, slot: 0 },
@@ -68,7 +82,7 @@ fn sample_harness() -> (
                 Target { node: 4, slot: 1 }, // price into the broker's price slot
             ],
         }],
-        vec![
+        edges: vec![
             Edge { from: 0, to: 2, slot: 0, from_field: 0 },
             Edge { from: 1, to: 2, slot: 1, from_field: 0 },
             Edge { from: 2, to: 3, slot: 0, from_field: 0 },
@@ -76,7 +90,7 @@ fn sample_harness() -> (
             Edge { from: 4, to: 5, slot: 0, from_field: 0 }, // equity -> sink 5
             Edge { from: 3, to: 6, slot: 0, from_field: 0 }, // exposure -> sink 6
         ],
-    )
+    })
     .expect("valid sample signal-quality DAG");
     (h, rx_eq, rx_ex)
 }
@@ -122,7 +136,7 @@ fn run_sample() -> RunReport {
 fn sma_cross(name: &str) -> Composite {
     Composite::new(
         name,
-        vec![Sma::factory().into(), Sma::factory().into(), Sub::factory().into()],
+        vec![Sma::builder().into(), Sma::builder().into(), Sub::builder().into()],
         vec![
             Edge { from: 0, to: 2, slot: 0, from_field: 0 },
             Edge { from: 1, to: 2, slot: 1, from_field: 0 },
@@ -130,6 +144,7 @@ fn sma_cross(name: &str) -> Composite {
         vec![Role {
             name: "price".into(),
             targets: vec![Target { node: 0, slot: 0 }, Target { node: 1, slot: 0 }],
+            source: None,
         }],
         vec![
             ParamAlias { name: "fast".into(), node: 0, slot: 0 }, // fast SMA length
@@ -143,35 +158,39 @@ fn sma_cross(name: &str) -> Composite {
 /// exposure scale are injected at compile via the point vector (see
 /// `sample_point`). Recorders need a channel to construct; the receivers are
 /// dropped because the render never runs the graph.
-fn build_sample() -> Blueprint {
+fn build_sample() -> Composite {
     let (tx_eq, _rx_eq) = mpsc::channel();
     let (tx_ex, _rx_ex) = mpsc::channel();
-    Blueprint::new(
+    Composite::new(
+        "sample",
         vec![
             BlueprintNode::Composite(sma_cross("sma_cross")),
-            Exposure::factory().into(),
-            SimBroker::factory(0.0001).into(),
-            Recorder::factory(vec![ScalarKind::F64], Firing::Any, tx_eq).into(),
-            Recorder::factory(vec![ScalarKind::F64], Firing::Any, tx_ex).into(),
+            Exposure::builder().into(),
+            SimBroker::builder(0.0001).into(),
+            Recorder::builder(vec![ScalarKind::F64], Firing::Any, tx_eq).into(),
+            Recorder::builder(vec![ScalarKind::F64], Firing::Any, tx_ex).into(),
         ],
-        vec![SourceSpec {
-            kind: ScalarKind::F64,
-            targets: vec![
-                Target { node: 0, slot: 0 }, // price -> sma_cross role 0
-                Target { node: 2, slot: 1 }, // price -> SimBroker price slot
-            ],
-        }],
         vec![
             Edge { from: 0, to: 1, slot: 0, from_field: 0 }, // spread   -> Exposure
             Edge { from: 1, to: 2, slot: 0, from_field: 0 }, // exposure -> broker slot 0
             Edge { from: 2, to: 3, slot: 0, from_field: 0 }, // equity   -> sink
             Edge { from: 1, to: 4, slot: 0, from_field: 0 }, // exposure -> sink
         ],
+        vec![Role {
+            name: "price".into(),
+            targets: vec![
+                Target { node: 0, slot: 0 }, // price -> sma_cross role 0
+                Target { node: 2, slot: 1 }, // price -> SimBroker price slot
+            ],
+            source: Some(ScalarKind::F64),
+        }],
+        vec![], // params: the interior sma_cross carries the aliases
+        vec![], // output: the root ends in sinks, no re-export
     )
 }
 
 /// The built-in sample rendered by `aura graph`.
-fn sample_blueprint() -> Blueprint {
+fn sample_blueprint() -> Composite {
     build_sample()
 }
 
@@ -195,11 +214,11 @@ fn macd(name: &str) -> Composite {
     Composite::new(
         name,
         vec![
-            Ema::factory().into(), // 0 fast EMA
-            Ema::factory().into(), // 1 slow EMA
-            Sub::factory().into(), // 2 MACD line = fast − slow
-            Ema::factory().into(), // 3 signal EMA of the MACD line
-            Sub::factory().into(), // 4 histogram = MACD line − signal
+            Ema::builder().into(), // 0 fast EMA
+            Ema::builder().into(), // 1 slow EMA
+            Sub::builder().into(), // 2 MACD line = fast − slow
+            Ema::builder().into(), // 3 signal EMA of the MACD line
+            Sub::builder().into(), // 4 histogram = MACD line − signal
         ],
         vec![
             Edge { from: 0, to: 2, slot: 0, from_field: 0 }, // fast   → line[0]
@@ -214,6 +233,7 @@ fn macd(name: &str) -> Composite {
                 Target { node: 0, slot: 0 }, // price → fast EMA
                 Target { node: 1, slot: 0 }, // price → slow EMA
             ],
+            source: None,
         }],
         vec![
             ParamAlias { name: "fast".into(), node: 0, slot: 0 },   // fast EMA length
@@ -234,34 +254,38 @@ fn macd(name: &str) -> Composite {
 fn macd_strategy_blueprint(
     tx_eq: mpsc::Sender<(Timestamp, Vec<Scalar>)>,
     tx_ex: mpsc::Sender<(Timestamp, Vec<Scalar>)>,
-) -> Blueprint {
-    Blueprint::new(
+) -> Composite {
+    Composite::new(
+        "macd_strategy",
         vec![
             BlueprintNode::Composite(macd("macd")),
-            Exposure::factory().into(),
-            SimBroker::factory(0.0001).into(),
-            Recorder::factory(vec![ScalarKind::F64], Firing::Any, tx_eq).into(),
-            Recorder::factory(vec![ScalarKind::F64], Firing::Any, tx_ex).into(),
+            Exposure::builder().into(),
+            SimBroker::builder(0.0001).into(),
+            Recorder::builder(vec![ScalarKind::F64], Firing::Any, tx_eq).into(),
+            Recorder::builder(vec![ScalarKind::F64], Firing::Any, tx_ex).into(),
         ],
-        vec![SourceSpec {
-            kind: ScalarKind::F64,
-            targets: vec![
-                Target { node: 0, slot: 0 }, // price → macd role 0
-                Target { node: 2, slot: 1 }, // price → SimBroker price slot
-            ],
-        }],
         vec![
             Edge { from: 0, to: 1, slot: 0, from_field: 2 }, // histogram → Exposure
             Edge { from: 1, to: 2, slot: 0, from_field: 0 }, // exposure  → broker slot 0
             Edge { from: 2, to: 3, slot: 0, from_field: 0 }, // equity    → sink
             Edge { from: 1, to: 4, slot: 0, from_field: 0 }, // exposure  → sink
         ],
+        vec![Role {
+            name: "price".into(),
+            targets: vec![
+                Target { node: 0, slot: 0 }, // price → macd role 0
+                Target { node: 2, slot: 1 }, // price → SimBroker price slot
+            ],
+            source: Some(ScalarKind::F64),
+        }],
+        vec![], // params: the interior macd carries the aliases
+        vec![], // output: the root ends in sinks, no re-export
     )
 }
 
 /// The MACD strategy blueprint for the structural render (receivers dropped, as
 /// the render never runs the graph).
-fn macd_blueprint() -> Blueprint {
+fn macd_blueprint() -> Composite {
     let (tx_eq, _rx_eq) = mpsc::channel();
     let (tx_ex, _rx_ex) = mpsc::channel();
     macd_strategy_blueprint(tx_eq, tx_ex)
@@ -296,10 +320,10 @@ fn macd_prices() -> Vec<(Timestamp, Scalar)> {
 fn run_macd() -> RunReport {
     let (tx_eq, rx_eq) = mpsc::channel();
     let (tx_ex, rx_ex) = mpsc::channel();
-    let (nodes, sources, edges) = macd_strategy_blueprint(tx_eq, tx_ex)
+    let flat = macd_strategy_blueprint(tx_eq, tx_ex)
         .compile_with_params(&macd_point())
         .expect("valid macd blueprint");
-    let mut h = Harness::bootstrap(nodes, sources, edges).expect("valid macd harness");
+    let mut h = Harness::bootstrap(flat).expect("valid macd harness");
 
     let prices = macd_prices();
     let window = (
@@ -333,9 +357,9 @@ fn run_macd() -> RunReport {
 
 /// Compile a blueprint under its point vector and render the flat post-inline
 /// (C23) view — the shared body of the `--compiled` paths.
-fn render_compiled(bp: Blueprint, point: &[Scalar], color: graph::Color) -> String {
-    let (nodes, sources, edges) = bp.compile_with_params(point).expect("valid blueprint");
-    graph::render_compilat(&nodes, &sources, &edges, color)
+fn render_compiled(bp: Composite, point: &[Scalar], color: graph::Color) -> String {
+    let flat = bp.compile_with_params(point).expect("valid blueprint");
+    graph::render_compilat(&flat.nodes, &flat.sources, &flat.edges, color)
 }
 
 const USAGE: &str = "usage: aura run [--macd] | aura graph [--compiled | --macd [--compiled]]";
@@ -415,25 +439,30 @@ mod tests {
         // structure only (no compile/validate), so a minimal fixture suffices.
         let inner = Composite::new(
             "inner",
-            vec![Sma::factory().into()],
+            vec![Sma::builder().into()],
             vec![],
-            vec![Role { name: "price".into(), targets: vec![Target { node: 0, slot: 0 }] }],
+            vec![Role { name: "price".into(), targets: vec![Target { node: 0, slot: 0 }], source: None }],
             vec![],
             vec![OutField { node: 0, field: 0, name: "out".into() }],
         );
         let outer = Composite::new(
             "outer",
-            vec![BlueprintNode::Composite(inner), Sub::factory().into()],
+            vec![BlueprintNode::Composite(inner), Sub::builder().into()],
             vec![Edge { from: 0, to: 1, slot: 0, from_field: 0 }],
-            vec![Role { name: "price".into(), targets: vec![Target { node: 0, slot: 0 }] }],
+            vec![Role { name: "price".into(), targets: vec![Target { node: 0, slot: 0 }], source: None }],
             vec![],
             vec![OutField { node: 1, field: 0, name: "out".into() }],
         );
-        let bp = Blueprint::new(
-            vec![BlueprintNode::Composite(outer)],
-            vec![SourceSpec { kind: ScalarKind::F64, targets: vec![Target { node: 0, slot: 0 }] }],
-            vec![],
-        );
+        let bp = Composite::new(
+        "root",
+        vec![BlueprintNode::Composite(outer)],
+        vec![],
+        vec![
+            Role { name: "src".into(), targets: vec![Target { node: 0, slot: 0 }], source: Some(ScalarKind::F64) },
+        ],
+        vec![], // params
+        vec![], // output
+    );
         let out = graph::render_blueprint(&bp, graph::Color::Plain); // must not panic (no unimplemented!)
         // outer shows the inner composite as an opaque node, and both get a definition
         assert!(out.contains("[outer]"), "missing opaque outer node:\n{out}");
@@ -445,21 +474,23 @@ mod tests {
     #[test]
     fn reused_composite_defined_once() {
         // the same composite type used twice: two opaque nodes, one definition.
-        let bp = Blueprint::new(
-            vec![
+        let bp = Composite::new(
+        "root",
+        vec![
                 BlueprintNode::Composite(sma_cross("dup")),
                 BlueprintNode::Composite(sma_cross("dup")),
-                Exposure::factory().into(),
+                Exposure::builder().into(),
             ],
-            vec![SourceSpec {
-                kind: ScalarKind::F64,
-                targets: vec![Target { node: 0, slot: 0 }, Target { node: 1, slot: 0 }],
-            }],
-            vec![
+        vec![
                 Edge { from: 0, to: 2, slot: 0, from_field: 0 },
                 Edge { from: 1, to: 2, slot: 0, from_field: 0 },
             ],
-        );
+        vec![
+            Role { name: "src".into(), targets: vec![Target { node: 0, slot: 0 }, Target { node: 1, slot: 0 }], source: Some(ScalarKind::F64) },
+        ],
+        vec![], // params
+        vec![], // output
+    );
         let out = graph::render_blueprint(&bp, graph::Color::Plain);
         assert_eq!(out.matches("[dup]").count(), 2, "two opaque uses expected:\n{out}");
         assert_eq!(out.matches("dup(").count(), 1, "body defined once:\n{out}");
@@ -468,9 +499,8 @@ mod tests {
     #[test]
     fn compiled_view_dissolves_the_composite_boundary() {
         let bp = sample_blueprint();
-        let (nodes, sources, edges) =
-            bp.compile_with_params(&sample_point()).expect("valid sample");
-        let out = graph::render_compilat(&nodes, &sources, &edges, graph::Color::Plain);
+        let flat = bp.compile_with_params(&sample_point()).expect("valid sample");
+        let out = graph::render_compilat(&flat.nodes, &flat.sources, &flat.edges, graph::Color::Plain);
         // node labels survive inlining...
         assert!(out.contains("SMA(2)") && out.contains("SMA(4)"), "labels lost:\n{out}");
         // ...but the composite cluster name does NOT (boundary dissolved, C23)
@@ -483,12 +513,12 @@ mod tests {
         // The blueprint is now param-generic (identical for both orderings), so the
         // swap is observable only after the vector is injected — in the COMPILED
         // view, where the flat nodes carry their valued labels (SMA(2)/SMA(4)).
-        let (cn, cs, ce) =
+        let cflat =
             sample_blueprint().compile_with_params(&sample_point()).expect("valid sample");
-        let correct = graph::render_compilat(&cn, &cs, &ce, graph::Color::Plain);
-        let (sn, ss, se) =
+        let correct = graph::render_compilat(&cflat.nodes, &cflat.sources, &cflat.edges, graph::Color::Plain);
+        let sflat =
             sample_blueprint().compile_with_params(&swapped_point()).expect("valid sample");
-        let swapped = graph::render_compilat(&sn, &ss, &se, graph::Color::Plain);
+        let swapped = graph::render_compilat(&sflat.nodes, &sflat.sources, &sflat.edges, graph::Color::Plain);
         assert_ne!(correct, swapped, "a fast/slow SMA swap must change the compiled render");
     }
 
@@ -534,9 +564,8 @@ sma_cross(fast:i64, slow:i64) -> (cross):
     #[test]
     fn compiled_view_golden() {
         let bp = sample_blueprint();
-        let (nodes, sources, edges) =
-            bp.compile_with_params(&sample_point()).expect("valid sample");
-        let out = graph::render_compilat(&nodes, &sources, &edges, graph::Color::Plain);
+        let flat = bp.compile_with_params(&sample_point()).expect("valid sample");
+        let out = graph::render_compilat(&flat.nodes, &flat.sources, &flat.edges, graph::Color::Plain);
         let expected = r#"            [source:F64]
          ┌────────└─┐──────┐
          ↓          ↓      │
@@ -615,13 +644,20 @@ sma_cross(fast:i64, slow:i64) -> (cross):
         //   #price (role name) and #Es
         let c = Composite::new(
             "roles",
-            vec![Ema::factory().into(), Sub::factory().into()],
+            vec![Ema::builder().into(), Sub::builder().into()],
             vec![Edge { from: 0, to: 1, slot: 1, from_field: 0 }],
-            vec![Role { name: "price".into(), targets: vec![Target { node: 1, slot: 0 }] }],
+            vec![Role { name: "price".into(), targets: vec![Target { node: 1, slot: 0 }], source: None }],
             vec![ParamAlias { name: "slow".into(), node: 0, slot: 0 }],
             vec![OutField { node: 1, field: 0, name: "o".into() }],
         );
-        let bp = Blueprint::new(vec![BlueprintNode::Composite(c)], vec![], vec![]);
+        let bp = Composite::new(
+        "root",
+        vec![BlueprintNode::Composite(c)],
+        vec![],
+        vec![],
+        vec![], // params
+        vec![], // output
+    );
         let out = graph::render_blueprint(&bp, graph::Color::Plain);
         assert!(out.contains("[o := Sub(#price,#Es)]"), "role name verbatim + source-derived, bound as output `o`: {out}");
     }
@@ -634,13 +670,13 @@ sma_cross(fast:i64, slow:i64) -> (cross):
         let c = Composite::new(
             "nest",
             vec![
-                Ema::factory().into(), // 0 fast
-                Ema::factory().into(), // 1 slow
-                Ema::factory().into(), // 2 up
-                Ema::factory().into(), // 3 down
-                Sub::factory().into(), // 4 = Sub(0,1)
-                Sub::factory().into(), // 5 = Sub(2,3)
-                Sub::factory().into(), // 6 = Sub(4,5)  (the outer fan-in)
+                Ema::builder().into(), // 0 fast
+                Ema::builder().into(), // 1 slow
+                Ema::builder().into(), // 2 up
+                Ema::builder().into(), // 3 down
+                Sub::builder().into(), // 4 = Sub(0,1)
+                Sub::builder().into(), // 5 = Sub(2,3)
+                Sub::builder().into(), // 6 = Sub(4,5)  (the outer fan-in)
             ],
             vec![
                 Edge { from: 0, to: 4, slot: 0, from_field: 0 },
@@ -657,8 +693,7 @@ sma_cross(fast:i64, slow:i64) -> (cross):
                     Target { node: 1, slot: 0 },
                     Target { node: 2, slot: 0 },
                     Target { node: 3, slot: 0 },
-                ],
-            }],
+                ], source: None, }],
             vec![
                 ParamAlias { name: "fast".into(), node: 0, slot: 0 },
                 ParamAlias { name: "slow".into(), node: 1, slot: 0 },
@@ -667,7 +702,14 @@ sma_cross(fast:i64, slow:i64) -> (cross):
             ],
             vec![OutField { node: 6, field: 0, name: "o".into() }],
         );
-        let bp = Blueprint::new(vec![BlueprintNode::Composite(c)], vec![], vec![]);
+        let bp = Composite::new(
+        "root",
+        vec![BlueprintNode::Composite(c)],
+        vec![],
+        vec![],
+        vec![], // params
+        vec![], // output
+    );
         let out = graph::render_blueprint(&bp, graph::Color::Plain);
         assert!(out.contains("[o := Sub(#SEf,#SEu)]"), "outer Sub descends into inner Subs, bound as output `o`: {out}");
         assert!(out.contains("[Sub(#Ef,#Es)]"), "inner Sub uses EMA aliases: {out}");

crates/aura-core/src/lib.rs

@@ -16,9 +16,10 @@
 //!
 //! Delivered in cycle 0002 — the node contract:
 //!
-//! - [`Node`] — the `schema`/`eval` contract every node implements (C8), with
-//!   [`NodeSchema`] / [`InputSpec`] declaring inputs (kind + lookback) and the
-//!   output record ([`FieldSpec`] columns; length 1 = scalar);
+//! - [`Node`] — the `lookbacks`/`eval` contract every node implements (C8); the
+//!   static signature ([`NodeSchema`] / [`PortSpec`] declaring input ports by kind
+//!   and the output record of [`FieldSpec`] columns; length 1 = scalar) is declared
+//!   on the node's [`PrimitiveBuilder`], pre-build;
 //! - [`Ctx`] — the per-`eval` read-side: zero-copy, financial-indexed [`Window`]
 //!   access into each input (closing the cycle-0001 read-side gap on
 //!   [`AnyColumn`]).
@@ -39,5 +40,5 @@ pub use any::AnyColumn;
 pub use column::{Column, Window};
 pub use ctx::Ctx;
 pub use error::KindMismatch;
-pub use node::{FieldSpec, Firing, InputSpec, LeafFactory, Node, NodeSchema, ParamSpec};
+pub use node::{FieldSpec, Firing, Node, NodeSchema, ParamSpec, PortSpec, PrimitiveBuilder};
 pub use scalar::{Scalar, ScalarKind, Timestamp};

crates/aura-core/src/node.rs

@@ -1,12 +1,14 @@
-//! The node contract (C8): the interface every node implements. A node declares
-//! its inputs (each with its scalar kind, lookback depth, and firing policy, C6)
-//! and its output record (0..K base columns, C7; an empty output declares a
-//! pure consumer / sink role, C8) via `schema`, and computes one cycle's row
-//! via `eval`.
-//! Tunable params (C12/C19) are declared via `params` (cycle 0015): each node's
-//! typed knobs, which `Blueprint::param_space` aggregates into the sweep's flat,
-//! path-qualified param-space (C8/C23). Identity is positional (slot); the name is
-//! a non-load-bearing debug symbol.
+//! The node contract (C8): the interface every node implements. A node's static
+//! signature (its input ports — each a scalar kind + firing policy, C6 — its output
+//! record of 0..K base columns, C7, and its tunable params) is declared once on the
+//! node's `PrimitiveBuilder` (`NodeSchema`), pre-build. The one signature-adjacent
+//! quantity that may depend on an injected param — the per-input buffer lookback —
+//! is answered by `Node::lookbacks()`, read only by bootstrap. `eval` computes one
+//! cycle's row.
+//! Tunable params (C12/C19) are declared via the builder's `NodeSchema.params`
+//! (cycle 0015): each node's typed knobs, which `Composite::param_space` aggregates
+//! into the sweep's flat, path-qualified param-space (C8/C23). Identity is
+//! positional (slot); the name is a non-load-bearing debug symbol.
 
 use crate::{Ctx, Scalar, ScalarKind};
 
@@ -24,12 +26,13 @@ pub enum Firing {
     Barrier(u8),
 }
 
-/// One declared input of a node: its scalar kind, the lookback depth the engine
-/// must pre-size for it (must be >= 1), and its firing policy (C6).
+/// One declared input **port** of a node: its scalar kind and firing policy (C6).
+/// The lookback depth is NOT here — it is a build-time *sizing* concern answered by
+/// `Node::lookbacks()`, not part of the static signature (a node's lookback can
+/// depend on an injected param, e.g. `Sma`'s window = its `length`).
 #[derive(Clone, Copy, Debug, PartialEq, Eq)]
-pub struct InputSpec {
+pub struct PortSpec {
     pub kind: ScalarKind,
-    pub lookback: usize,
     pub firing: Firing,
 }
 
@@ -57,49 +60,50 @@ pub struct ParamSpec {
     pub kind: ScalarKind,
 }
 
-/// A param-generic blueprint leaf (C19): a node's declared tunable params plus a
-/// closure that builds a sized instance through the node's own constructor (the
-/// single sizing/validation gate). A blueprint holds these recipes, never built
-/// instances, so it stays value-empty until a param-set is injected (C19/C23).
-pub struct LeafFactory {
+/// A param-generic blueprint **primitive** recipe (C19): a node's full declared
+/// signature (`NodeSchema` — inputs/output/params) plus a closure that builds a
+/// sized instance through the node's own constructor (the single sizing/validation
+/// gate). A blueprint holds these recipes, never built instances, so it stays
+/// value-empty until a param-set is injected (C19/C23). The signature is declared
+/// here ONCE — the built node no longer re-declares it (closes the param-declared-
+/// twice drift, #36) and a value-empty recipe now exposes its full I/O interface
+/// pre-build (#43).
+pub struct PrimitiveBuilder {
     name: &'static str,
-    params: Vec<ParamSpec>,
+    schema: NodeSchema,
     // The build closure's type is exactly the recipe contract (a param slice in, a
     // boxed node out); a type alias would not clarify it.
     #[allow(clippy::type_complexity)]
     build: Box<dyn Fn(&[Scalar]) -> Box<dyn Node>>,
 }
 
-impl LeafFactory {
+impl PrimitiveBuilder {
     /// `name` is the param-generic render label (the node type, e.g. `"SMA"`);
-    /// `params` the declared knobs; `build` constructs a sized node from a
+    /// `schema` the full declared signature; `build` constructs a sized node from a
     /// kind-checked param slice.
     pub fn new(
         name: &'static str,
-        params: Vec<ParamSpec>,
+        schema: NodeSchema,
         build: impl Fn(&[Scalar]) -> Box<dyn Node> + 'static,
     ) -> Self {
-        Self { name, params, build: Box::new(build) }
+        Self { name, schema, build: Box::new(build) }
     }
-    /// The declared tunable params (read by `Blueprint::param_space`, pre-build).
+    /// The full declared signature (read pre-build by `Composite::param_space`,
+    /// `BlueprintNode::signature`, and the renderer).
+    pub fn schema(&self) -> &NodeSchema {
+        &self.schema
+    }
+    /// The declared tunable params (a view into the signature; pre-build).
     pub fn params(&self) -> &[ParamSpec] {
-        &self.params
+        &self.schema.params
     }
     /// Build a sized node from its param slice (the slice is kind-checked by the
     /// caller before this runs).
     pub fn build(&self, params: &[Scalar]) -> Box<dyn Node> {
         (self.build)(params)
     }
-    /// The param-generic render label for the blueprint view (C22 "structure
-    /// before"): just the node type, e.g. `SMA`. A value-empty recipe has no values
-    /// to show; the tunable knobs are surfaced by `Blueprint::param_space`. The
-    /// factory label stays the bare type because alias / handle names are a
-    /// *composite-level* concept — the renderer folds them into the leaf label at
-    /// the composite boundary (`render_definition`'s `leaf_label`), where the
-    /// `(node, slot)` → name mapping lives, not on the standalone factory. (Wide
-    /// labels are safe: both `aura graph` views are flat since cycle 0017 — no
-    /// cluster subgraph, so no sibling-overlap garble.) The compiled view labels the
-    /// built node valued (`SMA(2)`) via `Node::label`, unaffected.
+    /// The param-generic render label for the blueprint view (C22): just the node
+    /// type, e.g. `SMA`.
     pub fn label(&self) -> String {
         self.name.to_string()
     }
@@ -112,7 +116,7 @@ impl LeafFactory {
 /// the hot path — the `Vec`s are fine here.
 #[derive(Clone, Debug, PartialEq, Eq)]
 pub struct NodeSchema {
-    pub inputs: Vec<InputSpec>,
+    pub inputs: Vec<PortSpec>,
     pub output: Vec<FieldSpec>,
     pub params: Vec<ParamSpec>,
 }
@@ -120,15 +124,22 @@ pub struct NodeSchema {
 /// The universal composable dataflow unit (C8): a **producer, consumer, or both**.
 /// A producer/transformer exposes one output port carrying a record of 1..K base
 /// columns; a **pure consumer (sink)** declares `output: vec![]` and records via
-/// an out-of-graph side effect in `eval` (cycle 0006). `schema` declares the
-/// interface; `eval` computes one cycle's row, returning a borrowed slice into a
-/// buffer the node owns. `None` = filter / not-yet-warmed-up / pure sink; a
-/// returned `Some(row)` must satisfy `row.len() == schema().output.len()` (so a
+/// an out-of-graph side effect in `eval` (cycle 0006). The static signature is
+/// declared on the node's `PrimitiveBuilder`; `eval` computes one cycle's row,
+/// returning a borrowed slice into a buffer the node owns. `None` = filter /
+/// not-yet-warmed-up / pure sink; a returned `Some(row)` must satisfy
+/// `row.len() == signature().output.len()` (so a
 /// pure consumer returns `None` or an empty `Some(&[])`) — the engine debug-asserts
 /// the width and forwards the row field-wise along the node's out-edges. `&mut self`
 /// because a node may keep its own derived state (and its output buffer).
 pub trait Node {
-    fn schema(&self) -> NodeSchema;
+    /// The per-input buffer **lookback** depth (each `>= 1`), in input-slot order —
+    /// the only signature-adjacent quantity that may depend on an injected param
+    /// (e.g. `Sma`'s window = its `length`). Read once by `Harness::bootstrap` to
+    /// size each input column; never on the hot path. Length MUST equal the node's
+    /// declared `signature().inputs.len()`.
+    fn lookbacks(&self) -> Vec<usize>;
+
     fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Scalar]>;
     /// A one-line, **non-load-bearing** render label (C23): a debug symbol for
     /// tracing / graph rendering (#13), never read by the run loop and never
@@ -147,11 +158,11 @@ pub trait Node {
 mod tests {
     use super::*;
 
-    /// A minimal node with an empty schema, reused across the label / factory tests.
+    /// A minimal node with no inputs, reused across the label / builder tests.
     struct Bare;
     impl Node for Bare {
-        fn schema(&self) -> NodeSchema {
-            NodeSchema { inputs: vec![], output: vec![], params: vec![] }
+        fn lookbacks(&self) -> Vec<usize> {
+            vec![]
         }
         fn eval(&mut self, _ctx: Ctx<'_>) -> Option<&[Scalar]> {
             None
@@ -159,9 +170,9 @@ mod tests {
     }
 
     #[test]
-    fn input_spec_carries_firing() {
-        let a = InputSpec { kind: ScalarKind::F64, lookback: 1, firing: Firing::Any };
-        let b = InputSpec { kind: ScalarKind::F64, lookback: 2, firing: Firing::Barrier(0) };
+    fn port_spec_carries_firing() {
+        let a = PortSpec { kind: ScalarKind::F64, firing: Firing::Any };
+        let b = PortSpec { kind: ScalarKind::F64, firing: Firing::Barrier(0) };
         assert_eq!(a.firing, Firing::Any);
         assert_eq!(b.firing, Firing::Barrier(0));
         assert_ne!(a.firing, b.firing);
@@ -174,34 +185,52 @@ mod tests {
     }
 
     #[test]
-    fn leaf_factory_label_is_the_bare_type() {
+    fn primitive_builder_label_is_the_bare_type() {
         // param-generic: the label is just the node type, no knob suffix (the
         // ascii-dag blueprint view cannot render wide cluster-sibling labels).
-        let with = LeafFactory::new(
+        let with = PrimitiveBuilder::new(
             "SMA",
-            vec![ParamSpec { name: "length".into(), kind: ScalarKind::I64 }],
+            NodeSchema {
+                inputs: vec![],
+                output: vec![],
+                params: vec![ParamSpec { name: "length".into(), kind: ScalarKind::I64 }],
+            },
             |_| Box::new(Bare),
         );
         assert_eq!(with.label(), "SMA");
-        let none = LeafFactory::new("Sub", vec![], |_| Box::new(Bare));
+        let none = PrimitiveBuilder::new(
+            "Sub",
+            NodeSchema { inputs: vec![], output: vec![], params: vec![] },
+            |_| Box::new(Bare),
+        );
         assert_eq!(none.label(), "Sub");
     }
 
     #[test]
-    fn leaf_factory_build_runs_the_closure() {
-        let f = LeafFactory::new("Bare", vec![], |_| Box::new(Bare));
-        assert_eq!(f.build(&[]).schema().params, Vec::<ParamSpec>::new());
+    fn primitive_builder_build_runs_the_closure() {
+        let f = PrimitiveBuilder::new(
+            "Bare",
+            NodeSchema { inputs: vec![], output: vec![], params: vec![] },
+            |_| Box::new(Bare),
+        );
+        assert_eq!(f.params(), Vec::<ParamSpec>::new());
+        assert_eq!(f.build(&[]).lookbacks(), Vec::<usize>::new());
     }
 
     #[test]
     fn schema_carries_declared_params() {
-        let s = NodeSchema {
-            inputs: vec![],
-            output: vec![],
-            params: vec![ParamSpec { name: "length".into(), kind: ScalarKind::I64 }],
-        };
-        assert_eq!(s.params.len(), 1);
-        assert_eq!(s.params[0].name, "length");
-        assert_eq!(s.params[0].kind, ScalarKind::I64);
+        let b = PrimitiveBuilder::new(
+            "SMA",
+            NodeSchema {
+                inputs: vec![],
+                output: vec![],
+                params: vec![ParamSpec { name: "length".into(), kind: ScalarKind::I64 }],
+            },
+            |_| Box::new(Bare),
+        );
+        assert_eq!(
+            b.schema().params,
+            vec![ParamSpec { name: "length".into(), kind: ScalarKind::I64 }]
+        );
     }
 }

crates/aura-engine/src/harness.rs

@@ -21,7 +21,7 @@
 //! 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, Ctx, Firing, Node, Scalar, ScalarKind, Timestamp};
+use aura_core::{AnyColumn, 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
@@ -51,6 +51,18 @@ pub struct SourceSpec {
     pub targets: Vec<Target>,
 }
 
+/// The flat, type-erased, index-wired output of `Composite::compile` — the target
+/// `Harness::bootstrap` consumes (C23's "compilat", 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>,
+}
+
 /// A wiring fault caught once, at bootstrap — C7's "type check paid at wiring"
 /// generalized to the whole topology.
 #[derive(Debug, PartialEq, Eq)]
@@ -107,41 +119,41 @@ impl core::fmt::Debug for Harness {
 }
 
 impl Harness {
-    /// Bind nodes + wiring into a frozen, runnable graph. Sizes each node's input
-    /// columns from its `schema`, 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(
-        nodes: Vec<Box<dyn Node>>,
-        sources: Vec<SourceSpec>,
-        edges: Vec<Edge>,
-    ) -> Result<Harness, BootstrapError> {
+    /// 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> {
+        let FlatGraph { nodes, signatures, sources, edges } = flat;
         let n = nodes.len();
 
-        let schemas: Vec<_> = nodes.iter().map(|nd| nd.schema()).collect();
-
-        // size each node's input columns from its schema; lift firing; init slots
+        // 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, schema) in nodes.into_iter().zip(schemas.iter()) {
-            let inputs: Vec<AnyColumn> = schema
+        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()
-                .map(|spec| AnyColumn::with_capacity(spec.kind, spec.lookback))
+                .zip(depths)
+                .map(|(spec, depth)| AnyColumn::with_capacity(spec.kind, depth))
                 .collect();
-            let firing: Vec<Firing> = schema.inputs.iter().map(|spec| spec.firing).collect();
-            let slots: Vec<SlotState> = schema
+            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 = schema.output.len();
+            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 = schemas.get(t.node).ok_or(BootstrapError::BadIndex)?;
+                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 {
@@ -155,8 +167,8 @@ impl Harness {
         // 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 = schemas.get(e.from).ok_or(BootstrapError::BadIndex)?;
-            let to = schemas.get(e.to).ok_or(BootstrapError::BadIndex)?;
+            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 {
@@ -333,10 +345,9 @@ fn group_id(f: &Firing) -> Option<u8> {
 #[cfg(test)]
 mod tests {
     use super::*;
-    // InputSpec / NodeSchema are not imported by harness.rs production code (it
-    // only reads `nd.schema()` fields, never naming the types), so they are not
-    // brought in by `use super::*` — the fixtures construct them, so import here.
-    use aura_core::{FieldSpec, InputSpec, NodeSchema};
+    // 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::{Exposure, Recorder, Sma, SimBroker, Sub};
     use std::sync::mpsc;
 
@@ -345,6 +356,33 @@ mod tests {
         points.iter().map(|&(t, v)| (Timestamp(t), Scalar::F64(v))).collect()
     }
 
+    /// 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 }
+    }
+
+    /// 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 })
+    }
+
+    /// The declared signature of a `Recorder` over `kinds` with the given firing.
+    fn recorder_sig(kinds: &[ScalarKind], firing: Firing) -> NodeSchema {
+        NodeSchema {
+            inputs: kinds.iter().map(|&kind| PortSpec { kind, firing }).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) ---
 
@@ -353,17 +391,19 @@ mod tests {
     struct AsOfSum {
         out: [Scalar; 1],
     }
-    impl Node for AsOfSum {
-        fn schema(&self) -> NodeSchema {
+    impl AsOfSum {
+        fn sig() -> NodeSchema {
             NodeSchema {
-                inputs: vec![
-                    InputSpec { kind: ScalarKind::F64, lookback: 1, firing: Firing::Any },
-                    InputSpec { kind: ScalarKind::F64, lookback: 1, firing: Firing::Any },
-                ],
+                inputs: vec![f64_port(Firing::Any), f64_port(Firing::Any)],
                 output: vec![FieldSpec { name: "value", kind: ScalarKind::F64 }],
                 params: vec![],
             }
         }
+    }
+    impl Node for AsOfSum {
+        fn lookbacks(&self) -> Vec<usize> {
+            vec![1, 1]
+        }
         fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Scalar]> {
             let a = ctx.f64_in(0);
             let b = ctx.f64_in(1);
@@ -380,17 +420,19 @@ mod tests {
     struct BarrierSum {
         out: [Scalar; 1],
     }
-    impl Node for BarrierSum {
-        fn schema(&self) -> NodeSchema {
+    impl BarrierSum {
+        fn sig() -> NodeSchema {
             NodeSchema {
-                inputs: vec![
-                    InputSpec { kind: ScalarKind::F64, lookback: 1, firing: Firing::Barrier(0) },
-                    InputSpec { kind: ScalarKind::F64, lookback: 1, firing: Firing::Barrier(0) },
-                ],
+                inputs: vec![f64_port(Firing::Barrier(0)), f64_port(Firing::Barrier(0))],
                 output: vec![FieldSpec { name: "value", kind: ScalarKind::F64 }],
                 params: vec![],
             }
         }
+    }
+    impl Node for BarrierSum {
+        fn lookbacks(&self) -> Vec<usize> {
+            vec![1, 1]
+        }
         fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Scalar]> {
             self.out[0] = Scalar::F64(ctx.f64_in(0)[0] + ctx.f64_in(1)[0]);
             Some(&self.out)
@@ -403,18 +445,23 @@ mod tests {
     struct MixedSum {
         out: [Scalar; 1],
     }
-    impl Node for MixedSum {
-        fn schema(&self) -> NodeSchema {
+    impl MixedSum {
+        fn sig() -> NodeSchema {
             NodeSchema {
                 inputs: vec![
-                    InputSpec { kind: ScalarKind::F64, lookback: 1, firing: Firing::Barrier(0) },
-                    InputSpec { kind: ScalarKind::F64, lookback: 1, firing: Firing::Barrier(0) },
-                    InputSpec { kind: ScalarKind::F64, lookback: 1, firing: Firing::Any },
+                    f64_port(Firing::Barrier(0)),
+                    f64_port(Firing::Barrier(0)),
+                    f64_port(Firing::Any),
                 ],
                 output: vec![FieldSpec { name: "value", 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<&[Scalar]> {
             let a = ctx.f64_in(0);
             let b = ctx.f64_in(1);
@@ -434,16 +481,10 @@ mod tests {
     struct Ohlcv {
         out: [Scalar; 5],
     }
-    impl Node for Ohlcv {
-        fn schema(&self) -> NodeSchema {
+    impl Ohlcv {
+        fn sig() -> NodeSchema {
             NodeSchema {
-                inputs: vec![
-                    InputSpec { kind: ScalarKind::F64, lookback: 1, firing: Firing::Barrier(0) },
-                    InputSpec { kind: ScalarKind::F64, lookback: 1, firing: Firing::Barrier(0) },
-                    InputSpec { kind: ScalarKind::F64, lookback: 1, firing: Firing::Barrier(0) },
-                    InputSpec { kind: ScalarKind::F64, lookback: 1, firing: Firing::Barrier(0) },
-                    InputSpec { kind: ScalarKind::F64, lookback: 1, firing: Firing::Barrier(0) },
-                ],
+                inputs: vec![f64_port(Firing::Barrier(0)); 5],
                 output: vec![
                     FieldSpec { name: "open", kind: ScalarKind::F64 },
                     FieldSpec { name: "high", kind: ScalarKind::F64 },
@@ -454,6 +495,11 @@ mod tests {
                 params: vec![],
             }
         }
+    }
+    impl Node for Ohlcv {
+        fn lookbacks(&self) -> Vec<usize> {
+            vec![1; 5]
+        }
         fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Scalar]> {
             for i in 0..5 {
                 let w = ctx.f64_in(i);
@@ -473,10 +519,10 @@ mod tests {
     struct TwoField {
         out: [Scalar; 2],
     }
-    impl Node for TwoField {
-        fn schema(&self) -> NodeSchema {
+    impl TwoField {
+        fn sig() -> NodeSchema {
             NodeSchema {
-                inputs: vec![InputSpec { kind: ScalarKind::F64, lookback: 1, firing: Firing::Any }],
+                inputs: vec![f64_port(Firing::Any)],
                 output: vec![
                     FieldSpec { name: "f", kind: ScalarKind::F64 },
                     FieldSpec { name: "i", kind: ScalarKind::I64 },
@@ -484,6 +530,11 @@ mod tests {
                 params: vec![],
             }
         }
+    }
+    impl Node for TwoField {
+        fn lookbacks(&self) -> Vec<usize> {
+            vec![1]
+        }
         fn eval(&mut self, _ctx: Ctx<'_>) -> Option<&[Scalar]> {
             self.out[0] = Scalar::F64(0.0);
             self.out[1] = Scalar::I64(0);
@@ -498,14 +549,19 @@ mod tests {
         out: [Scalar; 1],
         tx: mpsc::Sender<(Timestamp, Vec<Scalar>)>,
     }
-    impl Node for TapForward {
-        fn schema(&self) -> NodeSchema {
+    impl TapForward {
+        fn sig() -> NodeSchema {
             NodeSchema {
-                inputs: vec![InputSpec { kind: ScalarKind::F64, lookback: 1, firing: Firing::Any }],
+                inputs: vec![f64_port(Firing::Any)],
                 output: vec![FieldSpec { name: "value", kind: ScalarKind::F64 }],
                 params: vec![],
             }
         }
+    }
+    impl Node for TapForward {
+        fn lookbacks(&self) -> Vec<usize> {
+            vec![1]
+        }
         fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Scalar]> {
             let w = ctx.f64_in(0);
             if w.is_empty() {
@@ -522,11 +578,15 @@ mod tests {
     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 = Harness::bootstrap(
+        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 { kind: ScalarKind::F64, targets: vec![Target { node: 0, slot: 0 }] }],
             vec![Edge { from: 0, to: 1, slot: 0, from_field: 0 }],
         )
@@ -550,13 +610,19 @@ mod tests {
         // 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| {
-            Harness::bootstrap(
+            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 {
                     kind: ScalarKind::F64,
                     targets: vec![Target { node: 0, slot: 0 }, Target { node: 1, slot: 0 }],
@@ -598,11 +664,15 @@ mod tests {
         // 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| {
-            Harness::bootstrap(
+            boot(
                 vec![
                     Box::new(AsOfSum { out: [Scalar::F64(0.0)] }),
                     Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx)),
                 ],
+            vec![
+                AsOfSum::sig(),
+                recorder_sig(&[ScalarKind::F64], Firing::Any),
+            ],
                 vec![
                     SourceSpec { kind: ScalarKind::F64, targets: vec![Target { node: 0, slot: 0 }] },
                     SourceSpec { kind: ScalarKind::F64, targets: vec![Target { node: 0, slot: 1 }] },
@@ -640,11 +710,15 @@ mod tests {
     fn mode_b_barrier_fires_only_on_timestamp_coincidence() {
         // identical wiring to mode A, but both BarrierSum inputs are Barrier(0).
         let build = |tx| {
-            Harness::bootstrap(
+            boot(
                 vec![
                     Box::new(BarrierSum { out: [Scalar::F64(0.0)] }),
                     Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx)),
                 ],
+            vec![
+                BarrierSum::sig(),
+                recorder_sig(&[ScalarKind::F64], Firing::Any),
+            ],
                 vec![
                     SourceSpec { kind: ScalarKind::F64, targets: vec![Target { node: 0, slot: 0 }] },
                     SourceSpec { kind: ScalarKind::F64, targets: vec![Target { node: 0, slot: 1 }] },
@@ -683,13 +757,19 @@ mod tests {
         // 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| {
-            Harness::bootstrap(
+            boot(
                 vec![
                     Box::new(Sma::new(2)),
                     Box::new(Sma::new(4)),
                     Box::new(BarrierSum { out: [Scalar::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 {
                     kind: ScalarKind::F64,
                     targets: vec![Target { node: 0, slot: 0 }, Target { node: 1, slot: 0 }],
@@ -730,11 +810,15 @@ mod tests {
     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 = Harness::bootstrap(
+        let mut h = boot(
             vec![
                 Box::new(MixedSum { out: [Scalar::F64(0.0)] }),
                 Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx)),
             ],
+            vec![
+                MixedSum::sig(),
+                recorder_sig(&[ScalarKind::F64], Firing::Any),
+            ],
             vec![
                 SourceSpec { kind: ScalarKind::F64, targets: vec![Target { node: 0, slot: 0 }] },
                 SourceSpec { kind: ScalarKind::F64, targets: vec![Target { node: 0, slot: 1 }] },
@@ -763,8 +847,12 @@ mod tests {
     #[test]
     fn bootstrap_rejects_a_cycle() {
         // two SMA(1) nodes wired a -> b -> a
-        let err = Harness::bootstrap(
+        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 }],
         )
@@ -775,8 +863,11 @@ mod tests {
     #[test]
     fn bootstrap_rejects_a_kind_mismatch() {
         // SMA(1) declares an f64 input; an i64 source mismatches
-        let err = Harness::bootstrap(
+        let err = boot(
             vec![Box::new(Sma::new(1))],
+            vec![
+                Sma::builder().schema().clone(),
+            ],
             vec![SourceSpec { kind: ScalarKind::I64, targets: vec![Target { node: 0, slot: 0 }] }],
             vec![],
         )
@@ -792,8 +883,12 @@ mod tests {
         // 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 = Harness::bootstrap(
+        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 { kind: ScalarKind::F64, targets: vec![Target { node: 0, slot: 0 }] }],
             vec![Edge { from: 0, to: 9, slot: 0, from_field: 0 }],
         )
@@ -827,7 +922,7 @@ mod tests {
         // 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 = Harness::bootstrap(
+        let mut h = boot(
             vec![
                 Box::new(Ohlcv { out: [Scalar::F64(0.0); 5] }),
                 Box::new(Recorder::new(
@@ -836,6 +931,10 @@ mod tests {
                     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
@@ -876,12 +975,17 @@ mod tests {
         // 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| {
-            Harness::bootstrap(
+            boot(
                 vec![
                     Box::new(Ohlcv { out: [Scalar::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
@@ -917,12 +1021,17 @@ mod tests {
         // (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 = Harness::bootstrap(
+        let mut h = boot(
             vec![
                 Box::new(Ohlcv { out: [Scalar::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
@@ -947,8 +1056,12 @@ mod tests {
     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 = Harness::bootstrap(
+        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 { kind: ScalarKind::F64, targets: vec![Target { node: 0, slot: 0 }] }],
             vec![Edge { from: 0, to: 1, slot: 0, from_field: 9 }],
         )
@@ -961,8 +1074,12 @@ mod tests {
         // 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 = Harness::bootstrap(
+        let err = boot(
             vec![Box::new(TwoField { out: [Scalar::F64(0.0), Scalar::I64(0)] }), Box::new(Sma::new(1))],
+            vec![
+                TwoField::sig(),
+                Sma::builder().schema().clone(),
+            ],
             vec![SourceSpec { kind: ScalarKind::F64, targets: vec![Target { node: 0, slot: 0 }] }],
             vec![Edge { from: 0, to: 1, slot: 0, from_field: 1 }],
         )
@@ -979,13 +1096,19 @@ mod tests {
         // 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 = Harness::bootstrap(
+        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 {
                 kind: ScalarKind::F64,
                 targets: vec![Target { node: 0, slot: 0 }, Target { node: 1, slot: 0 }],
@@ -1023,7 +1146,7 @@ mod tests {
         // 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 = Harness::bootstrap(
+        let mut h = boot(
             vec![
                 Box::new(Ohlcv { out: [Scalar::F64(0.0); 5] }),
                 Box::new(Recorder::new(
@@ -1032,6 +1155,10 @@ mod tests {
                     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 },
@@ -1071,12 +1198,15 @@ mod tests {
         // 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 = Harness::bootstrap(
+        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 { kind: ScalarKind::I64, targets: vec![Target { node: 0, slot: 0 }] },
                 SourceSpec { kind: ScalarKind::F64, targets: vec![Target { node: 0, slot: 1 }] },
@@ -1111,11 +1241,15 @@ mod tests {
         // 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 = Harness::bootstrap(
+        let mut h = boot(
             vec![
                 Box::new(TapForward { out: [Scalar::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 { kind: ScalarKind::F64, targets: vec![Target { node: 0, slot: 0 }] }],
             vec![Edge { from: 0, to: 1, slot: 0, from_field: 0 }],
         )
@@ -1137,11 +1271,15 @@ mod tests {
         // Two fresh harnesses, two channels, identical input -> bit-identical
         // recorded streams (C1).
         let build = |tx| {
-            Harness::bootstrap(
+            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 { kind: ScalarKind::F64, targets: vec![Target { node: 0, slot: 0 }] }],
                 vec![Edge { from: 0, to: 1, slot: 0, from_field: 0 }],
             )
@@ -1168,12 +1306,15 @@ mod tests {
         // 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 = Harness::bootstrap(
+        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 { kind: ScalarKind::F64, targets: vec![Target { node: 0, slot: 0 }] },
                 SourceSpec { kind: ScalarKind::F64, targets: vec![Target { node: 0, slot: 1 }] },
@@ -1200,12 +1341,15 @@ mod tests {
         // 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 = Harness::bootstrap(
+        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 { kind: ScalarKind::F64, targets: vec![Target { node: 0, slot: 0 }] },
                 SourceSpec { kind: ScalarKind::F64, targets: vec![Target { node: 0, slot: 1 }] },
@@ -1236,11 +1380,15 @@ mod tests {
         // 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 = Harness::bootstrap(
+        let err = boot(
             vec![
                 Box::new(TwoField { out: [Scalar::F64(0.0), Scalar::I64(0)] }),
                 Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx)),
             ],
+            vec![
+                TwoField::sig(),
+                recorder_sig(&[ScalarKind::F64], Firing::Any),
+            ],
             vec![SourceSpec { kind: ScalarKind::F64, targets: vec![Target { node: 0, slot: 0 }] }],
             vec![Edge { from: 0, to: 1, slot: 0, from_field: 1 }],
         )
@@ -1265,13 +1413,19 @@ mod tests {
         // 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| {
-            Harness::bootstrap(
+            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 {
                     kind: ScalarKind::F64,
                     targets: vec![Target { node: 0, slot: 0 }],
@@ -1322,7 +1476,7 @@ mod tests {
         // 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| {
-            Harness::bootstrap(
+            boot(
                 vec![
                     Box::new(Sma::new(2)),                                  // 0: shared producer
                     Box::new(Sma::new(4)),                                  // 1: second producer
@@ -1330,6 +1484,13 @@ mod tests {
                     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 {
                     kind: ScalarKind::F64,
                     targets: vec![Target { node: 0, slot: 0 }, Target { node: 1, slot: 0 }],
@@ -1384,13 +1545,19 @@ mod tests {
         // 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| {
-            Harness::bootstrap(
+            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: [Scalar::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 { kind: ScalarKind::F64, targets: vec![Target { node: 0, slot: 0 }] }, // A
                     SourceSpec { kind: ScalarKind::F64, targets: vec![Target { node: 2, slot: 1 }] }, // B
@@ -1441,13 +1608,19 @@ mod tests {
         // -> SMA(2) -> recorder) propagates values correctly through depth, with
         // each stage's warm-up delaying the tail — closes "deep chains untested".
         let build = |tx| {
-            Harness::bootstrap(
+            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 {
                     kind: ScalarKind::F64,
                     targets: vec![Target { node: 0, slot: 0 }],
@@ -1486,7 +1659,7 @@ mod tests {
         // run, yields each parallel stream correctly — closes "wide layers
         // untested" and exercises multi-sink at width.
         let build = |t2, t3, t4| {
-            Harness::bootstrap(
+            boot(
                 vec![
                     Box::new(Sma::new(2)), // 0
                     Box::new(Sma::new(3)), // 1
@@ -1495,6 +1668,14 @@ mod tests {
                     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 {
                     kind: ScalarKind::F64,
                     targets: vec![
@@ -1559,7 +1740,7 @@ mod tests {
         // 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| {
-            Harness::bootstrap(
+            boot(
                 vec![
                     Box::new(Sma::new(2)),                                  // 0: shared f64 producer
                     Box::new(Sma::new(4)),                                  // 1: second f64 producer
@@ -1568,6 +1749,14 @@ mod tests {
                     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 {
                         kind: ScalarKind::F64,
@@ -1634,7 +1823,7 @@ mod tests {
     #[test]
     fn signal_quality_loop_records_pip_equity() {
         let (tx_eq, rx_eq) = mpsc::channel();
-        let mut h = Harness::bootstrap(
+        let mut h = boot(
             vec![
                 Box::new(Sma::new(2)),                              // 0 fast
                 Box::new(Sma::new(4)),                              // 1 slow
@@ -1643,6 +1832,14 @@ mod tests {
                 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(),
+                Exposure::builder().schema().clone(),
+                SimBroker::builder(0.0001).schema().clone(),
+                recorder_sig(&[ScalarKind::F64], Firing::Any),
+            ],
             vec![SourceSpec {
                 kind: ScalarKind::F64,
                 targets: vec![
@@ -1692,7 +1889,7 @@ mod tests {
     fn signal_quality_loop_is_deterministic() {
         let build = || {
             let (tx, rx) = mpsc::channel();
-            let mut h = Harness::bootstrap(
+            let mut h = boot(
                 vec![
                     Box::new(Sma::new(2)),
                     Box::new(Sma::new(4)),
@@ -1701,6 +1898,14 @@ mod tests {
                     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(),
+                Exposure::builder().schema().clone(),
+                SimBroker::builder(0.0001).schema().clone(),
+                recorder_sig(&[ScalarKind::F64], Firing::Any),
+            ],
                 vec![SourceSpec {
                     kind: ScalarKind::F64,
                     targets: vec![

crates/aura-engine/src/lib.rs

@@ -36,15 +36,14 @@ mod harness;
 mod report;
 
 pub use blueprint::{
-    aliases_on, signature_of, Blueprint, BlueprintNode, CompileError, Composite, OutField,
-    ParamAlias, Role,
+    aliases_on, signature_of, BlueprintNode, CompileError, Composite, OutField, ParamAlias, Role,
 };
-pub use harness::{BootstrapError, Edge, Harness, SourceSpec, Target};
+pub use harness::{BootstrapError, Edge, FlatGraph, Harness, SourceSpec, Target};
 pub use report::{f64_field, summarize, RunManifest, RunMetrics, RunReport};
 // #29: re-export the core scalar vocabulary a Blueprint builder needs
 // (SourceSpec.kind is a ScalarKind; sources/Recorder columns are Scalar /
 // Firing / Timestamp) so a graph builder has one import surface, not two.
-pub use aura_core::{Firing, ParamSpec, Scalar, ScalarKind, Timestamp};
+pub use aura_core::{Firing, NodeSchema, ParamSpec, PortSpec, Scalar, ScalarKind, Timestamp};
 
 #[cfg(test)]
 mod reexport_tests {

crates/aura-engine/src/report.rs

@@ -198,11 +198,21 @@ pub fn f64_field(rows: &[(Timestamp, Vec<Scalar>)], field: usize) -> Vec<(Timest
 #[cfg(test)]
 mod tests {
     use super::*;
-    use crate::{Edge, Harness, SourceSpec, Target};
-    use aura_core::{Firing, ScalarKind};
+    use crate::{Edge, FlatGraph, Harness, SourceSpec, Target};
+    use aura_core::{Firing, NodeSchema, PortSpec, ScalarKind};
     use aura_std::{Exposure, Recorder, SimBroker, Sma, Sub};
     use std::sync::mpsc;
 
+    /// The declared signature of a `Recorder` over one f64 column (the sink shape
+    /// the two-sink harness uses).
+    fn f64_recorder_sig() -> NodeSchema {
+        NodeSchema {
+            inputs: vec![PortSpec { kind: ScalarKind::F64, firing: Firing::Any }],
+            output: vec![],
+            params: vec![],
+        }
+    }
+
     /// Build an f64 source stream from (timestamp, value) points (mirrors the
     /// harness.rs test helper; the e2e test needs its own copy — the harness
     /// test module's is private to that module).
@@ -221,8 +231,8 @@ mod tests {
     ) {
         let (tx_eq, rx_eq) = mpsc::channel();
         let (tx_ex, rx_ex) = mpsc::channel();
-        let h = Harness::bootstrap(
-            vec![
+        let h = Harness::bootstrap(FlatGraph {
+            nodes: vec![
                 Box::new(Sma::new(2)),                                       // 0
                 Box::new(Sma::new(4)),                                       // 1
                 Box::new(Sub::new()),                                        // 2
@@ -231,7 +241,16 @@ mod tests {
                 Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx_eq)), // 5 equity sink
                 Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx_ex)), // 6 exposure sink
             ],
-            vec![SourceSpec {
+            signatures: vec![
+                Sma::builder().schema().clone(),
+                Sma::builder().schema().clone(),
+                Sub::builder().schema().clone(),
+                Exposure::builder().schema().clone(),
+                SimBroker::builder(0.0001).schema().clone(),
+                f64_recorder_sig(),
+                f64_recorder_sig(),
+            ],
+            sources: vec![SourceSpec {
                 kind: ScalarKind::F64,
                 targets: vec![
                     Target { node: 0, slot: 0 },
@@ -239,7 +258,7 @@ mod tests {
                     Target { node: 4, slot: 1 }, // price into the broker
                 ],
             }],
-            vec![
+            edges: vec![
                 Edge { from: 0, to: 2, slot: 0, from_field: 0 },
                 Edge { from: 1, to: 2, slot: 1, from_field: 0 },
                 Edge { from: 2, to: 3, slot: 0, from_field: 0 },
@@ -247,7 +266,7 @@ mod tests {
                 Edge { from: 4, to: 5, slot: 0, from_field: 0 }, // equity -> sink 5
                 Edge { from: 3, to: 6, slot: 0, from_field: 0 }, // exposure -> sink 6
             ],
-        )
+        })
         .expect("valid signal-quality DAG");
         (h, rx_eq, rx_ex)
     }

crates/aura-ingest/tests/real_bars.rs

@@ -7,9 +7,9 @@
 use std::sync::mpsc;
 use std::sync::Arc;
 
-use aura_core::{Firing, Scalar, ScalarKind, Timestamp};
+use aura_core::{Firing, NodeSchema, PortSpec, Scalar, ScalarKind, Timestamp};
 use aura_engine::{
-    f64_field, summarize, Edge, Harness, RunManifest, RunReport, SourceSpec, Target,
+    f64_field, summarize, Edge, FlatGraph, Harness, RunManifest, RunReport, SourceSpec, Target,
 };
 use aura_ingest::load_m1_window;
 use aura_std::{Exposure, Recorder, SimBroker, Sma, Sub};
@@ -22,8 +22,13 @@ use data_server::{DataServer, DEFAULT_DATA_PATH};
 fn run_sample_over(prices: Vec<(Timestamp, Scalar)>) -> RunReport {
     let (tx_eq, rx_eq) = mpsc::channel();
     let (tx_ex, rx_ex) = mpsc::channel();
-    let mut h = Harness::bootstrap(
-        vec![
+    let f64_recorder_sig = || NodeSchema {
+        inputs: vec![PortSpec { kind: ScalarKind::F64, firing: Firing::Any }],
+        output: vec![],
+        params: vec![],
+    };
+    let mut h = Harness::bootstrap(FlatGraph {
+        nodes: vec![
             Box::new(Sma::new(2)),                                            // 0
             Box::new(Sma::new(4)),                                            // 1
             Box::new(Sub::new()),                                             // 2
@@ -32,7 +37,16 @@ fn run_sample_over(prices: Vec<(Timestamp, Scalar)>) -> RunReport {
             Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx_eq)),  // 5 equity sink
             Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx_ex)),  // 6 exposure sink
         ],
-        vec![SourceSpec {
+        signatures: vec![
+            Sma::builder().schema().clone(),
+            Sma::builder().schema().clone(),
+            Sub::builder().schema().clone(),
+            Exposure::builder().schema().clone(),
+            SimBroker::builder(0.0001).schema().clone(),
+            f64_recorder_sig(),
+            f64_recorder_sig(),
+        ],
+        sources: vec![SourceSpec {
             kind: ScalarKind::F64,
             targets: vec![
                 Target { node: 0, slot: 0 },
@@ -40,7 +54,7 @@ fn run_sample_over(prices: Vec<(Timestamp, Scalar)>) -> RunReport {
                 Target { node: 4, slot: 1 }, // price into the broker
             ],
         }],
-        vec![
+        edges: vec![
             Edge { from: 0, to: 2, slot: 0, from_field: 0 },
             Edge { from: 1, to: 2, slot: 1, from_field: 0 },
             Edge { from: 2, to: 3, slot: 0, from_field: 0 },
@@ -48,7 +62,7 @@ fn run_sample_over(prices: Vec<(Timestamp, Scalar)>) -> RunReport {
             Edge { from: 4, to: 5, slot: 0, from_field: 0 }, // equity -> sink 5
             Edge { from: 3, to: 6, slot: 0, from_field: 0 }, // exposure -> sink 6
         ],
-    )
+    })
     .expect("valid signal-quality DAG");
 
     let window = (

crates/aura-std/src/add.rs

@@ -2,7 +2,7 @@
 //! Combines two signal streams into one — the most basic combinator for the
 //! north-star "combine one signal with another" research move (C10).
 
-use aura_core::{Ctx, FieldSpec, Firing, InputSpec, LeafFactory, Node, NodeSchema, Scalar, ScalarKind};
+use aura_core::{Ctx, FieldSpec, Firing, Node, NodeSchema, PortSpec, PrimitiveBuilder, Scalar, ScalarKind};
 
 /// Two-input f64 sum: input 0 plus input 1. Emits `None` until both inputs
 /// have a value.
@@ -26,10 +26,21 @@ impl Add {
         Self { out: [Scalar::F64(0.0)] }
     }
 
-    /// The param-generic recipe for a blueprint leaf: paramless, builds through
+    /// The param-generic recipe for a blueprint primitive: paramless, builds through
     /// `Add::new`.
-    pub fn factory() -> LeafFactory {
-        LeafFactory::new("Add", vec![], |_| Box::new(Add::new()))
+    pub fn builder() -> PrimitiveBuilder {
+        PrimitiveBuilder::new(
+            "Add",
+            NodeSchema {
+                inputs: vec![
+                    PortSpec { kind: ScalarKind::F64, firing: Firing::Any },
+                    PortSpec { kind: ScalarKind::F64, firing: Firing::Any },
+                ],
+                output: vec![FieldSpec { name: "value", kind: ScalarKind::F64 }],
+                params: vec![],
+            },
+            |_| Box::new(Add::new()),
+        )
     }
 }
 
@@ -40,15 +51,8 @@ impl Default for Add {
 }
 
 impl Node for Add {
-    fn schema(&self) -> NodeSchema {
-        NodeSchema {
-            inputs: vec![
-                InputSpec { kind: ScalarKind::F64, lookback: 1, firing: Firing::Any },
-                InputSpec { kind: ScalarKind::F64, lookback: 1, firing: Firing::Any },
-            ],
-            output: vec![FieldSpec { name: "value", kind: ScalarKind::F64 }],
-            params: vec![],
-        }
+    fn lookbacks(&self) -> Vec<usize> {
+        vec![1, 1]
     }
 
     fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Scalar]> {
@@ -71,13 +75,6 @@ mod tests {
     use super::*;
     use aura_core::{AnyColumn, Timestamp};
 
-    #[test]
-    fn factory_params_match_built_node_schema() {
-        let f = Add::factory();
-        let built = f.build(&[]);
-        assert_eq!(f.params(), built.schema().params.as_slice());
-    }
-
     #[test]
     fn add_is_sum_once_both_inputs_present() {
         let mut add = Add::new();

crates/aura-std/src/ema.rs

@@ -18,7 +18,8 @@
 //!   nothing on the hot path (the output buffer is reused).
 
 use aura_core::{
-    Ctx, FieldSpec, Firing, InputSpec, LeafFactory, Node, NodeSchema, ParamSpec, Scalar, ScalarKind,
+    Ctx, FieldSpec, Firing, Node, NodeSchema, ParamSpec, PortSpec, PrimitiveBuilder, Scalar,
+    ScalarKind,
 };
 
 /// Exponential moving average of one f64 input, smoothing `alpha = 2/(length+1)`,
@@ -52,31 +53,27 @@ impl Ema {
         }
     }
 
-    /// The param-generic recipe for a blueprint leaf: declares `length` and builds
+    /// The param-generic recipe for a blueprint primitive: declares `length` and builds
     /// through `Ema::new` (the single sizing/validation gate; the slice is
     /// kind-checked before `build` runs, so the typed read is total).
-    pub fn factory() -> LeafFactory {
-        LeafFactory::new(
+    pub fn builder() -> PrimitiveBuilder {
+        PrimitiveBuilder::new(
             "EMA",
-            vec![ParamSpec { name: "length".into(), kind: ScalarKind::I64 }],
+            NodeSchema {
+                inputs: vec![PortSpec { kind: ScalarKind::F64, firing: Firing::Any }],
+                output: vec![FieldSpec { name: "value", kind: ScalarKind::F64 }],
+                params: vec![ParamSpec { name: "length".into(), kind: ScalarKind::I64 }],
+            },
             |p| Box::new(Ema::new(p[0].as_i64().expect("length slot is I64") as usize)),
         )
     }
 }
 
 impl Node for Ema {
-    fn schema(&self) -> NodeSchema {
-        NodeSchema {
-            inputs: vec![InputSpec {
-                kind: ScalarKind::F64,
-                // recursive: the running average lives in internal state, so only
-                // the newest sample is read — `length` sizes alpha, not the window.
-                lookback: 1,
-                firing: Firing::Any,
-            }],
-            output: vec![FieldSpec { name: "value", kind: ScalarKind::F64 }],
-            params: vec![ParamSpec { name: "length".into(), kind: ScalarKind::I64 }],
-        }
+    // recursive: the running average lives in internal state, so only the newest
+    // sample is read — `length` sizes alpha, not the window.
+    fn lookbacks(&self) -> Vec<usize> {
+        vec![1]
     }
 
     fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Scalar]> {
@@ -157,12 +154,11 @@ mod tests {
     }
 
     #[test]
-    fn factory_params_match_built_node_schema() {
-        let f = Ema::factory();
-        let built = f.build(&[Scalar::I64(9)]);
-        assert_eq!(f.params(), built.schema().params.as_slice());
-        // the built node carries the declared I64 `length` knob
-        assert_eq!(built.schema().params[0].name, "length");
-        assert_eq!(built.schema().params[0].kind, ScalarKind::I64);
+    fn builder_declares_the_length_knob() {
+        // the param-generic recipe carries the declared I64 `length` knob
+        let b = Ema::builder();
+        assert_eq!(b.params(), b.schema().params.as_slice());
+        assert_eq!(b.schema().params[0].name, "length");
+        assert_eq!(b.schema().params[0].kind, ScalarKind::I64);
     }
 }

crates/aura-std/src/exposure.rs

@@ -4,7 +4,8 @@
 //! `scale` sets which signal magnitude maps to full exposure (sizing lives here).
 
 use aura_core::{
-    Ctx, FieldSpec, Firing, InputSpec, LeafFactory, Node, NodeSchema, ParamSpec, Scalar, ScalarKind,
+    Ctx, FieldSpec, Firing, Node, NodeSchema, ParamSpec, PortSpec, PrimitiveBuilder, Scalar,
+    ScalarKind,
 };
 
 /// Bounded exposure from a raw signal score: `clamp(signal / scale, -1.0, +1.0)`.
@@ -21,25 +22,25 @@ impl Exposure {
         Self { scale, out: [Scalar::F64(0.0)] }
     }
 
-    /// The param-generic recipe for a blueprint leaf: declares `scale` and builds
+    /// The param-generic recipe for a blueprint primitive: declares `scale` and builds
     /// through `Exposure::new` (the single sizing/validation gate; the slice is
     /// kind-checked before `build` runs, so the typed read is total).
-    pub fn factory() -> LeafFactory {
-        LeafFactory::new(
+    pub fn builder() -> PrimitiveBuilder {
+        PrimitiveBuilder::new(
             "Exposure",
-            vec![ParamSpec { name: "scale".into(), kind: ScalarKind::F64 }],
+            NodeSchema {
+                inputs: vec![PortSpec { kind: ScalarKind::F64, firing: Firing::Any }],
+                output: vec![FieldSpec { name: "exposure", kind: ScalarKind::F64 }],
+                params: vec![ParamSpec { name: "scale".into(), kind: ScalarKind::F64 }],
+            },
             |p| Box::new(Exposure::new(p[0].as_f64().expect("scale slot is F64"))),
         )
     }
 }
 
 impl Node for Exposure {
-    fn schema(&self) -> NodeSchema {
-        NodeSchema {
-            inputs: vec![InputSpec { kind: ScalarKind::F64, lookback: 1, firing: Firing::Any }],
-            output: vec![FieldSpec { name: "exposure", kind: ScalarKind::F64 }],
-            params: vec![ParamSpec { name: "scale".into(), kind: ScalarKind::F64 }],
-        }
+    fn lookbacks(&self) -> Vec<usize> {
+        vec![1]
     }
 
     fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Scalar]> {
@@ -82,13 +83,6 @@ mod tests {
         }
     }
 
-    #[test]
-    fn factory_params_match_built_node_schema() {
-        let f = Exposure::factory();
-        let built = f.build(&[Scalar::F64(0.5)]);
-        assert_eq!(f.params(), built.schema().params.as_slice());
-    }
-
     #[test]
     fn exposure_is_none_until_input_present() {
         let mut e = Exposure::new(0.5);

crates/aura-std/src/lincomb.rs

@@ -7,7 +7,8 @@
 //! indexed `F64` knobs that `Blueprint::param_space` aggregates (C8/C12/C19).
 
 use aura_core::{
-    Ctx, FieldSpec, Firing, InputSpec, LeafFactory, Node, NodeSchema, ParamSpec, Scalar, ScalarKind,
+    Ctx, FieldSpec, Firing, Node, NodeSchema, ParamSpec, PortSpec, PrimitiveBuilder, Scalar,
+    ScalarKind,
 };
 
 /// Weighted sum of `N` f64 inputs: `Σ weights[i] · input[i]`. The `weights` are
@@ -39,16 +40,19 @@ impl LinComb {
         Self { weights, out: [Scalar::F64(0.0)] }
     }
 
-    /// The param-generic recipe for a blueprint leaf. The `arity` is topology
-    /// (fixed per blueprint, C19), taken as a factory arg; only the weight *values*
+    /// The param-generic recipe for a blueprint primitive. The `arity` is topology
+    /// (fixed per blueprint, C19), taken as a builder arg; only the weight *values*
     /// are injected, slot by slot, through `LinComb::new` (the single sizing gate).
-    pub fn factory(arity: usize) -> LeafFactory {
+    pub fn builder(arity: usize) -> PrimitiveBuilder {
+        let inputs = (0..arity)
+            .map(|_| PortSpec { kind: ScalarKind::F64, firing: Firing::Any })
+            .collect();
         let params = (0..arity)
             .map(|i| ParamSpec { name: format!("weights[{i}]"), kind: ScalarKind::F64 })
             .collect();
-        LeafFactory::new(
+        PrimitiveBuilder::new(
             "LinComb",
-            params,
+            NodeSchema { inputs, output: vec![FieldSpec { name: "value", kind: ScalarKind::F64 }], params },
             |p| Box::new(LinComb::new(
                 p.iter().map(|s| s.as_f64().expect("weight slot is F64")).collect(),
             )),
@@ -57,18 +61,8 @@ impl LinComb {
 }
 
 impl Node for LinComb {
-    fn schema(&self) -> NodeSchema {
-        NodeSchema {
-            inputs: self
-                .weights
-                .iter()
-                .map(|_| InputSpec { kind: ScalarKind::F64, lookback: 1, firing: Firing::Any })
-                .collect(),
-            output: vec![FieldSpec { name: "value", kind: ScalarKind::F64 }],
-            params: (0..self.weights.len())
-                .map(|i| ParamSpec { name: format!("weights[{i}]"), kind: ScalarKind::F64 })
-                .collect(),
-        }
+    fn lookbacks(&self) -> Vec<usize> {
+        vec![1; self.weights.len()]
     }
 
     fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Scalar]> {
@@ -142,13 +136,6 @@ mod tests {
         assert_eq!(lc.eval(Ctx::new(&inputs, Timestamp(0))), Some([Scalar::F64(6.0)].as_slice()));
     }
 
-    #[test]
-    fn factory_params_match_built_node_schema() {
-        let f = LinComb::factory(2);
-        let built = f.build(&[Scalar::F64(1.0), Scalar::F64(-1.0)]);
-        assert_eq!(f.params(), built.schema().params.as_slice());
-    }
-
     #[test]
     #[should_panic(expected = "LinComb needs at least one weight")]
     fn lincomb_empty_weights_panics() {

crates/aura-std/src/recorder.rs

@@ -7,7 +7,7 @@
 //! four base scalar kinds so any column can be persisted; returns `None` (filters)
 //! until every input column is warm.
 
-use aura_core::{Ctx, Firing, InputSpec, LeafFactory, Node, NodeSchema, Scalar, ScalarKind, Timestamp};
+use aura_core::{Ctx, Firing, Node, NodeSchema, PortSpec, PrimitiveBuilder, Scalar, ScalarKind, Timestamp};
 use std::sync::mpsc::Sender;
 
 /// A recording sink over `kinds.len()` input columns. Each fired cycle it reads
@@ -16,42 +16,41 @@ use std::sync::mpsc::Sender;
 /// value.
 pub struct Recorder {
     kinds: Vec<ScalarKind>,
-    firing: Firing,
     tx: Sender<(Timestamp, Vec<Scalar>)>,
 }
 
 impl Recorder {
     /// A recorder over one input column per entry in `kinds`, each with the given
-    /// `firing` policy, sending recorded `(timestamp, row)` pairs to `tx`.
-    pub fn new(kinds: &[ScalarKind], firing: Firing, tx: Sender<(Timestamp, Vec<Scalar>)>) -> Self {
-        Self { kinds: kinds.to_vec(), firing, tx }
+    /// `firing` policy, sending recorded `(timestamp, row)` pairs to `tx`. The
+    /// `firing` policy is a property of the declared signature (carried by the
+    /// `PrimitiveBuilder`'s `PortSpec`s), not of the built sink, so it is part of
+    /// the construction contract but not stored on the instance.
+    pub fn new(kinds: &[ScalarKind], _firing: Firing, tx: Sender<(Timestamp, Vec<Scalar>)>) -> Self {
+        Self { kinds: kinds.to_vec(), tx }
     }
 
-    /// The param-generic recipe for a blueprint leaf. The channel + kinds + firing
+    /// The param-generic recipe for a blueprint primitive. The channel + kinds + firing
     /// are non-param construction args (captured), not tunable params; the node
-    /// declares none. `tx` is cloned per build (`mpsc::Sender: Clone`).
-    pub fn factory(
+    /// declares none. The input ports (one per kind) are threaded into the schema
+    /// statically; `tx`/`kinds` are cloned for the build closure.
+    pub fn builder(
         kinds: Vec<ScalarKind>,
         firing: Firing,
         tx: Sender<(Timestamp, Vec<Scalar>)>,
-    ) -> LeafFactory {
-        LeafFactory::new("Recorder", vec![], move |_| {
-            Box::new(Recorder::new(&kinds, firing, tx.clone()))
-        })
+    ) -> PrimitiveBuilder {
+        let inputs = kinds.iter().map(|&kind| PortSpec { kind, firing }).collect();
+        let build_kinds = kinds.clone();
+        PrimitiveBuilder::new(
+            "Recorder",
+            NodeSchema { inputs, output: vec![], params: vec![] }, // sink: empty output (C8)
+            move |_| Box::new(Recorder::new(&build_kinds, firing, tx.clone())),
+        )
     }
 }
 
 impl Node for Recorder {
-    fn schema(&self) -> NodeSchema {
-        NodeSchema {
-            inputs: self
-                .kinds
-                .iter()
-                .map(|&kind| InputSpec { kind, lookback: 1, firing: self.firing })
-                .collect(),
-            output: vec![],
-            params: vec![],
-        }
+    fn lookbacks(&self) -> Vec<usize> {
+        vec![1; self.kinds.len()]
     }
 
     fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Scalar]> {
@@ -81,26 +80,22 @@ mod tests {
     use aura_core::{AnyColumn, Timestamp};
     use std::sync::mpsc;
 
-    #[test]
-    fn factory_params_match_built_node_schema() {
-        let (tx, _rx) = mpsc::channel();
-        let f = Recorder::factory(vec![ScalarKind::F64], Firing::Any, tx);
-        let built = f.build(&[]);
-        assert_eq!(f.params(), built.schema().params.as_slice());
-    }
-
     #[test]
     fn recorder_captures_f64_stream_after_warmup() {
         let (tx, rx) = mpsc::channel();
         let mut rec = Recorder::new(&[ScalarKind::F64], Firing::Any, tx);
 
-        // size the one f64 input column from the schema, as the engine would.
-        let schema = rec.schema();
-        assert!(schema.output.is_empty(), "a sink declares no output (C8)");
-        let mut inputs = vec![AnyColumn::with_capacity(
-            schema.inputs[0].kind,
-            schema.inputs[0].lookback,
-        )];
+        // a sink declares no output (C8) — asserted on the param-generic builder
+        let (tx_b, _rx_b) = mpsc::channel();
+        assert!(
+            Recorder::builder(vec![ScalarKind::F64], Firing::Any, tx_b)
+                .schema()
+                .output
+                .is_empty(),
+            "a sink declares no output (C8)"
+        );
+        // size the one f64 input column from the node's lookback, as bootstrap would.
+        let mut inputs = vec![AnyColumn::with_capacity(ScalarKind::F64, rec.lookbacks()[0])];
 
         // cold: returns None and records nothing.
         assert_eq!(rec.eval(Ctx::new(&inputs, Timestamp(1))), None);

crates/aura-std/src/sim_broker.rs

@@ -4,7 +4,7 @@
 //! each cycle (decided at t-1) into a cumulative synthetic pip-equity output.
 //! Measures signal quality, not execution-modelled P&L.
 
-use aura_core::{Ctx, FieldSpec, Firing, InputSpec, LeafFactory, Node, NodeSchema, Scalar, ScalarKind};
+use aura_core::{Ctx, FieldSpec, Firing, Node, NodeSchema, PortSpec, PrimitiveBuilder, Scalar, ScalarKind};
 
 /// Integrates `exposure * price-return` into cumulative pips. `pip_size` is
 /// per-instrument reference metadata (beside the hot path, C7/C15), held here,
@@ -55,24 +55,28 @@ impl SimBroker {
         }
     }
 
-    /// The param-generic recipe for a blueprint leaf. `pip_size` is per-instrument
+    /// The param-generic recipe for a blueprint primitive. `pip_size` is per-instrument
     /// metadata (C10/C15), not a tunable param — it is captured by the closure, not
     /// injected; the node declares no params.
-    pub fn factory(pip_size: f64) -> LeafFactory {
-        LeafFactory::new("SimBroker", vec![], move |_| Box::new(SimBroker::new(pip_size)))
+    pub fn builder(pip_size: f64) -> PrimitiveBuilder {
+        PrimitiveBuilder::new(
+            "SimBroker",
+            NodeSchema {
+                inputs: vec![
+                    PortSpec { kind: ScalarKind::F64, firing: Firing::Any }, // 0 exposure
+                    PortSpec { kind: ScalarKind::F64, firing: Firing::Any }, // 1 price
+                ],
+                output: vec![FieldSpec { name: "equity", kind: ScalarKind::F64 }],
+                params: vec![],
+            },
+            move |_| Box::new(SimBroker::new(pip_size)),
+        )
     }
 }
 
 impl Node for SimBroker {
-    fn schema(&self) -> NodeSchema {
-        NodeSchema {
-            inputs: vec![
-                InputSpec { kind: ScalarKind::F64, lookback: 1, firing: Firing::Any }, // 0 exposure
-                InputSpec { kind: ScalarKind::F64, lookback: 1, firing: Firing::Any }, // 1 price
-            ],
-            output: vec![FieldSpec { name: "equity", kind: ScalarKind::F64 }],
-            params: vec![],
-        }
+    fn lookbacks(&self) -> Vec<usize> {
+        vec![1, 1]
     }
 
     fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Scalar]> {
@@ -122,13 +126,6 @@ mod tests {
         }
     }
 
-    #[test]
-    fn factory_params_match_built_node_schema() {
-        let f = SimBroker::factory(0.0001);
-        let built = f.build(&[]);
-        assert_eq!(f.params(), built.schema().params.as_slice());
-    }
-
     #[test]
     fn sim_broker_integrates_lagged_exposure_times_return() {
         let mut b = SimBroker::new(1.0);

crates/aura-std/src/sma.rs

@@ -4,7 +4,8 @@
 //! engine present (the test drives it by hand, as the sim loop later will).
 
 use aura_core::{
-    Ctx, FieldSpec, Firing, InputSpec, LeafFactory, Node, NodeSchema, ParamSpec, Scalar, ScalarKind,
+    Ctx, FieldSpec, Firing, Node, NodeSchema, ParamSpec, PortSpec, PrimitiveBuilder, Scalar,
+    ScalarKind,
 };
 
 /// Simple moving average over the last `length` values of one f64 input.
@@ -20,29 +21,25 @@ impl Sma {
         Self { length, out: [Scalar::F64(0.0)] }
     }
 
-    /// The param-generic recipe for a blueprint leaf: declares `length` and builds
+    /// The param-generic recipe for a blueprint primitive: declares `length` and builds
     /// through `Sma::new` (the single sizing/validation gate; the slice is
     /// kind-checked before `build` runs, so the typed read is total).
-    pub fn factory() -> LeafFactory {
-        LeafFactory::new(
+    pub fn builder() -> PrimitiveBuilder {
+        PrimitiveBuilder::new(
             "SMA",
-            vec![ParamSpec { name: "length".into(), kind: ScalarKind::I64 }],
+            NodeSchema {
+                inputs: vec![PortSpec { kind: ScalarKind::F64, firing: Firing::Any }],
+                output: vec![FieldSpec { name: "value", kind: ScalarKind::F64 }],
+                params: vec![ParamSpec { name: "length".into(), kind: ScalarKind::I64 }],
+            },
             |p| Box::new(Sma::new(p[0].as_i64().expect("length slot is I64") as usize)),
         )
     }
 }
 
 impl Node for Sma {
-    fn schema(&self) -> NodeSchema {
-        NodeSchema {
-            inputs: vec![InputSpec {
-                kind: ScalarKind::F64,
-                lookback: self.length,
-                firing: Firing::Any,
-            }],
-            output: vec![FieldSpec { name: "value", kind: ScalarKind::F64 }],
-            params: vec![ParamSpec { name: "length".into(), kind: ScalarKind::I64 }],
-        }
+    fn lookbacks(&self) -> Vec<usize> {
+        vec![self.length]
     }
 
     fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Scalar]> {
@@ -70,14 +67,12 @@ mod tests {
 
     #[test]
     fn sma_warms_up_then_tracks_the_window_mean() {
-        let mut sma = Sma::new(3);
-        let schema = sma.schema();
+        let sma_for_depth = Sma::new(3);
 
-        // size the input column from the schema, as the engine will at wiring
-        let mut inputs = vec![AnyColumn::with_capacity(
-            schema.inputs[0].kind,
-            schema.inputs[0].lookback,
-        )];
+        // size the input column from the node's lookback, as bootstrap will at wiring
+        let mut inputs =
+            vec![AnyColumn::with_capacity(ScalarKind::F64, sma_for_depth.lookbacks()[0])];
+        let mut sma = sma_for_depth;
 
         let feed = [1.0_f64, 2.0, 3.0, 4.0, 5.0];
         // means of [1,2,3], [2,3,4], [3,4,5] once warmed up
@@ -124,37 +119,35 @@ mod tests {
         assert_eq!(Recorder::new(&[ScalarKind::F64], Firing::Any, tx).label(), "Recorder");
     }
 
-    #[test]
-    fn factory_params_match_built_node_schema() {
-        let f = Sma::factory();
-        let built = f.build(&[Scalar::I64(3)]);
-        assert_eq!(f.params(), built.schema().params.as_slice());
-    }
-
     #[test]
     fn nodes_declare_expected_params() {
         use crate::{Add, Exposure, LinComb, Recorder, SimBroker, Sub};
         use aura_core::{Firing, ParamSpec, ScalarKind};
-        // single scalar knobs
+        // single scalar knobs (declared on the param-generic builder, pre-build)
         assert_eq!(
-            Sma::new(3).schema().params,
+            Sma::builder().schema().params,
             vec![ParamSpec { name: "length".into(), kind: ScalarKind::I64 }],
         );
         assert_eq!(
-            Exposure::new(0.5).schema().params,
+            Exposure::builder().schema().params,
             vec![ParamSpec { name: "scale".into(), kind: ScalarKind::F64 }],
         );
         // vector knob expands flat to N indexed F64 entries
-        let lc = LinComb::new(vec![1.0, -1.0]).schema().params;
+        let lc = LinComb::builder(2).schema().params.clone();
         assert_eq!(lc.len(), 2);
         assert_eq!(lc[0].name, "weights[0]");
         assert_eq!(lc[1].name, "weights[1]");
         assert!(lc.iter().all(|p| p.kind == ScalarKind::F64));
         // param-less nodes declare empty
-        assert!(Sub::new().schema().params.is_empty());
-        assert!(Add::new().schema().params.is_empty());
-        assert!(SimBroker::new(0.0001).schema().params.is_empty());
+        assert!(Sub::builder().schema().params.is_empty());
+        assert!(Add::builder().schema().params.is_empty());
+        assert!(SimBroker::builder(0.0001).schema().params.is_empty());
         let (tx, _rx) = std::sync::mpsc::channel();
-        assert!(Recorder::new(&[ScalarKind::F64], Firing::Any, tx).schema().params.is_empty());
+        assert!(
+            Recorder::builder(vec![ScalarKind::F64], Firing::Any, tx)
+                .schema()
+                .params
+                .is_empty()
+        );
     }
 }

crates/aura-std/src/sub.rs

@@ -3,7 +3,7 @@
 //! real fan-out + join to run (two SMAs joining into one node), exercising
 //! multi-input `Ctx` access inside a running graph.
 
-use aura_core::{Ctx, FieldSpec, Firing, InputSpec, LeafFactory, Node, NodeSchema, Scalar, ScalarKind};
+use aura_core::{Ctx, FieldSpec, Firing, Node, NodeSchema, PortSpec, PrimitiveBuilder, Scalar, ScalarKind};
 
 /// Two-input f64 difference: input 0 minus input 1. Emits `None` until both
 /// inputs have a value.
@@ -17,10 +17,21 @@ impl Sub {
         Self { out: [Scalar::F64(0.0)] }
     }
 
-    /// The param-generic recipe for a blueprint leaf: paramless, builds through
+    /// The param-generic recipe for a blueprint primitive: paramless, builds through
     /// `Sub::new`.
-    pub fn factory() -> LeafFactory {
-        LeafFactory::new("Sub", vec![], |_| Box::new(Sub::new()))
+    pub fn builder() -> PrimitiveBuilder {
+        PrimitiveBuilder::new(
+            "Sub",
+            NodeSchema {
+                inputs: vec![
+                    PortSpec { kind: ScalarKind::F64, firing: Firing::Any },
+                    PortSpec { kind: ScalarKind::F64, firing: Firing::Any },
+                ],
+                output: vec![FieldSpec { name: "value", kind: ScalarKind::F64 }],
+                params: vec![],
+            },
+            |_| Box::new(Sub::new()),
+        )
     }
 }
 
@@ -31,15 +42,8 @@ impl Default for Sub {
 }
 
 impl Node for Sub {
-    fn schema(&self) -> NodeSchema {
-        NodeSchema {
-            inputs: vec![
-                InputSpec { kind: ScalarKind::F64, lookback: 1, firing: Firing::Any },
-                InputSpec { kind: ScalarKind::F64, lookback: 1, firing: Firing::Any },
-            ],
-            output: vec![FieldSpec { name: "value", kind: ScalarKind::F64 }],
-            params: vec![],
-        }
+    fn lookbacks(&self) -> Vec<usize> {
+        vec![1, 1]
     }
 
     fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Scalar]> {
@@ -62,13 +66,6 @@ mod tests {
     use super::*;
     use aura_core::{AnyColumn, Timestamp};
 
-    #[test]
-    fn factory_params_match_built_node_schema() {
-        let f = Sub::factory();
-        let built = f.build(&[]);
-        assert_eq!(f.params(), built.schema().params.as_slice());
-    }
-
     #[test]
     fn sub_is_difference_once_both_inputs_present() {
         let mut sub = Sub::new();