AILang/docs/plans/embedding-abi-m5.2.md

embedding-abi-m5.2 — data-server adapter + symbol-fan swarm + leak-proof — Implementation Plan

Parent spec: docs/specs/2026-05-19-embedding-abi-m5.md

For agentic workers: REQUIRED SUB-SKILL: use skills/implement to run this plan. Steps use - [ ] checkboxes for tracking.

Goal: Add the data-server adapter layer to ail-embed and the symbol-fan thread-swarm existence proof: real Pepperstone ticks → (ask+bid)/2 mid → the shipped M3 kernel, one symbol per thread with its own ailang_ctx_t, each thread bit-exact vs an independent host reference, the whole run globally leak-free (AILANG_RC_STATS).

Architecture: Builds on shipped m5.1 (204c171: workspace- excluded ail-embed, data-server-free Kernel core). data-server promotes dev-dep → real dep (sanctioned: ail-embed is the sole data-server↔AILang meeting point; Invariant 1 governs the compiler crates, unchanged). A new additive adapter module maps TickParsed→mid and lazily streams chunks into Kernel::run. A swarm_runner bin spawns one thread per symbol (each thread owns its Kernel; Ctx: !Send makes one-ctx-per-thread a compile-time guarantee). An integration test runs that bin as a subprocess under AILANG_RC_STATS=1, asserting per-symbol bit-exactness and Σallocs==Σfrees. Time-shard + friction-harvest are deferred to m5.3.

Tech Stack: Rust (edition 2021); data-server (path ../../libs/data-server) — DataServer, SymbolChunkIter, records::{TickParsed,DataFormat}, DEFAULT_DATA_PATH; std::thread; the M3 staticlib via the m5.1 build.rs.

---

Files this plan creates or modifies:

• Modify: ail-embed/Cargo.toml — move data-server from [dev-dependencies] to [dependencies]; add explicit [[bin]] swarm_runner
• Modify: ail-embed/src/lib.rs:7 — add one line pub mod adapter; (core logic untouched)
• Create: ail-embed/src/adapter.rs — MidPriceStream (lazy Iterator<Item=f64> over SymbolChunkIter) + fold_symbol; the sole data-server-knowing code
• Create: ail-embed/src/bin/swarm_runner.rs — symbol-fan swarm harness; one thread per symbol, prints RESULT <sym> <acc_bits:016x> <n>
• Test: ail-embed/tests/swarm.rs — skip_if_no_data; spawn swarm_runner under AILANG_RC_STATS=1; assert per-symbol bit-exact vs independent reference + Σallocs==Σfrees

Cross-reference anchors (read-only, do not edit):

• crates/ail/tests/embed_tick_e2e.rs:84-104 — the dual-stat-line ailang_rc_stats: Σ-summation discipline to port (attribution-model comment :84-90; parse loop :94-103; seen guard :104)
• crates/ail/tests/embed_swarm_tsan.rs:23-42 — the green C-host per-thread-ctx TSan precedent that ratifies the race-freedom m5.2 reuses (cited, not re-run — see Boss decisions)
• runtime/rc.c:124-138 — ailang_rc_stats_atexit + the __attribute__((constructor)) ailang_rc_stats_install registration (host-language-agnostic; why the leak-proof needs a subprocess)
• ../../libs/data-server/src/lib.rs:166,174,179,236 — DataServer::new / has_symbol / symbols / stream_tick
• ../../libs/data-server/src/records.rs:85-89 — TickParsed{time_ms:i64,ask:f64,bid:f64} (pub fields, Copy); DataFormat::Tick
• ail-embed/src/lib.rs:77-100 — the Kernel API the adapter builds on (Kernel::new, Kernel::run<I:IntoIterator<Item=f64>>)

---

Task 1: Promote data-server to a real dep + the adapter module

Files:

• Modify: ail-embed/Cargo.toml

• Modify: ail-embed/src/lib.rs:7

• Create: ail-embed/src/adapter.rs

• Step 1: Write the failing unit test

Create ail-embed/src/adapter.rs containing only the test module below (the function it calls does not exist yet — RED):

