earthsci-ast 0.6.0

//! Fused-group execution: the per-chunk kernels, the monomorphized kernel
//! dispatch tables, and the strip-mined executor that runs one `FusedSpec`
//! (plus its `#[target_feature]` SIMD clones and the per-element
//! measurement arm).
//!
//! The dispatch macros live beside the chunk kernels rather than in
//! `interp` because BOTH executors expand them — the chunked loops here and
//! the unfused instruction arms in `interp` — and a single definition is
//! what makes an op monomorphized in one path impossible to miss in the
//! other.

use super::resolve::resolve_scalar;
use super::*;

// ---------------------------------------------------------------------------
// Step 4: fused-group execution.
// ---------------------------------------------------------------------------

/// Chunk length of the strip-mined fused executor: intermediates live in a
/// small register file (`n_regs × FCHUNK` doubles, L1-resident) instead of
/// round-tripping ~55 KB slab arrays between every op.
pub(super) const FCHUNK: usize = 1024;

/// A micro-op operand resolved for one chunk: a pointer to `c` contiguous
/// values, or a constant broadcast over the chunk.
#[derive(Clone, Copy)]
pub(super) enum MSrc {
    P(*const f64),
    C(f64),
}

/// `dst[k] = f(a[k])` over one chunk. `dst` is a register region; a `P`
/// operand is either a DIFFERENT register region (the allocator never assigns
/// an op's out register to one of its operands) or a slab/state/observed
/// buffer (disjoint from the register file), so slice-based loops are sound —
/// and vectorizable.
#[inline(always)]
pub(super) unsafe fn fch1(dst: *mut f64, c: usize, a: MSrc, f: impl Fn(f64) -> f64 + Copy) {
    unsafe {
        let d = std::slice::from_raw_parts_mut(dst, c);
        match a {
            MSrc::P(pa) => {
                let a = std::slice::from_raw_parts(pa, c);
                for (x, &v) in d.iter_mut().zip(a) {
                    *x = f(v);
                }
            }
            MSrc::C(v) => {
                let y = f(v);
                for x in d.iter_mut() {
                    *x = y;
                }
            }
        }
    }
}

/// `dst[k] = f(a[k], b[k])` over one chunk (see `fch1` for the aliasing
/// argument; the two operands may alias EACH OTHER, which shared slices
/// permit).
#[inline(always)]
pub(super) unsafe fn fch2(
    dst: *mut f64,
    c: usize,
    a: MSrc,
    b: MSrc,
    f: impl Fn(f64, f64) -> f64 + Copy,
) {
    unsafe {
        let d = std::slice::from_raw_parts_mut(dst, c);
        match (a, b) {
            (MSrc::P(pa), MSrc::P(pb)) => {
                let a = std::slice::from_raw_parts(pa, c);
                let b = std::slice::from_raw_parts(pb, c);
                for k in 0..c {
                    *d.get_unchecked_mut(k) = f(*a.get_unchecked(k), *b.get_unchecked(k));
                }
            }
            (MSrc::P(pa), MSrc::C(y)) => {
                let a = std::slice::from_raw_parts(pa, c);
                for (x, &v) in d.iter_mut().zip(a) {
                    *x = f(v, y);
                }
            }
            (MSrc::C(x0), MSrc::P(pb)) => {
                let b = std::slice::from_raw_parts(pb, c);
                for (x, &y) in d.iter_mut().zip(b) {
                    *x = f(x0, y);
                }
            }
            (MSrc::C(x0), MSrc::C(y0)) => {
                let v = f(x0, y0);
                for x in d.iter_mut() {
                    *x = v;
                }
            }
        }
    }
}

/// `dst[k] = g(a[k], b[k], c[k])` over one chunk (the Bin2 superop; see
/// `fch1` for the aliasing argument).
#[inline(always)]
unsafe fn fch3(
    dst: *mut f64,
    n: usize,
    a: MSrc,
    b: MSrc,
    cc: MSrc,
    g: impl Fn(f64, f64, f64) -> f64 + Copy,
) {
    unsafe {
        let d = std::slice::from_raw_parts_mut(dst, n);
        macro_rules! lp {
            ($ax:expr, $bx:expr, $cx:expr) => {
                for k in 0..n {
                    *d.get_unchecked_mut(k) = g($ax(k), $bx(k), $cx(k));
                }
            };
        }
        match (a, b, cc) {
            (MSrc::P(pa), MSrc::P(pb), MSrc::P(pc)) => {
                let (a, b, c) = (
                    std::slice::from_raw_parts(pa, n),
                    std::slice::from_raw_parts(pb, n),
                    std::slice::from_raw_parts(pc, n),
                );
                lp!(
                    |k: usize| *a.get_unchecked(k),
                    |k: usize| *b.get_unchecked(k),
                    |k: usize| *c.get_unchecked(k)
                );
            }
            (MSrc::P(pa), MSrc::P(pb), MSrc::C(z)) => {
                let (a, b) = (
                    std::slice::from_raw_parts(pa, n),
                    std::slice::from_raw_parts(pb, n),
                );
                lp!(
                    |k: usize| *a.get_unchecked(k),
                    |k: usize| *b.get_unchecked(k),
                    |_k: usize| z
                );
            }
            (MSrc::P(pa), MSrc::C(y), MSrc::P(pc)) => {
                let (a, c) = (
                    std::slice::from_raw_parts(pa, n),
                    std::slice::from_raw_parts(pc, n),
                );
                lp!(|k: usize| *a.get_unchecked(k), |_k: usize| y, |k: usize| *c
                    .get_unchecked(k));
            }
            (MSrc::P(pa), MSrc::C(y), MSrc::C(z)) => {
                let a = std::slice::from_raw_parts(pa, n);
                lp!(|k: usize| *a.get_unchecked(k), |_k: usize| y, |_k: usize| z);
            }
            (MSrc::C(x), MSrc::P(pb), MSrc::P(pc)) => {
                let (b, c) = (
                    std::slice::from_raw_parts(pb, n),
                    std::slice::from_raw_parts(pc, n),
                );
                lp!(|_k: usize| x, |k: usize| *b.get_unchecked(k), |k: usize| *c
                    .get_unchecked(k));
            }
            (MSrc::C(x), MSrc::P(pb), MSrc::C(z)) => {
                let b = std::slice::from_raw_parts(pb, n);
                lp!(|_k: usize| x, |k: usize| *b.get_unchecked(k), |_k: usize| z);
            }
            (MSrc::C(x), MSrc::C(y), MSrc::P(pc)) => {
                let c = std::slice::from_raw_parts(pc, n);
                lp!(|_k: usize| x, |_k: usize| y, |k: usize| *c.get_unchecked(k));
            }
            (MSrc::C(x), MSrc::C(y), MSrc::C(z)) => {
                let v = g(x, y, z);
                for w in d.iter_mut() {
                    *w = v;
                }
            }
        }
    }
}

/// `dst[k] = g(a[k], b[k], c[k], d[k])` over one chunk, all operands
/// pointers (the Bin3 superop; scalar/ghost operands were resolved to splat
/// registers by the caller). See `fch1` for the aliasing argument; operands
/// may alias each other, which shared slices permit.
#[inline(always)]
unsafe fn fch4(
    dst: *mut f64,
    n: usize,
    pa: *const f64,
    pb: *const f64,
    pc: *const f64,
    pd: *const f64,
    g: impl Fn(f64, f64, f64, f64) -> f64 + Copy,
) {
    unsafe {
        let o = std::slice::from_raw_parts_mut(dst, n);
        let a = std::slice::from_raw_parts(pa, n);
        let b = std::slice::from_raw_parts(pb, n);
        let c = std::slice::from_raw_parts(pc, n);
        let d = std::slice::from_raw_parts(pd, n);
        for k in 0..n {
            *o.get_unchecked_mut(k) = g(
                *a.get_unchecked(k),
                *b.get_unchecked(k),
                *c.get_unchecked(k),
                *d.get_unchecked(k),
            );
        }
    }
}

/// One chunk (`c` values at flat box offset `at`) of an absorbed scan along
/// rows of `row` elements: the running value restarts at `init` at each
/// row's first element and otherwise continues from `*carry`, which it
/// leaves holding the value after the chunk's last element.
#[inline(always)]
#[allow(clippy::too_many_arguments)]
unsafe fn fch_scan(
    dst: *mut f64,
    c: usize,
    at: usize,
    a: MSrc,
    row: usize,
    init: f64,
    inclusive: bool,
    carry: &mut f64,
    f: impl Fn(f64, f64) -> f64 + Copy,
) {
    let mut acc = *carry;
    let mut k = 0usize;
    while k < c {
        let pos = (at + k) % row;
        if pos == 0 {
            acc = init;
        }
        let end = k + (c - k).min(row - pos);
        // Monomorphized over the operand kind and the scan flavour, so each
        // piece is one plain dependency chain.
        unsafe {
            let d = std::slice::from_raw_parts_mut(dst.add(k), end - k);
            match (a, inclusive) {
                (MSrc::P(p), true) => {
                    for (o, &x) in d
                        .iter_mut()
                        .zip(std::slice::from_raw_parts(p.add(k), end - k))
                    {
                        acc = f(acc, x);
                        *o = acc;
                    }
                }
                (MSrc::P(p), false) => {
                    for (o, &x) in d
                        .iter_mut()
                        .zip(std::slice::from_raw_parts(p.add(k), end - k))
                    {
                        *o = acc;
                        acc = f(acc, x);
                    }
                }
                (MSrc::C(x), true) => {
                    for o in d.iter_mut() {
                        acc = f(acc, x);
                        *o = acc;
                    }
                }
                (MSrc::C(x), false) => {
                    for o in d.iter_mut() {
                        *o = acc;
                        acc = f(acc, x);
                    }
                }
            }
        }
        k = end;
    }
    *carry = acc;
}

