feat(engine): FlatTap type + FlatGraph.taps field (#282)

Second slice: the resolved-tap carrier. FlatTap { name, node, field } is the
output-side twin of SourceSpec — its name survives compile, load-bearing for
by-name binding (#275). FlatGraph gains a taps field, threaded (empty) through
every construction site the compiler enumerates. Bootstrap ignores it (taps
materialize as real recorder nodes/edges before bootstrap in a later slice).
Purely additive: the inert-producer pin and full suite stay green.

refs #282

crates/aura-engine/src/blueprint.rs

@@ -373,7 +373,13 @@ impl Composite {
             flat_sources.push(SourceSpec::named(role.name.clone(), kind, targets));
         }
 
-        Ok(FlatGraph { nodes: flat_nodes, signatures: flat_signatures, sources: flat_sources, edges: flat_edges })
+        Ok(FlatGraph {
+            nodes: flat_nodes,
+            signatures: flat_signatures,
+            sources: flat_sources,
+            edges: flat_edges,
+            taps: Vec::new(),
+        })
     }
 
     /// Compile from a self-describing param vector — the authoring-edge frontend
@@ -2806,6 +2812,7 @@ mod tests {
                 Edge { from: 4, to: 5, slot: 0, from_field: 0 },
                 Edge { from: 3, to: 6, slot: 0, from_field: 0 },
             ],
+            taps: Vec::new(),
         })
         .expect("valid hand-wired DAG");
         (h, rx_eq, rx_ex)

crates/aura-engine/src/harness.rs

@@ -286,6 +286,21 @@ impl SyntheticSpec {
     }
 }
 
+/// A resolved measurement tap in the flat graph: a name paired with a flat
+/// producer `(node, output-field)`. The output-side twin of `SourceSpec` — its
+/// `name` is meant to survive compile and be load-bearing for a future
+/// by-name binding step (like `SourceSpec.role`, #275), planned as `bind_tap`
+/// in a later slice of #282. This slice only carries the type; `compile`
+/// resolves no `Tap`s yet and always emits an empty `taps` list — the
+/// lowering remap that would populate it from a blueprint `Tap` lands with
+/// `bind_tap`.
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct FlatTap {
+    pub name: String,
+    pub node: usize,
+    pub field: usize,
+}
+
 /// The flat, type-erased, index-wired output of `Composite::compile` — the target
 /// `Harness::bootstrap` consumes (C23's "flat graph", now a named type). `signatures[i]`
 /// is the static signature of `nodes[i]` (gathered from each primitive's builder at
@@ -296,6 +311,7 @@ pub struct FlatGraph {
     pub signatures: Vec<NodeSchema>,
     pub sources: Vec<SourceSpec>,
     pub edges: Vec<Edge>,
+    pub taps: Vec<FlatTap>,
 }
 
 /// A wiring fault caught once, at bootstrap — C7's "type check paid at wiring"
@@ -360,7 +376,12 @@ impl Harness {
     /// 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;
+        // `taps` is ignored here by design: a future slice (#282's `bind_tap`) will
+        // materialize resolved taps into real recorder nodes/edges before bootstrap,
+        // so bootstrap will still have nothing of its own to do with the list. In
+        // this slice `taps` is always empty (no `Tap` resolution exists yet), so the
+        // ignore is presently a no-op, not yet load-bearing.
+        let FlatGraph { nodes, signatures, sources, edges, taps: _ } = flat;
         let n = nodes.len();
 
         // size each node's input columns: KIND/firing from the carried signature,
@@ -808,7 +829,7 @@ mod tests {
         sources: Vec<SourceSpec>,
         edges: Vec<Edge>,
     ) -> Result<Harness, BootstrapError> {
-        Harness::bootstrap(FlatGraph { nodes, signatures, sources, edges })
+        Harness::bootstrap(FlatGraph { nodes, signatures, sources, edges, taps: Vec::new() })
     }
 
     /// The declared signature of a `Recorder` over `kinds` with the given firing.
@@ -1485,6 +1506,54 @@ mod tests {
         );
     }
 