//! The sole `data-server`-knowing layer of `ail-embed`: maps
//! `data_server::records::TickParsed` to the single scalar the M3
//! `Tick` record carries (mid price), and lazily streams
//! `SymbolChunkIter` chunks into the data-server-free `Kernel`
//! (host-side chunk unroll — the adapter owns chunk iteration).

#[cfg(test)]
mod tests {
    use super::tick_to_px;
    use data_server::records::TickParsed;

    /// mid = (ask + bid) / 2. Exact in f64 for these inputs.
    #[test]
    fn tick_to_px_is_mid() {
        let t = TickParsed { time_ms: 0, ask: 2.0, bid: 4.0 };
        assert_eq!(tick_to_px(&t), 3.0);
    }
}

• Step 2: Wire the module + dependency so the test can build

In ail-embed/src/lib.rs, add the module declaration immediately after the use std::ffi::c_void; line (currently src/lib.rs:7). The exact existing line is:

use std::ffi::c_void;

Replace it verbatim with:

use std::ffi::c_void;

/// The sole `data-server`-knowing layer (spec §2). Additive; the
/// embedding core below has zero finance knowledge.
pub mod adapter;

In ail-embed/Cargo.toml, move the data-server line out of [dev-dependencies] into [dependencies]. The exact current block is:

[dependencies]

[dev-dependencies]
data-server = { path = "../../libs/data-server" }
zip = "2"
tempfile = "3"

Replace it verbatim with:

[dependencies]
# The adapter + swarm_runner bin link data-server (ail-embed is the
# sole data-server↔AILang meeting point — Invariant 1 governs the
# *compiler* crates, not this crate). Path is manifest-relative
# (data-server is at /home/brummel/dev/libs/data-server).
data-server = { path = "../../libs/data-server" }

[dev-dependencies]
zip = "2"
tempfile = "3"

[[bin]]
name = "swarm_runner"
path = "src/bin/swarm_runner.rs"

• Step 3: Run the test to verify it fails

Run: cargo test --manifest-path ail-embed/Cargo.toml --lib tick_to_px_is_mid Expected: FAIL — compile error cannot find function \tick_to_px` in this scope(orunresolved import `super::tick_to_px``). The adapter functions are unimplemented.

• Step 4: Write the adapter implementation

Prepend the following to ail-embed/src/adapter.rs, above the #[cfg(test)] mod tests block (keep the module doc comment at the top of the file):

use std::sync::Arc;

use data_server::records::TickParsed;
use data_server::{DataServer, SymbolChunkIter};

use crate::Kernel;

/// Mid price — the single scalar the M3 `Tick` record carries.
#[inline]
pub fn tick_to_px(t: &TickParsed) -> f64 {
    (t.ask + t.bid) / 2.0
}

/// Lazy `Iterator<Item = f64>` over a `data-server` tick stream,
/// yielding the per-tick mid. Pulls `next_chunk` on demand (the
/// adapter owns chunk iteration; chunk boundaries are invisible to
/// the fold) so a whole symbol is never materialised.
pub struct MidPriceStream {
    it: SymbolChunkIter<TickParsed>,
    chunk: Option<Arc<[TickParsed]>>,
    idx: usize,
}

impl MidPriceStream {
    pub fn new(it: SymbolChunkIter<TickParsed>) -> Self {
        Self { it, chunk: None, idx: 0 }
    }
}

impl Iterator for MidPriceStream {
    type Item = f64;
    fn next(&mut self) -> Option<f64> {
        loop {
            if let Some(c) = &self.chunk {
                if self.idx < c.len() {
                    let px = tick_to_px(&c[self.idx]);
                    self.idx += 1;
                    return Some(px);
                }
            }
            match self.it.next_chunk() {
                Some(c) => {
                    self.chunk = Some(c);
                    self.idx = 0;
                }
                None => return None,
            }
        }
    }
}

/// Fold up to `max_ticks` of `symbol`'s mid-price stream through the
/// M3 kernel. `None` if the symbol has no tick stream. Returns
/// `(Σ mid, count)` — the kernel's `acc += px; n += 1` over the
/// bounded prefix, identical-order to a host reference fold.
pub fn fold_symbol(
    server: &Arc<DataServer>,
    symbol: &str,
    max_ticks: usize,
) -> Option<(f64, i64)> {
    let it = server.stream_tick(symbol)?;
    Some(Kernel::new().run(MidPriceStream::new(it).take(max_ticks)))
}