/// One chunk of a [`ScanFuse`] (Float64): per element, in order, the scanned
/// value (`pre`'s `op(a, b)`, else `src`), the scan's combine `sop` with the
/// running value as in [`fch_scan`], and the stored value (`post`'s `op` of
/// the scan's value and `other`, the scan's value on the right when `swap`;
/// else the scan's value), and the scan's value again at `keep` when given.
/// `dst` may be the register of an operand of `pre` or `post`: each element
/// is read before it is written.
#[inline(always)]
#[allow(clippy::too_many_arguments)]
unsafe fn fch_scan_fused(
    dst: *mut f64,
    c: usize,
    at: usize,
    row: usize,
    init: f64,
    inclusive: bool,
    carry: &mut f64,
    sop: BinCode,
    pre: Option<(BinCode, MSrc, MSrc)>,
    src: MSrc,
    post: Option<(BinCode, MSrc, bool)>,
    keep: Option<*mut f64>,
) {
    // The pieces between row starts, each one plain loop over operands read
    // as (pointer, stride) — stride 0 for a constant — so the loop has no
    // operand-kind branch.
    #[inline(always)]
    #[allow(clippy::too_many_arguments)]
    unsafe fn pieces(
        c: usize,
        at: usize,
        row: usize,
        init: f64,
        inclusive: bool,
        carry: &mut f64,
        f: impl Fn(f64, f64) -> f64,
        x: impl Fn(usize) -> f64,
        y: impl Fn(f64, usize) -> f64,
        dst: *mut f64,
        keep: Option<*mut f64>,
    ) {
        let mut acc = *carry;
        let mut k = 0usize;
        while k < c {
            let pos = (at + k) % row;
            if pos == 0 {
                acc = init;
            }
            let end = k + (c - k).min(row - pos);
            for j in k..end {
                let v = if inclusive {
                    acc = f(acc, x(j));
                    acc
                } else {
                    let v = acc;
                    acc = f(acc, x(j));
                    v
                };
                if let Some(kp) = keep {
                    unsafe { *kp.add(j) = v };
                }
                unsafe { *dst.add(j) = y(v, j) };
            }
            k = end;
        }
        *carry = acc;
    }
    let mut consts = [0.0f64; 3];
    let mut strided = |m: MSrc, slot: usize| -> (*const f64, usize) {
        match m {
            MSrc::P(p) => (p, 1),
            MSrc::C(v) => {
                consts[slot] = v;
                (std::ptr::null(), 0)
            }
        }
    };
    let (pa, sa, pb, sb) = match pre {
        Some((_, a, b)) => {
            let (pa, sa) = strided(a, 0);
            let (pb, sb) = strided(b, 1);
            (pa, sa, pb, sb)
        }
        None => {
            let (pa, sa) = strided(src, 0);
            (pa, sa, std::ptr::null(), 0)
        }
    };
    let (po, so) = match post {
        Some((_, o, _)) => strided(o, 2),
        None => (std::ptr::null(), 0),
    };
    // A constant reads its slot of `consts`, which now holds every value.
    let kp = consts.as_ptr();
    let fix = |p: *const f64, slot: usize| {
        if p.is_null() {
            unsafe { kp.add(slot) }
        } else {
            p
        }
    };
    let (pa, pb, po) = (fix(pa, 0), fix(pb, 1), fix(po, 2));
    let ld = |p: *const f64, s: usize, j: usize| unsafe { *p.add(j * s) };
    macro_rules! kernel {
        (Add) => {
            |x: f64, y: f64| x + y
        };
        (Sub) => {
            |x: f64, y: f64| x - y
        };
        (Mul) => {
            |x: f64, y: f64| x * y
        };
        (Div) => {
            |x: f64, y: f64| x / y
        };
    }
    macro_rules! with_post {
        ($f:expr, $x:expr) => {
            match post {
                None => unsafe {
                    pieces(
                        c,
                        at,
                        row,
                        init,
                        inclusive,
                        carry,
                        $f,
                        $x,
                        |v, _| v,
                        dst,
                        keep,
                    )
                },
                Some((op, _, swap)) => {
                    macro_rules! go {
                        ($g:tt) => {{
                            let g = kernel!($g);
                            if swap {
                                unsafe {
                                    pieces(
                                        c,
                                        at,
                                        row,
                                        init,
                                        inclusive,
                                        carry,
                                        $f,
                                        $x,
                                        |v, j| g(ld(po, so, j), v),
                                        dst,
                                        keep,
                                    )
                                }
                            } else {
                                unsafe {
                                    pieces(
                                        c,
                                        at,
                                        row,
                                        init,
                                        inclusive,
                                        carry,
                                        $f,
                                        $x,
                                        |v, j| g(v, ld(po, so, j)),
                                        dst,
                                        keep,
                                    )
                                }
                            }
                        }};
                    }
                    match op {
                        BinCode::Add => go!(Add),
                        BinCode::Sub => go!(Sub),
                        BinCode::Mul => go!(Mul),
                        BinCode::Div => go!(Div),
                        other => unreachable!("scan fusion post-op {other:?}"),
                    }
                }
            }
        };
    }
    macro_rules! with_pre {
        ($f:expr) => {
            match pre {
                None => with_post!($f, |j| ld(pa, sa, j)),
                Some((op, _, _)) => {
                    macro_rules! go {
                        ($h:tt) => {{
                            let h = kernel!($h);
                            with_post!($f, |j| h(ld(pa, sa, j), ld(pb, sb, j)))
                        }};
                    }
                    match op {
                        BinCode::Add => go!(Add),
                        BinCode::Sub => go!(Sub),
                        BinCode::Mul => go!(Mul),
                        BinCode::Div => go!(Div),
                        other => unreachable!("scan fusion pre-op {other:?}"),
                    }
                }
            }
        };
    }
    match sop {
        BinCode::Add => with_pre!(kernel!(Add)),
        BinCode::Mul => with_pre!(kernel!(Mul)),
        other => unreachable!("scan fusion scan op {other:?}"),
    }
}

/// One chunk of an absorbed scan whose lanes are `post` elements wide (an
/// axis other than the last): position `at + k` is lane `(at + k) % post` at
/// step `((at + k) / post) % row`; a lane restarts at `init` on step 0 (on
/// the last step, `row - 1`, for a reverse scan, whose group visits its
/// rows from the last) and otherwise continues from its slot in `carry`.
#[inline(always)]
#[allow(clippy::too_many_arguments)]
unsafe fn fch_scan_lanes(
    dst: *mut f64,
    c: usize,
    at: usize,
    a: MSrc,
    row: usize,
    post: usize,
    init: f64,
    inclusive: bool,
    reverse: bool,
    carry: &mut [f64],
    f: impl Fn(f64, f64) -> f64 + Copy,
) {
    let first = if reverse { row - 1 } else { 0 };
    let mut k = 0usize;
    while k < c {
        let q0 = (at + k) % post;
        let step = ((at + k) / post) % row;
        let len = (c - k).min(post - q0);
        let acc = &mut carry[q0..q0 + len];
        if step == first {
            acc.fill(init);
        }
        unsafe {
            let d = std::slice::from_raw_parts_mut(dst.add(k), len);
            for (j, (o, y)) in d.iter_mut().zip(acc.iter_mut()).enumerate() {
                let x = match a {
                    MSrc::P(p) => *p.add(k + j),
                    MSrc::C(v) => v,
                };
                if inclusive {
                    *y = f(*y, x);
                    *o = *y;
                } else {
                    *o = *y;
                    *y = f(*y, x);
                }
            }
        }
        k += len;
    }
}

/// The `vec_select` pick over one chunk.
#[inline(always)]
pub(super) unsafe fn fch_sel(dst: *mut f64, c: usize, cond: MSrc, a: MSrc, b: MSrc) {
    // A constant condition is the filter-gate broadcast: whole-chunk pick.
    if let MSrc::C(cv) = cond {
        let pick = if cv != 0.0 { a } else { b };
        unsafe { fch1(dst, c, pick, |x| x) };
        return;
    }
    let MSrc::P(cp) = cond else { unreachable!() };
    unsafe {
        let d = std::slice::from_raw_parts_mut(dst, c);
        let cs = std::slice::from_raw_parts(cp, c);
        // Monomorphized over the operand kinds so each arm is a branch-free
        // (blendable) loop rather than a per-element kind dispatch.
        match (a, b) {
            (MSrc::P(pa), MSrc::P(pb)) => {
                let a = std::slice::from_raw_parts(pa, c);
                let b = std::slice::from_raw_parts(pb, c);
                for k in 0..c {
                    // Load both, then select VALUES (evaluate-both semantics;
                    // a value select vectorizes to a blend, where a pointer
                    // select stays a scalar cmov+load).
                    let av = *a.get_unchecked(k);
                    let bv = *b.get_unchecked(k);
                    *d.get_unchecked_mut(k) = if *cs.get_unchecked(k) != 0.0 { av } else { bv };
                }
            }
            (MSrc::P(pa), MSrc::C(y)) => {
                let a = std::slice::from_raw_parts(pa, c);
                for k in 0..c {
                    let av = *a.get_unchecked(k);
                    *d.get_unchecked_mut(k) = if *cs.get_unchecked(k) != 0.0 { av } else { y };
                }
            }
            (MSrc::C(x), MSrc::P(pb)) => {
                let b = std::slice::from_raw_parts(pb, c);
                for k in 0..c {
                    let bv = *b.get_unchecked(k);
                    *d.get_unchecked_mut(k) = if *cs.get_unchecked(k) != 0.0 { x } else { bv };
                }
            }
            (MSrc::C(x), MSrc::C(y)) => {
                for k in 0..c {
                    *d.get_unchecked_mut(k) = if *cs.get_unchecked(k) != 0.0 { x } else { y };
                }
            }
        }
    }
}

// ---------------------------------------------------------------------------
// Monomorphized kernel dispatch tables — expanded ONCE, used by BOTH the
// unfused executor (`run_range`, through the strided `ew1`/`ew2` loops) and
// the fused chunk executor (`exec_fused_runs`, through `fch1`/`fch2`), so an
// op given a monomorphized arm in one cannot be missed in the other.
// ---------------------------------------------------------------------------

/// Dispatch a [`BinCode`] to `$apply!(kernel)` with a monomorphized closure
/// for each hot op. Each closure body is the IDENTICAL expression to
/// `binary_kernel_of`'s arm for that op, so this is a pure dispatch hoist —
/// one indirect kernel call per node (unfused) or per chunk (fused) becomes
/// an inlined, vectorizable element loop. The arms mirror `vec_combine`'s,
/// plus the comparison relops feeding `Select` masks; any remaining op falls
/// through to the shared fn-pointer table itself, so semantics have a single
/// source either way.
macro_rules! dispatch_bin_kernel {
    ($op:expr, $apply:ident) => {
        // Under `element_type: "Float32"` the monomorphized closures below are
        // a SECOND, hand-copied definition of the arithmetic that never sees
        // the precision — the archetypal fast path that quietly stays f64. So
        // in that mode dispatch through the shared table instead, which returns
        // the binary32 kernel. The element loop then makes one indirect call
        // per element as it did before the monomorphization; correctness over
        // throughput is the only defensible trade when the answer differs.
        if $crate::precision::is_f32() {
            $apply!(binary_kernel_of(*$op))
        } else {
            match $op {
                BinCode::Add => $apply!(|x, y| x + y),
                BinCode::Sub => $apply!(|x, y| x - y),
                BinCode::Mul => $apply!(|x, y| x * y),
                BinCode::Div => $apply!(|x, y| x / y),
                BinCode::Pow => $apply!(|x: f64, y: f64| x.powf(y)),
                BinCode::Min => $apply!(|x: f64, y: f64| x.min(y)),
                BinCode::Max => $apply!(|x: f64, y: f64| x.max(y)),
                BinCode::Eq => $apply!(|x, y| (x == y) as i32 as f64),
                BinCode::Ne => $apply!(|x, y| (x != y) as i32 as f64),
                BinCode::Lt => $apply!(|x, y| (x < y) as i32 as f64),
                BinCode::Le => $apply!(|x, y| (x <= y) as i32 as f64),
                BinCode::Gt => $apply!(|x, y| (x > y) as i32 as f64),
                BinCode::Ge => $apply!(|x, y| (x >= y) as i32 as f64),
                BinCode::Atan2 => $apply!(|x: f64, y: f64| x.atan2(y)),
                BinCode::And => $apply!(|x: f64, y: f64| (x != 0.0 && y != 0.0) as i32 as f64),
                BinCode::Or => $apply!(|x: f64, y: f64| (x != 0.0 || y != 0.0) as i32 as f64),
                other => $apply!(binary_kernel_of(*other)),
            }
        }
    };
}
pub(super) use dispatch_bin_kernel;

/// Dispatch a [`UnCode`] to `$apply!(kernel)` with a monomorphized closure
/// for each hot op — the unary counterpart of [`dispatch_bin_kernel`]: each
/// closure body is the IDENTICAL expression to `unary_kernel_of`'s arm, and
/// the remaining ops fall through to the fn-pointer table.
macro_rules! dispatch_un_kernel {
    ($op:expr, $apply:ident) => {
        // See `dispatch_bin_kernel`: the monomorphized arms are f64-only, so
        // Float32 routes through the shared (precision-aware) table.
        if $crate::precision::is_f32() {
            $apply!(unary_kernel_of(*$op))
        } else {
            match $op {
                UnCode::Abs => $apply!(|x: f64| x.abs()),
                UnCode::Sqrt => $apply!(|x: f64| x.sqrt()),
                UnCode::Exp => $apply!(|x: f64| x.exp()),
                UnCode::Ln => $apply!(|x: f64| x.ln()),
                UnCode::Log10 => $apply!(|x: f64| x.log10()),
                UnCode::Sin => $apply!(|x: f64| x.sin()),
                UnCode::Cos => $apply!(|x: f64| x.cos()),
                UnCode::Tanh => $apply!(|x: f64| x.tanh()),
                UnCode::Floor => $apply!(|x: f64| x.floor()),
                UnCode::Ceil => $apply!(|x: f64| x.ceil()),
                UnCode::Tan => $apply!(|x: f64| x.tan()),
                UnCode::Asin => $apply!(|x: f64| x.asin()),
                UnCode::Acos => $apply!(|x: f64| x.acos()),
                UnCode::Atan => $apply!(|x: f64| x.atan()),
                UnCode::Sinh => $apply!(|x: f64| x.sinh()),
                UnCode::Cosh => $apply!(|x: f64| x.cosh()),
                UnCode::Asinh => $apply!(|x: f64| x.asinh()),
                UnCode::Acosh => $apply!(|x: f64| x.acosh()),
                UnCode::Atanh => $apply!(|x: f64| x.atanh()),
                UnCode::Not => $apply!(|x: f64| (x == 0.0) as i32 as f64),
                UnCode::Sign => $apply!(|x: f64| {
                    if x > 0.0 {
                        1.0
                    } else if x < 0.0 {
                        -1.0
                    } else {
                        0.0
                    }
                }),
                other => $apply!(unary_kernel_of(*other)),
            }
        }
    };
}
pub(super) use dispatch_un_kernel;

