feat(aura-ingest): GER40 session-breakout on real M1 bars (examples/)

Run the Phase-1 session-breakout node vocabulary over REAL GER40 M1 bars
from the data-server archive — the first real-data exercise of the shipped
strategy nodes (Resample/Delay/Gt/Session/EqConst/And/Latch + SimBroker).

It is a pure composition of shipped aura-std nodes (no project-specific
signal code), so it lives in the engine repo's examples/ carveout (C9), not
a separate consumer crate. The breakout FlatGraph is reused VERBATIM from
aura-engine/tests/ger40_breakout.rs; only the sources change — four real
M1FieldSource (open/high/low/close, in the fixed C4 merge order) replace the
synthetic VecSources, with pip_size = 1.0 (GER40 is an index).

- examples/ger40_breakout_real.rs — runnable demo over a Sept-2024 window:
  prints the RunReport metrics plus a per-session entry/exit trace.
- examples/shared/breakout_real.rs — the single 13-node wiring (the proven
  11 + two trace taps), shared by the example and the test via #[path] so it
  is never duplicated.
- tests/ger40_breakout_real.rs — gated determinism test (mirrors
  real_bars.rs): skips where the archive is absent; where present, asserts
  finite metrics, binary Latch exposure (held in {0.0, 1.0}), and
  bit-identical C1 reruns of both the report and the recorded series.

chrono / chrono-tz added as aura-ingest dev-dependencies (the Session
timezone and the UTC window bounds), test-only — never on the normal path.

Cargo.lock

@@ -86,6 +86,8 @@ dependencies = [
  "aura-core",
  "aura-engine",
  "aura-std",
+ "chrono",
+ "chrono-tz",
  "data-server",
 ]
 

crates/aura-ingest/Cargo.toml

@@ -18,3 +18,10 @@ aura-engine = { path = "../aura-engine" }
 [dev-dependencies]
 # the integration tests bootstrap a sample harness + fold a RunReport
 aura-std = { path = "../aura-std" }
+# chrono / chrono-tz build the Berlin-local-wall-clock window bounds + the
+# Session node's timezone for the real GER40 session-breakout example/test
+# (examples/ger40_breakout_real.rs, tests/ger40_breakout_real.rs). Pinned to
+# aura-std's / aura-engine's versions (the same DST-aware wall-clock math),
+# test-only — never on the ingestion crate's normal path.
+chrono = { version = "0.4", default-features = false, features = ["clock"] }
+chrono-tz = { version = "0.10", default-features = false }

crates/aura-ingest/examples/ger40_breakout_real.rs

