feat(aura-ingest): consolidate the real-data source-open seam

A real OHLC source now builds from aura-ingest alone, in the engine-native epoch-ns Timestamp currency, with the one ms<->ns crossing owned by the seam. Library surface (src/lib.rs), all additive — the ms-based `open` and `unix_ms_to_epoch_ns` are untouched, so every existing call site is preserved byte-for-byte: - epoch_ns_to_unix_ms (private): the seam-owned inverse; consumers never convert (#80). - M1FieldSource::open_window: the Timestamp-window mirror of `open`, mapping each bound through the private inverse and delegating to `open`. - open_ohlc: the canonical OHLC bundle opener — four M1FieldSources in the fixed open/high/low/close C4 merge order, sharing one Arc<DataServer> (one cache, C12); the order lives in one vetted place (#92). - default_data_server + `pub use data_server::{DataServer, DEFAULT_DATA_PATH}`: a real-data source builds without naming the external data_server crate (#81). Consumer migration: the shared breakout_real.rs helpers and every GER40 example/test move to the Timestamp-native surface — open_ohlc_sources removed, utc_month_window_ms -> utc_month_window (Timestamp), report_from_trace retyped, the per-consumer ns_to_ms hand-divides deleted, the direct data_server imports replaced by the re-exports. ger40_breakout_compare.rs (which opened OHLC by its own hand-loop, reached by neither retyped helper) is migrated by inspection. Return type is Vec, not [_; 4]: Harness::run consumes Vec<Box<dyn Source>>, so it feeds straight in; order-safety comes from the single helper either way. Tests: a hermetic round-trip pinning the inverse; a gated behaviour-preservation test (open_ohlc Timestamp path == ms-path open, bit-identical recorded series); a hermetic absent-archive fixture pinning the file-level None contract and the #81 re-export firewall. Verified: cargo build --workspace --all-targets clean; cargo test --workspace green (the gated GER40 tests ran against the local archive — open_ohlc_seam proves byte-identity on real data, not a skip); clippy --all-targets -D warnings clean. Acceptance greps pass: open_ohlc_sources gone, no consumer-side ns_to_ms, no data_server import in any migrated GER40 consumer, epoch_ns_to_unix_ms private. closes #80, #81, #92
2026-06-18 12:59:02 +02:00
parent 2ffcd9f03d
commit 8c9a1b4630
11 changed files with 355 additions and 163 deletions
@@ -23,12 +23,10 @@
 use std::sync::Arc;

 use aura_core::{Cell, Scalar, Timestamp};
-use aura_engine::{summarize, RunMetrics, Source};
-use chrono::TimeZone;
+use aura_engine::{summarize, RunMetrics};
 use chrono_tz::Europe::Berlin;
-use data_server::{DataServer, DEFAULT_DATA_PATH};

-use aura_ingest::{M1Field, M1FieldSource};
+use aura_ingest::{default_data_server, open_ohlc, DataServer, DEFAULT_DATA_PATH};

 #[path = "shared/breakout_real.rs"]
 mod breakout_real;
@@ -48,7 +46,7 @@ const INSTRUMENTS: &[(&str, f64, &str)] = &[
 ];

 /// Bootstrap the blueprint (at `bar_period`) at `{entry=3, exit=5}`, run it over
-/// `[from_ms, to_ms]` of real `symbol` OHLC, and summarize — or `None` if the
+/// `[from, to]` (epoch-ns) of real `symbol` OHLC, and summarize — or `None` if the
 /// symbol has no file overlapping the window. The `pip_size` rides on the
 /// blueprint's baked SimBroker (always 1.0 here; both indices). Returns the
 /// report plus the entry count (0→1 transitions of held).
@@ -56,8 +54,8 @@ fn run_cell(
    server: &Arc<DataServer>,
    symbol: &str,
    bar_period: i64,
-    from_ms: i64,
-    to_ms: i64,
+    from: Timestamp,
+    to: Timestamp,
 ) -> Option<(RunMetrics, u32)> {
    let (bp, taps) = ger40_breakout_blueprint(bar_period, SESSION_HOUR, SESSION_MINUTE, Berlin);
    let point: Vec<Cell> = bp
@@ -71,12 +69,7 @@ fn run_cell(
        .collect();
    let mut h = bp.bootstrap_with_cells(&point).expect("point kind-checked");

-    let fields = [M1Field::Open, M1Field::High, M1Field::Low, M1Field::Close];
-    let mut srcs: Vec<Box<dyn Source>> = Vec::with_capacity(4);
-    for &f in &fields {
-        let s = M1FieldSource::open(server, symbol, Some(from_ms), Some(to_ms), f)?;
-        srcs.push(Box::new(s));
-    }
+    let srcs = open_ohlc(server, symbol, from, to)?;
    h.run(srcs);
    drop(h);

@@ -104,21 +97,15 @@ fn run_cell(
 }

 fn main() {
-    let server = Arc::new(DataServer::new(DEFAULT_DATA_PATH));
+    let server = default_data_server();
    if !server.has_symbol(SYMBOL) {
        println!("skip: no local data at {DEFAULT_DATA_PATH} (symbol {SYMBOL} absent).");
        return;
    }

-    // Full year 2024 for body.
-    let from_ms = chrono::Utc
-        .with_ymd_and_hms(2024, 1, 1, 0, 0, 0)
-        .unwrap()
-        .timestamp_millis();
-    let to_ms = chrono::Utc
-        .with_ymd_and_hms(2024, 12, 31, 23, 59, 59)
-        .unwrap()
-        .timestamp_millis();
+    // Full year 2024 (the same UTC span, via the shared Timestamp helper).
+    let (from, _) = utc_month_window(2024, 1);
+    let (_, to) = utc_month_window(2024, 12);

    println!("=== GER40 session-breakout — World `compare` (structural matrix) ===");
    println!("window=2024 (UTC, full year), fixed tuning point entry={ENTRY_BAR}, exit={EXIT_BAR}\n");
@@ -135,7 +122,7 @@ fn main() {
            println!("{symbol:<8} (no local data — skip)");
            continue;
        }
-        match run_cell(&server, symbol, BAR_MINUTES, from_ms, to_ms) {
+        match run_cell(&server, symbol, BAR_MINUTES, from, to) {
            Some((m, entries)) => println!(
                "{symbol:<8} {note:<14} {entries:>9} {:>11.1} {:>10.1} {:>7}",
                m.total_pips, m.max_drawdown, m.exposure_sign_flips
@@ -156,7 +143,7 @@ fn main() {
    );
    println!("{}", "-".repeat(52));
    for &period in &[15_i64, 30] {
-        match run_cell(&server, SYMBOL, period, from_ms, to_ms) {
+        match run_cell(&server, SYMBOL, period, from, to) {
            Some((m, entries)) => println!(
                "{:<10} {entries:>9} {:>11.1} {:>10.1} {:>7}",
                format!("{period}m"),
@@ -20,11 +20,9 @@
 //! 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 chrono_tz::Europe::Berlin;
-use data_server::{DataServer, DEFAULT_DATA_PATH};
+use aura_ingest::{default_data_server, open_ohlc, 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.
@@ -37,7 +35,7 @@ const YEAR: i32 = 2024;
 const MONTH: u32 = 9;

 fn main() {
-    let server = Arc::new(DataServer::new(DEFAULT_DATA_PATH));
+    let server = default_data_server();
    if !server.has_symbol(SYMBOL) {
        println!(
            "skip: no local data at {DEFAULT_DATA_PATH} (symbol {SYMBOL} absent) — \
@@ -46,9 +44,9 @@ fn main() {
        return;
    }

-    let (from_ms, to_ms) = utc_month_window_ms(YEAR, MONTH);
+    let (from, to) = utc_month_window(YEAR, MONTH);

-    let Some(sources) = open_ohlc_sources(&server, from_ms, to_ms) else {
+    let Some(sources) = open_ohlc(&server, SYMBOL, from, to) else {
        println!(
            "skip: {SYMBOL} has no M1 file overlapping {YEAR}-{MONTH:02} — \
             nothing to demonstrate here."
@@ -72,7 +70,7 @@ fn main() {
    // 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 report = report_from_trace(&trace, from, to);
    let metrics = &report.metrics;

    println!("=== GER40 15m session-breakout — REAL bars ===");
@@ -18,7 +18,7 @@ use std::sync::Arc;
 use aura_core::{Cell, Scalar, Timestamp};
 use aura_engine::{sweep, summarize, GridSpace, RunManifest, RunReport};
 use chrono_tz::Europe::Berlin;
-use data_server::{DataServer, DEFAULT_DATA_PATH};
+use aura_ingest::{default_data_server, open_ohlc, DataServer, DEFAULT_DATA_PATH};

 #[path = "shared/breakout_real.rs"]
 mod breakout_real;
@@ -29,17 +29,18 @@ use breakout_real::*;
 const ENTRY_BARS: &[i64] = &[2, 3, 4];
 const EXIT_BARS: &[i64] = &[4, 5, 6];

-/// Bootstrap the blueprint at one grid point, run it over `[from_ms, to_ms]` of
-/// real GER40 OHLC, and fold the recorded equity/held taps into a `RunReport`
-/// (`summarize` over the drained channels). A fresh blueprint per call (fresh
-/// nodes + channels) keeps the disjoint sweep points independent (C1).
-fn run_point(server: &Arc<DataServer>, point: &[Cell], from_ms: i64, to_ms: i64) -> RunReport {
+/// Bootstrap the blueprint at one grid point, run it over the epoch-ns
+/// [`Timestamp`] window `[from, to]` of real GER40 OHLC, and fold the recorded
+/// equity/held taps into a `RunReport` (`summarize` over the drained channels).
+/// A fresh blueprint per call (fresh nodes + channels) keeps the disjoint sweep
+/// points independent (C1).
+fn run_point(server: &Arc<DataServer>, point: &[Cell], from: Timestamp, to: Timestamp) -> RunReport {
    let (bp, taps) = ger40_breakout_blueprint(BAR_MINUTES, SESSION_HOUR, SESSION_MINUTE, Berlin);
    let mut h = bp
        .bootstrap_with_cells(point)
        .expect("sweep point kind-checked against param_space");
    let sources =
-        open_ohlc_sources(server, from_ms, to_ms).expect("window overlaps GER40 data");
+        open_ohlc(server, SYMBOL, from, to).expect("window overlaps GER40 data");
    h.run(sources);
    drop(h);
    let equity: Vec<(Timestamp, f64)> = taps
@@ -56,10 +57,7 @@ fn run_point(server: &Arc<DataServer>, point: &[Cell], from_ms: i64, to_ms: i64)
        manifest: RunManifest {
            commit: "ger40-breakout-sweep".to_string(),
            params: vec![],
-            window: (
-                aura_ingest::unix_ms_to_epoch_ns(from_ms),
-                aura_ingest::unix_ms_to_epoch_ns(to_ms),
-            ),
+            window: (from, to),
            seed: 0,
            broker: "sim-optimal(pip_size=1)".to_string(),
        },
@@ -68,15 +66,15 @@ fn run_point(server: &Arc<DataServer>, point: &[Cell], from_ms: i64, to_ms: i64)
 }

 fn main() {
-    let server = Arc::new(DataServer::new(DEFAULT_DATA_PATH));
+    let server = default_data_server();
    if !server.has_symbol(SYMBOL) {
        println!("skip: no local data at {DEFAULT_DATA_PATH} (symbol {SYMBOL} absent).");
        return;
    }

    // Full-year 2024.
-    let (from_ms, _) = utc_month_window_ms(2024, 1);
-    let (_, to_ms) = utc_month_window_ms(2024, 12);
+    let (from, _) = utc_month_window(2024, 1);
+    let (_, to) = utc_month_window(2024, 12);

    // The grid is built from the blueprint's param_space() directly — the two
    // EqConst targets are the swept axes (no re-authoring).
@@ -109,7 +107,7 @@ fn main() {

    let server_for_closure = Arc::clone(&server);
    let family = sweep(&grid, |pt: &[Cell]| {
-        run_point(&server_for_closure, pt, from_ms, to_ms)
+        run_point(&server_for_closure, pt, from, to)
    });

    // Rank the family by total_pips, descending.
@@ -24,7 +24,7 @@ use aura_engine::{
 };
 use chrono::TimeZone;
 use chrono_tz::Europe::Berlin;
-use data_server::{DataServer, DEFAULT_DATA_PATH};
+use aura_ingest::{default_data_server, open_ohlc, DataServer, DEFAULT_DATA_PATH};

 #[path = "shared/breakout_real.rs"]
 mod breakout_real;
@@ -35,28 +35,22 @@ use breakout_real::*;
 const ENTRY_BARS: &[i64] = &[2, 3, 4];
 const EXIT_BARS: &[i64] = &[4, 5, 6];

-/// ns → ms: `open_ohlc_sources` opens by Unix-ms, but `WindowBounds` are
-/// epoch-ns. `aura-ingest` exports only the forward direction, so the inverse is
-/// a local hand-divide (#80, filed).
-fn ns_to_ms(ts: Timestamp) -> i64 {
-    ts.0 / 1_000_000
-}
-
-/// Bootstrap the blueprint at one grid point, run it over `[from_ms, to_ms]` of
-/// real GER40 OHLC, and fold the recorded equity/held taps into `(RunReport,
-/// equity-segment)`. Fresh blueprint per call (C1).
+/// Bootstrap the blueprint at one grid point, run it over the epoch-ns
+/// [`Timestamp`] window `[from, to]` of real GER40 OHLC, and fold the recorded
+/// equity/held taps into `(RunReport, equity-segment)`. Fresh blueprint per call
+/// (C1).
 fn run_point(
    server: &Arc<DataServer>,
    point: &[Cell],
-    from_ms: i64,
-    to_ms: i64,
+    from: Timestamp,
+    to: Timestamp,
 ) -> (RunReport, Vec<(Timestamp, f64)>) {
    let (bp, taps) = ger40_breakout_blueprint(BAR_MINUTES, SESSION_HOUR, SESSION_MINUTE, Berlin);
    let mut h = bp
        .bootstrap_with_cells(point)
        .expect("point kind-checked against param_space");
    let sources =
-        open_ohlc_sources(server, from_ms, to_ms).expect("window overlaps GER40 data");
+        open_ohlc(server, SYMBOL, from, to).expect("window overlaps GER40 data");
    h.run(sources);
    drop(h);
    let equity: Vec<(Timestamp, f64)> = taps
@@ -73,10 +67,7 @@ fn run_point(
        manifest: RunManifest {
            commit: "ger40-breakout-wfo".to_string(),
            params: vec![],
-            window: (
-                aura_ingest::unix_ms_to_epoch_ns(from_ms),
-                aura_ingest::unix_ms_to_epoch_ns(to_ms),
-            ),
+            window: (from, to),
            seed: 0,
            broker: "sim-optimal(pip_size=1)".to_string(),
        },
@@ -86,7 +77,7 @@ fn run_point(
 }

 fn main() {
-    let server = Arc::new(DataServer::new(DEFAULT_DATA_PATH));
+    let server = default_data_server();
    if !server.has_symbol(SYMBOL) {
        println!("skip: no local data at {DEFAULT_DATA_PATH} (symbol {SYMBOL} absent).");
        return;
@@ -143,9 +134,8 @@ fn main() {
                .collect(),
        )
        .expect("IS grid well-formed");
-        let (is_from, is_to) = (ns_to_ms(w.is.0), ns_to_ms(w.is.1));
        let is_family = sweep(&is_grid, |pt: &[Cell]| {
-            run_point(&server_for_closure, pt, is_from, is_to).0
+            run_point(&server_for_closure, pt, w.is.0, w.is.1).0
        });
        let best = is_family
            .points
@@ -161,9 +151,8 @@ fn main() {
        let chosen = best.params.clone();

        // Apply the chosen params on the OOS window — the honest out-of-sample run.
-        let (oos_from, oos_to) = (ns_to_ms(w.oos.0), ns_to_ms(w.oos.1));
        let (oos_report, oos_equity) =
-            run_point(&server_for_closure, &chosen, oos_from, oos_to);
+            run_point(&server_for_closure, &chosen, w.oos.0, w.oos.1);
        WindowRun { chosen_params: chosen, oos_equity, oos_report }
    });

@@ -37,19 +37,15 @@
 #![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, Composite, Edge, FlatGraph, GraphBuilder, Harness, RunManifest, RunReport, Source,
+    summarize, Composite, Edge, FlatGraph, GraphBuilder, Harness, RunManifest, RunReport,
    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
@@ -306,32 +302,14 @@ pub fn ger40_breakout_blueprint(
    (bp, 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) {
+/// Build the inclusive window for the whole calendar month `(year, month)` in
+/// UTC: `[first instant of the 1st, last ms of the last day]`, in aura's native
+/// epoch-ns [`Timestamp`] currency. UTC keeps the boundary trivial to reason
+/// about; the Session node still indexes bars in Berlin wall-clock inside the
+/// graph. The SAME UTC instants as before, typed as `Timestamp` (each bound
+/// wrapped through the one `unix_ms_to_epoch_ns` seam), so the consumer layer is
+/// Timestamp-native and no caller round-trips ms→Timestamp→ms.
+pub fn utc_month_window(year: i32, month: u32) -> (Timestamp, Timestamp) {
    let from = chrono::Utc
        .with_ymd_and_hms(year, month, 1, 0, 0, 0)
        .unwrap()
@@ -342,7 +320,10 @@ pub fn utc_month_window_ms(year: i32, month: u32) -> (i64, i64) {
        .with_ymd_and_hms(ny, nm, 1, 0, 0, 0)
        .unwrap()
        .timestamp_millis();
-    (from, next - 1)
+    (
+        aura_ingest::unix_ms_to_epoch_ns(from),
+        aura_ingest::unix_ms_to_epoch_ns(next - 1),
+    )
 }

 /// One bar's recorded trace row, fused across the four taps by **timestamp**.
@@ -403,8 +384,8 @@ pub fn drain_trace(taps: &Taps) -> Vec<BarTrace> {
 /// 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 {
+/// window is the requested `[from, to]` epoch-ns window.
+pub fn report_from_trace(trace: &[BarTrace], from: Timestamp, to: Timestamp) -> 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);
@@ -419,10 +400,7 @@ pub fn report_from_trace(trace: &[BarTrace], from_ms: i64, to_ms: i64) -> RunRep
                ("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),
-            ),
+            window: (from, to),
            seed: 0,
            broker: "sim-optimal(pip_size=1)".to_string(),
        },
@@ -22,7 +22,12 @@

 use aura_core::{Scalar, Timestamp};
 use data_server::records::M1Parsed;
-use data_server::{DataServer, SymbolChunkIter};
+use data_server::SymbolChunkIter;
+
+/// Re-export of the data-server archive entry points (#81): a real-data source
+/// builds from `aura-ingest` alone — a consumer never names the external
+/// `data_server` crate directly.
+pub use data_server::{DataServer, DEFAULT_DATA_PATH};
 use std::sync::Arc;

 /// Normalize data-server's Unix-millisecond time to aura's canonical epoch-ns
@@ -33,6 +38,14 @@ pub fn unix_ms_to_epoch_ns(time_ms: i64) -> Timestamp {
    Timestamp(time_ms * 1_000_000)
 }

+/// Inverse of [`unix_ms_to_epoch_ns`]: project aura's canonical epoch-ns
+/// [`Timestamp`] back to data-server's Unix-millisecond time. **Private** — the
+/// seam owns the ms↔ns convention (C3); a consumer threads `Timestamp`s and never
+/// converts. Floor division by 1e6 ns/ms (archived M1 data has no sub-ms instant).
+fn epoch_ns_to_unix_ms(ts: Timestamp) -> i64 {
+    ts.0 / 1_000_000
+}
+
 /// One M1 window transposed Array-of-Structs → Structure-of-Arrays (C7): the
 /// OHLCV bar as a bundle of base columns, time already normalized to epoch-ns.
 /// `volume` is the one `i64` column; the price/spread columns are `f64`. All
@@ -216,6 +229,28 @@ impl M1FieldSource {
        Some(s)
    }

+    /// Open over the `[from, to]` window in aura's native epoch-ns [`Timestamp`]
+    /// currency — the engine-side mirror of [`open`](Self::open), which takes
+    /// data-server's Unix-ms. Each bound is mapped through the seam-private
+    /// `epoch_ns_to_unix_ms` and delegated to `open`, so the ms↔ns crossing
+    /// happens once, here, and a consumer never divides. `None` bounds preserve
+    /// open-ended windows exactly as `open` does.
+    pub fn open_window(
+        server: &Arc<DataServer>,
+        symbol: &str,
+        from: Option<Timestamp>,
+        to: Option<Timestamp>,
+        field: M1Field,
+    ) -> Option<Self> {
+        Self::open(
+            server,
+            symbol,
+            from.map(epoch_ns_to_unix_ms),
+            to.map(epoch_ns_to_unix_ms),
+            field,
+        )
+    }
+
    /// Decode the head at the cursor, refilling chunks as needed. Sets
    /// `self.head = None` at exhaustion.
    fn advance(&mut self) {
@@ -271,6 +306,39 @@ impl aura_engine::Source for M1FieldSource {
    }
 }

+/// Open the four real OHLC [`M1FieldSource`]s for `symbol` over the closed
+/// epoch-ns window `[from, to]`, in the FIXED order open, high, low, close — the
+/// C4 merge tie-break order a resampler's `Barrier(0)` group depends on (#92).
+/// This is the single vetted home of that order; consumers never spell it out.
+/// All four sources share the one `Arc<DataServer>` (one `FileCache`), so a
+/// window's bars are parsed once and reused across the four field decodes, and
+/// the same `Arc` flows across the disjoint sims of a sweep / walk-forward (C12).
+///
+/// Returns `None` if any field has no archived file overlapping the window (the
+/// caller skips cleanly), propagating each [`open_window`](M1FieldSource::open_window)'s
+/// file-level `Option`. A window that overlaps a file but holds zero bars yields
+/// four sources whose first `peek` is `None`, not a `None` here.
+pub fn open_ohlc(
+    server: &Arc<DataServer>,
+    symbol: &str,
+    from: Timestamp,
+    to: Timestamp,
+) -> Option<Vec<Box<dyn aura_engine::Source>>> {
+    let open = M1FieldSource::open_window(server, symbol, Some(from), Some(to), M1Field::Open)?;
+    let high = M1FieldSource::open_window(server, symbol, Some(from), Some(to), M1Field::High)?;
+    let low = M1FieldSource::open_window(server, symbol, Some(from), Some(to), M1Field::Low)?;
+    let close = M1FieldSource::open_window(server, symbol, Some(from), Some(to), M1Field::Close)?;
+    Some(vec![Box::new(open), Box::new(high), Box::new(low), Box::new(close)])
+}
+
+/// Construct the default data-server over the local archive at
+/// [`DEFAULT_DATA_PATH`], wrapped in the `Arc` the streaming sources share (#81).
+/// Build it once and clone the `Arc` across a family's sims — one cache, never
+/// one server per field (C12).
+pub fn default_data_server() -> Arc<DataServer> {
+    Arc::new(DataServer::new(DEFAULT_DATA_PATH))
+}
+
 #[cfg(test)]
 mod tests {
    use super::*;
@@ -286,6 +354,16 @@ mod tests {
        assert_eq!(unix_ms_to_epoch_ns(0), Timestamp(0));
    }

+    #[test]
+    fn epoch_ns_to_unix_ms_inverts_unix_ms_to_epoch_ns() {
+        // The seam-owned inverse round-trips the forward normalization for every
+        // ms instant, so a Timestamp window bound fed back to data-server's ms
+        // contract recovers the exact ms (C3: one currency crossing, owned here).
+        for ms in [0_i64, 1, 1_488_326_400_000, 1_727_000_000_000] {
+            assert_eq!(epoch_ns_to_unix_ms(unix_ms_to_epoch_ns(ms)), ms);
+        }
+    }
+
    /// A hand-built M1 bar with distinct per-field values so a transpose test
    /// can tell the columns apart.
    fn full_bar(time_ms: i64) -> M1Parsed {
@@ -16,9 +16,9 @@

 use std::sync::Arc;

-use aura_core::Scalar;
+use aura_core::{Scalar, Timestamp};
 use chrono_tz::Europe::Berlin;
-use data_server::{DataServer, DEFAULT_DATA_PATH};
+use aura_ingest::{default_data_server, open_ohlc, DataServer, DEFAULT_DATA_PATH};

 #[path = "../examples/shared/breakout_real.rs"]
 mod breakout_real;
@@ -69,10 +69,11 @@ fn blueprint_param_space_is_construction_deterministic() {
    assert_eq!(a, b, "blueprint param_space() is a deterministic function of its args");
 }

-/// Run the blueprint-bootstrapped harness over `[from_ms, to_ms]` and drain the
-/// recorded per-bar `(held, equity)` series. Fresh build per call (fresh nodes +
-/// channels) keeps two runs disjoint for the C1 determinism assertion.
-fn run_blueprint(server: &Arc<DataServer>, from_ms: i64, to_ms: i64) -> Vec<(f64, f64)> {
+/// Run the blueprint-bootstrapped harness over the epoch-ns [`Timestamp`] window
+/// `[from, to]` and drain the recorded per-bar `(held, equity)` series. Fresh
+/// build per call (fresh nodes + channels) keeps two runs disjoint for the C1
+/// determinism assertion.
+fn run_blueprint(server: &Arc<DataServer>, from: Timestamp, to: Timestamp) -> Vec<(f64, f64)> {
    let (bp, taps) = ger40_breakout_blueprint(BAR_MINUTES, SESSION_HOUR, SESSION_MINUTE, Berlin);
    let mut h = bp
        .with("entry_bar.target", Scalar::i64(3))
@@ -80,7 +81,7 @@ fn run_blueprint(server: &Arc<DataServer>, from_ms: i64, to_ms: i64) -> Vec<(f64
        .bootstrap()
        .expect("blueprint bootstraps with entry=3, exit=5");
    let sources =
-        open_ohlc_sources(server, from_ms, to_ms).expect("GER40 has data in the Sept-2024 window");
+        open_ohlc(server, SYMBOL, from, to).expect("GER40 has data in the Sept-2024 window");
    h.run(sources);
    let trace = drain_trace(&taps);
    trace.iter().map(|b| (b.held, b.equity)).collect()
@@ -88,10 +89,10 @@ fn run_blueprint(server: &Arc<DataServer>, from_ms: i64, to_ms: i64) -> Vec<(f64

 /// Run the shipped hand-wired `FlatGraph` over the same window — the C23
 /// reference series.
-fn run_flatgraph(server: &Arc<DataServer>, from_ms: i64, to_ms: i64) -> Vec<(f64, f64)> {
+fn run_flatgraph(server: &Arc<DataServer>, from: Timestamp, to: Timestamp) -> Vec<(f64, f64)> {
    let (mut h, taps) = build_harness();
    let sources =
-        open_ohlc_sources(server, from_ms, to_ms).expect("GER40 has data in the Sept-2024 window");
+        open_ohlc(server, SYMBOL, from, to).expect("GER40 has data in the Sept-2024 window");
    h.run(sources);
    let trace = drain_trace(&taps);
    trace.iter().map(|b| (b.held, b.equity)).collect()
@@ -108,26 +109,26 @@ fn run_flatgraph(server: &Arc<DataServer>, from_ms: i64, to_ms: i64) -> Vec<(f64
 /// disjoint rerun (C1).
 #[test]
 fn composite_matches_flatgraph_bit_identical() {
-    let server = Arc::new(DataServer::new(DEFAULT_DATA_PATH));
+    let server = default_data_server();
    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 (from, to) = utc_month_window(YEAR, MONTH);

-    let blueprint_series = run_blueprint(&server, from_ms, to_ms);
+    let blueprint_series = run_blueprint(&server, from, to);
    assert!(!blueprint_series.is_empty(), "window resolved to zero bars");

    // C23: the Composite blueprint reproduces the hand-wired FlatGraph EXACTLY.
-    let flatgraph_series = run_flatgraph(&server, from_ms, to_ms);
+    let flatgraph_series = run_flatgraph(&server, from, to);
    assert_eq!(
        blueprint_series, flatgraph_series,
        "blueprint must reproduce the hand-wired FlatGraph bit-identically (C23)",
    );

    // C1: a fresh disjoint blueprint run over the identical window is bit-identical.
-    let blueprint_rerun = run_blueprint(&server, from_ms, to_ms);
+    let blueprint_rerun = run_blueprint(&server, from, to);
    assert_eq!(
        blueprint_series, blueprint_rerun,
        "blueprint recorded (held, equity) series bit-identical across runs (C1)",
@@ -12,7 +12,8 @@

 use std::sync::Arc;

-use data_server::{DataServer, DEFAULT_DATA_PATH};
+use aura_core::Timestamp;
+use aura_ingest::{default_data_server, open_ohlc, DataServer, DEFAULT_DATA_PATH};

 #[path = "../examples/shared/breakout_real.rs"]
 mod breakout_real;
@@ -22,16 +23,17 @@ use breakout_real::*;
 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).
+/// Build the real breakout harness, run it over the epoch-ns [`Timestamp`]
+/// window `[from, to]`, 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,
+    from: Timestamp,
+    to: Timestamp,
 ) -> (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");
+        open_ohlc(server, SYMBOL, from, to).expect("GER40 has data in the Sept-2024 window");
    let (mut h, taps) = build_harness();
    h.run(sources);

@@ -39,7 +41,7 @@ fn run_real(
    // 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 report = report_from_trace(&trace, from, to);
    let series: Vec<(f64, f64)> = trace.iter().map(|b| (b.held, b.equity)).collect();
    (report, series)
 }
@@ -56,15 +58,15 @@ fn run_real(
 /// 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));
+    let server = default_data_server();
    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 (from, to) = utc_month_window(YEAR, MONTH);

-    let (r1, s1) = run_real(&server, from_ms, to_ms);
+    let (r1, s1) = run_real(&server, from, to);
    assert!(!s1.is_empty(), "window resolved to zero bars");

    // Metrics are finite — a backtest actually ran over real bars.
@@ -82,7 +84,7 @@ fn ger40_breakout_real_bars_run_is_deterministic() {

    // 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);
+    let (r2, s2) = run_real(&server, from, to);
    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)");
 }
@@ -27,31 +27,24 @@ use aura_engine::{
 };
 use chrono::TimeZone;
 use chrono_tz::Europe::Berlin;
-use data_server::{DataServer, DEFAULT_DATA_PATH};
+use aura_ingest::{default_data_server, open_ohlc, DataServer, DEFAULT_DATA_PATH};

 #[path = "../examples/shared/breakout_real.rs"]
 mod breakout_real;
 use breakout_real::*;

-/// ns → ms: the shipped `open_ohlc_sources` opens by Unix-ms, but the World
-/// `WindowBounds` are epoch-ns. `aura-ingest` exports only the forward direction
-/// (`unix_ms_to_epoch_ns`), so the inverse is a local hand-divide (#80, filed).
-fn ns_to_ms(ts: Timestamp) -> i64 {
-    ts.0 / 1_000_000
-}
-
 /// Bootstrap the shipped blueprint at one `{entry_bar, exit_bar}` point, run it
-/// over `[from_ms, to_ms]` of real GER40 OHLC, and fold the recorded equity/held
-/// taps into a `RunReport` (`summarize` over the drained channels — exactly as
-/// the SMA-cross sweep fixture and the fieldtest's `run_point` do). A fresh
-/// blueprint per call (fresh nodes + channels) keeps the disjoint sweep points
-/// independent (C1).
-fn run_point(server: &Arc<DataServer>, point: &[Cell], from_ms: i64, to_ms: i64) -> RunReport {
+/// over the epoch-ns [`Timestamp`] window `[from, to]` of real GER40 OHLC, and
+/// fold the recorded equity/held taps into a `RunReport` (`summarize` over the
+/// drained channels — exactly as the SMA-cross sweep fixture and the fieldtest's
+/// `run_point` do). A fresh blueprint per call (fresh nodes + channels) keeps the
+/// disjoint sweep points independent (C1).
+fn run_point(server: &Arc<DataServer>, point: &[Cell], from: Timestamp, to: Timestamp) -> RunReport {
    let (bp, taps) = ger40_breakout_blueprint(BAR_MINUTES, SESSION_HOUR, SESSION_MINUTE, Berlin);
    let mut h = bp
        .bootstrap_with_cells(point)
        .expect("sweep point kind-checked against param_space");
-    let sources = open_ohlc_sources(server, from_ms, to_ms).expect("window overlaps GER40 data");
+    let sources = open_ohlc(server, SYMBOL, from, to).expect("window overlaps GER40 data");
    h.run(sources);
    drop(h);
    let equity: Vec<(Timestamp, f64)> = taps.equity.try_iter().collect::<Vec<_>>()
@@ -66,10 +59,7 @@ fn run_point(server: &Arc<DataServer>, point: &[Cell], from_ms: i64, to_ms: i64)
        manifest: RunManifest {
            commit: "ger40-breakout-world".to_string(),
            params: vec![],
-            window: (
-                aura_ingest::unix_ms_to_epoch_ns(from_ms),
-                aura_ingest::unix_ms_to_epoch_ns(to_ms),
-            ),
+            window: (from, to),
            seed: 0,
            broker: "sim-optimal(pip_size=1)".to_string(),
        },
@@ -91,13 +81,13 @@ fn scalar_f64(s: &Scalar) -> f64 {
 /// `param_space()` directly, so the two `EqConst` targets are the swept axes.
 #[test]
 fn sweep_consumes_blueprint_over_named_params() {
-    let server = Arc::new(DataServer::new(DEFAULT_DATA_PATH));
+    let server = default_data_server();
    if !server.has_symbol(SYMBOL) {
        eprintln!("skip: no local data at {DEFAULT_DATA_PATH} (symbol {SYMBOL} absent)");
        return;
    }

-    let (from_ms, to_ms) = utc_month_window_ms(2024, 9);
+    let (from, to) = utc_month_window(2024, 9);
    let space = ger40_breakout_blueprint(BAR_MINUTES, SESSION_HOUR, SESSION_MINUTE, Berlin)
        .0
        .param_space();
@@ -117,7 +107,7 @@ fn sweep_consumes_blueprint_over_named_params() {

    let server_for_closure = Arc::clone(&server);
    let family = sweep(&grid, |pt: &[Cell]| {
-        run_point(&server_for_closure, pt, from_ms, to_ms)
+        run_point(&server_for_closure, pt, from, to)
    });

    // A full 3×3 grid was enumerated over the two named params (no re-authoring).
@@ -152,7 +142,7 @@ fn sweep_consumes_blueprint_over_named_params() {
 /// over the raw FlatGraph, where `chosen_params == vec![]` everywhere).
 #[test]
 fn walk_forward_consumes_blueprint_non_degenerate() {
-    let server = Arc::new(DataServer::new(DEFAULT_DATA_PATH));
+    let server = default_data_server();
    if !server.has_symbol(SYMBOL) {
        eprintln!("skip: no local data at {DEFAULT_DATA_PATH} (symbol {SYMBOL} absent)");
        return;
@@ -201,9 +191,8 @@ fn walk_forward_consumes_blueprint_non_degenerate() {
                .collect(),
        )
        .expect("IS grid well-formed");
-        let (is_from, is_to) = (ns_to_ms(w.is.0), ns_to_ms(w.is.1));
        let is_family = sweep(&is_grid, |pt: &[Cell]| {
-            run_point(&server_for_closure, pt, is_from, is_to)
+            run_point(&server_for_closure, pt, w.is.0, w.is.1)
        });
        let best = is_family
            .points
@@ -219,13 +208,12 @@ fn walk_forward_consumes_blueprint_non_degenerate() {
        let chosen = best.params.clone();

        // Apply the chosen params on the OOS window.
-        let (oos_from, oos_to) = (ns_to_ms(w.oos.0), ns_to_ms(w.oos.1));
-        let oos_report = run_point(&server_for_closure, &chosen, oos_from, oos_to);
+        let oos_report = run_point(&server_for_closure, &chosen, w.oos.0, w.oos.1);
        // Re-run once on OOS to capture the recorded equity segment for stitching.
        let (bp, taps) =
            ger40_breakout_blueprint(BAR_MINUTES, SESSION_HOUR, SESSION_MINUTE, Berlin);
        let mut h = bp.bootstrap_with_cells(&chosen).expect("chosen point kind-checked");
-        let sources = open_ohlc_sources(&server_for_closure, oos_from, oos_to)
+        let sources = open_ohlc(&server_for_closure, SYMBOL, w.oos.0, w.oos.1)
            .expect("OOS window overlaps data");
        h.run(sources);
        drop(h);
@@ -0,0 +1,102 @@
+//! Hermetic coverage for the Task-1 OHLC source-open surface (#80/#81/#92): the
+//! documented *file-level* `None` contract of the canonical openers, and the #81
+//! re-export firewall (a real-data source builds from `aura-ingest` alone). Every
+//! item here is reached through `aura_ingest` ONLY — this test never names the
+//! external `data_server` crate, which is the firewall it pins.
+//!
+//! Unlike the gated GER40 tests (which skip without a local Pepperstone archive),
+//! these run everywhere: they drive a `DataServer` over a guaranteed-EMPTY base
+//! path, so no archived file overlaps any window and the openers take their
+//! documented "no data source present" branch deterministically — same input,
+//! same `None`, every run, on any machine.
+
+use std::sync::Arc;
+
+// Firewall (#81): the ENTIRE source-open stack is imported from `aura_ingest`.
+// `DataServer` / `DEFAULT_DATA_PATH` are re-exports; if a future change drops
+// the re-export, this `use` stops compiling — that is the test.
+use aura_core::Timestamp;
+use aura_ingest::{
+    default_data_server, open_ohlc, DataServer, M1Field, M1FieldSource, DEFAULT_DATA_PATH,
+};
+
+/// A `DataServer` over a base path that is guaranteed not to exist, so its symbol
+/// index is empty on every machine (data-server's directory scan returns `Err`
+/// for an absent path and yields no symbols). Gives a deterministic "no archive"
+/// server with no filesystem writes and no dependence on local data.
+fn empty_server() -> Arc<DataServer> {
+    // A path component that cannot exist as a real archive directory; uniqueness
+    // is irrelevant (an absent path scans empty regardless), but keep it obviously
+    // synthetic so an accidental match is impossible.
+    let absent = std::env::temp_dir().join("aura-ingest-no-such-archive-DO-NOT-CREATE");
+    Arc::new(DataServer::new(absent))
+}
+
+/// Property: `open_ohlc` returns the *file-level* `None` (the "no data source to
+/// read from" branch) when no archived file overlaps the window — distinct from
+/// the "overlaps a file but holds zero bars" case, which would return
+/// `Some(empty)`. Pins the documented Option semantics of the canonical OHLC
+/// bundle opener (#80/#92) so a future refactor cannot silently turn "no source"
+/// into a panic or an empty `Some`.
+#[test]
+fn open_ohlc_yields_file_level_none_when_no_archive_overlaps() {
+    let server = empty_server();
+    let from = Timestamp(0);
+    let to = Timestamp(1_000_000_000_000_000_000); // ~2001, well-formed epoch-ns window
+    assert!(
+        open_ohlc(&server, "GER40", from, to).is_none(),
+        "open_ohlc over an empty archive must return the file-level None (no source present)",
+    );
+}
+
+/// Property: `M1FieldSource::open_window` — the engine-side epoch-ns `Timestamp`
+/// mirror of the ms-path `open` — threads its `Timestamp` bounds through the
+/// seam-private ms<->ns crossing and propagates data-server's file-level `None`
+/// for an absent symbol, without panicking on the ns->ms divide. Pins that the
+/// Timestamp opener is a faithful, total mirror of `open` on the "no source"
+/// branch (the C3 one-crossing seam, #80).
+#[test]
+fn open_window_propagates_file_level_none_for_absent_symbol() {
+    let server = empty_server();
+    let from = Timestamp(0);
+    let to = Timestamp(1_000_000_000_000_000_000);
+    assert!(
+        M1FieldSource::open_window(&server, "GER40", Some(from), Some(to), M1Field::Close)
+            .is_none(),
+        "open_window over an empty archive must propagate the file-level None",
+    );
+    // An open-ended window (None bounds) takes the same no-source branch — the
+    // ns->ms mapping is a no-op on None, so the openness is preserved.
+    assert!(
+        M1FieldSource::open_window(&server, "GER40", None, None, M1Field::Close).is_none(),
+        "open-ended open_window over an empty archive must also propagate None",
+    );
+}
+
+/// Property: the #81 source-open surface is reachable from `aura_ingest` alone —
+/// `default_data_server()` constructs the canonical-path server and the
+/// re-exported `DEFAULT_DATA_PATH` is the path it points at, with the external
+/// `data_server` crate never named by a consumer. Observable: the convenience
+/// constructor agrees with constructing `DataServer::new(DEFAULT_DATA_PATH)` by
+/// hand on the symbol-presence query (both report the same absence of a synthetic
+/// symbol that no real archive can hold), so the convenience is a faithful alias.
+#[test]
+fn default_data_server_builds_from_ingest_alone_over_canonical_path() {
+    let convenience = default_data_server();
+    // The re-export is reachable and is what the convenience constructor uses.
+    let by_hand = Arc::new(DataServer::new(DEFAULT_DATA_PATH));
+    // A symbol no real Pepperstone archive holds: both servers agree it is absent,
+    // independent of whether the local archive exists — so the convenience
+    // constructor is a faithful alias for `DataServer::new(DEFAULT_DATA_PATH)`,
+    // and the whole stack was built without naming `data_server`.
+    const SYNTHETIC_ABSENT: &str = "AURA-INGEST-SYNTHETIC-NONEXISTENT-SYMBOL";
+    assert_eq!(
+        convenience.has_symbol(SYNTHETIC_ABSENT),
+        by_hand.has_symbol(SYNTHETIC_ABSENT),
+        "default_data_server() must agree with DataServer::new(DEFAULT_DATA_PATH)",
+    );
+    assert!(
+        !convenience.has_symbol(SYNTHETIC_ABSENT),
+        "a synthetic symbol no archive holds must be absent",
+    );
+}
@@ -0,0 +1,71 @@
+//! Gated behaviour-preservation proof for the consolidated OHLC opener (#80/#92):
+//! the library `open_ohlc` over an epoch-ns `Timestamp` window produces a recorded
+//! `(held, equity)` series BIT-IDENTICAL to opening the four `M1FieldSource`s by
+//! hand over the equivalent Unix-ms window via the untouched ms-path `open`. Same
+//! instants, same bars, same numbers — the migration changes the surface, not the
+//! behaviour.
+//!
+//! Mirrors the other gated GER40 tests: skips with a note where the local
+//! Pepperstone archive is absent, so `cargo test --workspace` stays green
+//! anywhere; exercises the real path where the files exist.
+
+use std::sync::Arc;
+
+use aura_engine::{Harness, Source};
+use chrono::TimeZone;
+use data_server::{DataServer, DEFAULT_DATA_PATH};
+
+use aura_ingest::{open_ohlc, M1Field, M1FieldSource};
+
+#[path = "../examples/shared/breakout_real.rs"]
+mod breakout_real;
+use breakout_real::*;
+
+/// Run a fresh harness over `sources` and drain the recorded `(held, equity)`
+/// series (the same fold the determinism tests use).
+fn series_from(mut h: Harness, taps: &Taps, sources: Vec<Box<dyn Source>>) -> Vec<(f64, f64)> {
+    h.run(sources);
+    drop(h);
+    let trace = drain_trace(taps);
+    trace.iter().map(|b| (b.held, b.equity)).collect()
+}
+
+/// Property: the Timestamp `open_ohlc` path and the hand-rolled ms-path `open`
+/// (×4, in O/H/L/C order) over the equivalent window yield bit-identical recorded
+/// series — behaviour-preserving consolidation (AC5).
+#[test]
+fn open_ohlc_timestamp_path_matches_ms_path_open() {
+    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;
+    }
+
+    // September 2024 (inclusive UTC), the window the other gated tests pin.
+    let from_ms = chrono::Utc.with_ymd_and_hms(2024, 9, 1, 0, 0, 0).unwrap().timestamp_millis();
+    let to_ms = chrono::Utc.with_ymd_and_hms(2024, 9, 30, 23, 59, 59).unwrap().timestamp_millis();
+    let from_ts = aura_ingest::unix_ms_to_epoch_ns(from_ms);
+    let to_ts = aura_ingest::unix_ms_to_epoch_ns(to_ms);
+
+    // Path A: the consolidated Timestamp opener.
+    let (h_a, taps_a) = build_harness();
+    let sources_a = open_ohlc(&server, SYMBOL, from_ts, to_ts).expect("window overlaps GER40 data");
+    let series_a = series_from(h_a, &taps_a, sources_a);
+    assert!(!series_a.is_empty(), "window resolved to zero bars");
+
+    // Path B: the untouched ms-path open, four fields by hand in O/H/L/C order.
+    let fields = [M1Field::Open, M1Field::High, M1Field::Low, M1Field::Close];
+    let mut sources_b: Vec<Box<dyn Source>> = Vec::with_capacity(4);
+    for &f in &fields {
+        let s = M1FieldSource::open(&server, SYMBOL, Some(from_ms), Some(to_ms), f)
+            .expect("window overlaps GER40 data");
+        sources_b.push(Box::new(s));
+    }
+    let (h_b, taps_b) = build_harness();
+    let series_b = series_from(h_b, &taps_b, sources_b);
+
+    assert_eq!(
+        series_a, series_b,
+        "open_ohlc (Timestamp) must reproduce the ms-path open bit-identically (behaviour-preserving, #80/#92)",
+    );
+}