/// A fused group's operands resolved for one execution: its scalar values,
/// its array inputs' base pointers and its live-outs' slab pointers. Held by
/// the executor with room for the program's largest group, so resolving a
/// group never allocates.
pub(super) struct FusedScratch {
    svals: Vec<f64>,
    bases: Vec<*const f64>,
    outs: Vec<(GroupIx, *mut f64)>,
    cursor: RunCursor,
    /// The boxes of a group's whole-read gathers ([`ChunkGather::whole`]),
    /// back to back, sized for the largest group's.
    whole: Vec<f64>,
    /// The sources of the current group's whole-read gathers, in input
    /// order (their `bases` entries point into `whole`).
    whole_src: Vec<*const f64>,
    /// Per group, [`node_elems`] of its schedule: what lets a worker step
    /// straight to the start of its share of the group.
    node_elems: Vec<Vec<usize>>,
    /// The register files and cursors of the workers a split group runs on
    /// (native targets only; see `par`).
    #[cfg(not(target_arch = "wasm32"))]
    pub(super) workers: super::par::FusedWorkers,
}

// SAFETY: the raw pointers in `bases`, `outs` and `whole_src` are addresses
// into the slab, `dy` and the state of the call that resolved them. Every
// group execution rewrites them before it reads one, and nothing reads them
// after the call returns, so moving the scratch to another thread between
// calls cannot carry a live borrow with it.
unsafe impl Send for FusedScratch {}

impl FusedScratch {
    pub(super) fn for_program(prog: &TapeProgram) -> Self {
        let most = |f: fn(&FusedSpec) -> usize| prog.fused.iter().map(f).max().unwrap_or(0);
        FusedScratch {
            svals: Vec::with_capacity(most(|f| f.scalars.len())),
            bases: Vec::with_capacity(most(|f| f.inputs.len())),
            outs: Vec::with_capacity(most(|f| f.outputs.len())),
            cursor: RunCursor::with_room(
                most(|f| f.schedule.depth),
                most(n_shifted),
                most(n_scans),
            ),
            whole: vec![0.0; most(whole_len)],
            whole_src: Vec::with_capacity(most(|f| f.inputs.len())),
            node_elems: prog
                .fused
                .iter()
                .map(|f| node_elems(&f.schedule.nodes))
                .collect(),
            #[cfg(not(target_arch = "wasm32"))]
            workers: super::par::FusedWorkers::for_program(prog),
        }
    }

    /// The scratch of a thread that runs groups beside the calling thread
    /// (see `plan`): its whole-read boxes start empty and grow the first
    /// time a group needs them.
    #[cfg(not(target_arch = "wasm32"))]
    pub(super) fn for_peer(prog: &TapeProgram) -> Self {
        FusedScratch {
            whole: Vec::new(),
            ..Self::for_program(prog)
        }
    }
}

pub(super) fn n_shifted(fs: &FusedSpec) -> usize {
    fs.inputs.iter().filter(|i| i.shifted_ix.is_some()).count()
}

/// The elements a group's whole-read gathers occupy in `FusedScratch::whole`.
fn whole_len(fs: &FusedSpec) -> usize {
    let n = fs.n_elems();
    fs.inputs
        .iter()
        .filter(|i| i.gather.as_deref().is_some_and(|g| g.whole))
        .count()
        * n
}

/// The carry slots a group's absorbed scans use (one per lane).
pub(super) fn n_scans(fs: &FusedSpec) -> usize {
    fs.micro
        .iter()
        .map(|m| match m {
            MicroOp::Scan { post, .. } => *post as usize,
            _ => 0,
        })
        .sum()
}

/// The elements one execution of each schedule node covers: a run's length,
/// or ONE repetition of a `Repeat`'s body (its whole span is that times its
/// count). Pre-order, like the nodes.
pub(super) fn node_elems(nodes: &[RunNode]) -> Vec<usize> {
    fn span(nodes: &[RunNode], out: &mut [usize], i: usize) -> (usize, usize) {
        match &nodes[i] {
            RunNode::Run(r) => {
                out[i] = r.len as usize;
                (out[i], i + 1)
            }
            RunNode::Repeat { count, body, .. } => {
                let end = i + 1 + *body as usize;
                let (mut rep, mut j) = (0usize, i + 1);
                while j < end {
                    let (t, next) = span(nodes, out, j);
                    rep += t;
                    j = next;
                }
                out[i] = rep;
                (rep * *count as usize, end)
            }
        }
    }
    let mut out = vec![0usize; nodes.len()];
    let mut j = 0usize;
    while j < nodes.len() {
        j = span(nodes, &mut out, j).1;
    }
    out
}

/// The share of a fused group one execution of the chunk loop runs: the
/// elements at flat positions `[lo, hi)` of the schedule's execution order,
/// or, with a `period`, every element whose position modulo `period` lies in
/// `[lo, hi)` (the inner positions of an absorbed reduction, each of which
/// one worker then folds over every leading position, in order). Every
/// element is computed by the same micro-ops whatever window it falls in, so
/// splitting a group into windows cannot change a bit.
#[derive(Clone, Copy)]
pub(super) struct Window<'a> {
    pub(super) lo: usize,
    pub(super) hi: usize,
    /// 0 for a plain window.
    pub(super) period: usize,
    /// [`node_elems`] of the group's schedule.
    pub(super) node_elems: &'a [usize],
}

impl<'a> Window<'a> {
    /// The whole group.
    pub(super) fn all(node_elems: &'a [usize]) -> Self {
        Window {
            lo: 0,
            hi: usize::MAX,
            period: 0,
            node_elems,
        }
    }

    /// The first piece `[s, e)` (relative to the run) at or after `from` of
    /// the run of `len` elements starting at schedule position `g0` that
    /// falls in the window.
    #[inline(always)]
    fn piece(&self, g0: usize, len: usize, from: usize) -> Option<(usize, usize)> {
        if self.period == 0 {
            let s = from.max(self.lo.saturating_sub(g0));
            let e = len.min(self.hi.saturating_sub(g0));
            return (s < e).then_some((s, e));
        }
        let mut p = g0 + from;
        while p < g0 + len {
            let base = p - p % self.period;
            let s = p.max(base + self.lo);
            let e = (g0 + len).min(base + self.hi);
            if s < e {
                return Some((s - g0, e - g0));
            }
            p = base + self.period;
        }
        None
    }
}

/// Unused bytes past the end of each buffer a split's worker writes (two
/// cache lines, the pair the adjacent-line prefetcher moves together).
#[cfg(not(target_arch = "wasm32"))]
pub(super) const PAD_BYTES: usize = 128;

/// The executor's position in a [`RunSchedule`]: one frame per open
/// [`RunNode::Repeat`], the offsets the open repetitions have stepped so far,
/// and the current run's shifted-input offsets with them applied. Sized for
/// the program's deepest schedule and widest group, so walking never
/// allocates.
pub(super) struct RunCursor {
    frames: Vec<RepeatFrame>,
    in_delta: Vec<i64>,
    in_off: Vec<i64>,
    /// Each absorbed scan's running value, carried from one chunk to the
    /// next.
    carries: Vec<f64>,
}

/// An open repetition: the `Repeat` node, its body's node range, and the
/// repetitions still to run (this one included).
#[derive(Clone, Copy)]
struct RepeatFrame {
    node: usize,
    end: usize,
    left: u32,
}

impl RunCursor {
    pub(super) fn with_room(depth: usize, n_shifted: usize, n_scans: usize) -> Self {
        RunCursor {
            frames: Vec::with_capacity(depth),
            in_delta: Vec::with_capacity(n_shifted),
            in_off: Vec::with_capacity(n_shifted),
            carries: Vec::with_capacity(n_scans),
        }
    }

    /// [`RunCursor::with_room`] for a split's worker: every buffer has
    /// unused room for [`PAD_BYTES`] past its end, so what one worker writes
    /// never shares a cache line with the next allocation (another worker's
    /// cursor or registers).
    #[cfg(not(target_arch = "wasm32"))]
    pub(super) fn for_worker(depth: usize, n_shifted: usize, n_scans: usize) -> Self {
        fn room<T>(n: usize) -> Vec<T> {
            Vec::with_capacity(n + PAD_BYTES.div_ceil(std::mem::size_of::<T>().max(1)))
        }
        RunCursor {
            frames: room(depth),
            in_delta: room(n_shifted),
            in_off: room(n_shifted),
            carries: room(n_scans),
        }
    }

    #[cfg(test)]
    pub(super) fn for_spec(fs: &FusedSpec) -> Self {
        Self::with_room(fs.schedule.depth, n_shifted(fs), n_scans(fs))
    }
}

/// A folded data-subscript gather whose subscript array is fixed for the
/// call: its source positions, resolved once when the sections that define
/// the subscript run, so a steady call reads one `u32` per element instead of
/// rounding and range-checking an `f64` subscript.
pub(super) struct IndexTable {
    /// The subscript array's slot (defined in the CONST or SEGMENT section).
    subscript: SlotId,
    /// How a data position becomes a source offset.
    fi: FoldedIndex,
    /// One position per element of the group box, or [`GATHER_GHOST`].
    pos: Vec<u32>,
    /// No position is the ghost, so the gather is a plain indexed load.
    all_in: bool,
}

/// Per fused group, per input: the [`IndexTable`] of every folded
/// data-subscript gather in a CONTINUOUS-section group whose subscript is a
/// slot the CONST or SEGMENT section defines (an empty list for a group with
/// none). Sized here, once; [`refill_index_tables`] only overwrites.
pub(super) fn index_tables_for(prog: &TapeProgram) -> Vec<Vec<Option<IndexTable>>> {
    let prime_end = (prog.n_const + prog.n_segment) as usize;
    let tables = prog.tables();
    let mut primed: Vec<bool> = vec![false; prog.slots.len()];
    for ins in &prog.instrs[..prime_end] {
        ins.for_each_def(&tables, |s| primed[s as usize] = true);
    }
    let mut out: Vec<Vec<Option<IndexTable>>> = prog.fused.iter().map(|_| Vec::new()).collect();
    for ins in &prog.instrs[prime_end..] {
        let Instr::Fused { spec } = ins else { continue };
        let fs = &prog.fused[*spec as usize];
        let n_elems = fs.n_elems();
        let per_input: Vec<Option<IndexTable>> = fs
            .inputs
            .iter()
            .map(|inp| {
                let fi = inp.index.filter(FoldedIndex::preloaded)?;
                let SrcRef::Slot(sub) = fs.inputs[fi.by as usize].src else {
                    return None;
                };
                (primed[sub as usize] && fi.src_len < GATHER_GHOST as usize).then(|| IndexTable {
                    subscript: sub,
                    fi,
                    pos: vec![GATHER_GHOST; n_elems],
                    all_in: false,
                })
            })
            .collect();
        if per_input.iter().any(Option::is_some) {
            out[*spec as usize] = per_input;
        }
    }
    out
}

/// Re-resolve every [`IndexTable`] from its subscript slot, after the
/// sections that define the subscripts ran. The positions are
/// [`data_subscript`]'s, so the gather reads the same elements it would
/// resolve per call.
///
/// # Safety
/// `slab_ptr` must be the slab `slot_off` lays out, with every table's
/// subscript slot holding the table's element count.
pub(super) unsafe fn refill_index_tables(
    tables: &mut [Vec<Option<IndexTable>>],
    slab_ptr: *const f64,
    slot_off: &[usize],
) {
    for t in tables.iter_mut().flatten().flatten() {
        let sub = unsafe {
            std::slice::from_raw_parts(slab_ptr.add(slot_off[t.subscript as usize]), t.pos.len())
        };
        let mut all_in = true;
        for (k, (p, &v)) in t.pos.iter_mut().zip(sub).enumerate() {
            *p = match data_subscript(v, t.fi.n) {
                Some(i) => t.fi.at(i, k) as u32,
                None => {
                    all_in = false;
                    GATHER_GHOST
                }
            };
        }
        t.all_in = all_in;
    }
}

/// A fused group's operands, resolved for one call (see [`resolve_fused`]).
pub(super) struct Resolved<'a> {
    pub(super) svals: &'a [f64],
    pub(super) bases: &'a [*const f64],
    pub(super) whole_src: &'a [*const f64],
    pub(super) outs: &'a [(GroupIx, *mut f64)],
    /// An absorbed reduction's accumulator (null without one).
    pub(super) red: *mut f64,
    pub(super) node_elems: &'a [usize],
    pub(super) cursor: &'a mut RunCursor,
    /// The group's [`PreReduce`], to run range by range (`None` without one,
    /// or when its source is not contiguous and it already ran whole).
    pub(super) pre: Option<PreTile>,
}