+    #[test]
+    fn flat_graph_taps_are_inert_scaffolding_at_bootstrap() {
+        // CURRENT-BEHAVIOUR PIN (#282, FlatTap + FlatGraph.taps): `bootstrap`
+        // destructures `taps` as `taps: _` and never reads it — populating the
+        // field with a `FlatTap` (even one naming a real producer) must not
+        // perturb bootstrap sizing/wiring or the run's recorded output one bit.
+        // This is the flat-graph-level twin of the compile-boundary pin
+        // (`flat_taps_e2e.rs`): here the field carries data, and the claim is
+        // that the data is inert, not merely that nothing populates it yet.
+        let prices = f64_stream(&[(1, 10.0), (2, 20.0), (3, 30.0)]);
+
+        fn graph(tx: mpsc::Sender<(Timestamp, Vec<Scalar>)>) -> FlatGraph {
+            FlatGraph {
+                nodes: vec![
+                    Box::new(Sma::new(1)),
+                    Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx)),
+                ],
+                signatures: vec![Sma::builder().schema().clone(), recorder_sig(&[ScalarKind::F64], Firing::Any)],
+                sources: vec![SourceSpec::raw(
+                    ScalarKind::F64,
+                    vec![Target { node: 0, slot: 0 }, Target { node: 1, slot: 0 }],
+                )],
+                edges: vec![],
+                taps: vec![],
+            }
+        }
+
+        let (tx, rx) = mpsc::channel();
+        let mut h_bare = Harness::bootstrap(graph(tx)).expect("empty taps list bootstraps");
+        h_bare.run(vec![Box::new(VecSource::new(prices.clone()))]);
+        let bare_out: Vec<(Timestamp, Vec<Scalar>)> = rx.try_iter().collect();
+
+        let (tx2, rx2) = mpsc::channel();
+        let mut with_taps = graph(tx2);
+        with_taps.taps = vec![
+            FlatTap { name: "p".into(), node: 0, field: 0 },
+            FlatTap { name: "q".into(), node: 1, field: 0 },
+        ];
+        let mut h_tapped = Harness::bootstrap(with_taps).expect("a populated taps list bootstraps identically");
+        h_tapped.run(vec![Box::new(VecSource::new(prices))]);
+        let tapped_out: Vec<(Timestamp, Vec<Scalar>)> = rx2.try_iter().collect();
+
+        assert_eq!(
+            bare_out, tapped_out,
+            "a populated FlatGraph.taps list must not change bootstrap/run behaviour in this slice"
+        );
+    }
+
     #[test]
     fn mixed_a_and_b_or_combine_on_one_node() {
         // MixedSum @0 (in0,in1 barrier group 0; in2 as-of); node 1 = Recorder taps it.

crates/aura-engine/src/report.rs

@@ -511,6 +511,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
             ],
+            taps: Vec::new(),
         })
         .expect("valid signal-quality DAG");
         (h, rx_eq, rx_ex)

crates/aura-engine/tests/flat_taps_e2e.rs

