//! Symbol-fan swarm E2E (spec Testing §2). Runs the `swarm_runner`
//! bin as a subprocess under `AILANG_RC_STATS=1` (the C-runtime
//! atexit stat line, runtime/rc.c:124-138, is only observable from a
//! separate process that exits). Skips when /mnt tick data is absent
//! — mirroring data-server's own tests/data_server.rs::skip_if_no_data().
//!
//! Two assertions, deliberately SPLIT into two tests:
//!
//! - `symbol_fan_swarm_bit_exact` — the existence proof. Per symbol,
//!   the kernel `(acc,n)` is BIT-EXACT vs an independent same-order
//!   host reference fold. This is GREEN and live: it proves the
//!   real-data-server → adapter → M3-kernel swarm computes correctly.
//!
//! - `symbol_fan_swarm_leak_free` — LIVE and GREEN. The global
//!   `Σallocs==Σfrees` measurement (ported from
//!   crates/ail/tests/embed_tick_e2e.rs:94-104) was non-deterministic
//!   under the multi-threaded host. Root cause was a build-dependency
//!   staleness gap, NOT a residual runtime defect: `ail-embed/build.rs`
//!   did not declare the `runtime/` C sources as `rerun-if-changed`,
//!   so the atomic-counter runtime fix `bugfix-rc-global-stats-race`
//!   (commit `7bfa11e`) never relinked `libailang_rt.a` and the swarm
//!   kept linking the stale pre-`7bfa11e` non-atomic
//!   `g_rc_alloc_count++` whose race produced the jittering
//!   `Σallocs` undercount (`Σfrees` stable-exact). `build.rs` now
//!   tracks `runtime/` (this iter, `bugfix-swarm-rc-alloc-undercount`);
//!   the fresh atomic archive relinks and the leak Σ balances
//!   deterministically (`Σallocs == Σfrees == 12000003`, every run).
//!   This test is the integration-level acceptance of both `7bfa11e`
//!   and the build-dependency fix. Body preserved verbatim — it was
//!   quarantined, never weakened. See
//!   docs/journals/2026-05-19-iter-bugfix-swarm-rc-alloc-undercount.md.

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

use data_server::{DataServer, DEFAULT_DATA_PATH};

// Must match swarm_runner.rs.
const MAX_TICKS: usize = 2_000_000;
const N_SYMBOLS: usize = 4;

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

/// Independent host reference: same symbols, same order, same cap,
/// pure-Rust fold (`acc += mid; n += 1`). Bit-exact with the kernel
/// fold because the addition order is identical.
fn reference(symbol: &str) -> (f64, i64) {
    let server = Arc::new(DataServer::new(DEFAULT_DATA_PATH));
    let mut it = server.stream_tick(symbol).expect("tick stream");
    let (mut acc, mut n) = (0.0_f64, 0_i64);
    'outer: while let Some(chunk) = it.next_chunk() {
        for r in chunk.iter() {
            if n as usize >= MAX_TICKS {
                break 'outer;
            }
            acc += (r.ask + r.bid) / 2.0;
            n += 1;
        }
    }
    (acc, n)
}

/// Spawn `swarm_runner` under `AILANG_RC_STATS=1`. `None` on the
/// skip-if-absent path (no /mnt data); otherwise `(stdout, stderr)`
/// after asserting a clean child exit.
fn run_swarm_or_skip() -> Option<(String, String)> {
    if skip_if_no_data() {
        eprintln!("skipping: {DEFAULT_DATA_PATH} absent (mirrors data-server)");
        return None;
    }
    let out = Command::new(env!("CARGO_BIN_EXE_swarm_runner"))
        .arg(DEFAULT_DATA_PATH)
        .env("AILANG_RC_STATS", "1")
        .output()
        .expect("spawn swarm_runner");
    let stdout = String::from_utf8_lossy(&out.stdout).into_owned();
    let stderr = String::from_utf8_lossy(&out.stderr).into_owned();
    assert!(
        out.status.success(),
        "swarm_runner exited {:?}\nstdout:\n{stdout}\nstderr:\n{stderr}",
        out.status.code()
    );
    Some((stdout, stderr))
}

/// Existence proof (GREEN, live): each symbol's kernel `(acc,n)` is
/// bit-exact vs an independent same-order host reference fold. Real
/// data-server → adapter → M3-frozen kernel, one symbol per thread.
#[test]
fn symbol_fan_swarm_bit_exact() {
    let Some((stdout, _stderr)) = run_swarm_or_skip() else { return };

    let mut symbols_checked = 0usize;
    for line in stdout.lines().filter(|l| l.starts_with("RESULT ")) {
        let mut t = line.split_whitespace();
        let _ = t.next(); // "RESULT"
        let sym = t.next().expect("symbol field");
        let acc_bits = u64::from_str_radix(
            t.next().expect("acc-bits field"), 16,
        ).expect("acc-bits hex");
        let n: i64 = t.next().expect("n field").parse().expect("n int");

        let kernel_acc = f64::from_bits(acc_bits);
        let (ref_acc, ref_n) = reference(sym);
        assert_eq!(
            kernel_acc.to_bits(), ref_acc.to_bits(),
            "{sym}: kernel acc bit-exact vs same-order host reference"
        );
        assert_eq!(n, ref_n, "{sym}: tick count matches reference");
        assert!(ref_n > 0, "{sym}: non-empty stream");
        symbols_checked += 1;
    }
    assert!(
        (2..=N_SYMBOLS).contains(&symbols_checked),
        "expected 2..={N_SYMBOLS} symbols fanned, got {symbols_checked}"
    );
}

/// Global leak-freedom: Σallocs == Σfrees across ALL `ailang_rc_stats:`
/// lines (the M2 dual-stat-line model, ported from
/// embed_tick_e2e.rs:94-104). The invariant is the global Σ, not
/// per-line balance (M2 TLS-ctx cross-attribution).
///
/// LIVE and GREEN. The non-determinism was a build-dependency
/// staleness gap, not a residual runtime defect: `ail-embed/build.rs`
/// did not track the `runtime/` C sources as `rerun-if-changed`, so
/// the atomic-counter runtime fix `bugfix-rc-global-stats-race`
/// (commit `7bfa11e`) never relinked `libailang_rt.a` — the swarm
/// kept linking the stale pre-`7bfa11e` non-atomic
/// `g_rc_alloc_count++` whose race undercounted `Σallocs` (`Σfrees`
/// stable-exact). Fixed in `build.rs` this iter
/// (`bugfix-swarm-rc-alloc-undercount`): the fresh atomic archive
/// relinks and the leak Σ now balances deterministically
/// (`Σallocs == Σfrees == 12000003`, every run). This test is the
/// integration-level acceptance of both `7bfa11e` and the
/// build-dependency fix. Body unchanged — it was quarantined, never
/// weakened. See
/// docs/journals/2026-05-19-iter-bugfix-swarm-rc-alloc-undercount.md.
#[test]
fn symbol_fan_swarm_leak_free() {
    let Some((_stdout, stderr)) = run_swarm_or_skip() else { return };

    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)"
    );
}