//! Time-shard boundary-invisibility E2E (spec Architecture §3
//! second assertion / Testing strategy §3). One symbol, N disjoint
//! single-month windows, one thread per window (each owning its
//! `Kernel` — `Ctx: !Send`). Asserts, in spec strength order:
//!  (a) each shard's `(acc,n)` == a single-thread host fold of THAT
//!      EXACT window in stream order, BIT-EXACT — the
//!      boundary-invisibility proof (the kernel result for a window
//!      is independent of how data-server chunked it);
//!  (b) the host-summed `(Σacc,Σn)` == the identically-summed
//!      reference partials, bit-exact by construction;
//!  (c) the whole-window single-thread total within an
//!      f64-reassociation tolerance only (cross-shard re-association
//!      is host arithmetic, NOT a kernel property — asserting it
//!      bit-exact would be WRONG, per the spec).
//! Plus the global leak-Σ (deterministic since dbd76e5) and a
//! stderr-only friction timing capture (NO bench gate).
//!
//! Concrete pin: EURUSD, months 2017-03 / 2017-04 / 2017-05 (all
//! present on /mnt, contiguous, no gaps — recon-verified). data-server
//! has no public month-listing API, so the months are hard-pinned.
//! Skips when /mnt absent — data-server's own precedent.

use std::process::Command;
use std::sync::Arc;

use chrono::{NaiveDate, TimeZone, Utc};
use data_server::{DataServer, DEFAULT_DATA_PATH};

const SYMBOL: &str = "EURUSD";
/// Three disjoint single-month windows. Contiguous so the whole-window
/// union is exactly 2017-03-01 .. 2017-05-31 23:59:59.999.
const SHARDS: [(i32, u32); 3] = [(2017, 3), (2017, 4), (2017, 5)];
/// Relative tolerance for the (c) whole-window cross-check. Derivation:
/// recursive f64 summation of n terms has error <= (n-1)*eps*Σ|x_i|.
/// One EURUSD month ~250k ticks => 3 shards ~750k adds of O(1) mids;
/// Σ|x| ~ 8.3e5, (n-1)*eps ~ 8.3e-11 => abs err ~ 6.8e-5, rel ~ 8e-11.
/// 1e-6 is ~4 orders of safety yet still catches any real bug
/// (a genuine miscount is off by >> 1 ppm).
const REL_TOL: f64 = 1e-6;

fn skip_if_no_data() -> bool {
    !std::path::Path::new(DEFAULT_DATA_PATH).exists()
}

/// Inclusive Unix-ms bounds for the whole of calendar month `(y, m)`.
/// `filter_files` windows at month granularity via
/// `unix_ms_to_year_month`, and `filter_chunk` is record-level
/// inclusive `[from, to]`; first-instant .. last-instant of the month
/// selects exactly that month's file and all its records.
fn month_window(y: i32, m: u32) -> (i64, i64) {
    let from = Utc
        .from_utc_datetime(
            &NaiveDate::from_ymd_opt(y, m, 1)
                .unwrap()
                .and_hms_opt(0, 0, 0)
                .unwrap(),
        )
        .timestamp_millis();
    let (ny, nm) = if m == 12 { (y + 1, 1) } else { (y, m + 1) };
    let next_from = Utc
        .from_utc_datetime(
            &NaiveDate::from_ymd_opt(ny, nm, 1)
                .unwrap()
                .and_hms_opt(0, 0, 0)
                .unwrap(),
        )
        .timestamp_millis();
    (from, next_from - 1)
}

/// Single-thread host reference: fold the mid stream over EXACTLY the
/// inclusive `[from_ms, to_ms]` window in data-server stream order
/// (`acc += mid; n += 1`). Bit-exact with `adapter::fold_window`
/// because the f64-add order is identical.
fn reference_window(server: &Arc<DataServer>, from_ms: i64, to_ms: i64) -> (f64, i64) {
    let mut it = server
        .stream_tick_windowed(SYMBOL, Some(from_ms), Some(to_ms))
        .expect("EURUSD has a tick stream in the pinned window");
    let (mut acc, mut n) = (0.0_f64, 0_i64);
    while let Some(chunk) = it.next_chunk() {
        for r in chunk.iter() {
            acc += (r.ask + r.bid) / 2.0;
            n += 1;
        }
    }
    (acc, n)
}