@@ -0,0 +1,127 @@
+//! Runnable demonstration: the GER40 15m session-breakout strategy — the
+//! Phase-1 `aura-std` node vocabulary, proven synthetically in
+//! `aura-engine/tests/ger40_breakout.rs` — driven over REAL GER40 M1 bars from
+//! the local data-server archive (one calendar month).
+//!
+//! This is a *pure composition* of SHIPPED `aura-std` nodes: no project-specific
+//! signal code, so it lives in the engine repo's `examples/` carveout (C9), not
+//! a separate project crate. It reuses the breakout FlatGraph VERBATIM (see
+//! `shared/breakout_real.rs`) — only the sources change: four real
+//! `M1FieldSource`s replace the synthetic `VecSource`s.
+//!
+//! Run:
+//! ```text
+//! cargo run -p aura-ingest --example ger40_breakout_real
+//! ```
+//! Prints the [`RunReport`] metrics (total pips / drawdown / sign-flips) and a
+//! compact entry/exit trace — each 0→1 entry and 1→0 exit with its timestamp —
+//! so the strategy's behaviour over the ~20 sessions in the month is visibly
+//! inspectable. Skips cleanly (no panic) where the archive is absent.
+
+use std::sync::Arc;
+
+use chrono::TimeZone;
+use data_server::{DataServer, DEFAULT_DATA_PATH};
+
+// The breakout wiring is defined once in the shared module and included by both
+// this example and the gated determinism test — never duplicated.
+#[path = "shared/breakout_real.rs"]
+mod breakout_real;
+use breakout_real::*;
+
+// --- The window: September 2024 (inclusive), trivially changeable. -----------
+const YEAR: i32 = 2024;
+const MONTH: u32 = 9;
+
+fn main() {
+    let server = Arc::new(DataServer::new(DEFAULT_DATA_PATH));
+    if !server.has_symbol(SYMBOL) {
+        println!(
+            "skip: no local data at {DEFAULT_DATA_PATH} (symbol {SYMBOL} absent) — \
+             nothing to demonstrate here."
+        );
+        return;
+    }
+
+    let (from_ms, to_ms) = utc_month_window_ms(YEAR, MONTH);
+
+    let Some(sources) = open_ohlc_sources(&server, from_ms, to_ms) else {
+        println!(
+            "skip: {SYMBOL} has no M1 file overlapping {YEAR}-{MONTH:02} — \
+             nothing to demonstrate here."
+        );
+        return;
+    };
+
+    let (mut h, taps) = build_harness();
+    h.run(sources);
+
+    // Drain the per-bar trace first (it consumes the equity/held/bars/breakout
+    // channels), then fold the report from that same trace so the mpsc channels
+    // are drained exactly once (shared with the determinism test).
+    let trace = drain_trace(&taps);
+    let report = report_from_trace(&trace, from_ms, to_ms);
+    let metrics = &report.metrics;
+
+    println!("=== GER40 15m session-breakout — REAL bars ===");
+    println!(
+        "symbol={SYMBOL}  window={YEAR}-{MONTH:02} (UTC, inclusive)  \
+         bars={}  pip_size={PIP_SIZE}",
+        trace.len()
+    );
+    println!("session: {SESSION_HOUR:02}:{SESSION_MINUTE:02} Europe/Berlin, {BAR_MINUTES}m bars; entry bar 3, exit bar 5");
+    println!();
+    println!("--- metrics (sim-optimal broker, pips = index points) ---");
+    println!("  total_pips          = {:.1}", metrics.total_pips);
+    println!("  max_drawdown        = {:.1}", metrics.max_drawdown);
+    println!("  exposure_sign_flips = {}", metrics.exposure_sign_flips);
+    println!();
+
+    // --- 2. The compact entry/exit trace. -----------------------------------
+    // Each 0→1 is an entry (a strict breakout on session bar 3), each 1→0 is the
+    // bar-5 exit. Printing only the transitions keeps the ~20 sessions readable.
+    println!("--- entry / exit transitions (held 0↔1) ---");
+    let mut prev_held = 0.0_f64;
+    let mut entries = 0u32;
+    let mut exits = 0u32;
+    let mut entry_equity = 0.0_f64;
+    for b in &trace {
+        if b.held == 1.0 && prev_held == 0.0 {
+            entries += 1;
+            entry_equity = b.equity;
+            println!(
+                "  ENTER  {}  bar={}  breakout={}  equity={:+.1}",
+                fmt_ts(b.ts),
+                b.bars_since_open,
+                b.breakout,
+                b.equity
+            );
+        } else if b.held == 0.0 && prev_held == 1.0 {
+            exits += 1;
+            println!(
+                "  EXIT   {}  bar={}  equity={:+.1}  (session pnl {:+.1})",
+                fmt_ts(b.ts),
+                b.bars_since_open,
+                b.equity,
+                b.equity - entry_equity
+            );
+        }
+        prev_held = b.held;
+    }
+    if prev_held == 1.0 {
+        println!("  (note: still held at end of window — a partial last session)");
+    }
+    println!();
+    println!("sessions entered = {entries}, exited = {exits}");
+}
+
+/// Render an epoch-ns [`Timestamp`](aura_core::Timestamp) as a UTC wall-clock
+/// `YYYY-MM-DD HH:MM` for the trace. Display only — never feeds the graph.
+fn fmt_ts(ts: aura_core::Timestamp) -> String {
+    let secs = ts.0.div_euclid(1_000_000_000);
+    let nanos = ts.0.rem_euclid(1_000_000_000) as u32;
+    match chrono::Utc.timestamp_opt(secs, nanos).single() {
+        Some(dt) => dt.format("%Y-%m-%d %H:%M UTC").to_string(),
+        None => format!("ts={}", ts.0),
+    }
+}

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