/// A [`PreReduce`]'s resolved source and result.
#[derive(Clone, Copy)]
pub(super) struct PreTile {
    pub(super) src: *const f64,
    pub(super) out: *mut f64,
}

/// Inner positions per range of a [`PreReduce`]: as many whole columns as
/// keep a range's source (`rows` elements each) within a few tens of KiB.
pub(super) fn pre_cols(rows: usize) -> usize {
    ((16384 / rows.max(1)) / 8 * 8).max(8)
}

/// Fold the cells `[a, b)` of a [`PreReduce`] (each from `init`, over every
/// leading position in order, as `Instr::Reduce` does) from a contiguous
/// row-major source.
///
/// # Safety
/// `pt` must address the source's `rows * n_inner` elements and the result's
/// `n_inner`.
unsafe fn pre_fold(pre: &PreReduce, pt: PreTile, a: usize, b: usize) {
    let n = pre.n_inner;
    unsafe {
        for k in a..b {
            *pt.out.add(k) = pre.init;
        }
    }
    let op = &pre.op;
    macro_rules! fold_rows {
        ($f:expr) => {
            for r in 0..pre.rows {
                unsafe {
                    super::kernels::reduce_rows(pt.out.add(a), b - a, pt.src.add(r * n + a), 1, $f)
                }
            }
        };
    }
    dispatch_bin_kernel!(op, fold_rows);
}

/// Run fused group `fs` over the inner positions `[lo, hi)` of its
/// [`PreReduce`] (every leading position of them), a range of columns at a
/// time: fold the range, then run the group's elements there.
///
/// # Safety
/// As for [`run_fused_window`], with `pt` resolved for `fs.pre`.
#[allow(clippy::too_many_arguments)]
pub(super) unsafe fn run_pre_tiled(
    simd: SimdLevel,
    fs: &FusedSpec,
    pt: PreTile,
    svals: &[f64],
    bases: &[*const f64],
    whole_src: &[*const f64],
    outs: &[(GroupIx, *mut f64)],
    red: *mut f64,
    idx: &[Option<IndexTable>],
    fregs: &mut [f64],
    cursor: &mut RunCursor,
    node_elems: &[usize],
    lo: usize,
    hi: usize,
) {
    let pre = fs.pre.as_ref().expect("a group with a pre-reduction");
    let period = pre.n_inner;
    let step = pre_cols(pre.rows);
    // A box whose every column fits one range: fold it all, then run the
    // group whole, as one window.
    if lo == 0 && hi == period && period <= step {
        unsafe {
            pre_fold(pre, pt, 0, period);
            fill_whole(fs, bases, whole_src, 0, fs.n_elems(), 0);
            run_fused_window(
                simd,
                fs,
                svals,
                bases,
                outs,
                red,
                idx,
                fregs,
                cursor,
                Window::all(node_elems),
            );
        }
        return;
    }
    let mut a = lo;
    while a < hi {
        let b = (a + step).min(hi);
        unsafe {
            pre_fold(pre, pt, a, b);
            fill_whole(fs, bases, whole_src, a, b, period);
            run_fused_window(
                simd,
                fs,
                svals,
                bases,
                outs,
                red,
                idx,
                fregs,
                cursor,
                Window {
                    lo: a,
                    hi: b,
                    period,
                    node_elems,
                },
            );
        }
        a = b;
    }
}

/// The workers of `scratch`, or a stand-in on wasm (which never splits).
#[cfg(not(target_arch = "wasm32"))]
pub(super) type Workers = super::par::FusedWorkers;
#[cfg(target_arch = "wasm32")]
pub(super) type Workers = ();

/// Resolve fused group `spec`'s operands for this call into `scratch`:
/// scalar values, input base pointers (a whole-read gather's pointing at its
/// scratch box, its source recorded in `whole_src`), output pointers (the
/// slab, `dy` for a slot homed there, null for an unstored one) and the
/// reduction's accumulator.
///
/// # Safety
/// As for [`exec_fused`].
#[allow(clippy::too_many_arguments)]
pub(super) unsafe fn resolve_fused<'a>(
    spec: usize,
    env: &Env,
    slab_ptr: *mut f64,
    slot_off: &[usize],
    scratch: &'a mut FusedScratch,
    dy_home: &[usize],
    dy: *mut f64,
) -> (Resolved<'a>, &'a mut Workers) {
    let fs = &env.prog.fused[spec];
    let FusedScratch {
        svals,
        bases,
        outs,
        cursor,
        whole,
        whole_src,
        node_elems,
        #[cfg(not(target_arch = "wasm32"))]
        workers,
    } = scratch;
    let node_elems = &node_elems[spec][..];
    // Resolve scalar inputs once.
    svals.clear();
    for op in &fs.scalars {
        svals.push(resolve_scalar(op, env, slab_ptr, slot_off));
    }
    // Resolve array input base pointers once.
    bases.clear();
    for inp in &fs.inputs {
        let p: *const f64 = match &inp.src {
            SrcRef::Slot(s) => {
                debug_assert_eq!(&env.prog.slots[*s as usize].shape, &inp.src_shape);
                unsafe { slab_ptr.add(slot_off[*s as usize]) as *const f64 }
            }
            SrcRef::State(ix) => {
                debug_assert_eq!(&env.prog.state_vars[*ix as usize].shape, &inp.src_shape);
                super::resolve::state_ptr(env, *ix)
            }
        };
        bases.push(p);
    }
    // Whole-read gathers: the box read once, then addressed like an aligned
    // input. The read itself happens with the group's chunks (each worker of
    // a split reads the positions of its own window).
    let n_elems = fs.n_elems();
    let mut next = 0usize;
    #[cfg(not(target_arch = "wasm32"))]
    if whole.len() < whole_len(fs) {
        whole.resize(whole_len(fs), 0.0);
    }
    whole_src.clear();
    for (inp, base) in fs.inputs.iter().zip(bases.iter_mut()) {
        if inp.gather.as_deref().is_some_and(|g| g.whole) {
            whole_src.push(*base);
            *base = whole[next..next + n_elems].as_ptr();
            next += n_elems;
        }
    }
    // Output pointers: the slab, or `dy` for a slot homed there.
    outs.clear();
    for &(reg, slot) in &fs.outputs {
        let p = match dy_home.get(slot as usize) {
            Some(&super::UNSTORED) => std::ptr::null_mut(),
            Some(&off) if off != usize::MAX => unsafe { dy.add(off) },
            _ => unsafe { slab_ptr.add(slot_off[slot as usize]) },
        };
        outs.push((reg, p));
    }
    // An absorbed reduction's accumulator: its slot, or `dy` when homed
    // there.
    let red: *mut f64 = match &fs.reduce {
        Some(r) => match dy_home.get(r.out as usize) {
            Some(&off) if off != usize::MAX => unsafe { dy.add(off) },
            _ => unsafe { slab_ptr.add(slot_off[r.out as usize]) },
        },
        None => std::ptr::null_mut(),
    };
    // The pre-reduction, range by range from a contiguous source; else now,
    // whole, by the instruction's own walk.
    let pre = fs.pre.as_ref().and_then(|p| {
        let sv = super::resolve::resolve_src(&p.src, env, slab_ptr, slot_off);
        let out = unsafe { slab_ptr.add(slot_off[p.out as usize]) };
        if sv.strides[..] == super::resolve::rm_strides(&sv.shape)[..] {
            return Some(PreTile { src: sv.ptr, out });
        }
        unsafe { pre_fold_strided(p, &sv, out) };
        None
    });
    // Zero-sized: no allocation.
    #[cfg(target_arch = "wasm32")]
    let workers: &'a mut Workers = Box::leak(Box::new(()));
    (
        Resolved {
            svals,
            bases,
            whole_src,
            outs,
            red,
            node_elems,
            cursor,
            pre,
        },
        workers,
    )
}

/// A [`PreReduce`] over a source that is not contiguous: every cell folded
/// whole, in the instruction's order, through the source's strides.
///
/// # Safety
/// `sv` must be a live view of the source and `out` the result's cells.
unsafe fn pre_fold_strided(pre: &PreReduce, sv: &super::resolve::SrcView, out: *mut f64) {
    let f = binary_kernel_of(pre.op);
    let nd = sv.shape.len();
    let lead = {
        let mut k = 0usize;
        let mut rows = 1usize;
        while k < nd && rows < pre.rows {
            rows *= sv.shape[k];
            k += 1;
        }
        k
    };
    let mut idx: SmallVec<[usize; 4]> = SmallVec::from_elem(0, nd);
    for c in 0..pre.n_inner {
        // The inner coordinates of cell `c`, row-major.
        let mut rest = c;
        for d in (lead..nd).rev() {
            idx[d] = rest % sv.shape[d];
            rest /= sv.shape[d];
        }
        let mut acc = pre.init;
        for r in 0..pre.rows {
            let mut rest = r;
            for d in (0..lead).rev() {
                idx[d] = rest % sv.shape[d];
                rest /= sv.shape[d];
            }
            let off: i64 = (0..nd).map(|d| sv.strides[d] * idx[d] as i64).sum();
            acc = f(acc, unsafe { *sv.ptr.offset(off as isize) });
        }
        unsafe { *out.add(c) = acc };
    }
}

/// Execute one fused group. Iterates the precompiled run schedule; each run
/// is strip-mined into `FCHUNK`-element chunks whose micro-ops execute over
/// the register file, then live-out registers store to the slab. Per element
/// this applies exactly the same scalar kernels in the same order as the
/// unfused instructions (elementwise maps — chunking cannot change a bit).
#[inline(never)]
#[allow(clippy::too_many_arguments)]
pub(super) unsafe fn exec_fused(
    spec: usize,
    env: &Env,
    slab_ptr: *mut f64,
    slot_off: &[usize],
    fregs: &mut [f64],
    scratch: &mut FusedScratch,
    idx: &[Option<IndexTable>],
    simd: SimdLevel,
    dy_home: &[usize],
    dy: *mut f64,
) {
    let fs = &env.prog.fused[spec];
    let (r, _workers) =
        unsafe { resolve_fused(spec, env, slab_ptr, slot_off, scratch, dy_home, dy) };
    let Resolved {
        svals,
        bases,
        whole_src,
        outs,
        red,
        node_elems,
        cursor,
        pre,
    } = r;
    // A large group is split into windows across the caller's worker
    // threads (see `par`).
    #[cfg(not(target_arch = "wasm32"))]
    {
        let ways = super::par::split_ways(fs, _workers.call_ways);
        if ways > 1 {
            unsafe {
                _workers.run_split(
                    fs, svals, bases, whole_src, outs, red, idx, node_elems, simd, ways, pre,
                )
            };
            return;
        }
    }
    if let Some(pt) = pre {
        let n = fs.pre.as_ref().map_or(0, |p| p.n_inner);
        unsafe {
            run_pre_tiled(
                simd, fs, pt, svals, bases, whole_src, outs, red, idx, fregs, cursor, node_elems,
                0, n,
            )
        };
        return;
    }
    // Seeded with the reduction's identity (a split's workers seed their
    // own cells).
    if let Some(r) = &fs.reduce
        && !seeds_itself(fs)
    {
        unsafe { std::slice::from_raw_parts_mut(red, r.n_inner).fill(r.init) };
    }
    unsafe { fill_whole(fs, bases, whole_src, 0, fs.n_elems(), 0) };
    // A group with a reverse scan visits its rows from the last.
    if let Some((rows, len)) = fs.rows_reversed() {
        for r in (0..rows).rev() {
            let win = Window {
                lo: r * len,
                hi: (r + 1) * len,
                period: 0,
                node_elems,
            };
            unsafe { run_fused_window(simd, fs, svals, bases, outs, red, idx, fregs, cursor, win) };
        }
        return;
    }
    unsafe {
        run_fused_window(
            simd,
            fs,
            svals,
            bases,
            outs,
            red,
            idx,
            fregs,
            cursor,
            Window::all(node_elems),
        )
    }
}

/// Run share `w` of fused group `spec` split `ways` wide (see
/// [`super::par::split_ways`]) on this thread, resolving its operands into
/// this thread's `scratch`.
///
/// # Safety
/// As for [`exec_fused`].
#[cfg(not(target_arch = "wasm32"))]
#[allow(clippy::too_many_arguments)]
pub(super) unsafe fn exec_fused_share(
    spec: usize,
    env: &Env,
    slab_ptr: *mut f64,
    slot_off: &[usize],
    scratch: &mut FusedScratch,
    idx: &[Option<IndexTable>],
    simd: SimdLevel,
    dy_home: &[usize],
    dy: *mut f64,
    w: usize,
    ways: usize,
) {
    let fs = &env.prog.fused[spec];
    let (r, workers) =
        unsafe { resolve_fused(spec, env, slab_ptr, slot_off, scratch, dy_home, dy) };
    unsafe {
        workers.run_own_share(
            fs,
            r.svals,
            r.bases,
            r.whole_src,
            r.outs,
            r.red,
            idx,
            r.node_elems,
            simd,
            w,
            ways,
            r.pre,
        )
    };
}