@@ -0,0 +1,70 @@
+//! End-to-end coverage for the `FlatTap` / `FlatGraph.taps` plumbing (#282,
+//! "thread all literal sites"): a `Composite` can declare a `Tap` (via
+//! `with_taps`), but this slice adds only the flat-graph *carrier* — no
+//! resolution step exists yet. These tests pin the two halves of that
+//! contract at the public `aura_engine` boundary: `compile()` never surfaces
+//! a declared `Tap` into the flat graph it hands to `Harness::bootstrap`
+//! (`compile_never_resolves_declared_taps_yet`), and the field's presence is
+//! otherwise perfectly transparent to compilation of the rest of the graph
+//! (`tapped_composite_compiles_identically_to_untapped_besides_the_taps_list`)
+//! — a future resolving slice (`bind_tap`) is expected to flip the first of
+//! these; until it does, a regression that started eagerly resolving taps,
+//! or one that let a `Tap` perturb node/edge flattening, would be caught
+//! here.
+
+use aura_core::ScalarKind;
+use aura_engine::{Composite, OutField, Role, Tap, TapWire, Target};
+use aura_std::Sub;
+
+/// A trivial one-node composite with a single declared output (`Sub`'s
+/// difference), matching the `taps` argument requested by each test. Both of
+/// `Sub`'s input slots are fed by one root-bound (source) role — enough to
+/// satisfy `compile()`'s port-connectivity + root-binding checks.
+fn leaf(taps: Vec<Tap>) -> Composite {
+    Composite::new(
+        "leaf",
+        vec![Sub::builder().into()],
+        vec![],
+        vec![Role {
+            name: "price".into(),
+            targets: vec![Target { node: 0, slot: 0 }, Target { node: 0, slot: 1 }],
+            source: Some(ScalarKind::F64),
+        }],
+        vec![OutField { node: 0, field: 0, name: "o".into() }],
+    )
+    .with_taps(taps)
+}
+
+/// Property: **`compile()` resolves no `Tap`s in this slice.** A composite
+/// declaring a valid `Tap` (naming a real interior producer, `node: 0, field:
+/// 0`) still compiles to a `FlatGraph` whose `taps` list is empty — the
+/// declaration is carried on the `Composite` only; nothing yet lowers it into
+/// the flat `FlatTap` the run loop could someday read. If a later change
+/// started eagerly resolving `Tap`s into `FlatGraph.taps` inside `compile()`
+/// without an explicit opt-in (`bind_tap`), this test would flip green->red,
+/// flagging the design-changing move for the ledger instead of it happening
+/// silently.
+#[test]
+fn compile_never_resolves_declared_taps_yet() {
+    let tapped = leaf(vec![Tap { name: "spread".into(), from: TapWire { node: 0, field: 0 } }]);
+    let flat = tapped.compile().expect("a tap-bearing composite still compiles");
+    assert!(flat.taps.is_empty(), "compile() must not resolve declared Taps in this slice: {:?}", flat.taps);
+}
+
+/// Property: **a declared `Tap` is otherwise inert at compile time** — it
+/// changes nothing about the flattened node/edge structure. Compiling the
+/// same composite with and without a `Tap` declaration must produce
+/// structurally identical flat graphs (same node count, same edges); the
+/// only difference tap-bearing scaffolding is *allowed* to make is the (here,
+/// still-empty) `taps` list itself.
+#[test]
+fn tapped_composite_compiles_identically_to_untapped_besides_the_taps_list() {
+    let untapped = leaf(vec![]).compile().expect("untapped compiles");
+    let tapped = leaf(vec![Tap { name: "spread".into(), from: TapWire { node: 0, field: 0 } }])
+        .compile()
+        .expect("tapped compiles");
+
+    assert_eq!(untapped.nodes.len(), tapped.nodes.len(), "tap declaration must not add/remove flat nodes");
+    assert_eq!(untapped.edges, tapped.edges, "tap declaration must not perturb flat edges");
+    assert_eq!(untapped.taps, tapped.taps, "both compile to the same (empty) taps list in this slice");
+}

crates/aura-engine/tests/ger40_breakout.rs

@@ -168,7 +168,8 @@ fn build_harness(
         .map(|slot| SourceSpec::raw(ScalarKind::F64, vec![Target { node: RESAMPLE, slot }]))
         .collect();
 
-    Harness::bootstrap(FlatGraph { nodes, signatures, sources, edges }).expect("valid strategy DAG")
+    Harness::bootstrap(FlatGraph { nodes, signatures, sources, edges, taps: Vec::new() })
+        .expect("valid strategy DAG")
 }
 
 /// The synthetic M1 ticks realising one Frankfurt session whose 15m bars are:

crates/aura-ingest/examples/shared/breakout_real.rs

@@ -187,7 +187,7 @@ pub fn build_harness() -> (Harness, Taps) {
         .map(|slot| SourceSpec::raw(ScalarKind::F64, vec![Target { node: RESAMPLE, slot }]))
         .collect();
 
-    let h = Harness::bootstrap(FlatGraph { nodes, signatures, sources, edges })
+    let h = Harness::bootstrap(FlatGraph { nodes, signatures, sources, edges, taps: Vec::new() })
         .expect("valid breakout DAG");
     (h, Taps { equity: rx_equity, held: rx_held, bars: rx_bars, breakout: rx_breakout })
 }

crates/aura-ingest/tests/real_bars.rs

@@ -60,6 +60,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
         ],
+        taps: Vec::new(),
     })
     .expect("valid signal-quality DAG");
 

crates/aura-ingest/tests/streaming_seam.rs

@@ -66,6 +66,7 @@ fn run_sample(window: (Timestamp, Timestamp), source: Box<dyn Source>) -> RunRep
             Edge { from: 4, to: 5, slot: 0, from_field: 0 },
             Edge { from: 3, to: 6, slot: 0, from_field: 0 },
         ],
+        taps: Vec::new(),
     })
     .expect("valid signal-quality DAG");