• Step 5: Run the test to verify it passes

Run: cargo test --manifest-path ail-embed/Cargo.toml --lib tick_to_px_is_mid Expected: PASS — test adapter::tests::tick_to_px_is_mid ... ok, 1 passed; 0 failed.

• Step 6: Verify the m5.1 suite still passes (no core regression)

Run: cargo test --manifest-path ail-embed/Cargo.toml Expected: PASS — kernel_run_sums_prices (lib), tick_to_px_is_mid (lib), hermetic_smoke_data_server_roundtrip (smoke) all ok; 0 failed total. The m5.1 core (Kernel, Ctx, box helpers) is unchanged; only pub mod adapter; was added.

---

Task 2: The symbol-fan swarm binary

A separate process so the C-runtime AILANG_RC_STATS atexit line is observable (an in-process test exits after the harness, never seeing it — runtime/rc.c:132-138). One thread per symbol; each thread constructs its own Kernel — Ctx: !Send makes one-ctx-per-thread a compile-time guarantee (a Kernel cannot be moved across a thread boundary).

Files:

• Create: ail-embed/src/bin/swarm_runner.rs

• Step 1: Write ail-embed/src/bin/swarm_runner.rs

//! Symbol-fan swarm harness (spec §3 symbol-fan / Testing §2). Run
//! as a subprocess by `tests/swarm.rs` under `AILANG_RC_STATS=1` so
//! the C-runtime atexit `ailang_rc_stats:` line is observable.
//!
//! argv[1] = data dir (defaults to `data_server::DEFAULT_DATA_PATH`).
//! Enumerates up to `N_SYMBOLS` symbols that have tick files
//! (requires >= 2 for a meaningful swarm), spawns one thread per
//! symbol — each thread owns its `Kernel` (`Ctx: !Send` forces
//! one ctx per thread) — folds at most `MAX_TICKS` mids, and prints
//! one `RESULT <symbol> <acc_bits:016x> <count>` line per symbol to
//! stdout. Exits 0 on success, non-zero (with a message) otherwise.

use std::sync::Arc;

use ail_embed::adapter::fold_symbol;
use data_server::records::DataFormat;
use data_server::{DataServer, DEFAULT_DATA_PATH};

/// Deterministic bounded prefix per symbol (chunk-aligned via
/// `Iterator::take`): enough real volume to be a genuine proof,
/// bounded so the E2E stays ~seconds. m5.3 owns the friction
/// timing; here it is only a runtime bound.
const MAX_TICKS: usize = 2_000_000;
/// Max threads/symbols fanned. Real data has >= 2 tick symbols
/// (EURUSD, XAUUSD); take min(N_SYMBOLS, available).
const N_SYMBOLS: usize = 4;

fn main() {
    let data_dir = std::env::args()
        .nth(1)
        .unwrap_or_else(|| DEFAULT_DATA_PATH.to_string());

    let server = Arc::new(DataServer::new(&data_dir));

    let mut tick_syms: Vec<String> = server
        .symbols()
        .into_iter()
        .filter(|s| {
            server.file_count(s, DataFormat::Tick).unwrap_or(0) > 0
        })
        .map(|s| s.to_string())
        .collect();
    tick_syms.truncate(N_SYMBOLS);

    if tick_syms.len() < 2 {
        eprintln!(
            "swarm_runner: need >= 2 tick symbols under {data_dir}, found {}",
            tick_syms.len()
        );
        std::process::exit(2);
    }

    let handles: Vec<_> = tick_syms
        .into_iter()
        .map(|sym| {
            let srv = Arc::clone(&server);
            // Each thread builds its own Kernel inside the closure.
            // `Kernel`/`Ctx` are `!Send`, so this is the ONLY way
            // this compiles — one ctx per thread, enforced by the
            // type system, not by convention.
            std::thread::spawn(move || {
                let (acc, n) = fold_symbol(&srv, &sym, MAX_TICKS)
                    .expect("symbol has a tick stream");
                (sym, acc, n)
            })
        })
        .collect();

    for h in handles {
        let (sym, acc, n) = h.join().expect("worker thread panicked");
        println!("RESULT {sym} {:016x} {n}", acc.to_bits());
    }
}