#[test]
fn timeshard_boundary_invisibility_and_leak_free() {
    if skip_if_no_data() {
        eprintln!("skipping: {DEFAULT_DATA_PATH} absent (mirrors data-server)");
        return;
    }

    let windows: Vec<(i64, i64)> =
        SHARDS.iter().map(|&(y, m)| month_window(y, m)).collect();

    // Spawn the time-shard runner: <data_dir> <symbol> then one
    // "from:to" arg per shard. It folds each window on its own thread
    // (own Kernel) and prints `RESULT <idx> <from> <to> <accbits> <n>`
    // + a final `TIMING total_ticks=<N> elapsed_ms=<M>` line.
    let mut cmd = Command::new(env!("CARGO_BIN_EXE_timeshard_runner"));
    cmd.arg(DEFAULT_DATA_PATH).arg(SYMBOL);
    for (f, t) in &windows {
        cmd.arg(format!("{f}:{t}"));
    }
    let out = cmd
        .env("AILANG_RC_STATS", "1")
        .output()
        .expect("spawn timeshard_runner");
    let stdout = String::from_utf8_lossy(&out.stdout);
    let stderr = String::from_utf8_lossy(&out.stderr);
    assert!(
        out.status.success(),
        "timeshard_runner exited {:?}\nstdout:\n{stdout}\nstderr:\n{stderr}",
        out.status.code()
    );

    let server = Arc::new(DataServer::new(DEFAULT_DATA_PATH));

    // Parse RESULT lines into (idx -> (from,to,acc,n)).
    let mut shards: Vec<(usize, i64, i64, f64, i64)> = Vec::new();
    for line in stdout.lines().filter(|l| l.starts_with("RESULT ")) {
        let mut t = line.split_whitespace();
        let _ = t.next(); // RESULT
        let idx: usize = t.next().unwrap().parse().unwrap();
        let from: i64 = t.next().unwrap().parse().unwrap();
        let to: i64 = t.next().unwrap().parse().unwrap();
        let acc = f64::from_bits(
            u64::from_str_radix(t.next().unwrap(), 16).unwrap(),
        );
        let n: i64 = t.next().unwrap().parse().unwrap();
        shards.push((idx, from, to, acc, n));
    }
    shards.sort_by_key(|s| s.0);
    assert_eq!(
        shards.len(),
        SHARDS.len(),
        "expected {} shard RESULT lines, got {}",
        SHARDS.len(),
        shards.len()
    );

    // (a) Per-shard BIT-EXACT vs a single-thread host fold of that
    //     exact window. THE boundary-invisibility proof.
    let mut sum_acc = 0.0_f64;
    let mut sum_n = 0_i64;
    let mut ref_sum_acc = 0.0_f64;
    let mut ref_sum_n = 0_i64;
    for (i, &(idx, from, to, acc, n)) in shards.iter().enumerate() {
        assert_eq!(idx, i, "shard indices contiguous 0..N");
        assert_eq!((from, to), windows[i], "shard {i} window matches pin");
        let (r_acc, r_n) = reference_window(&server, from, to);
        assert_eq!(
            acc.to_bits(),
            r_acc.to_bits(),
            "shard {i} ({from}..{to}): kernel acc BIT-EXACT vs \
             single-thread host fold of that exact window \
             (boundary-invisibility)"
        );
        assert_eq!(n, r_n, "shard {i}: tick count == reference");
        assert!(r_n > 0, "shard {i}: non-empty window");
        sum_acc += acc;
        sum_n += n;
        ref_sum_acc += r_acc;
        ref_sum_n += r_n;
    }

    // (b) Host-summed partials == identically-summed reference
    //     partials, BIT-EXACT by construction (same values, same
    //     host summation order).
    assert_eq!(
        sum_acc.to_bits(),
        ref_sum_acc.to_bits(),
        "Σ shard acc bit-exact vs Σ reference partials"
    );
    assert_eq!(sum_n, ref_sum_n, "Σ shard n == Σ reference n");

    // (c) Whole-window single-thread total within REL_TOL only —
    //     cross-shard re-association is host arithmetic, NOT a kernel
    //     property; asserting bit-exact here would be WRONG (spec §3).
    let whole_from = windows.first().unwrap().0;
    let whole_to = windows.last().unwrap().1;
    let (whole_acc, whole_n) = reference_window(&server, whole_from, whole_to);
    assert_eq!(
        whole_n, sum_n,
        "whole-window tick count == Σ shard n (windows are an exact \
         disjoint partition of the union)"
    );
    let rel = ((whole_acc - sum_acc) / whole_acc).abs();
    assert!(
        rel < REL_TOL,
        "whole-window total {whole_acc} vs Σ partials {sum_acc}: \
         relative diff {rel:e} must be < {REL_TOL:e} (f64 \
         reassociation only — NOT a kernel discrepancy)"
    );

    // Global leak-Σ across ALL ailang_rc_stats: lines (deterministic
    // since dbd76e5: atomic global counters + build.rs runtime-track).
    let mut allocs: u64 = 0;
    let mut frees: u64 = 0;
    let mut seen = 0usize;
    for line in stderr.lines().filter(|l| l.starts_with("ailang_rc_stats:")) {
        seen += 1;
        for tok in line.split_whitespace() {
            if let Some(v) = tok.strip_prefix("allocs=") {
                allocs += v.parse::<u64>().expect("allocs= u64");
            } else if let Some(v) = tok.strip_prefix("frees=") {
                frees += v.parse::<u64>().expect("frees= u64");
            }
        }
    }
    assert!(seen > 0, "no ailang_rc_stats line; stderr:\n{stderr}");
    assert_eq!(
        allocs, frees,
        "globally leak-free: Σallocs==Σfrees across {seen} stat line(s)"
    );

    // Friction harvest: stderr-only observation, NO assertion (the
    // value feeds the P2 batch-crossing decision; it is not a gate).
    if let Some(tl) = stdout.lines().find(|l| l.starts_with("TIMING ")) {
        eprintln!("m5.3 friction-harvest: {tl} (Σn={sum_n})");
    }
}