/// Read a group's whole-read gathers (sources `whole_src`, destinations their
/// `bases` entries) at the flat positions `[lo, hi)`, or with a `period`, at
/// the positions whose remainder modulo `period` lies in `[lo, hi)`: the
/// positions a [`Window`] with the same bounds runs.
///
/// # Safety
/// `bases` must be the group's resolved inputs with every whole-read gather's
/// entry pointing at its `n_elems` scratch box, and `whole_src` their sources.
pub(super) unsafe fn fill_whole(
    fs: &FusedSpec,
    bases: &[*const f64],
    whole_src: &[*const f64],
    lo: usize,
    hi: usize,
    period: usize,
) {
    if whole_src.is_empty() {
        return;
    }
    let n = fs.n_elems();
    let mut k = 0usize;
    for (inp, &dst) in fs.inputs.iter().zip(bases) {
        let Some(g) = inp.gather.as_deref().filter(|g| g.whole) else {
            continue;
        };
        let (src, dst) = (whole_src[k], dst as *mut f64);
        k += 1;
        if period == 0 {
            let hi = hi.min(n);
            if lo < hi {
                unsafe { g.fill(src, lo, hi - lo, dst.add(lo)) };
            }
            continue;
        }
        let mut base = 0usize;
        while base < n {
            let (a, b) = (base + lo, (base + hi).min(base + period).min(n));
            if a < b {
                unsafe { g.fill(src, a, b - a, dst.add(a)) };
            }
            base += period;
        }
    }
}

/// Run one window of a fused group through the SIMD clone selected at
/// executor construction (Step 4b). Same source, same scalar semantics — the
/// `#[target_feature]` wrappers only widen the auto-vectorized lanes.
#[allow(clippy::too_many_arguments)]
#[inline(always)]
pub(super) unsafe fn run_fused_window(
    simd: SimdLevel,
    fs: &FusedSpec,
    svals: &[f64],
    bases: &[*const f64],
    outs: &[(GroupIx, *mut f64)],
    red: *mut f64,
    idx: &[Option<IndexTable>],
    fregs: &mut [f64],
    cursor: &mut RunCursor,
    win: Window,
) {
    match simd {
        SimdLevel::Generic => unsafe {
            exec_fused_runs_generic(fs, svals, bases, outs, red, idx, fregs, cursor, win)
        },
        #[cfg(target_arch = "x86_64")]
        SimdLevel::Avx2 => unsafe {
            exec_fused_runs_avx2(fs, svals, bases, outs, red, idx, fregs, cursor, win)
        },
        #[cfg(target_arch = "x86_64")]
        SimdLevel::Avx512 => unsafe {
            exec_fused_runs_avx512(fs, svals, bases, outs, red, idx, fregs, cursor, win)
        },
    }
}

/// Fold one chunk (`c` values at flat box offset `at`) into an absorbed
/// reduction's accumulator: `acc[(at + k) % n_inner] = f(acc[..], v[k])`,
/// ascending in `k` — contiguous pieces of the accumulator, one per crossing
/// of a leading-axes position. `acc` is where cell 0 would be: a split's
/// worker backs only its own cells, so it may point before that buffer
/// (hence the wrapping offset).
///
/// Every cell's first term sits at the first leading position (flat offset
/// below `n_inner`), which the walk visits before the others, so there the
/// cell is written as `f(init, v)` without reading it: the accumulator needs
/// no seeding pass (see [`seeds_itself`]).
#[inline(always)]
#[allow(clippy::too_many_arguments)]
unsafe fn fold_chunk(
    acc: *mut f64,
    n_inner: usize,
    at: usize,
    init: f64,
    v: *const f64,
    c: usize,
    f: impl Fn(f64, f64) -> f64 + Copy,
) {
    let mut k = 0usize;
    while k < c {
        let p = at + k;
        let o = p % n_inner;
        let len = (c - k).min(n_inner - o);
        unsafe {
            let a = std::slice::from_raw_parts_mut(acc.wrapping_add(o), len);
            let x = std::slice::from_raw_parts(v.add(k), len);
            if p < n_inner {
                for (y, &t) in a.iter_mut().zip(x) {
                    *y = f(init, t);
                }
            } else {
                for (y, &t) in a.iter_mut().zip(x) {
                    *y = f(*y, t);
                }
            }
        }
        k += len;
    }
}

/// [`fold_chunk`] of the values `g(a[k], b[k])`, computed in the same loop
/// instead of read back from a register: the group's last micro-op fused
/// into its reduction (see [`fold_fused_op`]). Each element applies `g`
/// then `f`, exactly the two kernels the unfused path applies.
#[inline(always)]
#[allow(clippy::too_many_arguments)]
unsafe fn fold_chunk_bin(
    acc: *mut f64,
    n_inner: usize,
    at: usize,
    init: f64,
    a: MSrc,
    b: MSrc,
    c: usize,
    g: impl Fn(f64, f64) -> f64 + Copy,
    f: impl Fn(f64, f64) -> f64 + Copy,
) {
    // One piece (inside the `unsafe` block below): `$x` / `$y` read the
    // operands at piece element `$i`.
    macro_rules! piece {
        ($ac:expr, $len:expr, $first:expr, |$i:ident| $x:expr, $y:expr) => {{
            let ac: &mut [f64] = $ac;
            if $first {
                for $i in 0..$len {
                    *ac.get_unchecked_mut($i) = f(init, g($x, $y));
                }
            } else {
                for $i in 0..$len {
                    let r = ac.get_unchecked_mut($i);
                    *r = f(*r, g($x, $y));
                }
            }
        }};
    }
    let mut k = 0usize;
    while k < c {
        let p = at + k;
        let o = p % n_inner;
        let len = (c - k).min(n_inner - o);
        let first = p < n_inner;
        unsafe {
            let ac = std::slice::from_raw_parts_mut(acc.wrapping_add(o), len);
            match (a, b) {
                (MSrc::P(pa), MSrc::P(pb)) => {
                    let xa = std::slice::from_raw_parts(pa.add(k), len);
                    let xb = std::slice::from_raw_parts(pb.add(k), len);
                    piece!(
                        ac,
                        len,
                        first,
                        |i| *xa.get_unchecked(i),
                        *xb.get_unchecked(i)
                    )
                }
                (MSrc::P(pa), MSrc::C(y)) => {
                    let xa = std::slice::from_raw_parts(pa.add(k), len);
                    piece!(ac, len, first, |i| *xa.get_unchecked(i), y)
                }
                (MSrc::C(x), MSrc::P(pb)) => {
                    let xb = std::slice::from_raw_parts(pb.add(k), len);
                    piece!(ac, len, first, |i| x, *xb.get_unchecked(i))
                }
                (MSrc::C(x), MSrc::C(y)) => fold_piece_const(ac, first, init, x, y, g, f),
            }
        }
        k += len;
    }
}

/// One piece of [`fold_chunk_bin`] with both operands constant. Out of line:
/// inlined, the compiler computes `g(x, y)` ahead of the operand match from
/// whatever the operands hold, pointers included, and a pointer's bits read
/// as a subnormal `f64` make that one multiply cost a microcode assist on
/// every piece.
#[cold]
#[inline(never)]
fn fold_piece_const(
    ac: &mut [f64],
    first: bool,
    init: f64,
    x: f64,
    y: f64,
    g: impl Fn(f64, f64) -> f64,
    f: impl Fn(f64, f64) -> f64,
) {
    let v = g(x, y);
    for r in ac.iter_mut() {
        *r = f(if first { init } else { *r }, v);
    }
}

/// The micro-op a group's absorbed reduction folds straight from: its last
/// one, when it is a `+ - * /` [`MicroOp::Bin`] (or a [`MicroOp::Bin2`] of
/// one scaled by a scalar, into a sum) writing the reduction's register,
/// nothing else reads that register (not a stored output, not a direct
/// store), the fold is a sum or product, and the arithmetic is Float64 (the
/// fused loop composes the f64 kernels directly, as the scan fusion does).
/// `None` runs the op and the fold as two passes.
pub(in crate::simulate_array::tape) fn fold_fused_op(fs: &FusedSpec) -> Option<usize> {
    use BinCode::{Add, Div, Mul, Sub};
    let r = fs.reduce.as_ref()?;
    let mi = fs.micro.len().checked_sub(1)?;
    let (out, kinds_ok) = match &fs.micro[mi] {
        MicroOp::Bin { op, out, .. } => (
            out,
            matches!(op, Add | Sub | Mul | Div) && matches!(r.op, Add | Mul),
        ),
        // A superop scaled by a scalar (`(a op1 b) op2 s`, either order),
        // folded into a sum.
        MicroOp::Bin2 {
            op1,
            op2,
            c: MRef::Scal(_),
            out,
            ..
        } => (
            out,
            matches!(op1, Add | Sub | Mul | Div) && matches!(op2, Mul | Div) && r.op == Add,
        ),
        _ => return None,
    };
    let ok = *out == r.reg
        && kinds_ok
        && fs.scan_fuse.is_empty()
        && !fs.outputs.iter().any(|&(reg, _)| reg == r.reg)
        && !fs.direct.iter().any(|&(m, _)| m as usize == mi)
        && !crate::precision::is_f32();
    ok.then_some(mi)
}

/// Whether a group's absorbed reduction writes every accumulator cell before
/// reading it (see [`fold_chunk`]), so nothing seeds it: the group visits at
/// least one leading position.
pub(super) fn seeds_itself(fs: &FusedSpec) -> bool {
    fs.reduce
        .as_ref()
        .is_some_and(|r| r.n_inner > 0 && fs.shape.iter().product::<usize>() >= r.n_inner)
}