• Step 2: Verify the swarm binary compiles

Run: cargo build --manifest-path ail-embed/Cargo.toml --bin swarm_runner Expected: PASS — exit 0, Finished line. That it compiles is the compile-time per-thread-ctx proof: fold_symbol constructs a Kernel (!Send) inside each thread::spawn closure; a shared-ctx design would be error[E0277]: \*mut c_void` cannot be sent between threads safely` and fail here.

---

Task 3: Integration test — symbol-fan leak-free + bit-exact

Adds no production code; spawns the Task 2 bin and asserts the spec acceptance. Skip-if-absent mirrors data-server's own precedent.

Files:

• Create: ail-embed/tests/swarm.rs

• Step 1: Write 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). Asserts, per symbol, the kernel
//! `(acc,n)` is BIT-EXACT vs an independent same-order host
//! reference fold, and globally Σallocs==Σfrees across all
//! `ailang_rc_stats:` lines (the M1/M2/M3 dual-stat-line model,
//! ported from crates/ail/tests/embed_tick_e2e.rs:84-104). Skips
//! when /mnt tick data is absent — mirroring data-server's own
//! tests/data_server.rs::skip_if_no_data() precedent.

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

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

    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);
    let stderr = String::from_utf8_lossy(&out.stderr);
    assert!(
        out.status.success(),
        "swarm_runner exited {:?}\nstdout:\n{stdout}\nstderr:\n{stderr}",
        out.status.code()
    );

    // Per-symbol: kernel (acc,n) bit-exact vs independent reference.
    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 stat lines
    // (the swarm_runner child has exited, so its atexit g_rc line +
    // every per-ctx ailang_ctx_free readback are present). Ported
    // from embed_tick_e2e.rs:94-104; the invariant is the global Σ,
    // not per-line balance (M2 TLS-ctx cross-attribution).
    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)"
    );
}

• Step 2: Run the swarm E2E

Run: cargo test --manifest-path ail-embed/Cargo.toml --test swarm symbol_fan_swarm_leak_free_and_bit_exact -- --nocapture Expected (real data present on this machine): PASS — test symbol_fan_swarm_leak_free_and_bit_exact ... ok, 1 passed; 0 failed. (~seconds: a few million per-tick FFI calls × N symbols.) If /mnt/tickdata/Pepperstone were absent it would print the skip line and still pass — but it IS present here, so it must actually run and assert.

• Step 3: Full crate suite green

Run: cargo test --manifest-path ail-embed/Cargo.toml Expected: PASS — lib (kernel_run_sums_prices, adapter::tests::tick_to_px_is_mid), smoke (hermetic_smoke_data_server_roundtrip), swarm (symbol_fan_swarm_leak_free_and_bit_exact) all ok; 0 failed across all four test binaries.

• Step 4: Invariant 1 — AILang workspace graph still data-server-free

Run: cargo metadata --format-version 1 --manifest-path Cargo.toml --no-deps 2>/dev/null | grep -c data-server | cat Expected: 0 — promoting data-server to a real dep of ail-embed does NOT enter the AILang workspace graph (ail-embed is its own workspace root via the empty [workspace] table). Invariant 1 holds.

• Step 5: No compiler-surface diff