@@ -0,0 +1,338 @@
+//! Shared wiring for the real-data GER40 15m session-breakout demonstration.
+//!
+//! This is the *single* definition of the breakout harness used by BOTH the
+//! runnable example (`examples/ger40_breakout_real.rs`) and the gated
+//! determinism test (`tests/ger40_breakout_real.rs`), so the 11-node FlatGraph
+//! is wired exactly once and never drifts between the two. It is included into
+//! each consumer via `#[path = "..."] mod`, not compiled as its own target.
+//!
+//! The DAG is the VERBATIM breakout graph proven synthetically in
+//! `aura-engine/tests/ger40_breakout.rs` — only the sources change: real
+//! `M1FieldSource`s over a data-server window replace the synthetic
+//! `VecSource`s. Source declaration order (open, high, low, close) is the C4
+//! merge tie-break order Resample's `Barrier(0)` group depends on, so it is
+//! load-bearing and preserved.
+//!
+//! ```text
+//! open,high,low,close (4 real M1 sources, FIXED order)
+//!   -> Resample(15)  ⇒ {open15,high15,low15,close15}
+//!   close15 -> Gt.a
+//!   high15  -> Delay(1) -> Gt.b           (prevHigh15)
+//!   Gt -> breakout:bool                   (close15 > prevHigh15, STRICT)
+//!   close15 -> Session(9,0,Berlin,15) ⇒ bars_since_open:i64
+//!       bars_since_open -> EqConst(3) -> isBar3
+//!       bars_since_open -> EqConst(5) -> isBar5Close
+//!   And(breakout, isBar3) -> entry
+//!   Latch(set=entry, reset=isBar5Close) -> held:f64
+//!   held -> SimBroker(exposure=held, price=close15) ⇒ equity   [pip_size=1.0]
+//!   equity -> Recorder(A);  held -> Recorder(B)
+//!   bars_since_open -> Recorder(C);  breakout -> Recorder(D)
+//! ```
+//!
+//! The synthetic test taps two channels (equity, held). The real demonstration
+//! taps two more — `bars_since_open` (i64) and `breakout` (bool) — so the
+//! entry/exit trace can name *why* each session did or did not fire, without
+//! reaching into engine internals (it reads only what sinks record, C8/C18).
+
+#![allow(dead_code)] // each consumer (example / test) uses a subset of this API
+
+use std::sync::mpsc;
+use std::sync::Arc;
+
+use aura_core::{NodeSchema, Scalar, ScalarKind, Timestamp};
+use aura_engine::{
+    summarize, Edge, FlatGraph, Harness, RunManifest, RunReport, Source, SourceSpec, Target,
+};
+use aura_std::{And, Delay, EqConst, Gt, Latch, Recorder, Resample, Session, SimBroker};
+use chrono::TimeZone;
+use chrono_tz::Europe::Berlin;
+use data_server::DataServer;
+
+use aura_ingest::{M1Field, M1FieldSource};
+
+// ---------------------------------------------------------------------------
+// Node indices in the FlatGraph (one fixed layout, the same nodes 0..=8 as
+// `aura-engine/tests/ger40_breakout.rs`, plus two extra trace taps 11/12).
+// ---------------------------------------------------------------------------
+pub const RESAMPLE: usize = 0;
+pub const DELAY: usize = 1;
+pub const GT: usize = 2;
+pub const SESSION: usize = 3;
+pub const EQ3: usize = 4;
+pub const EQ5: usize = 5;
+pub const AND: usize = 6;
+pub const LATCH: usize = 7;
+pub const BROKER: usize = 8;
+pub const REC_EQUITY: usize = 9; // tap A — SimBroker equity (f64, pips)
+pub const REC_HELD: usize = 10; // tap B — Latch exposure (f64, 0.0/1.0)
+pub const REC_BARS: usize = 11; // tap C — Session bars_since_open (i64)
+pub const REC_BREAKOUT: usize = 12; // tap D — Gt breakout flag (bool)
+
+/// The instrument. GER40 is an index, so pips are index points: `pip_size = 1.0`.
+pub const SYMBOL: &str = "GER40";
+
+/// SimBroker pip size for GER40. An index quotes in points, so one pip is one
+/// point — `1.0`, exactly as the synthetic capstone test uses.
+pub const PIP_SIZE: f64 = 1.0;
+
+/// Session config: Frankfurt cash open 09:00 Europe/Berlin, 15m bars. The same
+/// `Session::new(9, 0, Europe::Berlin, 15)` the synthetic test pins.
+pub const SESSION_HOUR: u32 = 9;
+pub const SESSION_MINUTE: u32 = 0;
+pub const BAR_MINUTES: i64 = 15;
+
+/// The four recorder channels a built harness drains after a run.
+pub struct Taps {
+    pub equity: mpsc::Receiver<(Timestamp, Vec<Scalar>)>,
+    pub held: mpsc::Receiver<(Timestamp, Vec<Scalar>)>,
+    pub bars: mpsc::Receiver<(Timestamp, Vec<Scalar>)>,
+    pub breakout: mpsc::Receiver<(Timestamp, Vec<Scalar>)>,
+}
+
+/// Hand-wire the breakout DAG into a frozen [`Harness`] with four recording
+/// taps. A fresh build per call (fresh nodes + channels) keeps two runs
+/// disjoint for the determinism assertion (C1). The wiring of nodes 0..=10 is
+/// VERBATIM from `aura-engine/tests/ger40_breakout.rs`; nodes 11/12 add the two
+/// trace taps.
+pub fn build_harness() -> (Harness, Taps) {
+    use aura_core::Firing;
+
+    let (tx_equity, rx_equity) = mpsc::channel();
+    let (tx_held, rx_held) = mpsc::channel();
+    let (tx_bars, rx_bars) = mpsc::channel();
+    let (tx_breakout, rx_breakout) = mpsc::channel();
+
+    // Each node's declared signature (kind/firing per port), parallel to `nodes`.
+    let signatures: Vec<NodeSchema> = vec![
+        Resample::builder().schema().clone(), // 0
+        Delay::builder().schema().clone(),     // 1
+        Gt::builder().schema().clone(),        // 2
+        Session::builder(SESSION_HOUR, SESSION_MINUTE, Berlin, BAR_MINUTES)
+            .schema()
+            .clone(), // 3
+        EqConst::builder().schema().clone(), // 4
+        EqConst::builder().schema().clone(), // 5
+        And::builder().schema().clone(),     // 6
+        Latch::builder().schema().clone(),   // 7
+        SimBroker::builder(PIP_SIZE).schema().clone(), // 8
+        Recorder::builder(vec![ScalarKind::F64], Firing::Any, tx_equity.clone())
+            .schema()
+            .clone(), // 9
+        Recorder::builder(vec![ScalarKind::F64], Firing::Any, tx_held.clone())
+            .schema()
+            .clone(), // 10
+        Recorder::builder(vec![ScalarKind::I64], Firing::Any, tx_bars.clone())
+            .schema()
+            .clone(), // 11
+        Recorder::builder(vec![ScalarKind::Bool], Firing::Any, tx_breakout.clone())
+            .schema()
+            .clone(), // 12
+    ];
+
+    let nodes: Vec<Box<dyn aura_core::Node>> = vec![
+        Box::new(Resample::new(15)),                                            // 0
+        Box::new(Delay::new(1)),                                                // 1
+        Box::new(Gt::new()),                                                    // 2
+        Box::new(Session::new(SESSION_HOUR, SESSION_MINUTE, Berlin, BAR_MINUTES)), // 3
+        Box::new(EqConst::new(3)),                                              // 4
+        Box::new(EqConst::new(5)),                                              // 5
+        Box::new(And::new()),                                                   // 6
+        Box::new(Latch::new()),                                                 // 7
+        Box::new(SimBroker::new(PIP_SIZE)),                                     // 8
+        Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx_equity)),    // 9
+        Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx_held)),      // 10
+        Box::new(Recorder::new(&[ScalarKind::I64], Firing::Any, tx_bars)),      // 11
+        Box::new(Recorder::new(&[ScalarKind::Bool], Firing::Any, tx_breakout)), // 12
+    ];
+
+    // Resample output fields: 0=open15, 1=high15, 2=low15, 3=close15.
+    const F_HIGH15: usize = 1;
+    const F_CLOSE15: usize = 3;
+
+    let edges = vec![
+        // close15 -> Gt.a (slot 0)
+        Edge { from: RESAMPLE, to: GT, slot: 0, from_field: F_CLOSE15 },
+        // high15 -> Delay.series (slot 0) -> Gt.b (slot 1)
+        Edge { from: RESAMPLE, to: DELAY, slot: 0, from_field: F_HIGH15 },
+        Edge { from: DELAY, to: GT, slot: 1, from_field: 0 },
+        // close15 -> Session.trigger (value ignored)
+        Edge { from: RESAMPLE, to: SESSION, slot: 0, from_field: F_CLOSE15 },
+        // bars_since_open -> EqConst(3) / EqConst(5)
+        Edge { from: SESSION, to: EQ3, slot: 0, from_field: 0 },
+        Edge { from: SESSION, to: EQ5, slot: 0, from_field: 0 },
+        // And(breakout, isBar3) -> entry
+        Edge { from: GT, to: AND, slot: 0, from_field: 0 },
+        Edge { from: EQ3, to: AND, slot: 1, from_field: 0 },
+        // Latch(set=entry, reset=isBar5Close)
+        Edge { from: AND, to: LATCH, slot: 0, from_field: 0 },
+        Edge { from: EQ5, to: LATCH, slot: 1, from_field: 0 },
+        // held -> SimBroker.exposure (slot 0); close15 -> SimBroker.price (slot 1)
+        Edge { from: LATCH, to: BROKER, slot: 0, from_field: 0 },
+        Edge { from: RESAMPLE, to: BROKER, slot: 1, from_field: F_CLOSE15 },
+        // equity -> tap A; held -> tap B
+        Edge { from: BROKER, to: REC_EQUITY, slot: 0, from_field: 0 },
+        Edge { from: LATCH, to: REC_HELD, slot: 0, from_field: 0 },
+        // trace taps: bars_since_open -> tap C; breakout -> tap D
+        Edge { from: SESSION, to: REC_BARS, slot: 0, from_field: 0 },
+        Edge { from: GT, to: REC_BREAKOUT, slot: 0, from_field: 0 },
+    ];
+
+    // Four f64 sources, one per OHLC field, each feeding one Resample slot. The
+    // declaration order open(0), high(1), low(2), close(3) is the merge tie-break
+    // order for the Barrier(0) group — identical to the synthetic capstone.
+    let sources = (0..4)
+        .map(|slot| SourceSpec {
+            kind: ScalarKind::F64,
+            targets: vec![Target { node: RESAMPLE, slot }],
+        })
+        .collect();
+
+    let h = Harness::bootstrap(FlatGraph { nodes, signatures, sources, edges })
+        .expect("valid breakout DAG");
+    (h, Taps { equity: rx_equity, held: rx_held, bars: rx_bars, breakout: rx_breakout })
+}
+
+/// Open the four real OHLC `M1FieldSource`s over `[from_ms, to_ms]` (inclusive
+/// Unix-ms) in the FIXED order open, high, low, close — the C4 merge tie-break
+/// order Resample's `Barrier(0)` group expects. Returns `None` if any field has
+/// no file overlapping the window (so the caller can skip cleanly).
+///
+/// This is exactly the multi-field OHLC open that `M1FieldSource` makes the
+/// caller spell out four times by hand — see the friction note in the report.
+pub fn open_ohlc_sources(
+    server: &Arc<DataServer>,
+    from_ms: i64,
+    to_ms: i64,
+) -> Option<Vec<Box<dyn Source>>> {
+    let open = M1FieldSource::open(server, SYMBOL, Some(from_ms), Some(to_ms), M1Field::Open)?;
+    let high = M1FieldSource::open(server, SYMBOL, Some(from_ms), Some(to_ms), M1Field::High)?;
+    let low = M1FieldSource::open(server, SYMBOL, Some(from_ms), Some(to_ms), M1Field::Low)?;
+    let close = M1FieldSource::open(server, SYMBOL, Some(from_ms), Some(to_ms), M1Field::Close)?;
+    let srcs: Vec<Box<dyn Source>> =
+        vec![Box::new(open), Box::new(high), Box::new(low), Box::new(close)];
+    Some(srcs)
+}
+
+/// Build the inclusive Unix-ms window for the whole calendar month `(year,
+/// month)` in UTC: `[first instant of the 1st, last ms of the last day]`. UTC
+/// keeps the window boundary trivial to reason about; the Session node still
+/// indexes bars in Berlin wall-clock inside the graph.
+pub fn utc_month_window_ms(year: i32, month: u32) -> (i64, i64) {
+    let from = chrono::Utc
+        .with_ymd_and_hms(year, month, 1, 0, 0, 0)
+        .unwrap()
+        .timestamp_millis();
+    // first instant of the next month, minus one ms => last ms of this month.
+    let (ny, nm) = if month == 12 { (year + 1, 1) } else { (year, month + 1) };
+    let next = chrono::Utc
+        .with_ymd_and_hms(ny, nm, 1, 0, 0, 0)
+        .unwrap()
+        .timestamp_millis();
+    (from, next - 1)
+}
+
+/// One bar's recorded trace row, fused across the four taps by **timestamp**.
+///
+/// The four sinks do NOT all fire on the same cardinality: `equity` and `held`
+/// fire once per completed Resample bar (downstream of Latch/SimBroker), but the
+/// `breakout` (Gt) tap is one bar shorter — `Gt` depends on `Delay(1)` of
+/// high15, which is cold on the very first bar and emits nothing — and the
+/// `bars_since_open` (Session) tap filters bars that precede a session open. So
+/// they cannot be fused by positional index; they are joined on the recorded
+/// timestamp. `equity`/`held` are the spine (one row per bar); the other two are
+/// looked up by ts, defaulting where they did not fire this bar.
+pub struct BarTrace {
+    pub ts: Timestamp,
+    pub equity: f64,
+    pub held: f64,
+    /// Session bar index, or `-1` where Session did not emit this bar (outside a
+    /// tracked session).
+    pub bars_since_open: i64,
+    /// The strict-breakout flag, or `false` where Gt did not emit (the cold
+    /// first bar, before Delay(1) is warm).
+    pub breakout: bool,
+}
+
+/// Drain the four taps into a per-bar trace, joining on the recorded timestamp
+/// (the sinks fire on different cardinalities — see [`BarTrace`]). Reads only
+/// recorded sink output (C8/C18) — never engine internals.
+pub fn drain_trace(taps: &Taps) -> Vec<BarTrace> {
+    use std::collections::HashMap;
+
+    let equity: Vec<(Timestamp, Vec<Scalar>)> = taps.equity.try_iter().collect();
+    let held: Vec<(Timestamp, Vec<Scalar>)> = taps.held.try_iter().collect();
+    let bars: Vec<(Timestamp, Vec<Scalar>)> = taps.bars.try_iter().collect();
+    let breakout: Vec<(Timestamp, Vec<Scalar>)> = taps.breakout.try_iter().collect();
+
+    // held shares the equity spine (same node-firing cadence); align by index is
+    // safe between those two, but we still fuse the side taps by ts.
+    let held_by_ts: HashMap<i64, f64> =
+        held.iter().map(|(t, r)| (t.0, as_f64(&r[0]))).collect();
+    let bars_by_ts: HashMap<i64, i64> =
+        bars.iter().map(|(t, r)| (t.0, as_i64(&r[0]))).collect();
+    let breakout_by_ts: HashMap<i64, bool> =
+        breakout.iter().map(|(t, r)| (t.0, as_bool(&r[0]))).collect();
+
+    equity
+        .iter()
+        .map(|(ts, row)| BarTrace {
+            ts: *ts,
+            equity: as_f64(&row[0]),
+            held: held_by_ts.get(&ts.0).copied().unwrap_or(0.0),
+            bars_since_open: bars_by_ts.get(&ts.0).copied().unwrap_or(-1),
+            breakout: breakout_by_ts.get(&ts.0).copied().unwrap_or(false),
+        })
+        .collect()
+}
+
+/// Fold an already-drained per-bar trace into a [`RunReport`] over the real
+/// window, exactly as `tests/real_bars.rs` does for the SMA sample. The trace is
+/// the single drain of the recording channels (`drain_trace`); this reduction is
+/// pure over it, so the example and the test share one report definition. The
+/// window is the requested `[from_ms, to_ms]` normalized to epoch-ns.
+pub fn report_from_trace(trace: &[BarTrace], from_ms: i64, to_ms: i64) -> RunReport {
+    let equity: Vec<(Timestamp, f64)> = trace.iter().map(|b| (b.ts, b.equity)).collect();
+    let exposure: Vec<(Timestamp, f64)> = trace.iter().map(|b| (b.ts, b.held)).collect();
+    let metrics = summarize(&equity, &exposure);
+
+    RunReport {
+        manifest: RunManifest {
+            commit: "ger40-breakout-real".to_string(),
+            params: vec![
+                ("session_hour".to_string(), Scalar::i64(SESSION_HOUR as i64)),
+                ("session_minute".to_string(), Scalar::i64(SESSION_MINUTE as i64)),
+                ("bar_minutes".to_string(), Scalar::i64(BAR_MINUTES)),
+                ("entry_bar".to_string(), Scalar::i64(3)),
+                ("exit_bar".to_string(), Scalar::i64(5)),
+            ],
+            window: (
+                aura_ingest::unix_ms_to_epoch_ns(from_ms),
+                aura_ingest::unix_ms_to_epoch_ns(to_ms),
+            ),
+            seed: 0,
+            broker: "sim-optimal(pip_size=1)".to_string(),
+        },
+        metrics,
+    }
+}
+
+fn as_f64(s: &Scalar) -> f64 {
+    match s {
+        Scalar::F64(v) => *v,
+        other => panic!("expected f64, got {other:?}"),
+    }
+}
+
+fn as_i64(s: &Scalar) -> i64 {
+    match s {
+        Scalar::I64(v) => *v,
+        other => panic!("expected i64, got {other:?}"),
+    }
+}
+
+fn as_bool(s: &Scalar) -> bool {
+    match s {
+        Scalar::Bool(v) => *v,
+        other => panic!("expected bool, got {other:?}"),
+    }
+}