/// The strip-mined chunk loop of [`exec_fused`], monomorphized per SIMD
/// level through the `#[target_feature]` wrappers below (`#[inline(always)]`
/// so each wrapper compiles the WHOLE loop nest — micro-op dispatch, chunk
/// kernels and stores — under its feature set).
#[inline(always)]
#[allow(clippy::too_many_arguments)]
unsafe fn exec_fused_runs(
    fs: &FusedSpec,
    svals: &[f64],
    bases: &[*const f64],
    outs: &[(GroupIx, *mut f64)],
    red: *mut f64,
    idx: &[Option<IndexTable>],
    fregs: &mut [f64],
    cursor: &mut RunCursor,
    win: Window,
) {
    let rp = fregs.as_mut_ptr();
    let cs = FCHUNK;
    // Scan fusions run under Float64 only (their loop composes the f64
    // arithmetic directly); otherwise the three passes run as written.
    let scan_fused = !fs.scan_fuse.is_empty() && !crate::precision::is_f32();
    // The last micro-op folded straight into the reduction, if it can be.
    let fold_op = fold_fused_op(fs);
    debug_assert!(fs.direct.windows(2).all(|w| w[0].0 < w[1].0));
    // The outputs a micro-op stores directly (no end-of-chunk copy).
    let direct_mask = fs
        .direct
        .iter()
        .filter(|&&(_, k)| k < 128)
        .fold(0u128, |m, &(_, k)| m | 1 << k);
    let is_direct = |k: usize| {
        if k < 128 {
            direct_mask >> k & 1 != 0
        } else {
            fs.direct.iter().any(|&(_, d)| d as usize == k)
        }
    };
    // Bin3 splat registers: one FCHUNK broadcast per scalar plus a zero
    // register (the ghost read), filled once per call. The values are the
    // EXACT scalars / the exact `+0.0` ghost, so an all-pointer superop
    // reads identical bits to the `MSrc::C` broadcast it replaces.
    let splat_base = fs.n_regs as usize + fs.n_load_regs as usize;
    let zero_ix = splat_base + svals.len();
    if fs.n_splat_regs > 0 {
        debug_assert_eq!(fs.n_splat_regs as usize, svals.len() + 1);
        unsafe {
            for (i, &v) in svals.iter().enumerate() {
                let p = rp.add((splat_base + i) * cs);
                for k in 0..cs {
                    *p.add(k) = v;
                }
            }
            let z = rp.add(zero_ix * cs);
            for k in 0..cs {
                *z.add(k) = 0.0;
            }
        }
    }
    let RunCursor {
        frames,
        in_delta,
        in_off,
        carries,
    } = cursor;
    // Every scan restarts at the first element of a lane the walk visits,
    // so the carries need no reset here; size them only. A group that visits
    // its rows from the last runs one window per row, each continuing the
    // carries the row before it left.
    let n_carries = n_scans(fs);
    if carries.len() != n_carries {
        carries.clear();
        carries.resize(n_carries, 0.0);
    }
    // One chunk: `c` elements at flat box offset `at`, `done` elements
    // into a run whose shifted inputs start at `in_off`.
    macro_rules! chunk {
        ($in_off:expr, $done:expr, $at:expr, $c:expr) => {{
            let (in_off, done, at, c): (&[i64], usize, usize, usize) = ($in_off, $done, $at, $c);
            // Pre-load strided shifted inputs into their dedicated chunk
            // registers (a ghost run needs no load — reads resolve to 0.0).
            for (i, inp) in fs.inputs.iter().enumerate() {
                if inp.load_reg == GroupIx::MAX {
                    continue;
                }
                // A gather read through its plan, one chunk at a time.
                if let Some(g) = inp.gather.as_deref()
                    && !g.whole
                {
                    unsafe { g.fill(bases[i], at, c, rp.add(inp.load_reg as usize * cs)) };
                    continue;
                }
                // A folded data-subscript gather: one random read per element,
                // through the subscript array's aligned chunk.
                if let Some(fi) = inp.index {
                    if let Some(Some(t)) = idx.get(i) {
                        unsafe {
                            let dst = std::slice::from_raw_parts_mut(
                                rp.add(inp.load_reg as usize * cs),
                                c,
                            );
                            let pos = &t.pos[at..at + c];
                            let src = bases[i];
                            if t.all_in {
                                for (d, &p) in dst.iter_mut().zip(pos) {
                                    *d = *src.add(p as usize);
                                }
                            } else {
                                for (d, &p) in dst.iter_mut().zip(pos) {
                                    *d = if p == GATHER_GHOST {
                                        0.0
                                    } else {
                                        *src.add(p as usize)
                                    };
                                }
                            }
                        }
                        continue;
                    }
                    unsafe {
                        let dst = rp.add(inp.load_reg as usize * cs);
                        let sub = bases[fi.by as usize].add(at);
                        let src = bases[i];
                        if fi.s_inner == 0 {
                            let src = src.add(fi.base);
                            for k in 0..c {
                                *dst.add(k) = match data_subscript(*sub.add(k), fi.n) {
                                    Some(p) => *src.add(p * fi.s_data),
                                    None => 0.0,
                                };
                            }
                        } else {
                            // The innermost position, carried instead of
                            // divided out per element.
                            let mut j = (at % fi.period) * fi.s_inner + fi.base;
                            let wrap = fi.period * fi.s_inner + fi.base;
                            for k in 0..c {
                                *dst.add(k) = match data_subscript(*sub.add(k), fi.n) {
                                    Some(p) => *src.add(p * fi.s_data + j),
                                    None => 0.0,
                                };
                                j += fi.s_inner;
                                if j == wrap {
                                    j = fi.base;
                                }
                            }
                        }
                    }
                    continue;
                }
                let sx = inp.shifted_ix.expect("load_reg implies shifted");
                let o = in_off[sx as usize];
                if o == GHOST_OFF {
                    continue;
                }
                unsafe {
                    let dst = rp.add(inp.load_reg as usize * cs);
                    let base =
                        bases[i].offset(o as isize + done as isize * inp.elem_stride as isize);
                    for k in 0..c {
                        *dst.add(k) = *base.offset(k as isize * inp.elem_stride as isize);
                    }
                }
            }
            let msrc = |m: &MRef| -> MSrc {
                match m {
                    MRef::Reg(r) => MSrc::P(unsafe { rp.add(*r as usize * cs) as *const f64 }),
                    MRef::Scal(i) => MSrc::C(svals[*i as usize]),
                    MRef::In(i) => {
                        let inp = &fs.inputs[*i as usize];
                        match inp.shifted_ix {
                            None if inp.index.is_some()
                                || inp.gather.as_deref().is_some_and(|g| !g.whole) =>
                            {
                                MSrc::P(unsafe { rp.add(inp.load_reg as usize * cs) as *const f64 })
                            }
                            None => MSrc::P(unsafe { bases[*i as usize].add(at) }),
                            Some(s) => {
                                let o = in_off[s as usize];
                                if o == GHOST_OFF {
                                    MSrc::C(0.0)
                                } else if inp.load_reg != GroupIx::MAX {
                                    MSrc::P(unsafe {
                                        rp.add(inp.load_reg as usize * cs) as *const f64
                                    })
                                } else if inp.elem_stride == 0 {
                                    // Constant along the run: one source
                                    // element, broadcast like a scalar.
                                    MSrc::C(unsafe { *bases[*i as usize].offset(o as isize) })
                                } else {
                                    MSrc::P(unsafe {
                                        bases[*i as usize].offset(o as isize + done as isize)
                                    })
                                }
                            }
                        }
                    }
                }
            };
            // All-pointer operand resolution for the Bin3 superop: scalar
            // operands point at their splat registers, ghost reads at the
            // zero register — same values, so same bits, with ONE loop shape
            // per (op1, op2, op3) instead of 2^4 operand-kind arms.
            let msrc_p = |m: &MRef| -> *const f64 {
                match msrc(m) {
                    MSrc::P(p) => p,
                    MSrc::C(v) => {
                        if v == 0.0 && v.is_sign_positive() {
                            unsafe { rp.add(zero_ix * cs) as *const f64 }
                        } else {
                            match m {
                                MRef::Scal(i) => unsafe {
                                    rp.add((splat_base + *i as usize) * cs) as *const f64
                                },
                                _ => unreachable!("non-scalar constant operand"),
                            }
                        }
                    }
                }
            };
            // Where micro-op `mi` writes: its register, or straight into its
            // output when it is that output's last writer and nothing after
            // it reads the register (`FusedSpec::direct`).
            // `direct` is in ascending micro-op order, and the micro loop asks
            // in that order too, so a cursor finds each entry in one step; a
            // query behind the cursor (a scan fusion asks for its post-op
            // first) starts over.
            let dnext = std::cell::Cell::new(0usize);
            let dst_of = |mi: usize, out: GroupIx| -> *mut f64 {
                let d = &fs.direct;
                let mut p = dnext.get();
                if p > 0 && d[p - 1].0 as usize >= mi {
                    p = 0;
                }
                while p < d.len() && (d[p].0 as usize) < mi {
                    p += 1;
                }
                dnext.set(p);
                if p < d.len() && d[p].0 as usize == mi {
                    let o = outs[d[p].1 as usize].1;
                    if !o.is_null() {
                        return unsafe { o.add(at) };
                    }
                }
                unsafe { rp.add(out as usize * cs) }
            };
            for (mi, op) in fs.micro.iter().enumerate() {
                if fold_op == Some(mi) {
                    continue;
                }
                if scan_fused {
                    let mut role = None;
                    for f in &fs.scan_fuse {
                        let s = f.scan as usize;
                        if f.pre && s == mi + 1 || f.post && s + 1 == mi {
                            role = Some(None);
                        } else if s == mi {
                            role = Some(Some(*f));
                        }
                    }
                    match role {
                        // Absorbed into its scan's loop.
                        Some(None) => continue,
                        Some(Some(f)) => {
                            let MicroOp::Scan {
                                op: sop,
                                a,
                                init,
                                inclusive,
                                row,
                                carry,
                                out,
                                ..
                            } = op
                            else {
                                unreachable!("a scan fusion names a scan")
                            };
                            let pre = f.pre.then(|| match &fs.micro[mi - 1] {
                                MicroOp::Bin { op, a, b, .. } => (*op, msrc(a), msrc(b)),
                                _ => unreachable!("a fused pre-op is a Bin"),
                            });
                            let (post, dst) = if f.post {
                                let MicroOp::Bin {
                                    op,
                                    a,
                                    b,
                                    out: pout,
                                } = &fs.micro[mi + 1]
                                else {
                                    unreachable!("a fused post-op is a Bin")
                                };
                                // `swap`: the scan's value is the right operand.
                                let swap = *b == MRef::Reg(*out);
                                let other = if swap { msrc(a) } else { msrc(b) };
                                (Some((*op, other, swap)), dst_of(mi + 1, *pout))
                            } else {
                                (None, dst_of(mi, *out))
                            };
                            let src = if f.pre { MSrc::C(0.0) } else { msrc(a) };
                            // The scan's own value, when it is also a live-out
                            // that this call stores.
                            let keep = (f.keep
                                && outs.iter().any(|o| o.0 == *out && !o.1.is_null()))
                            .then(|| dst_of(mi, *out));
                            unsafe {
                                fch_scan_fused(
                                    dst,
                                    c,
                                    at,
                                    *row as usize,
                                    *init,
                                    *inclusive,
                                    &mut carries[*carry as usize],
                                    *sop,
                                    pre,
                                    src,
                                    post,
                                    keep,
                                )
                            };
                            continue;
                        }
                        None => {}
                    }
                }
                match op {
                    MicroOp::Bin { op, a, b, out } => {
                        let (a, b) = (msrc(a), msrc(b));
                        let dst = dst_of(mi, *out);
                        // Monomorphized over the shared table — the same
                        // kernel bodies as the unfused `Instr::Bin` arm.
                        macro_rules! chunk {
                            ($f:expr) => {
                                unsafe { fch2(dst, c, a, b, $f) }
                            };
                        }
                        dispatch_bin_kernel!(op, chunk);
                    }
                    MicroOp::Un { op, a, out } => {
                        let a = msrc(a);
                        let dst = dst_of(mi, *out);
                        macro_rules! chunk {
                            ($f:expr) => {
                                unsafe { fch1(dst, c, a, $f) }
                            };
                        }
                        dispatch_un_kernel!(op, chunk);
                    }
                    MicroOp::Neg { a, out } => {
                        let a = msrc(a);
                        unsafe { fch1(dst_of(mi, *out), c, a, |x| -x) };
                    }
                    MicroOp::Select { cond, a, b, out } => {
                        let (cv, av, bv) = (msrc(cond), msrc(a), msrc(b));
                        unsafe { fch_sel(dst_of(mi, *out), c, cv, av, bv) };
                    }
                    MicroOp::Mov { a, out } => {
                        let a = msrc(a);
                        unsafe { fch1(dst_of(mi, *out), c, a, |x| x) };
                    }
                    MicroOp::Load { a, src, out } => {
                        let fi = fs.inputs[*src as usize]
                            .index
                            .expect("a Load reads a folded index gather");
                        let a = msrc(a);
                        unsafe { fch_load(dst_of(mi, *out), c, at, a, bases[*src as usize], &fi) };
                    }
                    MicroOp::Scan {
                        op,
                        a,
                        init,
                        inclusive,
                        reverse,
                        row,
                        post,
                        carry,
                        out,
                    } => {
                        let a = msrc(a);
                        let dst = dst_of(mi, *out);
                        let (row, post, init, inclusive, reverse) =
                            (*row as usize, *post as usize, *init, *inclusive, *reverse);
                        let cv = &mut carries[*carry as usize..*carry as usize + post];
                        if post == 1 {
                            debug_assert!(!reverse, "a reverse scan has lanes");
                            let cv = &mut cv[0];
                            macro_rules! chunk {
                                ($f:expr) => {
                                    unsafe { fch_scan(dst, c, at, a, row, init, inclusive, cv, $f) }
                                };
                            }
                            dispatch_bin_kernel!(op, chunk);
                        } else {
                            macro_rules! chunk {
                                ($f:expr) => {
                                    unsafe {
                                        fch_scan_lanes(
                                            dst, c, at, a, row, post, init, inclusive, reverse, cv,
                                            $f,
                                        )
                                    }
                                };
                            }
                            dispatch_bin_kernel!(op, chunk);
                        }
                    }
                    MicroOp::Bin2 {
                        op1,
                        a,
                        b,
                        op2,
                        c: c3,
                        swap,
                        out,
                    } => {
                        let (av, bv, cv) = (msrc(a), msrc(b), msrc(c3));
                        let dst = dst_of(mi, *out);
                        // Under `element_type: "Float32"` the monomorphized
                        // closures below are hand-copied f64 arithmetic that
                        // never sees the precision, so compose the SAME two
                        // kernels in the SAME order out of the shared,
                        // precision-aware table instead. Float64 falls through
                        // to the untouched match.
                        if crate::precision::is_f32() {
                            let f1 = binary_kernel_of(*op1);
                            let f2 = binary_kernel_of(*op2);
                            if *swap {
                                unsafe { fch3(dst, c, av, bv, cv, |x, y, z| f2(z, f1(x, y))) };
                            } else {
                                unsafe { fch3(dst, c, av, bv, cv, |x, y, z| f2(f1(x, y), z)) };
                            }
                            continue;
                        }
                        use BinCode::{Add, Div, Ge, Gt, Le, Lt, Max, Min, Mul, Sub};
                        // Monomorphized composition of the same two kernel
                        // bodies, applied in the same order (t = op1(a, b);
                        // out = swap ? op2(c, t) : op2(t, c)).
                        macro_rules! b2 {
                            ($f1:expr, $f2:expr) => {
                                if *swap {
                                    unsafe { fch3(dst, c, av, bv, cv, |x, y, z| $f2(z, $f1(x, y))) }
                                } else {
                                    unsafe { fch3(dst, c, av, bv, cv, |x, y, z| $f2($f1(x, y), z)) }
                                }
                            };
                        }
                        let add = |x: f64, y: f64| x + y;
                        let sub = |x: f64, y: f64| x - y;
                        let mul = |x: f64, y: f64| x * y;
                        let div = |x: f64, y: f64| x / y;
                        let min = |x: f64, y: f64| x.min(y);
                        let max = |x: f64, y: f64| x.max(y);
                        let gt = |x: f64, y: f64| (x > y) as i32 as f64;
                        let ge = |x: f64, y: f64| (x >= y) as i32 as f64;
                        let lt = |x: f64, y: f64| (x < y) as i32 as f64;
                        let le = |x: f64, y: f64| (x <= y) as i32 as f64;
                        match (op1, op2) {
                            (Add, Add) => b2!(add, add),
                            (Add, Sub) => b2!(add, sub),
                            (Add, Mul) => b2!(add, mul),
                            (Add, Div) => b2!(add, div),
                            (Sub, Add) => b2!(sub, add),
                            (Sub, Sub) => b2!(sub, sub),
                            (Sub, Mul) => b2!(sub, mul),
                            (Sub, Div) => b2!(sub, div),
                            (Mul, Add) => b2!(mul, add),
                            (Mul, Sub) => b2!(mul, sub),
                            (Mul, Mul) => b2!(mul, mul),
                            (Mul, Div) => b2!(mul, div),
                            (Div, Add) => b2!(div, add),
                            (Div, Sub) => b2!(div, sub),
                            (Div, Mul) => b2!(div, mul),
                            (Div, Div) => b2!(div, div),
                            // Step 4b extended pairs (`bin2_pair_ok`): the
                            // multiply-into-mask and min/max clamp idioms the
                            // adjacency histogram showed material.
                            (Mul, Gt) => b2!(mul, gt),
                            (Mul, Ge) => b2!(mul, ge),
                            (Mul, Lt) => b2!(mul, lt),
                            (Mul, Le) => b2!(mul, le),
                            (Min, Max) => b2!(min, max),
                            (Max, Min) => b2!(max, min),
                            (Mul, Min) => b2!(mul, min),
                            (Min, Mul) => b2!(min, mul),
                            (Mul, Max) => b2!(mul, max),
                            (Max, Mul) => b2!(max, mul),
                            other => unreachable!("Bin2 pair not monomorphized ({other:?})"),
                        }
                    }
                    MicroOp::Bin3 {
                        op1,
                        a,
                        b,
                        op2,
                        c: c3,
                        swap2,
                        op3,
                        d: d4,
                        swap3,
                        out,
                    } => {
                        let (pa, pb, pc, pd) = (msrc_p(a), msrc_p(b), msrc_p(c3), msrc_p(d4));
                        let dst = dst_of(mi, *out);
                        // See `Bin2`: Float32 composes the three shared kernels
                        // in the same order rather than the f64-only closures.
                        if crate::precision::is_f32() {
                            let f1 = binary_kernel_of(*op1);
                            let f2 = binary_kernel_of(*op2);
                            let f3 = binary_kernel_of(*op3);
                            match (*swap2, *swap3) {
                                (false, false) => unsafe {
                                    fch4(dst, c, pa, pb, pc, pd, |x, y, z, w| {
                                        f3(f2(f1(x, y), z), w)
                                    })
                                },
                                (true, false) => unsafe {
                                    fch4(dst, c, pa, pb, pc, pd, |x, y, z, w| {
                                        f3(f2(z, f1(x, y)), w)
                                    })
                                },
                                (false, true) => unsafe {
                                    fch4(dst, c, pa, pb, pc, pd, |x, y, z, w| {
                                        f3(w, f2(f1(x, y), z))
                                    })
                                },
                                (true, true) => unsafe {
                                    fch4(dst, c, pa, pb, pc, pd, |x, y, z, w| {
                                        f3(w, f2(z, f1(x, y)))
                                    })
                                },
                            }
                            continue;
                        }
                        use BinCode::{Add, Div, Mul, Sub};
                        // Monomorphized composition of the same three kernel
                        // bodies applied in order:
                        // t1 = op1(a, b); t2 = swap2 ? op2(c, t1) : op2(t1, c);
                        // out = swap3 ? op3(d, t2) : op3(t2, d).
                        macro_rules! b3 {
                            ($f1:expr, $f2:expr, $f3:expr) => {
                                match (*swap2, *swap3) {
                                    (false, false) => unsafe {
                                        fch4(dst, c, pa, pb, pc, pd, |x, y, z, w| {
                                            $f3($f2($f1(x, y), z), w)
                                        })
                                    },
                                    (true, false) => unsafe {
                                        fch4(dst, c, pa, pb, pc, pd, |x, y, z, w| {
                                            $f3($f2(z, $f1(x, y)), w)
                                        })
                                    },
                                    (false, true) => unsafe {
                                        fch4(dst, c, pa, pb, pc, pd, |x, y, z, w| {
                                            $f3(w, $f2($f1(x, y), z))
                                        })
                                    },
                                    (true, true) => unsafe {
                                        fch4(dst, c, pa, pb, pc, pd, |x, y, z, w| {
                                            $f3(w, $f2(z, $f1(x, y)))
                                        })
                                    },
                                }
                            };
                        }
                        let add = |x: f64, y: f64| x + y;
                        let sub = |x: f64, y: f64| x - y;
                        let mul = |x: f64, y: f64| x * y;
                        let div = |x: f64, y: f64| x / y;
                        macro_rules! b3_op12 {
                            ($f3:expr) => {
                                match (op1, op2) {
                                    (Add, Add) => b3!(add, add, $f3),
                                    (Add, Sub) => b3!(add, sub, $f3),
                                    (Add, Mul) => b3!(add, mul, $f3),
                                    (Add, Div) => b3!(add, div, $f3),
                                    (Sub, Add) => b3!(sub, add, $f3),
                                    (Sub, Sub) => b3!(sub, sub, $f3),
                                    (Sub, Mul) => b3!(sub, mul, $f3),
                                    (Sub, Div) => b3!(sub, div, $f3),
                                    (Mul, Add) => b3!(mul, add, $f3),
                                    (Mul, Sub) => b3!(mul, sub, $f3),
                                    (Mul, Mul) => b3!(mul, mul, $f3),
                                    (Mul, Div) => b3!(mul, div, $f3),
                                    (Div, Add) => b3!(div, add, $f3),
                                    (Div, Sub) => b3!(div, sub, $f3),
                                    (Div, Mul) => b3!(div, mul, $f3),
                                    (Div, Div) => b3!(div, div, $f3),
                                    other => {
                                        unreachable!("Bin3 restricted to + - * / ({other:?})")
                                    }
                                }
                            };
                        }
                        match op3 {
                            Add => b3_op12!(add),
                            Sub => b3_op12!(sub),
                            Mul => b3_op12!(mul),
                            Div => b3_op12!(div),
                            other => unreachable!("Bin3 restricted to + - * / ({other:?})"),
                        }
                    }
                }
            }
            for (k, &(reg, optr)) in outs.iter().enumerate() {
                if optr.is_null() || is_direct(k) {
                    continue;
                }
                unsafe {
                    std::ptr::copy_nonoverlapping(
                        rp.add(reg as usize * cs) as *const f64,
                        optr.add(at),
                        c,
                    );
                }
            }
            if let Some(r) = &fs.reduce {
                if let Some(MicroOp::Bin2 {
                    op1,
                    a,
                    b,
                    op2,
                    c: MRef::Scal(z),
                    swap,
                    ..
                }) = fold_op.map(|mi| &fs.micro[mi])
                {
                    let (a, b, z) = (msrc(a), msrc(b), svals[*z as usize]);
                    use BinCode::{Add, Div, Mul, Sub};
                    // `g` composes the superop's two kernels in its order.
                    macro_rules! ff {
                        ($f1:expr, $f2:expr) => {{
                            let (f1, f2) = ($f1, $f2);
                            if *swap {
                                unsafe {
                                    fold_chunk_bin(
                                        red,
                                        r.n_inner,
                                        at,
                                        r.init,
                                        a,
                                        b,
                                        c,
                                        move |x, y| f2(z, f1(x, y)),
                                        |x: f64, y: f64| x + y,
                                    )
                                }
                            } else {
                                unsafe {
                                    fold_chunk_bin(
                                        red,
                                        r.n_inner,
                                        at,
                                        r.init,
                                        a,
                                        b,
                                        c,
                                        move |x, y| f2(f1(x, y), z),
                                        |x: f64, y: f64| x + y,
                                    )
                                }
                            }
                        }};
                    }
                    let add = |x: f64, y: f64| x + y;
                    let sub = |x: f64, y: f64| x - y;
                    let mul = |x: f64, y: f64| x * y;
                    let div = |x: f64, y: f64| x / y;
                    match (op1, op2) {
                        (Add, Mul) => ff!(add, mul),
                        (Sub, Mul) => ff!(sub, mul),
                        (Mul, Mul) => ff!(mul, mul),
                        (Div, Mul) => ff!(div, mul),
                        (Add, Div) => ff!(add, div),
                        (Sub, Div) => ff!(sub, div),
                        (Mul, Div) => ff!(mul, div),
                        (Div, Div) => ff!(div, div),
                        other => unreachable!(
                            "fold fusion admits a + - * / then * / superop ({other:?})"
                        ),
                    }
                } else if let Some(MicroOp::Bin { op, a, b, .. }) = fold_op.map(|mi| &fs.micro[mi])
                {
                    let (a, b) = (msrc(a), msrc(b));
                    use BinCode::{Add, Div, Mul, Sub};
                    macro_rules! ff {
                        ($g:expr, $f:expr) => {
                            unsafe { fold_chunk_bin(red, r.n_inner, at, r.init, a, b, c, $g, $f) }
                        };
                    }
                    let add = |x: f64, y: f64| x + y;
                    let sub = |x: f64, y: f64| x - y;
                    let mul = |x: f64, y: f64| x * y;
                    let div = |x: f64, y: f64| x / y;
                    match (op, &r.op) {
                        (Add, Add) => ff!(add, add),
                        (Sub, Add) => ff!(sub, add),
                        (Mul, Add) => ff!(mul, add),
                        (Div, Add) => ff!(div, add),
                        (Add, Mul) => ff!(add, mul),
                        (Sub, Mul) => ff!(sub, mul),
                        (Mul, Mul) => ff!(mul, mul),
                        (Div, Mul) => ff!(div, mul),
                        other => unreachable!("fold fusion admits + - * / into + * ({other:?})"),
                    }
                } else {
                    let v = unsafe { rp.add(r.reg as usize * cs) as *const f64 };
                    macro_rules! fold {
                        ($f:expr) => {
                            unsafe { fold_chunk(red, r.n_inner, at, r.init, v, c, $f) }
                        };
                    }
                    dispatch_bin_kernel!(&r.op, fold);
                }
            }
        }};
    }
    // A reduction over a few leading positions, one run each, folds each
    // chunk of its accumulator for every position before moving on, so the
    // accumulator chunk stays in the fastest cache; each output still folds
    // its terms in ascending position order.
    if let (Some(runs), Some(r)) = (&fs.interleave, &fs.reduce) {
        // A split runs the inner positions `[lo, hi)` of its window.
        let (lo, hi) = if win.period == 0 {
            (0, r.n_inner)
        } else {
            (win.lo, win.hi.min(r.n_inner))
        };
        let mut c0 = lo;
        while c0 < hi {
            let c = (hi - c0).min(cs);
            for run in runs {
                chunk!(&run.in_off, c0, run.out_off as usize + c0, c);
            }
            c0 += c;
        }
        return;
    }
    // Walk the schedule in execution order: a `Repeat` opens a frame over
    // its body, and each time the walk reaches the body's end it either
    // steps every offset once more and goes back to the body's start, or
    // (the last repetition done) takes the steps back off and closes.
    let nodes = &fs.schedule.nodes;
    frames.clear();
    in_delta.clear();
    in_delta.resize(n_shifted(fs), 0);
    in_off.clear();
    in_off.resize(in_delta.len(), 0);
    let mut out_delta = 0i64;
    let mut i = 0usize;
    // Elements of the schedule before node `i` (executed or stepped over).
    let mut g = 0usize;
    loop {
        if win.period == 0 && g >= win.hi {
            break;
        }
        while let Some(fr) = frames.last_mut()
            && i == fr.end
        {
            let RunNode::Repeat {
                count,
                body,
                out_step,
                in_step,
            } = &nodes[fr.node]
            else {
                unreachable!("a frame is opened by a Repeat node")
            };
            fr.left -= 1;
            if fr.left > 0 {
                out_delta += out_step;
                for (d, s) in in_delta.iter_mut().zip(in_step) {
                    *d += s;
                }
                i = fr.end - *body as usize;
                break;
            }
            let back = *count as i64 - 1;
            out_delta -= back * out_step;
            for (d, s) in in_delta.iter_mut().zip(in_step) {
                *d -= back * s;
            }
            frames.pop();
        }
        let run = match nodes.get(i) {
            None => break,
            Some(RunNode::Repeat {
                count,
                body,
                out_step,
                in_step,
            }) => {
                let rep = win.node_elems[i];
                let span = rep * *count as usize;
                if win.period == 0 && g + span <= win.lo {
                    // Wholly before the window: step over the subtree.
                    g += span;
                    i += 1 + *body as usize;
                    continue;
                }
                // Whole repetitions before the window are stepped over at
                // once; the frame then counts down only the rest, and closing
                // it still takes back `count - 1` steps in all.
                let skip = if win.period == 0 && win.lo > g {
                    (win.lo - g) / rep
                } else {
                    0
                };
                if skip > 0 {
                    out_delta += skip as i64 * out_step;
                    for (d, s) in in_delta.iter_mut().zip(in_step) {
                        *d += skip as i64 * s;
                    }
                    g += skip * rep;
                }
                frames.push(RepeatFrame {
                    node: i,
                    end: i + 1 + *body as usize,
                    left: *count - skip as u32,
                });
                i += 1;
                continue;
            }
            Some(RunNode::Run(run)) => run,
        };
        i += 1;
        let len = run.len as usize;
        let g0 = g;
        g += len;
        let Some(first) = win.piece(g0, len, 0) else {
            continue;
        };
        for ((o, &r), &d) in in_off.iter_mut().zip(&run.in_off).zip(in_delta.iter()) {
            *o = if r == GHOST_OFF { r } else { r + d };
        }
        let in_off: &[i64] = in_off;
        let out_off = (run.out_off as i64 + out_delta) as usize;
        let (mut done, mut end) = first;
        loop {
            if done >= end {
                match win.piece(g0, len, end) {
                    Some(next) => (done, end) = next,
                    None => break,
                }
                continue;
            }
            let c = (end - done).min(cs);
            chunk!(in_off, done, out_off + done, c);
            done += c;
        }
    }
}