Run: git status --porcelain | grep -E '^\?\?|^ ?M' | grep -vE '^.. (ail-embed/|docs/|bench/orchestrator-stats/)' Expected: empty output — m5.2 touches only ail-embed/**, docs/, and the stats file. Zero diff to crates/ailang-*, crates/ail/, runtime/, examples/*.ail, and (unlike m5.1) not even the root Cargo.toml. The no-language-change guarantee.

---

Boss decisions (recorded for the audit)

• TSan: no new harness; the property is compile-time-enforced + already green-tested. The spec's "sanitiser-clean (no data race across per-thread ctxs)" acceptance is discharged by two facts, not a new nightly-Rust-TSan binary: (1) Ctx/Kernel are !Send (raw *mut c_void), so the swarm is structurally forced into one-ctx-per-thread — a shared-ctx design is error[E0277] and cannot compile (Task 2 Step 2 is therefore a compile-time proof); (2) the C-runtime race-freedom of that exact per-thread-ctx pattern is ratified green by crates/ail/tests/embed_swarm_tsan.rs (C host, 8 pthreads, own ctx each, clang -fsanitize=thread, exit-clean). m5.2 introduces no new AILang-side concurrency primitive (data-server's internal cache concurrency is data-server's own concern, not an AILang invariant). Re-proving with -Z sanitizer=thread would add a nightly-toolchain dependency to ail-embed for zero additional AILang coverage — speculative infra by the project's own criterion. The architect may re-examine this at milestone close.
• data-server dev-dep → real dep is sanctioned. Invariant 1 governs the compiler crates (ailang-*, crates/ail, runtime/), which gain nothing. ail-embed is explicitly the sole data-server↔AILang meeting point; the Invariant-1 gate is therefore "AILang workspace graph has zero data-server" (Step 4) + "zero compiler-surface diff" (Step 5), NOT "ail-embed has zero deps" (which only held for m5.1's pre-adapter state).
• Deferred to m5.3 (not this plan): the time-shard boundary-invisibility proof (spec §3 time-shard / Testing §3) and the friction-harvest timing measurement + milestone close-out.

Self-review (planner Step 5)

1. Spec coverage. §2 adapter → Task 1 (adapter.rs: tick_to_px/MidPriceStream/fold_symbol). §3 symbol-fan → Task 2 (swarm_runner). Testing §2 (symbol-fan, skip-if-absent) → Task 3. Leak-freedom (AILANG_RC_STATS Σ) → Task 3 Step 1 stat-parse + Step 2. Sanitiser/per-ctx-race → Boss decisions (compile-time !Send + the cited green embed_swarm_tsan.rs), verified by Task 2 Step 2's compile gate. Invariant 1 → Task 3 Steps 4-5. Time-shard / friction explicitly deferred to m5.3 — not a coverage gap for this iteration.
2. Placeholder scan. No "TBD/TODO/implement later/similar to/add appropriate". Constants MAX_TICKS/N_SYMBOLS are concrete values, duplicated verbatim in bin and test (Task 2/3), not placeholders.
3. Type/name consistency. adapter, tick_to_px, MidPriceStream, MidPriceStream::new, fold_symbol, Kernel, swarm_runner, CARGO_BIN_EXE_swarm_runner, RESULT line format (RESULT <sym> {:016x} <n> written in Task 2 / parsed in Task 3), MAX_TICKS=2_000_000, N_SYMBOLS=4, symbol_fan_swarm_leak_free_and_bit_exact, tick_to_px_is_mid — identical across tasks and verification filters.
4. Step granularity. Each step is one file write or one command; all 2-5 min. Largest paste (Task 3 Step 1) is a single verbatim test file.
5. No commit steps. None. Work stays in the working tree.
6. Pin/replacement substring contiguity. The two verbatim replacements (Task 1 Step 2: the use std::ffi::c_void; line and the [dependencies]/[dev-dependencies] block) are not paired with any contains(...)/grep over their own body. The greps (Task 3 Steps 4-5) assert over cargo metadata/git status output, not over a verbatim text body this plan writes — no contiguity hazard.
7. Compile-gate vs deferred-caller. No signature change to an existing function with callers. The only existing-file edits are additive (pub mod adapter;; Cargo dep move). Task 1 Step 6's crate build gate is satisfied within Task 1 (adapter complete). No later task is required to make an earlier compile gate pass.
8. Verification filter strings resolve. tick_to_px_is_mid, --bin swarm_runner, --test swarm, symbol_fan_swarm_leak_free_and_bit_exact are all defined by this plan in the same tasks that filter on them — resolve by construction. Task 1 Step 6 / Task 3 Step 3 use the unfiltered suite with explicit 0 failed assertions; Steps 4-5 assert an explicit grep -c … == 0 / empty-output — "nothing ran" cannot masquerade as success.