AILang/ail-embed/tests/swarm.rs

//! 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` — STILL `#[ignore]`d. The global
//!   `Σallocs==Σfrees` measurement (ported from
//!   crates/ail/tests/embed_tick_e2e.rs:94-104) was non-deterministic
//!   under the multi-threaded host. The runtime fix
//!   `bugfix-rc-global-stats-race` (commit `7bfa11e`) made the global
//!   `g_rc_*` fallback counters `_Atomic` — that fix is real,
//!   committed, and NECESSARY, but it is **not sufficient**: a Boss
//!   verification (4 swarm runs) showed the leak Σ is still
//!   non-deterministic — `Σfrees` is stable-exact (12000003) every
//!   run while `Σallocs` is short by a jittering ~1600–2260 in ~1/3
//!   of runs (e.g. left:11998383 right:12000003 across the 4 stat
//!   lines). A SECOND, distinct alloc-undercount remains that the
//!   isolated 8×2M pure-global RED did not model. Un-`#[ignore]` is
//!   now gated on that residual's own root-cause/debug cycle, NOT on
//!   `7bfa11e`. The swarm is still *actually* leak-free (no box
//!   crosses a thread — `Ctx: !Send`; `symbol_fan_swarm_bit_exact`
//!   GREEN every run); only the alloc *accounting* is still wrong.
//!   Body preserved verbatim — quarantined, NOT weakened. See
//!   docs/journals/2026-05-19-iter-embedding-abi-m5.2-resume-attempt.md.

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

use data_server::records::DataFormat;
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).
///
/// STILL `#[ignore]`d. The runtime fix `bugfix-rc-global-stats-race`
/// (commit `7bfa11e`) made the global `g_rc_*` fallback counters
/// `_Atomic` — necessary and committed, but Boss verification proved
/// it INSUFFICIENT for this swarm: `Σfrees` is stable-exact
/// (12000003) every run, `Σallocs` short by a jittering ~1600–2260
/// in ~1/3 of runs (e.g. left:11998383 right:12000003, 4 stat
/// lines). A second, distinct alloc-undercount remains — NOT pure
/// global-counter contention (the isolated 8×2M pure-global RED
/// `embed_rc_global_stats_race` is deterministically green). The
/// swarm is still actually leak-free (no box crosses a thread —
/// `Ctx: !Send`; bit-exact GREEN every run); only the alloc
/// accounting is wrong. Un-`#[ignore]` is gated on the residual's
/// own root-cause/debug cycle, not on `7bfa11e`. Body unchanged.
/// See docs/journals/2026-05-19-iter-embedding-abi-m5.2-resume-attempt.md.
#[test]
#[ignore = "still blocked: atomic-global fix 7bfa11e is necessary but \
            INSUFFICIENT — a residual swarm alloc-undercount remains \
            (Σfrees exact, Σallocs short ~1600 jittering, ~1/3 runs; \
            NOT pure global contention — the isolated RED is green). \
            Un-ignore on the residual's own debug cycle, not on \
            7bfa11e. See the m5.2-resume-attempt journal."]
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)"
    );
}