/// Baseline-codegen instantiation of the fused chunk loop.
#[inline(never)]
#[allow(clippy::too_many_arguments)]
pub(super) unsafe fn exec_fused_runs_generic(
    fs: &FusedSpec,
    svals: &[f64],
    bases: &[*const f64],
    outs: &[(GroupIx, *mut f64)],
    red: *mut f64,
    idx: &[Option<IndexTable>],
    fregs: &mut [f64],
    cursor: &mut RunCursor,
    win: Window,
) {
    unsafe { exec_fused_runs(fs, svals, bases, outs, red, idx, fregs, cursor, win) }
}

/// AVX2 clone: identical Rust source compiled under `avx2` (+`fma` is NOT
/// enabled — LLVM must not contract mul+add into fused multiply-add, which
/// would change bits). Reached only when `is_x86_feature_detected!` proved
/// support, so the `unsafe` target-feature contract holds.
#[cfg(target_arch = "x86_64")]
#[target_feature(enable = "avx2")]
#[allow(clippy::too_many_arguments)]
pub(super) unsafe fn exec_fused_runs_avx2(
    fs: &FusedSpec,
    svals: &[f64],
    bases: &[*const f64],
    outs: &[(GroupIx, *mut f64)],
    red: *mut f64,
    idx: &[Option<IndexTable>],
    fregs: &mut [f64],
    cursor: &mut RunCursor,
    win: Window,
) {
    unsafe { exec_fused_runs(fs, svals, bases, outs, red, idx, fregs, cursor, win) }
}

