Aura/fieldtests/milestone-walking-skeleton/ms_3_epoch_ns_coherence.rs

//! Milestone fieldtest — scenario 3: EPOCH-NS COHERENCE END TO END.
//!
//! Ties C3's one unit normalization (Unix-ms -> canonical epoch-ns) from the
//! ingestion boundary all the way through to the recorded sink trace that the
//! metric stream is folded from. The promise under test: the timestamps a
//! newcomer reads off a recorded trace are aura's canonical epoch-ns, derived
//! once at ingestion and carried unchanged through the run loop.
//!
//! Source records are hand-built `M1Parsed` (data-server's public AoS record —
//! exactly the shape a recorder / live-source edge materializes), with known
//! `time_ms`. We assert:
//!   1. close_stream timestamps == unix_ms_to_epoch_ns(time_ms) for each bar
//!      (C3: the boundary normalizes; ms * 1_000_000).
//!   2. the Recorder sink trace timestamps (captured during the run) equal the
//!      same epoch-ns values — the normalization survives ingest -> run -> sink.
//!
//! No external data needed: this is a pure-construction coherence check.

use std::sync::mpsc;

use aura_core::{Firing, Scalar, ScalarKind, Timestamp};
use aura_engine::{Edge, Harness, SourceSpec, Target};
use aura_ingest::{transpose_m1, unix_ms_to_epoch_ns};
use aura_std::{Recorder, Sma};
use data_server::records::M1Parsed;

fn bar(time_ms: i64, close: f64) -> M1Parsed {
    M1Parsed { time_ms, open: close, high: close, low: close, close, spread: 0.0, volume: 1 }
}

fn main() {
    // Known minute timestamps (arbitrary but realistic Unix-ms, one minute apart).
    let times_ms = [1_154_390_400_000_i64, 1_154_390_460_000, 1_154_390_520_000, 1_154_390_580_000];
    let bars: Vec<M1Parsed> = times_ms
        .iter()
        .enumerate()
        .map(|(i, &t)| bar(t, 1.0 + i as f64))
        .collect();

    // INGEST: AoS -> SoA, normalizing time at the one boundary (C3).
    let cols = transpose_m1(&bars);
    let prices = cols.close_stream();

    // (1) close_stream carries epoch-ns, not Unix-ms.
    let expected_ns: Vec<Timestamp> = times_ms.iter().map(|&t| unix_ms_to_epoch_ns(t)).collect();
    let got_ns: Vec<Timestamp> = prices.iter().map(|p| p.0).collect();
    assert_eq!(got_ns, expected_ns, "close_stream must carry canonical epoch-ns (C3)");
    // sanity: epoch-ns == ms * 1_000_000
    for (&ms, ns) in times_ms.iter().zip(&expected_ns) {
        assert_eq!(ns.0, ms * 1_000_000, "epoch-ns is ms*1e6");
    }
    eprintln!("ingest ts: {:?} (epoch-ns)", got_ns);

    // RUN: price -> SMA(2) -> Recorder. We do not need the full strategy here;
    // a one-node-plus-sink graph is enough to prove the timestamp carries
    // through the run loop into the recorded trace.
    let (tx, rx) = mpsc::channel::<(Timestamp, Vec<Scalar>)>();
    let nodes: Vec<Box<dyn aura_core::Node>> = vec![
        Box::new(Sma::new(2)),                                       // 0
        Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx)), // 1
    ];
    let sources = vec![SourceSpec {
        kind: ScalarKind::F64,
        targets: vec![Target { node: 0, slot: 0 }],
    }];
    let edges = vec![Edge { from: 0, to: 1, slot: 0, from_field: 0 }];

    let mut harness = Harness::bootstrap(nodes, sources, edges).expect("bootstrap");
    harness.run(vec![prices.clone()]);
    drop(harness);

    let recorded: Vec<(Timestamp, Vec<Scalar>)> = rx.iter().collect();
    let recorded_ts: Vec<Timestamp> = recorded.iter().map(|r| r.0).collect();
    eprintln!("sink ts:   {:?} (epoch-ns)", recorded_ts);

    // (2) Every recorded timestamp is one of the ingested epoch-ns stamps
    // (SMA(2) warms after the first bar, so the sink trace is a suffix of the
    // ingested timeline — but every stamp it carries is the canonical epoch-ns).
    assert!(!recorded_ts.is_empty(), "sink recorded at least one warmed cycle");
    for ts in &recorded_ts {
        assert!(
            expected_ns.contains(ts),
            "recorded sink ts {ts:?} is not a canonical ingested epoch-ns stamp"
        );
    }
    // and they are strictly the warmed suffix, in order
    assert_eq!(recorded_ts, expected_ns[1..].to_vec(), "sink trace = warmed suffix, in epoch-ns order");

    println!("SCENARIO 3: PASS (epoch-ns coherent ingest -> run -> sink)");
}