crates/aura-ingest/tests/ger40_breakout_real.rs

@@ -0,0 +1,88 @@
+//! Gated integration test: the GER40 15m session-breakout strategy — the
+//! Phase-1 `aura-std` node vocabulary wired VERBATIM from
+//! `aura-engine/tests/ger40_breakout.rs` — driven over REAL GER40 M1 bars from
+//! the local data-server archive (September 2024). Mirrors `real_bars.rs`:
+//! skips with a note where the local Pepperstone archive is absent, so
+//! `cargo test --workspace` stays green anywhere; exercises the real ingestion
+//! path where the files exist.
+//!
+//! The breakout FlatGraph is built once in the shared module (`shared/
+//! breakout_real.rs`) and reused by both this test and the runnable example, so
+//! the 11-node wiring is never duplicated.
+
+use std::sync::Arc;
+
+use data_server::{DataServer, DEFAULT_DATA_PATH};
+
+#[path = "../examples/shared/breakout_real.rs"]
+mod breakout_real;
+use breakout_real::*;
+
+// September 2024 (inclusive), the same window the example demonstrates.
+const YEAR: i32 = 2024;
+const MONTH: u32 = 9;
+
+/// Build the real breakout harness, run it over `[from_ms, to_ms]`, and return
+/// the folded `RunReport` plus the drained per-bar `(held, equity)` series. A
+/// fresh build per call keeps the two determinism runs disjoint (C1).
+fn run_real(
+    server: &Arc<DataServer>,
+    from_ms: i64,
+    to_ms: i64,
+) -> (aura_engine::RunReport, Vec<(f64, f64)>) {
+    let sources =
+        open_ohlc_sources(server, from_ms, to_ms).expect("GER40 has data in the Sept-2024 window");
+    let (mut h, taps) = build_harness();
+    h.run(sources);
+
+    // Drain the trace once; fold the report from the same trace via the shared
+    // helper so the mpsc channels are consumed exactly once and the report shape
+    // matches the example's byte-for-byte.
+    let trace = drain_trace(&taps);
+    let report = report_from_trace(&trace, from_ms, to_ms);
+    let series: Vec<(f64, f64)> = trace.iter().map(|b| (b.held, b.equity)).collect();
+    (report, series)
+}
+
+/// Property: the real-bar GER40 breakout harness runs end-to-end over a real
+/// data-server month and is **deterministic** (C1) — two disjoint runs over the
+/// byte-identical archive window produce bit-identical reports AND bit-identical
+/// recorded series. Also pins two observable invariants of the composed
+/// vocabulary on real data: the metrics are finite, and the Latch output
+/// (`held`) is *exactly* 0.0 or 1.0 every bar (a binary exposure register, never
+/// a fractional or NaN value). This protects the whole node composition —
+/// Resample / Delay / Gt / Session / EqConst / And / Latch / SimBroker — against
+/// any regression that would only surface on real (irregular, gap-laden) M1
+/// data, which the synthetic single-session capstone cannot reach.
+#[test]
+fn ger40_breakout_real_bars_run_is_deterministic() {
+    let server = Arc::new(DataServer::new(DEFAULT_DATA_PATH));
+    if !server.has_symbol(SYMBOL) {
+        eprintln!("skip: no local data at {DEFAULT_DATA_PATH} (symbol {SYMBOL} absent)");
+        return; // hermetic elsewhere; exercises the real path where files exist
+    }
+
+    let (from_ms, to_ms) = utc_month_window_ms(YEAR, MONTH);
+
+    let (r1, s1) = run_real(&server, from_ms, to_ms);
+    assert!(!s1.is_empty(), "window resolved to zero bars");
+
+    // Metrics are finite — a backtest actually ran over real bars.
+    assert!(r1.metrics.total_pips.is_finite(), "total_pips finite");
+    assert!(r1.metrics.max_drawdown.is_finite(), "max_drawdown finite");
+
+    // Latch output is a binary exposure register: every recorded `held` is
+    // exactly 0.0 or 1.0 — never fractional, never NaN.
+    for (held, _) in &s1 {
+        assert!(
+            *held == 0.0 || *held == 1.0,
+            "held must be exactly 0.0 or 1.0 (Latch register), got {held}"
+        );
+    }
+
+    // C1 determinism: a fresh disjoint run over the identical window is
+    // bit-identical in both the folded report and the raw recorded series.
+    let (r2, s2) = run_real(&server, from_ms, to_ms);
+    assert_eq!(r1.to_json(), r2.to_json(), "report bit-identical across runs (C1)");
+    assert_eq!(s1, s2, "recorded (held, equity) series bit-identical across runs (C1)");
+}