/// AVX-512 clone (f+vl+dq+bw, all runtime-checked), selected only on request
/// (see `simd_level`): LLVM vectorizes these loops in 512-bit registers.
#[cfg(target_arch = "x86_64")]
#[target_feature(
    enable = "avx512f",
    enable = "avx512vl",
    enable = "avx512dq",
    enable = "avx512bw"
)]
#[allow(clippy::too_many_arguments)]
pub(super) unsafe fn exec_fused_runs_avx512(
    fs: &FusedSpec,
    svals: &[f64],
    bases: &[*const f64],
    outs: &[(GroupIx, *mut f64)],
    red: *mut f64,
    idx: &[Option<IndexTable>],
    fregs: &mut [f64],
    cursor: &mut RunCursor,
    win: Window,
) {
    unsafe { exec_fused_runs(fs, svals, bases, outs, red, idx, fregs, cursor, win) }
}

/// The SINGLE definition of micro-op scalar semantics: one element of one
/// [`MicroOp`], evaluated over a scalar register file. The cold per-element
/// reference executor (`refexec`, test-only) dispatches through this
/// function, so it cannot drift from what it checks. The hot
/// chunked executor ([`exec_fused_runs`]) applies the SAME kernels in the
/// same per-element order through its monomorphized chunk loops (a structure
/// a scalar evaluator cannot back without disturbing it); the A/B tests and
/// `simd_clone_bit_identity` pin it bit-identical to this definition.
///
/// Semantics (see [`MicroOp`] for the full contract):
/// * `Bin`/`Un` apply [`binary_kernel_of`] / [`unary_kernel_of`] — the same
///   fn-pointer tables the unfused instructions dispatch through.
/// * `Neg` is `x → -x`, NOT `0 - x` (which differs on signed zero).
/// * `Bin2`/`Bin3` operand orientation: the fused intermediate always feeds
///   the NEXT kernel, and `swap*` records which SIDE it enters on —
///   `t = op1(a, b); out = swap ? op2(c, t) : op2(t, c)` (likewise `swap2`
///   then `swap3` for `Bin3`). The constituent kernels are applied strictly
///   in order — never contracted into a hardware FMA, which would change
///   bits.
/// * `Scan` folds into `carries[carry + at % post]`, restarting at `init`
///   where the element's flat box offset `at` is at step 0 of its lane; the
///   caller visits the box in ascending flat order.
/// * `get` resolves an [`MRef`] operand (register / broadcast scalar / array
///   input at the current element, including the [`GHOST_OFF`] `+0.0` read).
///   Operand reads are pure, so `Select` reading only the taken operand is
///   value-identical to the chunked executor's load-both blend.
///
/// Test-only since the chunked executor became the only production one: its
/// single caller is the reference executor that checks it.
#[cfg(test)]
#[inline(always)]
pub(in crate::simulate_array::tape) fn eval_micro_op(
    op: &MicroOp,
    regs: &mut [f64],
    at: usize,
    carries: &mut [f64],
    get: impl Fn(&MRef, &[f64]) -> f64,
    load: impl Fn(GroupIx, f64) -> f64,
) {
    match op {
        MicroOp::Load { a, src, out } => {
            regs[*out as usize] = load(*src, get(a, regs));
        }
        MicroOp::Bin { op, a, b, out } => {
            regs[*out as usize] = binary_kernel_of(*op)(get(a, regs), get(b, regs));
        }
        MicroOp::Un { op, a, out } => {
            regs[*out as usize] = unary_kernel_of(*op)(get(a, regs));
        }
        MicroOp::Neg { a, out } => {
            regs[*out as usize] = -get(a, regs);
        }
        MicroOp::Select { cond, a, b, out } => {
            regs[*out as usize] = if get(cond, regs) != 0.0 {
                get(a, regs)
            } else {
                get(b, regs)
            };
        }
        MicroOp::Mov { a, out } => {
            regs[*out as usize] = get(a, regs);
        }
        MicroOp::Scan {
            op,
            a,
            init,
            inclusive,
            reverse,
            row,
            post,
            carry,
            out,
        } => {
            let (row, post) = (*row as usize, *post as usize);
            let acc = &mut carries[*carry as usize + at % post];
            let first = if *reverse { row - 1 } else { 0 };
            if (at / post) % row == first {
                *acc = *init;
            }
            let x = get(a, regs);
            if *inclusive {
                *acc = binary_kernel_of(*op)(*acc, x);
                regs[*out as usize] = *acc;
            } else {
                regs[*out as usize] = *acc;
                *acc = binary_kernel_of(*op)(*acc, x);
            }
        }
        MicroOp::Bin2 {
            op1,
            a,
            b,
            op2,
            c,
            swap,
            out,
        } => {
            let t = binary_kernel_of(*op1)(get(a, regs), get(b, regs));
            let cv = get(c, regs);
            regs[*out as usize] = if *swap {
                binary_kernel_of(*op2)(cv, t)
            } else {
                binary_kernel_of(*op2)(t, cv)
            };
        }
        MicroOp::Bin3 {
            op1,
            a,
            b,
            op2,
            c,
            swap2,
            op3,
            d,
            swap3,
            out,
        } => {
            let t1 = binary_kernel_of(*op1)(get(a, regs), get(b, regs));
            let cv = get(c, regs);
            let t2 = if *swap2 {
                binary_kernel_of(*op2)(cv, t1)
            } else {
                binary_kernel_of(*op2)(t1, cv)
            };
            let dv = get(d, regs);
            regs[*out as usize] = if *swap3 {
                binary_kernel_of(*op3)(dv, t2)
            } else {
                binary_kernel_of(*op3)(t2, dv)
            };
        }
    }
}

/// `MicroOp::Load` over a chunk: element `k` (box offset `at + k`) reads
/// `src` at the data position subscript `a` names, through `fi`, or the
/// zero ghost.
///
/// # Safety
/// `dst` must hold `c` elements, an array `a` must hold them too, and `src`
/// must address the gather's `fi.src_len` elements.
#[inline(always)]
unsafe fn fch_load(dst: *mut f64, c: usize, at: usize, a: MSrc, src: *const f64, fi: &FoldedIndex) {
    let (n, s_data) = (fi.n, fi.s_data);
    if fi.s_inner == 0 {
        let src = unsafe { src.add(fi.base) };
        let one = |v: f64| match data_subscript(v, n) {
            Some(p) => unsafe { *src.add(p * s_data) },
            None => 0.0,
        };
        match a {
            MSrc::P(sub) => {
                for k in 0..c {
                    unsafe { *dst.add(k) = one(*sub.add(k)) };
                }
            }
            MSrc::C(v) => {
                let x = one(v);
                for k in 0..c {
                    unsafe { *dst.add(k) = x };
                }
            }
        }
        return;
    }
    // The innermost position, carried instead of divided out per element.
    let (s_inner, base) = (fi.s_inner, fi.base);
    let wrap = fi.period * s_inner + base;
    let mut j = (at % fi.period) * s_inner + base;
    let mut k = 0usize;
    while k < c {
        // A stretch with no wrap inside it.
        let len = ((wrap - j) / s_inner).min(c - k);
        let row = unsafe { src.add(j) };
        match a {
            MSrc::P(sub) => {
                for q in 0..len {
                    let v = unsafe { *sub.add(k + q) };
                    let x = match data_subscript(v, n) {
                        Some(p) => unsafe { *row.add(p * s_data + q * s_inner) },
                        None => 0.0,
                    };
                    unsafe { *dst.add(k + q) = x };
                }
            }
            MSrc::C(v) => {
                let p = data_subscript(v, n);
                for q in 0..len {
                    let x = match p {
                        Some(p) => unsafe { *row.add(p * s_data + q * s_inner) },
                        None => 0.0,
                    };
                    unsafe { *dst.add(k + q) = x };
                }
            }
        }
        k += len;
        j += len * s_inner;
        if j == wrap {
            j = base;
        }
    }
}