cortiq-engine 0.8.82

//! The Qwen3.5-family hybrid on CUDA (Qwen3.5 / 3.6 / 3.8 dense: GDN layers
//! between gated full-attention layers with Q/K norms and a partial rotary
//! turn), behind the same token and batch graph contracts the dense arm of
//! `gpu_cuda.rs` serves. The pipeline's MTP rounds, prompt lookup and server
//! run on it unchanged.
//!
//! Two paths:
//! - **small** (a decode step, a speculative verify window, an MTP warm-up:
//!   up to `NMAX` rows): weights read in their codec by `hy_mvn_N`, which
//!   reads a matrix once for all the rows of the window; every row goes
//!   through the arithmetic of a one-row step in the same order, so a verify
//!   window's rows agree to the bit with the decode steps they stand for, and
//!   a greedy answer is the same with drafts and without;
//! - **big** (prompt chunks): each weight expanded to an f16 plane per chunk,
//!   cuBLAS on the tensor cores, attention as two batched GEMMs a KV group.
//!
//! GDN state (the delta-rule matrix and the conv ring) lives on the card per
//! (sequence, layer). A verify window keeps the state it started from and a
//! tape of its rows' inputs; `gdn_spec_restore` puts the start back and
//! replays the accepted rows.

use std::collections::HashMap;
use std::ffi::c_void;
use std::sync::Arc;
use std::sync::atomic::Ordering;

use cortiq_core::CmfModel;

use super::{Buf, Ctx, DevMat, Gemm, KvMirror, Res, State, compile, cs, cu, debug, decline, derr, func, prec};
use crate::gpu::{BatchGraphOutcome, GraphAct, GraphAttn, GraphFfn, GraphLayer, GraphW, SpecTail, TokenGraphOutcome};

const HYB_SRC: &str = include_str!("hyb.cu");

/// The DFlash 2 block on the card, on this arm's kernels.
pub(super) mod dflash;
pub(super) mod dory;
/// Rows of the small path.
pub(super) const NMAX: usize = 16;
/// Rows of a prompt chunk read past a seal (`big` over sealed layers): the far
/// field's mirror keeps room for them after its near rows.
pub(super) const SEAL_ROWS: usize = 1024;
/// A prompt chunk's weight expands to f16 in bands of at most this many bytes
/// (`CMF_HYB_WD_MB`): the gate/up pair of a 27B layer is 356 MB whole.
fn wd_cap() -> usize {
    static CAP: std::sync::OnceLock<usize> = std::sync::OnceLock::new();
    *CAP.get_or_init(|| std::env::var("CMF_HYB_WD_MB").ok().and_then(|v| v.parse::<usize>().ok()).unwrap_or(64) << 20)
}
/// Rows of a chunk one attention GEMM pass takes at most (`CMF_HYB_ATT_ROWS`):
/// its score and weight buffers grow with rows x keys.
fn att_rows() -> usize {
    static N: std::sync::OnceLock<usize> = std::sync::OnceLock::new();
    *N.get_or_init(|| std::env::var("CMF_HYB_ATT_ROWS").ok().and_then(|v| v.parse::<usize>().ok()).filter(|&n| n >= 16).unwrap_or(128))
}
/// Stages of `CMF_CUDA_PROF=1` on the small path (each closed by a stream
/// sync, so the numbers are the card's).
const HSTAGES: [&str; 12] = [
    "norms", "gdn in", "gdn core", "gdn out", "attn in", "attn core", "attn out", "gate+up", "silu", "down",
    "head", "copies",
];
static HSTAGE_NS: [std::sync::atomic::AtomicU64; 12 * 17] = [const { std::sync::atomic::AtomicU64::new(0) }; 12 * 17];
static HCALLS_N: [std::sync::atomic::AtomicU64; 17] = [const { std::sync::atomic::AtomicU64::new(0) }; 17];
static HCALLS: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
thread_local! {
    /// The row count of the small-path call being timed (`hst`).
    static HROWS: std::cell::Cell<usize> = const { std::cell::Cell::new(1) };
}
/// Small-path graphs recorded and replayed (`CMF_CUDA_DEBUG=1` reports them).
static GRAPH_REC: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
static GRAPH_RUN: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
/// Stages of the prompt path (`CMF_CUDA_PROF=1`), summed over its chunks.
const BSTAGES: [&str; 8] = ["norms", "expand", "gemm", "gdn glue", "gdn scan", "attention", "act", "copies"];
static BSTAGE_NS: [std::sync::atomic::AtomicU64; 8] = [const { std::sync::atomic::AtomicU64::new(0) }; 8];
static BROWS: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);

fn bst(c: &Ctx, s: usize, t0: &std::cell::Cell<std::time::Instant>) -> Res<()> {
    if super::prof_on() {
        c.sync()?;
        BSTAGE_NS[s].fetch_add(t0.get().elapsed().as_nanos() as u64, Ordering::Relaxed);
        t0.set(std::time::Instant::now());
    }
    Ok(())
}

fn hst(c: &Ctx, s: usize, t0: &mut std::time::Instant) -> Res<()> {
    if super::prof_on() {
        c.sync()?;
        let n = HROWS.with(|r| r.get()).min(16);
        HSTAGE_NS[n * 12 + s].fetch_add(t0.elapsed().as_nanos() as u64, Ordering::Relaxed);
        *t0 = std::time::Instant::now();
    }
    Ok(())
}

/// The small path's stage times, per call (`CMF_CUDA_PROF=1`).
pub(super) fn report() {
    if debug() {
        eprintln!(
            "cuda hybrid graphs: {} recorded, {} replays",
            GRAPH_REC.load(Ordering::Relaxed),
            GRAPH_RUN.load(Ordering::Relaxed)
        );
    }
    let calls = HCALLS.load(Ordering::Relaxed);
    if calls == 0 || !super::prof_on() {
        return;
    }
    eprintln!("cuda hybrid small path: {calls} calls");
    for n in 1..=16 {
        let k = HCALLS_N[n].load(Ordering::Relaxed);
        if k == 0 {
            continue;
        }
        let row: Vec<String> = HSTAGES
            .iter()
            .enumerate()
            .map(|(i, name)| format!("{name} {:.3}", HSTAGE_NS[n * 12 + i].load(Ordering::Relaxed) as f64 / 1e6 / k as f64))
            .collect();
        let tot: f64 = (0..12).map(|i| HSTAGE_NS[n * 12 + i].load(Ordering::Relaxed) as f64 / 1e6 / k as f64).sum();
        eprintln!("  {n:>2} rows, {k} calls, {tot:.2} ms a call: {}", row.join(" | "));
    }
    let rows = BROWS.load(Ordering::Relaxed);
    if rows > 0 {
        eprintln!("cuda hybrid prompt path: {rows} rows");
        for (name, ns) in BSTAGES.iter().zip(&BSTAGE_NS) {
            eprintln!("  {name:>10}: {:8.3} ms per 1000 rows", ns.load(Ordering::Relaxed) as f64 / 1e6 / rows as f64 * 1000.0);
        }
    }
}

/// `struct HyJob` of `hyb.cu`.
#[repr(C)]
#[derive(Clone, Copy)]
struct HyJob {
    w: u64,
    aux0: u64,
    aux1: u64,
    y: u64,
    res: u64,
    rows: i32,
    kind: i32,
    blk0: i32,
    ystride: i32,
}

pub(super) struct HFuncs {
    /// `hy_mmp_{NT}_{TB}_{KS}` by [NT - 1][TB - 1][KS == 16].
    mmp: [[[cs::CUfunction; 2]; 2]; 2],
    split: cs::CUfunction,
    expand_p: cs::CUfunction,
    rmsnorm: cs::CUfunction,
    to_half: cs::CUfunction,
    ring: cs::CUfunction,
    scan: cs::CUfunction,
    front: cs::CUfunction,
    out32: cs::CUfunction,
    rmsnorm2: cs::CUfunction,
    conv32: cs::CUfunction,
    prep32: cs::CUfunction,
    gb32: cs::CUfunction,
    rs_scan: cs::CUfunction,
    rs_ring: cs::CUfunction,
    scan4: cs::CUfunction,
    rs_scan4: cs::CUfunction,
    gather: cs::CUfunction,
    argmax: cs::CUfunction,
    silu_pair: cs::CUfunction,
    repack: cs::CUfunction,
    qk: cs::CUfunction,
    scores: cs::CUfunction,
    values: cs::CUfunction,
    finish: cs::CUfunction,
    attn_out: cs::CUfunction,
    softmax_h: cs::CUfunction,
    far_insert: cs::CUfunction,
    far_prep: cs::CUfunction,
    far_prep_b: cs::CUfunction,
    expand_pr: cs::CUfunction,
    far_rows: cs::CUfunction,
    softmax_hf: cs::CUfunction,
    attn_out_f: cs::CUfunction,
    far_insert_n: cs::CUfunction,
}
unsafe impl Send for HFuncs {}
unsafe impl Sync for HFuncs {}

fn load_funcs(c: &Ctx, hpk: usize, hd: usize) -> Res<HFuncs> {
    if hd % 32 != 0 || hd > 512 || hpk == 0 || hpk > 16 {
        return Err(format!("attention geometry {hpk} heads per KV head x {hd} outside the kernels"));
    }
    let module = compile(HYB_SRC, vec![format!("-DHPK={hpk}"), format!("-DHD={hd}")], c.arch)?;
    let mut mmp = [[[std::ptr::null_mut(); 2]; 2]; 2];
    for nt in 1..=2usize {
        for tb in 1..=2usize {
            for (ki, ks) in [8usize, 16].into_iter().enumerate() {
                mmp[nt - 1][tb - 1][ki] = func(module, &format!("hy_mmp_{nt}_{tb}_{ks}"))?;
            }
        }
    }
    Ok(HFuncs {
        mmp,
        split: func(module, "hy_split")?,
        expand_p: func(module, "hy_expand_p")?,
        rmsnorm: func(module, "hy_rmsnorm")?,
        to_half: func(module, "hy_to_half")?,
        ring: func(module, "hy_gdn_ring")?,
        scan: func(module, "hy_gdn_scan")?,
        front: func(module, "hy_gdn_front")?,
        out32: func(module, "hy_gdn_out32")?,
        rmsnorm2: func(module, "hy_rmsnorm2")?,
        conv32: func(module, "hy_gdn_conv32")?,
        prep32: func(module, "hy_gdn_prep32")?,
        gb32: func(module, "hy_gdn_gb32")?,
        rs_scan: func(module, "hy_gdn_restore_scan")?,
        rs_ring: func(module, "hy_gdn_restore_ring")?,
        scan4: func(module, "hy_gdn_scan4")?,
        rs_scan4: func(module, "hy_gdn_restore_scan4")?,
        gather: func(module, "hy_embed_gather")?,
        argmax: func(module, "hy_argmax")?,
        silu_pair: func(module, "hy_silu_pair_half")?,
        repack: func(module, "hy_repack")?,
        qk: func(module, "hy_qk")?,
        scores: func(module, "hy_scores")?,
        values: func(module, "hy_values")?,
        finish: func(module, "hy_finish")?,
        attn_out: func(module, "hy_attn_out")?,
        softmax_h: func(module, "hy_softmax_h")?,
        far_insert: func(module, "hy_far_insert")?,
        far_prep: func(module, "hy_far_prep")?,
        far_prep_b: func(module, "hy_far_prep_b")?,
        expand_pr: func(module, "hy_expand_pr")?,
        far_rows: func(module, "hy_far_rows")?,
        softmax_hf: func(module, "hy_softmax_hf")?,
        attn_out_f: func(module, "hy_attn_out_f")?,
        far_insert_n: func(module, "hy_far_insert_n")?,
    })
}

/// (matrix, output, residual or 0, output row stride, rows to compute)
type JobSpec<'a> = (&'a DevMat, u64, u64, usize, usize);

/// One launch's q4tp matrices for the tensor-core kernel: job tables for one
/// and two tiles a block, and the split of the groups (`KS` warps) the
/// arithmetic of every output depends on.
struct MmaSet {
    bufs: [Buf; 2],
    blocks: [u32; 2],
    n: i32,
    /// 16: a matrix of under 16 384 rows (more warps a tile); 8 otherwise.
    ks16: bool,
    /// One tile a block when the launch's matrices are short (K/V).
    tb1: bool,
    /// One tile a block for a one-row call over 12 288 rows or more: the
    /// tail of a launch costs less than half the tiles' parallelism there
    /// (3090: gate/up 801 against 745 GB/s, GDN input 746 / 679, head
    /// 863 / 839; 5120-row matrices keep two). The arithmetic does not
    /// depend on it.
    tb1_n1: bool,
}

fn mmaset(list: &[JobSpec]) -> Res<MmaSet> {
    mmaset_kind(list, 6)
}

/// One gate/up pair (see `hmat_pair`) whose output is silu(gate) * up,
/// written as the two bf16 planes `p0`, `p1` (`pstride` a row).
fn mmaset_pair(m: &DevMat, p0: u64, p1: u64, pstride: usize) -> Res<MmaSet> {
    mmaset_kind(&[(m, p0, p1, pstride, m.rows)], 6 | 0x100)
}

fn mmaset_kind(list: &[JobSpec], kind: i32) -> Res<MmaSet> {
    let t = std::time::Instant::now();
    let r = mmaset_kind_(list, kind);
    SETS_NS.fetch_add(t.elapsed().as_nanos() as u64, Ordering::Relaxed);
    r
}

fn mmaset_kind_(list: &[JobSpec], kind: i32) -> Res<MmaSet> {
    if list.iter().any(|(m, ..)| m.kind != 6 || m.cols % 32 != 0) {
        return Err("a tensor-core job over a matrix that is not repacked q4tp".into());
    }
    let max_rows = list.iter().map(|&(_, _, _, _, r)| r).max().unwrap_or(0);
    let mut bufs = Vec::with_capacity(2);
    let mut blocks = [0u32; 2];
    for tb in 1..=2usize {
        let mut jobs = Vec::with_capacity(list.len());
        let mut blk = 0u32;
        for &(m, y, res, ystride, rows) in list {
            jobs.push(HyJob {
                w: m.w.ptr,
                aux0: m.aux0(),
                aux1: m.aux1(),
                y,
                res,
                rows: rows as i32,
                kind,
                blk0: blk as i32,
                ystride: ystride as i32,
            });
            blk += rows.div_ceil(16).div_ceil(tb) as u32;
        }
        blocks[tb - 1] = blk;
        bufs.push(Buf::from_slice(&jobs)?);
    }
    let b1 = bufs.pop().unwrap();
    let b0 = bufs.pop().unwrap();
    // measured on a 3090 (q4tp, 1/8/16 rows): KS 8 wins from ~16k rows
    // (gate/up 67.5/71.6/124 us against 69.4/76.3/146), KS 16 below it
    // (down 5120 x 17408: 61.1/66.8/113 against 76.3/79.3/117)
    let total_rows: usize = list.iter().map(|&(_, _, _, _, r)| r).sum();
    Ok(MmaSet { bufs: [b0, b1], blocks, n: list.len() as i32, ks16: max_rows < 16384, tb1: max_rows <= 2048, tb1_n1: total_rows >= 12288 })
}

/// The row planes a tensor-core launch reads: two bf16 parts of each f32
/// (`put3` in hyb.cu), `stride` elements a row.
#[derive(Clone, Copy)]
struct Planes {
    p0: u64,
    p1: u64,
    p2: u64,
    stride: usize,
}

impl MmaSet {
    /// The window's rows of x (as split planes) through every matrix.
    unsafe fn run(&self, c: &Ctx, fx: &HFuncs, p: &Planes, cols: usize, n: usize) -> Res<()> {
        let nt = if n <= 8 { 1 } else { 2 };
        let tb = if self.tb1 || (n == 1 && self.tb1_n1) { 1 } else { 2 };
        let ks = if self.ks16 { 16 } else { 8 };
        let f = fx.mmp[nt - 1][tb - 1][self.ks16 as usize];
        let (nj, xs_i, n_i, cols_i) = (self.n, p.stride as i32, n as i32, cols as i32);
        unsafe {
            c.launch(
                f,
                (self.blocks[tb - 1], 1),
                (32 * ks) as u32,
                0,
                kargs!(self.bufs[tb - 1].ptr, nj, p.p0, p.p1, p.p2, xs_i, n_i, cols_i),
            )
        }
    }
}

/// x [n][xstride] f32 into the planes, the first `cols` columns (for an
/// input no kernel of this arm writes as planes itself).
unsafe fn split(c: &Ctx, fx: &HFuncs, p: &Planes, x: u64, xstride: usize, cols: usize, n: usize) -> Res<()> {
    let (xs_i, cols_i, os_i, n_i) = (xstride as i32, cols as i32, p.stride as i32, n as i32);
    let tot = (n * cols) as u64;
    unsafe { c.launch(fx.split, (tot.div_ceil(256) as u32, 1), 256, 0, kargs!(x, xs_i, cols_i, p.p0, p.p1, p.p2, os_i, n_i)) }
}

/// A q4tp tensor in the tensor-core layout (see `hyb.cu`), on the host: the
/// reference `hy_repack` is tested against. The nibbles as
/// [tile][group][lane] pairs of words, the codes [tile][group][16 rows], each
/// row's ladder as its two numbers (lo, step). Rows are padded to a multiple
/// of 16 with zero weights.
#[cfg(test)]
fn repack_q4tp(bytes: &[u8], rows: usize, cols: usize) -> Res<(Vec<u32>, Vec<u8>, Vec<f32>)> {
    repack_rows(&[(bytes, rows)], rows, cols, &|r| (0, r))
}

/// `repack_q4tp` over rows drawn from several q4tp payloads of one width:
/// logical row r is row `map(r).1` of payload `srcs[map(r).0]` (a payload
/// and its row count).
#[cfg(test)]
fn repack_rows(srcs: &[(&[u8], usize)], rows: usize, cols: usize, map: &(dyn Fn(usize) -> (usize, usize) + Sync)) -> Res<(Vec<u32>, Vec<u8>, Vec<f32>)> {
    use cortiq_core::quant::{GROUP_SIZE, f16_to_f32, q4tp_code, q4tp_sections};
    if cols % GROUP_SIZE != 0 {
        return Err("q4tp width not a multiple of 32".into());
    }
    let gpr = cols / GROUP_SIZE;
    // per payload: (nibble plane, params, codes, code stride)
    let mut planes = Vec::with_capacity(srcs.len());
    for &(bytes, srows) in srcs {
        let (params_off, codes_off, stride) = q4tp_sections(srows, cols);
        if bytes.len() < codes_off + srows * stride {
            return Err("q4tp payload shorter than its planes".into());
        }
        planes.push((&bytes[..params_off], &bytes[params_off..params_off + srows * 4], &bytes[codes_off..], stride, srows));
    }
    for r in 0..rows {
        let (i, sr) = map(r);
        if i >= planes.len() || sr >= planes[i].4 {
            return Err("repack: a row past its payload".into());
        }
    }
    let ntile = rows.div_ceil(16);
    let mut nib = vec![0u32; ntile * gpr * 64];
    let mut codes = vec![0u8; ntile * gpr * 16];
    // a row's ladder is exp2(lo + code * step) (q4tp_ladder's rungs, up to
    // rounding): the kernels take its two numbers
    let mut lad = vec![0f32; rows * 2];
    for r in 0..rows {
        let (i, sr) = map(r);
        let p = planes[i].1;
        lad[r * 2] = f16_to_f32(u16::from_le_bytes([p[sr * 4], p[sr * 4 + 1]]));
        lad[r * 2 + 1] = f16_to_f32(u16::from_le_bytes([p[sr * 4 + 2], p[sr * 4 + 3]]));
    }
    let threads = std::thread::available_parallelism().map_or(4, |n| n.get()).clamp(1, 32);
    let per = ntile.div_ceil(threads).max(1);
    let planes = &planes;
    std::thread::scope(|sc| {
        for (ci, (nch, cch)) in nib.chunks_mut(per * gpr * 64).zip(codes.chunks_mut(per * gpr * 16)).enumerate() {
            sc.spawn(move || {
                let t0 = ci * per;
                // the tile's 16 rows: their nibble planes and code rows
                let mut rsrc: [Option<(&[u8], &[u8])>; 16] = [None; 16];
                for (ti, (tn, tc)) in nch.chunks_mut(gpr * 64).zip(cch.chunks_mut(gpr * 16)).enumerate() {
                    let t = t0 + ti;
                    for (i, slot) in rsrc.iter_mut().enumerate() {
                        let r = t * 16 + i;
                        *slot = (r < rows).then(|| {
                            let (pi, sr) = map(r);
                            let (nbp, _, cp, stride, _) = planes[pi];
                            (&nbp[sr * gpr * 16..(sr + 1) * gpr * 16], &cp[sr * stride..(sr + 1) * stride])
                        });
                    }
                    for g in 0..gpr {
                        // the group's four words of each of the 16 rows (a
                        // missing row: nibbles 8, a zero weight)
                        let mut w = [[0x8888_8888u32; 4]; 16];
                        for (i, row) in rsrc.iter().enumerate() {
                            if let Some((nb, _)) = row {
                                let b = &nb[g * 16..g * 16 + 16];
                                for q in 0..4 {
                                    w[i][q] = u32::from_le_bytes([b[q * 4], b[q * 4 + 1], b[q * 4 + 2], b[q * 4 + 3]]);
                                }
                            }
                        }
                        for l in 0..32 {
                            let (gid, tig) = (l >> 2, l & 3);
                            for s in 0..2 {
                                let mut x = 0u32;
                                for m in 0..4 {
                                    // a_m: row gid (+8 for odd m), k = 16s + 2tig (+8 for m >= 2):
                                    // word 2s + (m >> 1) of the row, its byte tig
                                    let byte = (w[gid + (m & 1) * 8][2 * s + (m >> 1)] >> (8 * tig)) & 0xFF;
                                    x |= (byte & 0xF) << (4 * m);
                                    x |= (byte >> 4) << (4 * m + 16);
                                }
                                tn[(g * 32 + l) * 2 + s] = x;
                            }
                        }
                        for (i, row) in rsrc.iter().enumerate() {
                            tc[g * 16 + i] = row.map_or(0, |(_, cr)| q4tp_code(cr, g) as u8);
                        }
                    }
                }
            });
        }
    });
    Ok((nib, codes, lad))
}

/// `count` GEMMs of one shape: A and B f16 (A transposed when `ta`), C f32,
/// f32 accumulation; operand i sits `stride_*` elements after i - 1.
#[allow(clippy::too_many_arguments)]
unsafe fn gemm_batched_h(
    c: &Ctx,
    ta: bool,
    m: usize,
    n: usize,
    k: usize,
    alpha: f32,
    a: u64,
    lda: usize,
    stride_a: usize,
    b: u64,
    ldb: usize,
    stride_b: usize,
    cptr: u64,
    ldc: usize,
    stride_c: usize,
    count: usize,
) -> Res<()> {
    use super::bl;
    let op = |t| if t { bl::cublasOperation_t::CUBLAS_OP_T } else { bl::cublasOperation_t::CUBLAS_OP_N };
    let beta = 0.0f32;
    let st = unsafe {
        bl::cublasGemmStridedBatchedEx(
            c.blas,
            op(ta),
            op(false),
            m as i32,
            n as i32,
            k as i32,
            std::ptr::from_ref(&alpha).cast(),
            a as *const c_void,
            bl::cudaDataType::CUDA_R_16F,
            lda as i32,
            stride_a as i64,
            b as *const c_void,
            bl::cudaDataType::CUDA_R_16F,
            ldb as i32,
            stride_b as i64,
            std::ptr::from_ref(&beta).cast(),
            cptr as *mut c_void,
            bl::cudaDataType::CUDA_R_32F,
            ldc as i32,
            stride_c as i64,
            count as i32,
            bl::cublasComputeType_t::CUBLAS_COMPUTE_32F,
            bl::cublasGemmAlgo_t::CUBLAS_GEMM_DEFAULT_TENSOR_OP,
        )
    };
    if st as u32 != 0 {
        return Err(format!("cuBLAS f16 batched gemm {count} x {m}x{n}x{k}: {st:?}"));
    }
    Ok(())
}

/// A q4tp matrix (one payload, or the gate/up pair of two) uploaded as it
/// is stored and repacked on the card (`hy_repack`): (nibbles, codes, row
/// params) of `rows` logical rows.
fn repack_on_card(c: &Ctx, fx: &HFuncs, srcs: &[&[u8]], src_rows: usize, rows: usize, cols: usize) -> Res<(Buf, Buf, Buf)> {
    use cortiq_core::quant::{GROUP_SIZE, f16_to_f32, q4tp_sections};
    if cols % GROUP_SIZE != 0 || srcs.is_empty() || srcs.len() > 2 {
        return Err("repack: a q4tp payload (or a pair) 32 columns wide".into());
    }
    let pair = srcs.len() == 2;
    let gpr = cols / GROUP_SIZE;
    let (params_off, codes_off, stride) = q4tp_sections(src_rows, cols);
    for b in srcs {
        if b.len() < codes_off + src_rows * stride {
            return Err("q4tp payload shorter than its planes".into());
        }
    }
    let map = |r: usize| -> (usize, usize) { if pair { (((r & 15) >= 8) as usize, (r >> 4) * 8 + (r & 7)) } else { (0, r) } };
    let mut lad = vec![0f32; rows * 2];
    for r in 0..rows {
        let (i, sr) = map(r);
        let p = &srcs[i][params_off + sr * 4..params_off + sr * 4 + 4];
        lad[r * 2] = f16_to_f32(u16::from_le_bytes([p[0], p[1]]));
        lad[r * 2 + 1] = f16_to_f32(u16::from_le_bytes([p[2], p[3]]));
    }
    // the stored planes on the card for the kernel's reads
    let up = |b: &[u8]| -> Res<(Buf, Buf)> {
        Ok((Buf::from_slice(&b[..params_off])?, Buf::from_slice(&b[codes_off..codes_off + src_rows * stride])?))
    };
    let (na, ca) = up(srcs[0])?;
    let second = if pair { Some(up(srcs[1])?) } else { None };
    let (nb_p, cb_p) = second.as_ref().map_or((na.ptr, ca.ptr), |(n, c)| (n.ptr, c.ptr));
    let ntile = rows.div_ceil(16);
    let (t, cc, ld) = mat_planes(ntile * gpr * 64 * 4, ntile * gpr * 16, &lad)?;
    let (stride_i, pair_i, rows_i, cols_i) = (stride as i32, pair as i32, rows as i32, cols as i32);
    let threads = (ntile * gpr * 32) as u64;
    unsafe {
        c.launch(fx.repack, (threads.div_ceil(256) as u32, 1), 256, 0, kargs!(na.ptr, ca.ptr, nb_p, cb_p, stride_i, pair_i, t.ptr, cc.ptr, rows_i, cols_i))?;
    }
    // the stored planes are freed on return: the repack must be done
    c.sync()?;
    Ok((t, cc, ld))
}

thread_local! {
    /// A plan build's weight arena (`build`): base, capacity, used. Matrices
    /// come out of it while it has room, so the card's rounding of each
    /// allocation (to 2 MiB) costs once instead of once a matrix.
    static ARENA: std::cell::Cell<(u64, usize, usize)> = const { std::cell::Cell::new((0, 0, 0)) };
}

/// The bytes `mat_planes` takes for a rows x cols q4tp matrix.
fn planes_bytes(rows: usize, cols: usize) -> usize {
    let (ntile, gpr) = (rows.div_ceil(16), cols / 32);
    let cc_at = (ntile * gpr * 256).next_multiple_of(256);
    let lad_at = (cc_at + ntile * gpr * 16).next_multiple_of(256);
    (lad_at + rows * 8 + 64).next_multiple_of(256)
}

fn arena_view(bytes: usize) -> Option<Buf> {
    ARENA.with(|a| {
        let (p, cap, used) = a.get();
        if p == 0 || used + bytes > cap {
            return None;
        }
        a.set((p, cap, used + bytes.next_multiple_of(256)));
        Some(Buf { ptr: p + used as u64, bytes, owned: false, slab: None })
    })
}

/// A repacked matrix's three planes (tiles, rung codes, ladders) in one
/// allocation (the plan's arena when it has room): the tiles hold it, the
/// other two are views into it.
fn mat_planes(t_bytes: usize, cc_bytes: usize, lad: &[f32]) -> Res<(Buf, Buf, Buf)> {
    let cc_at = t_bytes.next_multiple_of(256);
    let lad_at = (cc_at + cc_bytes).next_multiple_of(256);
    let lad_bytes = std::mem::size_of_val(lad) + 64;
    let all = match arena_view(lad_at + lad_bytes) {
        Some(v) => v,
        None => Buf::new(lad_at + lad_bytes)?,
    };
    let (cc, ld) = (all.view(cc_at, cc_bytes), all.view(lad_at, lad_bytes));
    unsafe { cu::memcpy_htod_sync(ld.ptr, lad) }.map_err(derr)?;
    Ok((all, cc, ld))
}

/// `repack_on_card` for the leading `rows` rows of one q4tp payload, its
/// three sections given apart (nibbles, ladders, codes of those rows), so
/// they go to the card straight from the file's pages.
fn repack_rows_on_card(c: &Ctx, fx: &HFuncs, nib: &[u8], params: &[u8], codes: &[u8], stride: usize, rows: usize, cols: usize) -> Res<(Buf, Buf, Buf)> {
    use cortiq_core::quant::{GROUP_SIZE, f16_to_f32};
    let gpr = cols / GROUP_SIZE;
    if cols % GROUP_SIZE != 0 || nib.len() < rows * gpr * 16 || params.len() < rows * 4 || codes.len() < rows * stride {
        return Err("repack: q4tp sections shorter than their rows".into());
    }
    let mut lad = vec![0f32; rows * 2];
    for r in 0..rows {
        let p = &params[r * 4..r * 4 + 4];
        lad[r * 2] = f16_to_f32(u16::from_le_bytes([p[0], p[1]]));
        lad[r * 2 + 1] = f16_to_f32(u16::from_le_bytes([p[2], p[3]]));
    }
    let (na, ca) = (Buf::from_slice(&nib[..rows * gpr * 16])?, Buf::from_slice(&codes[..rows * stride])?);
    let ntile = rows.div_ceil(16);
    let (t, cc, ld) = mat_planes(ntile * gpr * 64 * 4, ntile * gpr * 16, &lad)?;
    let (stride_i, pair_i, rows_i, cols_i) = (stride as i32, 0i32, rows as i32, cols as i32);
    let threads = (ntile * gpr * 32) as u64;
    unsafe {
        c.launch(fx.repack, (threads.div_ceil(256) as u32, 1), 256, 0, kargs!(na.ptr, ca.ptr, na.ptr, ca.ptr, stride_i, pair_i, t.ptr, cc.ptr, rows_i, cols_i))?;
    }
    c.sync()?;
    Ok((t, cc, ld))
}

/// Plan-build time spent repacking and uploading weights (ns).
static REPACK_NS: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
static UPLOAD_NS: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
static HMAT_NS: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
static SETS_NS: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);

/// A graph weight resolved for this arm: q4tp repacked for the tensor-core
/// kernels (cached here, apart from the dense arm's layout), f32 as the
/// dense arm keeps it.
fn hmat(c: &Ctx, st: &mut State, model: &Arc<CmfModel>, gw: &GraphW, rows: usize, cols: usize) -> Res<Arc<DevMat>> {
    let t = std::time::Instant::now();
    let r = hmat_(c, st, model, gw, rows, cols);
    HMAT_NS.fetch_add(t.elapsed().as_nanos() as u64, Ordering::Relaxed);
    r
}

fn hmat_(c: &Ctx, st: &mut State, model: &Arc<CmfModel>, gw: &GraphW, rows: usize, cols: usize) -> Res<Arc<DevMat>> {
    if gw.kind != 6 {
        if gw.kind == 4 {
            return st.mat(c, model, gw, rows, cols);
        }
        return Err(format!("weight codec kind {} has no hybrid kernel", gw.kind));
    }
    if gw.prism != crate::gpu::GraphPrismOp::None || gw.affine {
        return Err("Prism / affine weight".into());
    }
    let key = (model.uid(), gw.idx);
    if let Some(m) = st.hyb.mats.get(&key) {
        if m.rows == rows && m.cols == cols {
            return Ok(m.clone());
        }
    }
    let entry = model.tensors.get(gw.idx).ok_or("tensor index out of range")?;
    if entry.shape.first() != Some(&rows) || entry.shape.get(1) != Some(&cols) {
        return Err(format!("tensor {} is {:?}, the graph wants {rows}x{cols}", entry.name, entry.shape));
    }
    let bytes = model.entry_bytes(entry);
    let budget = super::vram_budget();
    let held = c.resident.load(Ordering::Relaxed);
    if held + bytes.len() as u64 > budget {
        return Err(format!(
            "weights do not fit the card ({:.1} GB held, budget {:.1} GB)",
            held as f64 / 1e9,
            budget as f64 / 1e9
        ));
    }
    let t0 = std::time::Instant::now();
    let fx = st.hyb.funcs.values().next().cloned().ok_or("repack before the hybrid kernels")?;
    let (w, aux0, aux1) = repack_on_card(c, &fx, &[bytes], rows, rows, cols)?;
    let m = DevMat { kind: 6, rows, cols, w: Arc::new(w), aux0: Some(aux0), aux1: Some(aux1), held: bytes.len() as u64 };
    REPACK_NS.fetch_add(t0.elapsed().as_nanos() as u64, Ordering::Relaxed);
    c.resident.fetch_add(m.held, Ordering::Relaxed);
    let m = Arc::new(m);
    st.hyb.mats.insert(key, m.clone());
    Ok(m)
}

/// A dense FFN's gate and up as one repacked matrix of 2 inter rows, its
/// 16-row tiles 8 gate rows then the same 8 up rows: an mma tile holds a
/// row's gate and up in one lane, and the kernel's epilogue makes
/// silu(gate) * up there.
fn hmat_pair(c: &Ctx, st: &mut State, model: &Arc<CmfModel>, gw: &GraphW, uw: &GraphW, inter: usize, cols: usize) -> Res<Arc<DevMat>> {
    use cortiq_core::quant::GROUP_SIZE;
    if gw.kind != 6 || uw.kind != 6 || inter % 8 != 0 || cols % GROUP_SIZE != 0 {
        return Err("a gate/up pair of q4tp matrices with rows a multiple of 8".into());
    }
    let key = (model.uid(), gw.idx, uw.idx);
    if let Some(m) = st.hyb.pairs.get(&key) {
        return Ok(m.clone());
    }
    let ge = model.tensors.get(gw.idx).ok_or("tensor index out of range")?;
    let ue = model.tensors.get(uw.idx).ok_or("tensor index out of range")?;
    for e in [ge, ue] {
        if e.shape.first() != Some(&inter) || e.shape.get(1) != Some(&cols) {
            return Err(format!("tensor {} is {:?}, the pair wants {inter}x{cols}", e.name, e.shape));
        }
    }
    let t0 = std::time::Instant::now();
    let (gb, ub) = (model.entry_bytes(ge), model.entry_bytes(ue));
    let held = c.resident.load(Ordering::Relaxed);
    if held + (gb.len() + ub.len()) as u64 > super::vram_budget() {
        return Err("weights do not fit the card".into());
    }
    // the pair's rows straight from the two payloads, on the card
    let rows = 2 * inter;
    let fx = st.hyb.funcs.values().next().cloned().ok_or("repack before the hybrid kernels")?;
    let (w, aux0, aux1) = repack_on_card(c, &fx, &[gb, ub], inter, rows, cols)?;
    let m = DevMat { kind: 6, rows, cols, w: Arc::new(w), aux0: Some(aux0), aux1: Some(aux1), held: (gb.len() + ub.len()) as u64 };
    REPACK_NS.fetch_add(t0.elapsed().as_nanos() as u64, Ordering::Relaxed);
    c.resident.fetch_add(m.held, Ordering::Relaxed);
    let m = Arc::new(m);
    st.hyb.pairs.insert(key, m.clone());
    Ok(m)
}

struct GdnL {
    qkv: Arc<DevMat>,
    z: Arc<DevMat>,
    /// Held for the job tables' pointers into them.
    #[allow(dead_code)]
    a: Arc<DevMat>,
    #[allow(dead_code)]
    b: Arc<DevMat>,
    out: Arc<DevMat>,
    /// f16 copies of the f32 a/b projections (the prompt's GEMM operands).
    a_h: Buf,
    b_h: Buf,
    taps: Arc<Buf>,
    a_log: Arc<Buf>,
    dt_bias: Arc<Buf>,
    norm: Arc<Buf>,
    nv: usize,
    nk: usize,
    dk: usize,
    dv: usize,
    kk: usize,
    cdim: usize,
    /// qkv and z (tensor cores); a and b run in the front kernel.
    ms_in: MmaSet,
    ms_out: MmaSet,
}

struct AttnL {
    q: Arc<DevMat>,
    k: Arc<DevMat>,
    v: Arc<DevMat>,
    o: Arc<DevMat>,
    qn: Option<Arc<Buf>>,
    kn: Option<Arc<Buf>>,
    gated: bool,
    ms_in: MmaSet,
    ms_out: MmaSet,
}

enum Mix {
    Gdn(GdnL),
    Attn(AttnL),
}

struct HLayer {
    mix: Mix,
    /// gate and up as one pair (`hmat_pair`)
    gu: Arc<DevMat>,
    down: Arc<DevMat>,
    in_norm: Arc<Buf>,
    post_norm: Arc<Buf>,
    inter: usize,
    ms_gu: MmaSet,
    ms_dn: MmaSet,
}

/// A hybrid stack resolved for the card.
pub(super) struct HPlan {
    /// Unique among the plans of the process (graph keys hold it).
    id: u64,
    /// The call records of the attention layers (8 words each).
    ci: Buf,
    sig: Vec<usize>,
    consts: Vec<(usize, usize, u64)>,
    fx: Arc<HFuncs>,
    layers: Vec<HLayer>,
    lm: Option<(Arc<DevMat>, usize, MmaSet)>,
    final_norm: Option<Arc<Buf>>,
    invf: Vec<f32>,
    half: usize,
    nh: usize,
    nkv: usize,
    hd: usize,
    hpk: usize,
    hidden: usize,
    gemma: bool,
    eps: f32,
    scale: f32,
    // the small path's rows (NMAX each), fixed: the job tables hold them
    h: Buf,
    xn: Buf,
    q: Buf,
    k: Buf,
    v: Buf,
    qg: Buf,
    gate: Buf,
    #[allow(dead_code)]
    ao: Buf,
    qkv: Buf,
    z: Buf,
    o: Buf,
    of: Buf,
    logits: Buf,
    tab: Buf,
    #[allow(dead_code)]
    ids: Buf,
    /// The bf16 planes every tensor-core launch reads (`NMAX` rows of
    /// `pstride` each).
    pp: [Buf; 3],
    /// The FFN middle's two planes.
    qq: [Buf; 2],
    pstride: usize,
    qrow: usize,
    cdmax: usize,
    vdmax: usize,
    nvmax: usize,
}

/// One GDN layer's recurrent state on the card, with what a verify window
/// needs to step back.
struct GdnState {
    dims: (usize, usize, usize, usize, usize),
    /// The allocation the buffers below are views into (one a layer: the card
    /// rounds each allocation up, and five of the seven are small).
    _all: Buf,
    s: Buf,
    ring: Buf,
    next_pos: usize,
    // the window's tape: raw rows (`t_stride` floats apart), conv rows,
    // gates (NMAX rows each)
    t_qkv: Buf,
    t_stride: usize,
    t_cq: Buf,
    t_gb: Buf,
    s_save: Buf,
    ring_save: Buf,
    /// (first position, rows) of the last window that kept a tape.
    tape: Option<(usize, usize)>,
    /// A copy of `[s | ring]` at a prompt position (`gdn_snapshot`): the
    /// next turn of a chat goes back there instead of reading the
    /// conversation again.
    snap: Option<(usize, Buf, Buf)>,
}

#[derive(Default)]
pub(super) struct HState {
    funcs: HashMap<(usize, usize), Arc<HFuncs>>,
    /// q4tp weights in the tensor-core layout.
    mats: HashMap<(u64, usize), Arc<DevMat>>,
    /// Gate/up pairs (`hmat_pair`) by (model, gate, up).
    pairs: HashMap<(u64, usize, usize), Arc<DevMat>>,
    /// The far fields of sealed layers (turbo files), by sequence and layer.
    o1: HashMap<(u64, usize), HO1>,
    /// K/V mirrors this arm keeps in f16 (by sequence and layer): a mirror's
    /// `row` counts f32 slots, half its f16 values.
    pub(super) half: std::collections::HashSet<(u64, usize)>,
    /// `rows_matvec`'s job tables, by tensor, with the output they point at.
    rowsets: HashMap<(u64, usize), (MmaSet, u64)>,
    /// The restore tables of `spec_restore`, by sequence: the layers' state
    /// pointers (7 a layer) and the keys they were made for.
    rtabs: HashMap<u64, (Buf, Vec<(u64, usize)>)>,
    /// Recorded small-path graphs by (plan, sequence, layer base, rows, tape,
    /// head).
    graphs: HashMap<(u64, u64, usize, usize, bool, bool, u64), GraphRec>,
    plans: HashMap<(u64, usize, usize, bool), Arc<HPlan>>,
    gdn: HashMap<(u64, usize), GdnState>,
    /// The MTP draft chain's table and norms (`mtp_chain`).
    chain: Option<ChainRes>,
    /// Plan builds' weight arenas (the matrices in them are views).
    arenas: Vec<Buf>,
    /// The chain's eh_proj job tables by (plan, tensor): their output is
    /// that plan's `h`.
    chain_sets: HashMap<(u64, usize), MmaSet>,
}

/// The draft chain's resources for one model and shortlist: the shortlist
/// rows of the embedding (a draft's token is always one of them), repacked
/// for the tensor-core layout, and the block input's two norms.
struct ChainRes {
    key: (u64, usize, usize, u64, u64),
    emb: DevMat,
    enorm: Buf,
    hnorm: Buf,
}

/// Does this stack belong to the hybrid arm? (A GDN layer, or full attention
/// with Q/K norms or the element-wise output gate.) An empty stack with a
/// head goes here when the dense arm's attention kernels cannot take the
/// geometry.
pub(super) fn wants(layers: &[GraphLayer], nh: usize, nkv: usize, hd: usize) -> bool {
    if layers.is_empty() {
        let hpk = if nkv > 0 { nh / nkv } else { 0 };
        return 8 * hpk * hd * 4 > 44_000;
    }
    layers.iter().any(|l| match &l.attn {
        GraphAttn::Gdn { .. } => true,
        GraphAttn::Full { q_norm, k_norm, output_gate, .. } => *output_gate || q_norm.is_some() || k_norm.is_some(),
        _ => false,
    })
}

fn rows_of(model: &CmfModel, gw: &GraphW) -> Option<usize> {
    if gw.kind == 4 {
        return None;
    }
    model.tensors.get(gw.idx).and_then(|e| e.shape.first().copied())
}

#[allow(clippy::too_many_arguments)]
fn hsig(
    model: &Arc<CmfModel>,
    layers: &[GraphLayer],
    lm: Option<(&GraphW, usize)>,
    final_norm: &[f32],
    nh: usize,
    nkv: usize,
    hd: usize,
    rd: usize,
    hidden: usize,
    gemma: bool,
    invf: &[f32],
) -> Vec<usize> {
    let mut s = Vec::with_capacity(layers.len() * 24 + 16);
    let w = |s: &mut Vec<usize>, gw: &GraphW| s.extend([gw.idx, gw.kind as usize, gw.data.as_ptr() as usize, gw.data.len()]);
    let f = |s: &mut Vec<usize>, d: &[f32]| s.extend([d.as_ptr() as usize, d.len()]);
    s.extend([model.uid() as usize, nh, nkv, hd, rd, hidden, gemma as usize]);
    f(&mut s, invf);
    // by content: a verify passes a fresh copy of the final norm every round
    s.extend([final_norm.len(), crate::gpu::fp_f32(final_norm) as usize]);
    if let Some((gw, rows)) = lm {
        w(&mut s, gw);
        s.push(rows);
    }
    for l in layers {
        f(&mut s, l.input_norm);
        f(&mut s, l.post_norm);
        match &l.attn {
            GraphAttn::Gdn { qkv, z, a, b, out, conv1d, a_log, dt_bias, norm, nv, nk, dk, dv, kk, .. } => {
                s.push(1);
                for g in [qkv, z, a, b, out] {
                    w(&mut s, g);
                }
                for d in [*conv1d, *a_log, *dt_bias, *norm] {
                    f(&mut s, d);
                }
                s.extend([*nv, *nk, *dk, *dv, *kk]);
            }
            GraphAttn::Full { wq, wk, wv, wo, q_norm, k_norm, output_gate, .. } => {
                s.push(2);
                for g in [wq, wk, wv, wo] {
                    w(&mut s, g);
                }
                f(&mut s, q_norm.unwrap_or(&[]));
                f(&mut s, k_norm.unwrap_or(&[]));
                s.push(*output_gate as usize);
            }
            _ => s.push(3),
        }
        if let GraphFfn::Dense { gate, up, down, .. } = &l.ffn {
            for g in [gate, up, down] {
                w(&mut s, g);
            }
        }
    }
    s
}

#[allow(clippy::too_many_arguments)]
fn build(
    c: &Ctx,
    st: &mut State,
    model: &Arc<CmfModel>,
    layers: &[GraphLayer],
    lm: Option<(&GraphW, usize)>,
    final_norm: &[f32],
    nh: usize,
    nkv: usize,
    hd: usize,
    rd: usize,
    hidden: usize,
    gemma: bool,
    eps: f32,
    scale: f32,
    invf: &[f32],
    sig: Vec<usize>,
) -> Res<HPlan> {
    if nkv == 0 || nh % nkv != 0 || (layers.is_empty() && lm.is_none()) {
        return Err("empty stack or head geometry".into());
    }
    if hidden % 32 != 0 {
        return Err("hidden width not a multiple of 32".into());
    }
    let hpk = nh / nkv;
    // a layer-less head pass turns nothing
    let rd = if layers.is_empty() { 0 } else { rd };
    if rd % 2 != 0 || rd > hd || invf.len() < rd / 2 {
        return Err(format!("rotary width {rd}"));
    }
    let half = rd / 2;
    let fx = match st.hyb.funcs.get(&(hpk, hd)) {
        Some(f) => f.clone(),
        None => {
            let t = std::time::Instant::now();
            let f = Arc::new(load_funcs(c, hpk, hd)?);
            tracing::info!("cuda hybrid kernels compiled in {:.1} s", t.elapsed().as_secs_f64());
            st.hyb.funcs.insert((hpk, hd), f.clone());
            f
        }
    };
    let mut consts = Vec::new();
    let mut note = |d: &[f32]| consts.push((d.as_ptr() as usize, d.len(), crate::gpu::fp_f32(d)));
    // geometry first: the row buffers the job tables point into
    let mut qrow = nh * hd;
    let (mut cdmax, mut vdmax, mut nvmax, mut intermax) = (0usize, 0usize, 0usize, 0usize);
    for (li, l) in layers.iter().enumerate() {
        if l.mhc.is_some() || l.attn_out_norm.is_some() || l.ffn_out_norm.is_some() || l.layer_scale.is_some() {
            return Err(format!("layer {li}: mHC streams, sandwich norms or a layer scale"));
        }
        if l.input_norm.len() != hidden || l.post_norm.len() != hidden {
            return Err(format!("layer {li}: norm width"));
        }
        match &l.attn {
            GraphAttn::Gdn { nv, nk, dk, dv, kk, preview, .. } => {
                if preview.is_some() {
                    return Err(format!("layer {li}: block preview"));
                }
                if *dk != 128 || *dv != *dk || *nk == 0 || nv % nk != 0 || *kk < 1 || *kk > 9 {
                    return Err(format!("layer {li}: GDN geometry nv {nv} nk {nk} dk {dk} dv {dv}"));
                }
                cdmax = cdmax.max(2 * nk * dk + nv * dv);
                vdmax = vdmax.max(nv * dv);
                nvmax = nvmax.max(*nv);
            }
            GraphAttn::Full { late_qk_norm, bias, output_gate, geom, head_gate, .. } => {
                if *late_qk_norm || bias.is_some() || geom.is_some() || head_gate.is_some() {
                    return Err(format!("layer {li}: late Q/K norm, biases, own geometry or a head-wise gate"));
                }
                qrow = qrow.max(nh * hd * (1 + *output_gate as usize));
            }
            _ => return Err(format!("layer {li}: a token mixer the hybrid arm has no kernels for")),
        }
        let GraphFfn::Dense { gate, act, limit, hot, .. } = &l.ffn else {
            return Err(format!("layer {li}: an FFN other than a dense one"));
        };
        if *act != GraphAct::Silu || *limit != 0.0 || hot.is_some() {
            return Err(format!("layer {li}: an FFN other than plain SwiGLU"));
        }
        intermax = intermax.max(rows_of(model, gate).ok_or("f32 FFN")?);
    }
    let lm_rows = lm.map_or(0, |(_, r)| r);
    let f4 = |n: usize| Buf::new(n.max(1) * 4);
    let h = f4(NMAX * hidden)?;
    let xn = f4(NMAX * hidden)?;
    let q = f4(NMAX * qrow)?;
    let k = f4(NMAX * nkv * hd)?;
    let v = f4(NMAX * nkv * hd)?;
    let qg = f4(NMAX * nh * hd)?;
    let gate = f4(NMAX * nh * hd)?;
    let ao = f4(NMAX * nh * hd)?;
    let qkv = f4(NMAX * cdmax)?;
    let z = f4(NMAX * vdmax)?;
    let o = f4(NMAX * vdmax)?;
    let of = f4(NMAX * vdmax)?;
    let tab = f4(NMAX * half * 2)?;
    let ids = f4(NMAX)?;
    let pstride = hidden.max(intermax).max(vdmax).max(nh * hd).next_multiple_of(32);
    // the FFN middle's planes (the gate/up pair writes them, down reads them)
    let qq = [Buf::new(NMAX * pstride * 2)?, Buf::new(NMAX * pstride * 2)?];
    let mut out = Vec::with_capacity(layers.len());
    mem_report(c, st, "a plan's row buffers");
    // the q4tp matrices this plan uploads, in one arena
    let uid = model.uid();
    let q6 = |w: &GraphW| w.kind == 6 && w.prism == crate::gpu::GraphPrismOp::None && !w.affine;
    let mut need = 0usize;
    for l in layers {
        let GraphFfn::Dense { gate: gw, up: uw, down: dw, .. } = &l.ffn else { continue };
        let inter = rows_of(model, gw).unwrap_or(0);
        if q6(gw) && q6(uw) && !st.hyb.pairs.contains_key(&(uid, gw.idx, uw.idx)) {
            need += planes_bytes(2 * inter, hidden);
        }
        if q6(dw) && !st.hyb.mats.contains_key(&(uid, dw.idx)) {
            need += planes_bytes(hidden, inter);
        }
        let mut one = |w: &GraphW, rows: usize, cols: usize| {
            if q6(w) && !st.hyb.mats.contains_key(&(uid, w.idx)) {
                need += planes_bytes(rows, cols);
            }
        };
        match &l.attn {
            GraphAttn::Gdn { qkv: qw, z: zw, out: ow, nv, nk, dk, dv, .. } => {
                one(qw, 2 * nk * dk + nv * dv, hidden);
                one(zw, nv * dv, hidden);
                one(ow, hidden, nv * dv);
            }
            GraphAttn::Full { wq, wk, wv, wo, output_gate, .. } => {
                one(wq, nh * hd * (1 + *output_gate as usize), hidden);
                one(wk, nkv * hd, hidden);
                one(wv, nkv * hd, hidden);
                one(wo, hidden, nh * hd);
            }
            _ => {}
        }
    }
    if let Some((gw, rows)) = lm {
        if q6(gw) && !st.hyb.mats.contains_key(&(uid, gw.idx)) {
            need += planes_bytes(rows, hidden);
        }
    }
    if need > 0 && std::env::var("CMF_HYB_ARENA").as_deref() != Ok("0") {
        let a = Buf::new(need)?;
        ARENA.with(|x| x.set((a.ptr, need, 0)));
        st.hyb.arenas.push(a);
    }
    struct ArenaEnd;
    impl Drop for ArenaEnd {
        fn drop(&mut self) {
            ARENA.with(|x| x.set((0, 0, 0)));
        }
    }
    let _arena_end = ArenaEnd;
    for (li, l) in layers.iter().enumerate() {
        if li % 16 == 1 {
            mem_report(c, st, &format!("{li} layers of a plan"));
        }
        let GraphFfn::Dense { gate: gw, up: uw, down: dw, .. } = &l.ffn else { unreachable!() };
        let inter = rows_of(model, gw).ok_or("f32 FFN")?;
        let gum = hmat_pair(c, st, model, gw, uw, inter, hidden)?;
        let dm = hmat(c, st, model, dw, hidden, inter)?;
        note(l.input_norm);
        note(l.post_norm);
        let in_norm = st.f32_buf(l.input_norm)?;
        let post_norm = st.f32_buf(l.post_norm)?;
        let mix = match &l.attn {
            GraphAttn::Gdn { qkv: qw, z: zw, a: aw, b: bw, out: ow, conv1d, a_log, dt_bias, norm, nv, nk, dk, dv, kk, .. } => {
                let (nv, nk, dk, dv, kk) = (*nv, *nk, *dk, *dv, *kk);
                let cdim = 2 * nk * dk + nv * dv;
                if conv1d.len() != cdim * kk || a_log.len() != nv || dt_bias.len() != nv || norm.len() != dv {
                    return Err(format!("layer {li}: GDN constants of unexpected lengths"));
                }
                let qm = hmat(c, st, model, qw, cdim, hidden)?;
                let zm = hmat(c, st, model, zw, nv * dv, hidden)?;
                let am = hmat(c, st, model, aw, nv, hidden)?;
                let bm = hmat(c, st, model, bw, nv, hidden)?;
                let om = hmat(c, st, model, ow, hidden, nv * dv)?;
                for d in [*conv1d, *a_log, *dt_bias, *norm] {
                    note(d);
                }
                let half_of = |m: &DevMat, st: &mut State| -> Res<Buf> {
                    let n = m.rows * m.cols;
                    let hb = st.slab_buf(n * 2 + 64)?;
                    let src = if m.kind == 4 {
                        m.w.ptr
                    } else {
                        return Err("a GDN gate projection that is not f32".into());
                    };
                    let n64 = n as u64;
                    unsafe { c.launch(fx.to_half, (n.div_ceil(256) as u32, 1), 256, 0, kargs!(src, hb.ptr, n64))? };
                    Ok(hb)
                };
                let a_h = half_of(&am, st)?;
                let b_h = half_of(&bm, st)?;
                let ms_in = mmaset(&[(&qm, qkv.ptr, 0, cdmax, cdim), (&zm, z.ptr, 0, vdmax, nv * dv)])?;
                let ms_out = mmaset(&[(&om, h.ptr, h.ptr, hidden, hidden)])?;
                Mix::Gdn(GdnL {
                    qkv: qm,
                    z: zm,
                    a: am,
                    b: bm,
                    out: om,
                    a_h,
                    b_h,
                    taps: st.f32_buf(conv1d)?,
                    a_log: st.f32_buf(a_log)?,
                    dt_bias: st.f32_buf(dt_bias)?,
                    norm: st.f32_buf(norm)?,
                    nv,
                    nk,
                    dk,
                    dv,
                    kk,
                    cdim,
                    ms_in,
                    ms_out,
                })
            }
            GraphAttn::Full { wq, wk, wv, wo, q_norm, k_norm, output_gate, .. } => {
                let qr = nh * hd * (1 + *output_gate as usize);
                let qm = hmat(c, st, model, wq, qr, hidden)?;
                let km = hmat(c, st, model, wk, nkv * hd, hidden)?;
                let vm = hmat(c, st, model, wv, nkv * hd, hidden)?;
                let om = hmat(c, st, model, wo, hidden, nh * hd)?;
                let nrm = |d: Option<&[f32]>, st: &mut State| -> Res<Option<Arc<Buf>>> {
                    match d {
                        Some(d) if d.len() == hd => Ok(Some(st.f32_buf(d)?)),
                        Some(_) => Err(format!("layer {li}: Q/K norm width")),
                        None => Ok(None),
                    }
                };
                if let Some(d) = q_norm {
                    note(d);
                }
                if let Some(d) = k_norm {
                    note(d);
                }
                let qn = nrm(*q_norm, st)?;
                let kn = nrm(*k_norm, st)?;
                let ms_in = mmaset(&[
                    (&qm, q.ptr, 0, qrow, qr),
                    (&km, k.ptr, 0, nkv * hd, nkv * hd),
                    (&vm, v.ptr, 0, nkv * hd, nkv * hd),
                ])?;
                let ms_out = mmaset(&[(&om, h.ptr, h.ptr, hidden, hidden)])?;
                Mix::Attn(AttnL { q: qm, k: km, v: vm, o: om, qn, kn, gated: *output_gate, ms_in, ms_out })
            }
            _ => unreachable!(),
        };
        let ms_gu = mmaset_pair(&gum, qq[0].ptr, qq[1].ptr, pstride)?;
        let ms_dn = mmaset(&[(&dm, h.ptr, h.ptr, hidden, hidden)])?;
        out.push(HLayer { mix, gu: gum, down: dm, in_norm, post_norm, inter, ms_gu, ms_dn });
    }
    let (lm, logits, final_norm_buf) = match lm {
        Some((gw, rows)) => {
            let m = hmat(c, st, model, gw, rows_of(model, gw).ok_or("an f32 head")?, hidden)?;
            if rows == 0 || rows > m.rows {
                return Err(format!("head of {rows} rows over a {}-row matrix", m.rows));
            }
            if final_norm.len() != hidden {
                return Err("final norm width".into());
            }
            // (in the signature by content: no pointer of it is kept)
            let logits = f4(NMAX * rows)?;
            let js = mmaset(&[(&m, logits.ptr, 0, rows, rows)])?;
            (Some((m, rows, js)), logits, Some(st.f32_buf(final_norm)?))
        }
        None => (None, f4(1)?, None),
    };
    let _ = lm_rows;
    let pp = [Buf::new(NMAX * pstride * 2)?, Buf::new(NMAX * pstride * 2)?, Buf::new(NMAX * pstride * 2)?];
    static PLAN_ID: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(1);
    let n_attn = layers.iter().filter(|l| matches!(l.attn, GraphAttn::Full { .. })).count();
    Ok(HPlan {
        id: PLAN_ID.fetch_add(1, Ordering::Relaxed),
        ci: Buf::new(n_attn.max(1) * 8 * 4)?,
        sig,
        consts,
        fx,
        layers: out,
        lm,
        final_norm: final_norm_buf,
        invf: invf[..half].to_vec(),
        half,
        nh,
        nkv,
        hd,
        hpk,
        hidden,
        gemma,
        eps,
        scale,
        h,
        xn,
        q,
        k,
        v,
        qg,
        gate,
        ao,
        qkv,
        z,
        o,
        of,
        logits,
        tab,
        ids,
        pp,
        qq,
        pstride,
        qrow,
        cdmax,
        vdmax,
        nvmax,
    })
}

#[allow(clippy::too_many_arguments)]
fn plan_for(
    c: &Ctx,
    st: &mut State,
    model: &Arc<CmfModel>,
    layers: &[GraphLayer],
    layer_base: usize,
    lm: Option<(&GraphW, usize)>,
    final_norm: &[f32],
    nh: usize,
    nkv: usize,
    hd: usize,
    rd: usize,
    hidden: usize,
    gemma: bool,
    eps: f32,
    scale: f32,
    invf: &[f32],
    first_pos: usize,
) -> Res<Arc<HPlan>> {
    let key = (model.uid(), layer_base, layers.len(), lm.is_some());
    let sig = hsig(model, layers, lm, final_norm, nh, nkv, hd, rd, hidden, gemma, invf);
    if let Some(p) = st.hyb.plans.get(&key) {
        if p.sig == sig && p.eps == eps && p.scale == scale {
            // SAFETY: the signature holds these addresses and lengths, and the
            // caller's slices are alive for the call.
            let same = first_pos != 0
                || p.consts.iter().all(|&(ptr, len, fp)| {
                    crate::gpu::fp_f32(unsafe { std::slice::from_raw_parts(ptr as *const f32, len) }) == fp
                });
            if same {
                return Ok(p.clone());
            }
        }
    }
    c.sync()?;
    let t0 = std::time::Instant::now();
    let p = Arc::new(build(c, st, model, layers, lm, final_norm, nh, nkv, hd, rd, hidden, gemma, eps, scale, invf, sig)?);
    mem_report(c, st, &format!("a plan of {} layers", p.layers.len()));
    tracing::info!(
        "cuda hybrid graph: {} layers resident, {:.2} GB of weights, built in {:.1} s (repack {:.1} s, upload {:.1} s, single weights {:.1} s, job tables {:.1} s)",
        p.layers.len(),
        c.resident.load(Ordering::Relaxed) as f64 / 1e9,
        t0.elapsed().as_secs_f64(),
        REPACK_NS.swap(0, Ordering::Relaxed) as f64 / 1e9,
        UPLOAD_NS.swap(0, Ordering::Relaxed) as f64 / 1e9,
        HMAT_NS.swap(0, Ordering::Relaxed) as f64 / 1e9,
        SETS_NS.swap(0, Ordering::Relaxed) as f64 / 1e9
    );
    st.hyb.plans.insert(key, p.clone());
    Ok(p)
}

/// (sin, cos) per row and frequency, `[row][pair][2]`, as the host's table.
fn rope_rows(invf: &[f32], positions: impl Iterator<Item = usize>) -> Vec<f32> {
    let mut tab = Vec::new();
    for p in positions {
        for &f in invf {
            let (s, co) = (p as f32 * f).sin_cos();
            tab.push(s);
            tab.push(co);
        }
    }
    tab
}

/// `KvMirror::sync_to` for an f16 mirror: rewind past rows of another
/// continuation, or take the rows the host holds (as f16) and the card does
/// not. Ok(false): the host does not hold them either.
#[allow(clippy::too_many_arguments)]
fn sync_half(c: &Ctx, m: &mut KvMirror, cpu_k: &[Vec<f32>], cpu_v: &[Vec<f32>], cpu_base: usize, nkv: usize, hd: usize, upto: usize) -> Res<bool> {
    if m.stored > upto {
        if upto >= m.base {
            m.stored = upto;
        } else {
            m.base = 0;
            m.stored = 0;
        }
    }
    if m.stored == upto {
        return Ok(true);
    }
    let from = m.stored;
    let n = upto - from;
    let have = cpu_k.first().map_or(0, |k| k.len() / hd);
    if cpu_k.len() != nkv || cpu_v.len() != nkv || from < cpu_base || upto > cpu_base + have {
        return Ok(false);
    }
    m.reserve(c, n, 0)?;
    let row = nkv * hd;
    let mut kb = vec![0u16; n * row];
    let mut vb = vec![0u16; n * row];
    for r in 0..n {
        let src = (from - cpu_base + r) * hd;
        for g in 0..nkv {
            for d in 0..hd {
                kb[r * row + g * hd + d] = cortiq_core::quant::f32_to_f16(cpu_k[g][src + d]);
                vb[r * row + g * hd + d] = cortiq_core::quant::f32_to_f16(cpu_v[g][src + d]);
            }
        }
    }
    let off = ((from - m.base) * row * 2) as u64;
    unsafe {
        cu::memcpy_htod_async(m.kptr() + off, &kb, c.stream).map_err(derr)?;
        cu::memcpy_htod_async(m.vptr() + off, &vb, c.stream).map_err(derr)?;
    }
    // the host vectors go when this returns
    c.sync()?;
    m.stored = upto;
    Ok(true)
}

/// Bring every layer's state to `pos0` rows before anything is launched:
/// the K/V mirrors (from the host's rows where the card lacks them) and the
/// GDN states (fresh at a sequence start, seeded from the host's state after
/// a host prefill). Err(true): the card holds state this call cannot use.
#[allow(clippy::too_many_arguments)]
fn admit(
    c: &Ctx,
    st: &mut State,
    plan: &HPlan,
    layers: &[GraphLayer],
    kv_id: u64,
    layer_base: usize,
    pos0: usize,
    n: usize,
    hint: usize,
    sealed: &[bool],
) -> Result<(), (bool, String)> {
    let soft = |e: String| (false, e);
    for (li, (pl, l)) in plan.layers.iter().zip(layers).enumerate() {
        let key = (kv_id, layer_base + li);
        if sealed.get(li).copied().unwrap_or(false) {
            continue;
        }
        match (&pl.mix, &l.attn) {
            (Mix::Attn(_), GraphAttn::Full { cpu_k, cpu_v, cpu_base, .. }) => {
                // f16 rows: `row` counts f32 slots of a mirror, so half the values
                let row = plan.nkv * plan.hd / 2;
                let fresh_half = !st.hyb.half.contains(&key);
                let m = st.kv.entry(key).or_insert_with(|| KvMirror::new(row, 0));
                if m.row != row || m.window != 0 || fresh_half {
                    let _ = c.sync();
                    c.kv_bytes.fetch_sub((m.cap * m.row * 8) as u64, Ordering::Relaxed);
                    *m = KvMirror::new(row, 0);
                }
                st.hyb.half.insert(key);
                let m = st.kv.get_mut(&key).unwrap();
                match sync_half(c, m, cpu_k, cpu_v, *cpu_base, plan.nkv, plan.hd, pos0) {
                    Ok(true) => m.reserve(c, n, hint).map_err(soft)?,
                    Ok(false) if m.stored < pos0 && m.stored >= m.base => {
                        // Rows neither on the card nor on the host: positions
                        // a draft block never saw (the plain tokens of a
                        // speculation trial). The wgpu graph reads whatever
                        // its mirror holds there; here they are zero rows,
                        // which a draft's attention sees as keys of score 0.
                        let gap = pos0 - m.stored;
                        m.reserve(c, gap + n, hint).map_err(soft)?;
                        let off = ((m.stored - m.base) * m.row * 4) as u64;
                        unsafe {
                            cu::memset_d8_async(m.kptr() + off, 0, gap * m.row * 4, c.stream).map_err(|e| soft(derr(e)))?;
                            cu::memset_d8_async(m.vptr() + off, 0, gap * m.row * 4, c.stream).map_err(|e| soft(derr(e)))?;
                        }
                        if debug() {
                            eprintln!("cuda hybrid: layer {li} of sequence {kv_id}: {gap} K/V rows to {pos0} filled with zeros");
                        }
                        m.stored = pos0;
                    }
                    Ok(false) => return Err(soft("a layer's K/V rows are neither on the card nor on the host".into())),
                    Err(e) => return Err(soft(e)),
                }
            }
            (Mix::Gdn(g), GraphAttn::Gdn { cpu_state, .. }) => {
                let dims = (g.nv, g.nk, g.dk, g.dv, g.kk);
                let ring_n = (g.kk - 1) * g.cdim;
                let s_n = g.nv * g.dk * g.dv;
                if let Some(gs) = st.hyb.gdn.get_mut(&key) {
                    if gs.dims == dims && gs.next_pos == pos0 {
                        continue;
                    }
                    if gs.dims == dims && pos0 == 0 {
                        // a new sequence on the same id: no history
                        unsafe {
                            cu::memset_d8_async(gs.s.ptr, 0, s_n * 4, c.stream).map_err(|e| soft(derr(e)))?;
                            cu::memset_d8_async(gs.ring.ptr, 0, ring_n.max(1) * 4, c.stream).map_err(|e| soft(derr(e)))?;
                        }
                        gs.next_pos = 0;
                        gs.tape = None;
                        continue;
                    }
                    return Err((true, format!("GDN state of layer {li} is at row {}, the call at {pos0}", gs.next_pos)));
                }
                if pos0 > 0 && cpu_state.len() != ring_n + s_n {
                    return Err(soft("GDN host seed missing for a nonzero position".into()));
                }
                // the seven buffers of a layer in one allocation, zeroed
                let sizes = [s_n, ring_n, NMAX * plan.cdmax, NMAX * g.cdim, NMAX * 2 * g.nv, s_n, ring_n].map(|n| (n.max(1) * 4).next_multiple_of(256));
                let alloc = || -> Res<GdnState> {
                    let all = Buf::new(sizes.iter().sum())?;
                    unsafe { cu::memset_d8_async(all.ptr, 0, all.bytes, c.stream).map_err(derr)? };
                    let mut at = 0usize;
                    let mut v = Vec::with_capacity(sizes.len());
                    for &n in &sizes {
                        v.push(all.view(at, n));
                        at += n;
                    }
                    let mut v = v.into_iter();
                    let mut next = || v.next().ok_or("GDN state views");
                    let (s, ring, t_qkv, t_cq, t_gb, s_save, ring_save) = (next()?, next()?, next()?, next()?, next()?, next()?, next()?);
                    if pos0 > 0 {
                        unsafe {
                            if ring_n > 0 {
                                cu::memcpy_htod_async(ring.ptr, &cpu_state[..ring_n], c.stream).map_err(derr)?;
                            }
                            cu::memcpy_htod_async(s.ptr, &cpu_state[ring_n..], c.stream).map_err(derr)?;
                        }
                    }
                    Ok(GdnState {
                        dims,
                        _all: all,
                        s,
                        ring,
                        next_pos: pos0,
                        t_qkv,
                        t_stride: plan.cdmax,
                        t_cq,
                        t_gb,
                        s_save,
                        ring_save,
                        tape: None,
                        snap: None,
                    })
                };
                let gs = alloc().map_err(soft)?;
                c.kv_bytes.fetch_add(((s_n * 2 + ring_n * 2 + NMAX * (2 * g.cdim + 2 * g.nv)) * 4) as u64, Ordering::Relaxed);
                st.hyb.gdn.insert(key, gs);
            }
            _ => return Err(soft(format!("layer {li}: descriptors changed under the plan"))),
        }
    }
    Ok(())
}

/// A sealed layer's far field on the card (attn.cu's `O1Dev` with f16 near
/// rows: the sinks and the rows kept exact, then the ring of the window, laid
/// out as a K/V mirror; the landmarks and each head's far accumulators). It
/// goes up once per seal epoch and moves with every row from there.
struct HO1 {
    epoch: u64,
    /// The position of the next row that may move this state; None until the
    /// first one has.
    next_pos: Option<usize>,
    bytes: u64,
    nk: Buf,
    nv: Buf,
    kt: Buf,
    qt: Buf,
    mu: Buf,
    mz: Buf,
    th: Buf,
    ns: usize,
    w: usize,
    m: usize,
    win_len: usize,
    win_head: usize,
    far_len: usize,
    scale: f32,
    rect_fm: bool,
    /// A verify window's rows (the first row's index, rows): written after the
    /// near rows, not yet in the window; `spec_commit` moves the accepted ones.
    spec: Option<(usize, usize)>,
    /// The layer's geometry: KV heads, query heads a KV head, head width.
    nkv: usize,
    hpk: usize,
    hd: usize,
}

/// A sealed layer's far field up for this epoch (unless it is already).
#[allow(clippy::too_many_arguments)]
fn o1_ensure(c: &Ctx, st: &mut State, key: (u64, usize), views: &[crate::nystrom::O1DeviceView<'_>], epoch: u64, hpk: usize, nkv: usize, hd: usize) -> Res<()> {
    if st.hyb.o1.get(&key).is_some_and(|d| d.epoch == epoch) {
        return Ok(());
    }
    let g0 = views.first().ok_or("far field: a layer without groups")?;
    let (m, ns, w) = (g0.m_eff, g0.sink_len, g0.w);
    let rect_fm = g0.heads.first().is_some_and(|h| h.rect_fm);
    if views.len() != nkv
        || views.iter().any(|v| {
            v.exact_only
                || v.d != hd
                || v.dv != hd
                || v.m_eff != m
                || v.sink_len != ns
                || v.w != w
                || v.heads.len() != hpk
                || v.win_len != g0.win_len
                || v.win_head != g0.win_head
                || v.far_len != g0.far_len
                || v.scale != g0.scale
                || v.heads.iter().any(|h| h.rect_fm != rect_fm)
        })
    {
        return Err("far field: the groups of a layer disagree, or one is exact-only".into());
    }
    if m == 0 || m > 64 || hpk * m > 1024 || w == 0 || g0.win_len > w {
        return Err(format!("far field: {m} landmarks x {hpk} heads a group, window {w}: outside the kernels"));
    }
    let row = nkv * hd;
    let h16 = cortiq_core::quant::f32_to_f16;
    // room after the near rows for a verify window's rows, or a prompt chunk's
    let extra = NMAX.max(SEAL_ROWS);
    let mut nk = vec![0u16; (ns + w + extra) * row];
    let mut nv = vec![0u16; (ns + w + extra) * row];
    let (mut kt, mut qt, mut mu, mut mz, mut th) = (Vec::new(), Vec::new(), Vec::new(), Vec::new(), Vec::new());
    for (g, v) in views.iter().enumerate() {
        let have = (v.win_k.len() / hd).min(v.win_v.len() / hd).min(w);
        if v.sink_k.len() != ns * hd || v.sink_v.len() != ns * hd || v.k_tilde.len() != m * hd || have < v.win_len {
            return Err("far field: sink, window or landmark rows of an unexpected length".into());
        }
        for r in 0..ns {
            for d in 0..hd {
                nk[(r * nkv + g) * hd + d] = h16(v.sink_k[r * hd + d]);
                nv[(r * nkv + g) * hd + d] = h16(v.sink_v[r * hd + d]);
            }
        }
        for r in 0..have {
            for d in 0..hd {
                nk[((ns + r) * nkv + g) * hd + d] = h16(v.win_k[r * hd + d]);
                nv[((ns + r) * nkv + g) * hd + d] = h16(v.win_v[r * hd + d]);
            }
        }
        kt.extend_from_slice(v.k_tilde);
        for h in &v.heads {
            if h.q_tilde.len() != m * hd || h.mu.len() != m * m || h.m_max.len() != m || h.z_hat.len() != m || h.t_hat.len() != m * hd {
                return Err("far field: a head's skeleton of an unexpected length".into());
            }
            qt.extend_from_slice(h.q_tilde);
            mu.extend_from_slice(h.mu);
            mz.extend_from_slice(h.m_max);
            mz.extend_from_slice(h.z_hat);
            th.extend_from_slice(h.t_hat);
        }
    }
    // queued work may still read the epoch being replaced
    c.sync()?;
    let bytes = ((nk.len() + nv.len()) * 2 + (kt.len() + qt.len() + mu.len() + mz.len() + th.len()) * 4) as u64;
    let d = HO1 {
        epoch,
        next_pos: None,
        bytes,
        nk: Buf::from_slice(&nk)?,
        nv: Buf::from_slice(&nv)?,
        kt: Buf::from_slice(&kt)?,
        qt: Buf::from_slice(&qt)?,
        mu: Buf::from_slice(&mu)?,
        mz: Buf::from_slice(&mz)?,
        th: Buf::from_slice(&th)?,
        ns,
        w,
        m,
        win_len: g0.win_len,
        win_head: g0.win_head,
        far_len: g0.far_len,
        scale: g0.scale,
        rect_fm,
        spec: None,
        nkv,
        hpk,
        hd,
    };
    if debug() {
        eprintln!(
            "cuda hybrid: far field of layer {} on the card (epoch {epoch}): {ns} exact rows, window {} of {w}, {m} landmarks, {} keys far, {:.1} MB",
            key.1,
            d.win_len,
            d.far_len,
            bytes as f64 / 1e6
        );
    }
    c.kv_bytes.fetch_add(bytes, Ordering::Relaxed);
    if let Some(old) = st.hyb.o1.insert(key, d) {
        c.kv_bytes.fetch_sub(old.bytes, Ordering::Relaxed);
    }
    Ok(())
}

/// One small-path call's launch inputs, made by `book`, which also moves
/// the host-side cursors (mirror rows, GDN positions and tapes).
struct Book {
    /// Per layer: the GDN state's buffers, or the mirror's K and V.
    layers: Vec<[u64; 7]>,
    /// The call record of every attention layer, 8 words each:
    /// n0 (mirror rows before the call), the mirror row of the first row,
    /// rb, ncap, nbcap.
    ci: Vec<i32>,
    /// The attention scratch (S, Mx, P, E).
    scratch: [u64; 4],
    /// Per layer: 0 its exact mirror, 1 a far field, 2 a far field whose ring
    /// row goes into the far sums first.
    far: Vec<u8>,
    /// Per layer: a far field's landmarks, scale and rectification.
    farp: Vec<(usize, f32, bool)>,
}

/// Rows a key block of the decode attention takes for `keys` keys (64 up to
/// 8192 keys, so a window's rows match their one-row steps there).
fn attn_rb(keys: usize) -> usize {
    64 * keys.div_ceil(8192).max(1)
}

/// Key blocks the attention grid covers: every count `attn_rb` allows.
const ATTN_NBLK: usize = 129;

#[allow(clippy::too_many_arguments)]
fn book(
    c: &Ctx,
    st: &mut State,
    plan: &HPlan,
    kv_id: u64,
    layer_base: usize,
    pos0: usize,
    n: usize,
    tape: bool,
    kcap: usize,
    sealed: &[bool],
) -> Res<Book> {
    let (nh, hd) = (plan.nh, plan.hd);
    let mut layers = Vec::with_capacity(plan.layers.len());
    let mut far = Vec::with_capacity(plan.layers.len());
    let mut farp = Vec::with_capacity(plan.layers.len());
    let mut ci = Vec::new();
    let mut keys_max = 0usize;
    for (li, pl) in plan.layers.iter().enumerate() {
        let key = (kv_id, layer_base + li);
        if !matches!(&pl.mix, Mix::Attn(_)) || !sealed.get(li).copied().unwrap_or(false) || !st.hyb.o1.contains_key(&key) {
            far.push(0);
            farp.push((0, 0.0, false));
        }
        match &pl.mix {
            Mix::Gdn(_) => {
                let gs = st.hyb.gdn.get_mut(&key).ok_or("GDN state missing")?;
                layers.push([gs.s.ptr, gs.ring.ptr, gs.t_qkv.ptr, gs.t_cq.ptr, gs.t_gb.ptr, gs.s_save.ptr, gs.ring_save.ptr]);
                gs.next_pos = pos0 + n;
                gs.tape = if tape { Some((pos0, n)) } else { None };
            }
            Mix::Attn(_) if st.hyb.o1.contains_key(&key) && sealed.get(li).copied().unwrap_or(false) && (tape || n > 1) => {
                // a verify window over the far field: its rows after the near
                // ones, read by the rows after them; the state moves over the
                // accepted ones only (`spec_commit`)
                if n > NMAX {
                    return Err("a verify window over a far field of more than 16 rows".into());
                }
                let d = st.hyb.o1.get_mut(&key).unwrap();
                let base = d.ns + d.win_len;
                d.spec = Some((base, n));
                let keys = base + n;
                let rb = attn_rb(keys);
                let ncap = keys.next_multiple_of(4096);
                let nbcap = ncap.div_ceil(rb) + 1;
                ci.extend([base as i32, base as i32, rb as i32, ncap as i32, nbcap as i32, 1, base as i32, (d.far_len > 0) as i32]);
                layers.push([d.nk.ptr, d.nv.ptr, d.kt.ptr, d.qt.ptr, d.mu.ptr, d.mz.ptr, d.th.ptr]);
                far.push(1);
                farp.push((d.m, d.scale, d.rect_fm));
                keys_max = keys_max.max(keys);
                continue;
            }
            Mix::Attn(_) if st.hyb.o1.contains_key(&key) && sealed.get(li).copied().unwrap_or(false) => {
                // the far field (`nystrom.rs` advance): the ring row the new
                // key replaces goes into the far sums first, once the ring is full
                let d = st.hyb.o1.get_mut(&key).unwrap();
                let full = d.win_len == d.w;
                let row = d.ns + if full { d.win_head } else { d.win_len };
                if full {
                    d.far_len += 1;
                    d.win_head = (d.win_head + 1) % d.w;
                } else {
                    d.win_len += 1;
                }
                d.next_pos = Some(pos0 + 1);
                let keys = d.ns + d.win_len;
                let rb = attn_rb(keys);
                let ncap = keys.next_multiple_of(4096);
                let nbcap = ncap.div_ceil(rb) + 1;
                ci.extend([(keys - 1) as i32, row as i32, rb as i32, ncap as i32, nbcap as i32, 1, row as i32, (d.far_len > 0) as i32]);
                layers.push([d.nk.ptr, d.nv.ptr, d.kt.ptr, d.qt.ptr, d.mu.ptr, d.mz.ptr, d.th.ptr]);
                far.push(if full { 2 } else { 1 });
                farp.push((d.m, d.scale, d.rect_fm));
                keys_max = keys_max.max(keys);
                continue;
            }
            Mix::Attn(_) => {
                let m = st.kv.get_mut(&key).ok_or("K/V mirror missing")?;
                let n0 = m.stored - m.base;
                layers.push([m.kptr(), m.vptr(), 0, 0, 0, 0, 0]);
                m.stored += n;
                let keys = n0 + n;
                let rb = attn_rb(keys);
                let ncap = keys.next_multiple_of(4096);
                let nbcap = ncap.div_ceil(rb);
                ci.extend([n0 as i32, n0 as i32, rb as i32, ncap as i32, nbcap as i32, 0, 0, 0]);
                keys_max = keys_max.max(keys);
            }
        }
    }
    // the attention scratch sized for this call's records: it grows (by 4096
    // keys at a time) as the rows do, and a graph whose pointers moved is
    // recorded again (its signature holds them). `CMF_HYB_KCAP=1` sizes it for
    // the context cap from the start.
    let kc = if std::env::var("CMF_HYB_KCAP").as_deref() == Ok("1") { kcap.max(keys_max) } else { keys_max }.next_multiple_of(4096).max(4096);
    let nbc = ci.chunks(8).map(|r| r[4].max(0) as usize).max().unwrap_or(0).max(1).min(ATTN_NBLK + 64);
    let scratch = [
        st.scr(c, "hy_S", NMAX * nh * kc * 4)?,
        st.scr(c, "hy_Mx", NMAX * nh * nbc * 4)?,
        st.scr(c, "hy_P", NMAX * nbc * nh * hd * 4)?,
        st.scr(c, "hy_E", NMAX * nbc * nh * 4)?,
    ];
    Ok(Book { layers, ci, scratch, far, farp })
}

/// The small path's launches over the rows in `plan.h` (`n <= NMAX`), as
/// `book` laid them out: kernels and device copies only, nothing on the
/// host, so a stream capture can record it. `tape`: keep what a verify
/// needs to step back.
#[allow(clippy::too_many_arguments)]
unsafe fn launch(c: &Ctx, plan: &HPlan, bk: &Book, n: usize, tape: bool, cib: u64, tabp: u64, taps: &[(usize, u64)]) -> Res<()> {
    let fx = &plan.fx;
    let (hidden, nh, nkv, hd) = (plan.hidden, plan.nh, plan.nkv, plan.hd);
    let (eps, gemma_i, n_i) = (plan.eps, plan.gemma as i32, n as i32);
    let hid_i = hidden as i32;
    let zero = 0i32;
    HCALLS.fetch_add(1, Ordering::Relaxed);
    HCALLS_N[n.min(16)].fetch_add(1, Ordering::Relaxed);
    HROWS.with(|r| r.set(n));
    let pl_ = Planes { p0: plan.pp[0].ptr, p1: plan.pp[1].ptr, p2: plan.pp[2].ptr, stride: plan.pstride };
    let planes = &pl_;
    let (pp0, pp1, pp2, ps_i) = (pl_.p0, pl_.p1, pl_.p2, plan.pstride as i32);
    let [s_p, mx_p, p_p, e_p] = bk.scratch;
    let mut ai = 0usize;
    let mut t0 = std::time::Instant::now();
    hst(c, 11, &mut t0)?;
    for (li, pl) in plan.layers.iter().enumerate() {
        let lb = bk.layers[li];
        unsafe {
            c.launch(
                fx.rmsnorm2,
                (hidden.div_ceil(1024) as u32, n as u32),
                256,
                0,
                kargs!(plan.h.ptr, pl.in_norm.ptr, plan.xn.ptr, hid_i, eps, gemma_i, pp0, pp1, pp2, ps_i),
            )?;
        }
        hst(c, 0, &mut t0)?;
        match &pl.mix {
            Mix::Gdn(g) => {
                let [gs_s, gs_ring, t_qkv, t_cq, t_gb, s_save, ring_save] = lb;
                if plan.nvmax != g.nv {
                    return Err("GDN layers of different widths in one stack".into());
                }
                let (cd_i, kk_i, nv_i, nk_i, dk_i, dv_i) =
                    (g.cdim as i32, g.kk as i32, g.nv as i32, g.nk as i32, g.dk as i32, g.dv as i32);
                let qs_i = plan.cdmax as i32;
                let s_bytes = g.nv * g.dk * g.dv * 4;
                let r_bytes = (g.kk - 1) * g.cdim * 4;
                unsafe {
                    g.ms_in.run(c, fx, planes, hidden, n)?;
                    hst(c, 1, &mut t0)?;
                    // dk = 128 (every Qwen3.5-family file): the column scan,
                    // which keeps the tape's state copy itself
                    let col_scan = g.dk == 128 && g.dv % 32 == 0;
                    if tape {
                        if !col_scan {
                            cu::memcpy_dtod_async(s_save, gs_s, s_bytes, c.stream).map_err(derr)?;
                        }
                        if r_bytes > 0 {
                            cu::memcpy_dtod_async(ring_save, gs_ring, r_bytes, c.stream).map_err(derr)?;
                        }
                        // the raw rows (the ring replays from them), at the
                        // projection's own row stride
                        cu::memcpy_dtod_async(t_qkv, plan.qkv.ptr, n * plan.cdmax * 4, c.stream).map_err(derr)?;
                    }
                    let units = (g.cdim / g.dk) as u32;
                    c.launch(
                        fx.front,
                        (units + 2 * g.nv as u32, n as u32),
                        g.dk.max(64) as u32,
                        0,
                        kargs!(plan.qkv.ptr, qs_i, gs_ring, g.taps.ptr, t_cq, n_i, cd_i, kk_i, nk_i, dk_i, plan.xn.ptr, hid_i, g.a.w.ptr, g.b.w.ptr, g.a_log.ptr, g.dt_bias.ptr, t_gb, nv_i),
                    )?;
                    let cols = g.nv * g.dv;
                    // the scan, with blocks after it that move the conv ring
                    let ring_p = if g.kk > 1 { gs_ring } else { 0 };
                    if col_scan {
                        let ring_blocks = if g.kk > 1 { (g.cdim as u32).div_ceil(128) } else { 0 };
                        let save = if tape { s_save } else { 0 };
                        c.launch(
                            fx.scan4,
                            ((g.nv * g.dv / 32) as u32 + ring_blocks, 1),
                            128,
                            0,
                            kargs!(t_cq, t_gb, gs_s, plan.o.ptr, n_i, cd_i, nv_i, nk_i, dv_i, ring_p, plan.qkv.ptr, qs_i, kk_i, save),
                        )?;
                    } else {
                        let ring_blocks = if g.kk > 1 { (g.cdim as u32).div_ceil(256) } else { 0 };
                        c.launch(
                            fx.scan,
                            ((cols as u32).div_ceil(8) + ring_blocks, 1),
                            256,
                            0,
                            kargs!(t_cq, t_gb, gs_s, plan.o.ptr, n_i, cd_i, nv_i, nk_i, dk_i, dv_i, ring_p, plan.qkv.ptr, qs_i, kk_i),
                        )?;
                    }
                    let zs_i = plan.vdmax as i32;
                    let sig0 = 0i32;
                    c.launch(
                        fx.out32,
                        ((n * g.nv) as u32, 1),
                        g.dv as u32,
                        0,
                        kargs!(plan.o.ptr, plan.z.ptr, zs_i, g.norm.ptr, plan.of.ptr, nv_i, dv_i, eps, sig0, zero, pp0, pp1, pp2, ps_i),
                    )?;
                    hst(c, 2, &mut t0)?;
                    g.ms_out.run(c, fx, planes, cols, n)?;
                    hst(c, 3, &mut t0)?;
                }
            }
            Mix::Attn(a) => {
                let (kp, vp) = (lb[0], lb[1]);
                let far = bk.far[li];
                let (fm, fscale, frect) = bk.farp[li];
                let cip = cib + (ai * 8 * 4) as u64;
                let w = &bk.ci[ai * 8..ai * 8 + 8];
                ai += 1;
                let (n0_i, slot0, rb_i, ncap_i, nbcap_i, r0) = (w[0], w[1], w[2], w[3], w[4], 0i32);
                let (nh_i, nkv_i, half_i, gated_i) = (nh as i32, nkv as i32, plan.half as i32, a.gated as i32);
                let (qs_i, kvs_i) = (plan.qrow as i32, (nkv * hd) as i32);
                let (qn, kn) = (a.qn.as_ref().map_or(0, |b| b.ptr), a.kn.as_ref().map_or(0, |b| b.ptr));
                let gate_p = if a.gated { plan.gate.ptr } else { 0 };
                let nul16 = 0u64;
                unsafe {
                    a.ms_in.run(c, fx, planes, hidden, n)?;
                    hst(c, 4, &mut t0)?;
                    if far == 2 {
                        // the ring row this key replaces into the far sums,
                        // before the key takes its place
                        let (fm_i, nkv_i2) = (fm as i32, nkv as i32);
                        c.launch(
                            fx.far_insert,
                            ((nkv * plan.hpk * fm).div_ceil(8) as u32, 1),
                            256,
                            0,
                            kargs!(kp, vp, lb[3], lb[5], lb[6], cip, fm_i, nkv_i2, fscale),
                        )?;
                    }
                    c.launch(
                        fx.qk,
                        ((nh + nkv) as u32, n as u32),
                        hd as u32,
                        0,
                        kargs!(
                            plan.q.ptr, qs_i, plan.k.ptr, plan.v.ptr, kvs_i, qn, kn, plan.qg.ptr, plan.gate.ptr, kp, vp,
                            tabp, nh_i, nkv_i, half_i, eps, gemma_i, gated_i, slot0, n_i, cip, nul16
                        ),
                    )?;
                    // a grid for any record: blocks past the call's keys leave
                    let nblk = ATTN_NBLK as u32;
                    c.launch_3d(
                        fx.scores,
                        (nblk, nkv as u32, n as u32),
                        256,
                        0,
                        kargs!(plan.qg.ptr, kp, s_p, mx_p, n0_i, r0, nkv_i, plan.scale, rb_i, ncap_i, nbcap_i, n_i, cip),
                    )?;
                    if far != 0 {
                        // the far part: one more block of P and E, and its maximum
                        // (each row of a verify window its own)
                        let (fm_i, frect_i) = (fm as i32, frect as i32);
                        if n == 1 {
                            c.launch(
                                fx.far_prep,
                                (nh as u32, 1),
                                256,
                                0,
                                kargs!(plan.qg.ptr, lb[2], lb[4], lb[5], lb[6], mx_p, p_p, e_p, cip, fm_i, nh_i, fscale, frect_i),
                            )?;
                        } else {
                            c.launch(
                                fx.far_prep_b,
                                (nh as u32, n as u32),
                                256,
                                0,
                                kargs!(plan.qg.ptr, lb[2], lb[4], lb[5], lb[6], mx_p, p_p, e_p, cip, fm_i, nh_i, fscale, frect_i, n_i),
                            )?;
                        }
                    }
                    c.launch_3d(
                        fx.values,
                        (nblk, nkv as u32, n as u32),
                        256,
                        0,
                        kargs!(s_p, mx_p, vp, p_p, e_p, n0_i, r0, nkv_i, rb_i, ncap_i, nbcap_i, n_i, cip),
                    )?;
                    c.launch(
                        fx.finish,
                        (nh as u32, n as u32),
                        256,
                        0,
                        kargs!(p_p, e_p, gate_p, pp0, pp1, pp2, ps_i, n0_i, rb_i, nh_i, nbcap_i, cip),
                    )?;
                    hst(c, 5, &mut t0)?;
                    a.ms_out.run(c, fx, planes, nh * hd, n)?;
                    hst(c, 6, &mut t0)?;
                }
            }
        }
        unsafe {
            let nullp = 0u64;
            c.launch(
                fx.rmsnorm2,
                (hidden.div_ceil(1024) as u32, n as u32),
                256,
                0,
                kargs!(plan.h.ptr, pl.post_norm.ptr, nullp, hid_i, eps, gemma_i, pp0, pp1, pp2, ps_i),
            )?;
            hst(c, 0, &mut t0)?;
            // gate/up with silu(gate) * up made in its epilogue, into the
            // middle's planes; down reads them
            pl.ms_gu.run(c, fx, planes, hidden, n)?;
            hst(c, 7, &mut t0)?;
            let mid = Planes { p0: plan.qq[0].ptr, p1: plan.qq[1].ptr, p2: plan.qq[1].ptr, stride: plan.pstride };
            pl.ms_dn.run(c, fx, &mid, pl.inter, n)?;
            hst(c, 9, &mut t0)?;
            // a DFlash drafter's taps: the residual stream after this layer
            for &(tl, dst) in taps {
                if tl == li {
                    cu::memcpy_dtod_async(dst, plan.h.ptr, n * hidden * 4, c.stream).map_err(derr)?;
                }
            }
        }
    }
    Ok(())
}

/// A replayable small-path call: the instantiated graph and the pointers it
/// was recorded with.
pub(super) struct GraphRec {
    exec: cs::CUgraphExec,
    sig: Vec<u64>,
}
unsafe impl Send for GraphRec {}

impl Drop for GraphRec {
    fn drop(&mut self) {
        unsafe {
            let _ = cs::cuGraphExecDestroy(self.exec);
        }
    }
}

fn graphs_on() -> bool {
    static ON: std::sync::OnceLock<bool> = std::sync::OnceLock::new();
    *ON.get_or_init(|| std::env::var("CMF_CUDA_GRAPH").as_deref() != Ok("0")) && !super::prof_on() && !KT_TOKEN.with(|k| k.get())
}

thread_local! {
    /// This token's kernels are timed (`CMF_HYB_KT=N`): launched one by one.
    static KT_TOKEN: std::cell::Cell<bool> = const { std::cell::Cell::new(false) };
}

fn curesult(r: cs::CUresult, what: &str) -> Res<()> {
    if r == cs::CUresult::CUDA_SUCCESS { Ok(()) } else { Err(format!("{what}: {r:?}")) }
}

/// Record `f`'s launches on the stream into an instantiated graph.
unsafe fn capture(c: &Ctx, f: impl FnOnce() -> Res<()>) -> Res<cs::CUgraphExec> {
    unsafe {
        curesult(
            cs::cuStreamBeginCapture_v2(c.stream, cs::CUstreamCaptureMode::CU_STREAM_CAPTURE_MODE_THREAD_LOCAL),
            "begin capture",
        )?;
        let r = f();
        let mut g: cs::CUgraph = std::ptr::null_mut();
        let e = cs::cuStreamEndCapture(c.stream, &mut g);
        r?;
        curesult(e, "end capture")?;
        let mut ex: cs::CUgraphExec = std::ptr::null_mut();
        let ie = cs::cuGraphInstantiateWithFlags(&mut ex, g, 0);
        let _ = cs::cuGraphDestroy(g);
        curesult(ie, "graph instantiate")?;
        Ok(ex)
    }
}

/// The small path, with the head when `head`: book, then the launches —
/// replayed from a recorded graph when one with the same pointers is there,
/// else recorded now (or launched directly, `CMF_CUDA_GRAPH=0` and under
/// `CMF_CUDA_PROF=1`). The rows in `plan.h` and the RoPE table in
/// `plan.tab` must be on the stream already.
#[allow(clippy::too_many_arguments)]
unsafe fn run_small(
    c: &Ctx,
    st: &mut State,
    plan: &HPlan,
    kv_id: u64,
    layer_base: usize,
    pos0: usize,
    n: usize,
    tape: bool,
    head: bool,
    kcap: usize,
    sealed: &[bool],
    taps: &[(usize, u64)],
) -> Res<usize> {
    let bk = book(c, st, plan, kv_id, layer_base, pos0, n, tape, kcap, sealed)?;
    let rows = plan.lm.as_ref().map_or(0, |l| l.1);
    unsafe {
        if !bk.ci.is_empty() {
            cu::memcpy_htod_async(plan.ci.ptr, &bk.ci, c.stream).map_err(derr)?;
        }
        if !graphs_on() {
            launch(c, plan, &bk, n, tape, plan.ci.ptr, plan.tab.ptr, taps)?;
            if head {
                head_launch(c, plan, n)?;
            }
            return Ok(if head { rows } else { 0 });
        }
        let mut far_key = 0u64;
        for (i, &f) in bk.far.iter().enumerate() {
            far_key ^= (f as u64).wrapping_mul(0x9E37_79B9_7F4A_7C15).rotate_left(i as u32 % 64);
        }
        // a call with taps records copies a call without them has not
        for &(tl, _) in taps {
            far_key ^= ((tl as u64 + 1) << 40).wrapping_mul(0xD6E8_FEB8_6659_FD93);
        }
        let key = (plan.id, kv_id, layer_base, n, tape, head, far_key);
        let mut sig: Vec<u64> = bk.layers.iter().flatten().copied().collect();
        sig.extend(bk.scratch);
        sig.extend(bk.far.iter().map(|&f| f as u64));
        sig.extend(bk.farp.iter().map(|&(m, sc, r)| (m as u64) << 33 | (sc.to_bits() as u64) << 1 | r as u64));
        sig.extend(taps.iter().flat_map(|&(tl, dst)| [tl as u64, dst]));
        let hit = st.hyb.graphs.get(&key).is_some_and(|g| g.sig == sig);
        GRAPH_RUN.fetch_add(1, Ordering::Relaxed);
        if !hit {
            GRAPH_REC.fetch_add(1, Ordering::Relaxed);
            if debug() {
                let why = match st.hyb.graphs.get(&key) {
                    None => "new key".to_string(),
                    Some(g) => format!("{} of {} pointers moved", g.sig.iter().zip(&sig).filter(|(a, b)| a != b).count(), sig.len()),
                };
                eprintln!("cuda hybrid: recording a graph for {n} rows (tape {tape}, head {head}): {why}");
            }
            let exec = capture(c, || {
                launch(c, plan, &bk, n, tape, plan.ci.ptr, plan.tab.ptr, taps)?;
                if head {
                    head_launch(c, plan, n)?;
                }
                Ok(())
            })?;
            st.hyb.graphs.insert(key, GraphRec { exec, sig });
        }
        let g = &st.hyb.graphs[&key];
        curesult(cs::cuGraphLaunch(g.exec, c.stream), "graph launch")?;
    }
    Ok(if head { rows } else { 0 })
}

/// The DFlash taps a small-path call keeps (`super::layer_taps_arm`): for
/// every asked layer of a call over the whole stack, its plan index and a
/// device slot of `NMAX` rows. None when nothing was asked or the call does
/// not reach every asked layer.
fn tap_slots(c: &Ctx, st: &mut State, nlayers: usize, layer_base: usize, hidden: usize) -> Res<Option<(Vec<(usize, u64)>, u64)>> {
    let want = super::taps_wanted();
    if want.is_empty() || layer_base != 0 || want.iter().any(|&l| l >= nlayers) {
        return Ok(None);
    }
    let buf = st.scr(c, "hy_taps", want.len() * NMAX * hidden * 4)?;
    let slots = want.iter().enumerate().map(|(s, &l)| (l, buf + (s * NMAX * hidden * 4) as u64)).collect();
    Ok(Some((slots, buf)))
}

/// The kept tap rows (`n` a slot) to the host as `[slot][row][hidden]`, for
/// `super::layer_taps_take` once the stream has run.
unsafe fn taps_read(c: &Ctx, slots: &[(usize, u64)], n: usize, hidden: usize) -> Res<Vec<f32>> {
    let mut raw = vec![0f32; slots.len() * n * hidden];
    for (s, &(_, dst)) in slots.iter().enumerate() {
        unsafe { cu::memcpy_dtoh_async(&mut raw[s * n * hidden..(s + 1) * n * hidden], dst, c.stream).map_err(derr)? };
    }
    Ok(raw)
}

/// The head over the rows in `plan.h`: logits [n][rows] into `plan.logits`.
unsafe fn head_launch(c: &Ctx, plan: &HPlan, n: usize) -> Res<usize> {
    let Some((_, rows, js)) = &plan.lm else { return Ok(0) };
    let fx = &plan.fx;
    let (hid_i, eps, gemma_i) = (plan.hidden as i32, plan.eps, plan.gemma as i32);
    let fnorm = plan.final_norm.as_ref().unwrap().ptr;
    let mut t0 = std::time::Instant::now();
    let planes = Planes { p0: plan.pp[0].ptr, p1: plan.pp[1].ptr, p2: plan.pp[2].ptr, stride: plan.pstride };
    let (pp0, pp1, pp2, ps_i, nullp) = (planes.p0, planes.p1, planes.p2, plan.pstride as i32, 0u64);
    unsafe {
        c.launch(
            fx.rmsnorm2,
            (plan.hidden.div_ceil(1024) as u32, n as u32),
            256,
            0,
            kargs!(plan.h.ptr, fnorm, nullp, hid_i, eps, gemma_i, pp0, pp1, pp2, ps_i),
        )?;
        js.run(c, fx, &planes, plan.hidden, n)?;
    }
    hst(c, 10, &mut t0)?;
    Ok(*rows)
}

#[allow(clippy::too_many_arguments)]
pub(super) fn token(
    c: &'static Ctx,
    model: &Arc<CmfModel>,
    kv_id: u64,
    layers: &[GraphLayer],
    o1: &[Option<Vec<crate::nystrom::O1DeviceView<'_>>>],
    o1_epoch: u64,
    invf: &[f32],
    h: &mut [f32],
    nh: usize,
    nkv: usize,
    hd: usize,
    attn_scale: f32,
    rd: usize,
    hidden: usize,
    position: usize,
    cap: usize,
    gemma: bool,
    eps: f32,
    lm_head: Option<(&GraphW, usize)>,
    final_norm: &[f32],
    logits: &mut Vec<f32>,
    loop_norm_at: &[usize],
    steps: usize,
    layers_run: Option<&mut usize>,
    layer_base: usize,
    hidden_too: bool,
) -> TokenGraphOutcome {
    if steps != 1 || !loop_norm_at.is_empty() || (!o1.is_empty() && o1.len() != layers.len()) {
        decline("hybrid token", "a multi-step burst, a mid-stack norm or far-field views short of the layers");
        return TokenGraphOutcome::Declined;
    }
    let sealed: Vec<bool> = (0..layers.len()).map(|li| o1.get(li).is_some_and(Option::is_some)).collect();
    if position >= cap || h.len() != hidden {
        decline("hybrid token", "position past the context cap");
        return TokenGraphOutcome::Declined;
    }
    let mut st = c.st.lock().unwrap();
    if c.current().is_err() {
        return TokenGraphOutcome::Declined;
    }
    let plan = match plan_for(c, &mut st, model, layers, layer_base, lm_head, final_norm, nh, nkv, hd, rd, hidden, gemma, eps, attn_scale, invf, position) {
        Ok(p) => p,
        Err(e) => {
            decline("hybrid token", &e);
            return TokenGraphOutcome::Declined;
        }
    };
    // the far fields first: a state the card has moved has no host copy to
    // go back to, so from there a refusal is a failure
    let moved = sealed.iter().enumerate().any(|(li, &s)| s && st.hyb.o1.get(&(kv_id, layer_base + li)).is_some_and(|d| d.next_pos.is_some()));
    for (li, views) in o1.iter().enumerate() {
        let Some(views) = views else { continue };
        let key = (kv_id, layer_base + li);
        let at = st.hyb.o1.get(&key).filter(|d| d.epoch == o1_epoch).and_then(|d| d.next_pos);
        let res = match at {
            Some(next) if next != position => Err(format!("far field: the card is at row {next}, the call at {position}")),
            _ => o1_ensure(c, &mut st, key, views, o1_epoch, plan.hpk, nkv, hd),
        };
        if let Err(e) = res {
            decline("hybrid token", &e);
            return if moved { TokenGraphOutcome::Failed } else { TokenGraphOutcome::Declined };
        }
        // a sealed layer reads no mirror again
        if let Some(m) = st.kv.remove(&key) {
            let _ = c.sync();
            c.kv_bytes.fetch_sub((m.cap * m.row * 8) as u64, Ordering::Relaxed);
        }
    }
    if let Err((hard, e)) = admit(c, &mut st, &plan, layers, kv_id, layer_base, position, 1, 0, &sealed) {
        decline("hybrid token", &e);
        return if hard || moved { TokenGraphOutcome::Failed } else { TokenGraphOutcome::Declined };
    }
    let t0 = std::time::Instant::now();
    // `CMF_HYB_KT=N`: the kernel times of the N-th token
    static TOKENS_SEEN: std::sync::atomic::AtomicUsize = std::sync::atomic::AtomicUsize::new(0);
    let kt = {
        let n = TOKENS_SEEN.fetch_add(1, Ordering::Relaxed);
        std::env::var("CMF_HYB_KT").ok().and_then(|v| v.parse::<usize>().ok()) == Some(n)
    };
    if kt {
        KT_TOKEN.with(|k| k.set(true));
        super::kt_start();
    }
    let mut run = || -> Res<()> {
        unsafe {
            cu::memcpy_htod_async(plan.h.ptr, h, c.stream).map_err(derr)?;
            let tab = rope_rows(&plan.invf, std::iter::once(position));
            if !tab.is_empty() {
                cu::memcpy_htod_async(plan.tab.ptr, &tab, c.stream).map_err(derr)?;
            }
            let taps = tap_slots(c, &mut st, layers.len(), layer_base, hidden)?;
            let tl = taps.as_ref().map_or(&[][..], |t| &t.0[..]);
            let rows = run_small(c, &mut st, &plan, kv_id, layer_base, position, 1, false, plan.lm.is_some(), cap, &sealed, tl)?;
            if taps.is_some() {
                // a pageable read: the rows are there when it returns
                super::taps_put(taps_read(c, tl, 1, hidden)?);
            }
            if rows > 0 {
                logits.resize(rows, 0.0);
                cu::memcpy_dtoh_async(logits.as_mut_slice(), plan.logits.ptr, c.stream).map_err(derr)?;
                if hidden_too {
                    cu::memcpy_dtoh_async(h, plan.h.ptr, c.stream).map_err(derr)?;
                }
            } else {
                cu::memcpy_dtoh_async(h, plan.h.ptr, c.stream).map_err(derr)?;
            }
        }
        c.sync()
    };
    let res = run();
    if kt {
        super::kt_stop();
        KT_TOKEN.with(|k| k.set(false));
    }
    match res {
        Ok(()) => {
            if let Some(lr) = layers_run {
                *lr = layers.len();
            }
            if debug() {
                super::TOKEN_US.lock().unwrap().push(t0.elapsed().as_micros() as u32);
            }
            if super::TOKENS.fetch_add(1, Ordering::Relaxed) >= 16 {
                super::TOKEN_NS.fetch_add(t0.elapsed().as_nanos() as u64, Ordering::Relaxed);
            }
            if position % 256 == 0 {
                mem_report(c, &st, &format!("a token at {position}"));
            }
            TokenGraphOutcome::Completed
        }
        Err(e) => {
            tracing::error!("cuda hybrid token graph failed: {e}");
            let _ = c.sync();
            TokenGraphOutcome::Failed
        }
    }
}

/// The round's `k` greedy drafts of an MTP block in ONE graph: per step,
/// the block input (the token's embedding and the hidden row, each RMS-normed,
/// through eh_proj), the block over its K/V mirror at `position + j`, the head
/// over the shortlist and its argmax, whose token the next step's embedding
/// is gathered for on the card. One upload, one launch, the k ids back.
/// `e0` is the embedding of `t0` (the host's); `embed_row(id, out)` fills a
/// shortlist row the first time the table is built. Err(false): declined
/// before any state changed; Err(true): failed after the mirror moved.
#[allow(clippy::too_many_arguments)]
pub(super) fn mtp_chain(
    c: &'static Ctx,
    model: &Arc<CmfModel>,
    kv_id: u64,
    layer: &GraphLayer,
    invf: &[f32],
    h0: &[f32],
    e0: &[f32],
    t0: u32,
    k: usize,
    nh: usize,
    nkv: usize,
    hd: usize,
    attn_scale: f32,
    rd: usize,
    hidden: usize,
    position: usize,
    cap: usize,
    gemma: bool,
    eps: f32,
    lm_head: (&GraphW, usize),
    final_norm: &[f32],
    eh: &GraphW,
    enorm: &[f32],
    hnorm: &[f32],
    embed: &GraphW,
    layer_base: usize,
) -> Result<Vec<u32>, bool> {
    let rows = lm_head.1;
    if k == 0 || k > NMAX || h0.len() != hidden || e0.len() != hidden || enorm.len() != hidden || hnorm.len() != hidden {
        return Err(false);
    }
    if position + k >= cap || eh.kind != 6 || embed.kind != 6 || embed.prism != crate::gpu::GraphPrismOp::None || hidden % 1024 != 0 {
        return Err(false);
    }
    let layers = std::slice::from_ref(layer);
    let mut st = c.st.lock().unwrap();
    if c.current().is_err() {
        return Err(false);
    }
    let st = &mut *st;
    let plan = match plan_for(c, st, model, layers, layer_base, Some(lm_head), final_norm, nh, nkv, hd, rd, hidden, gemma, eps, attn_scale, invf, position) {
        Ok(p) => p,
        Err(e) => {
            decline("hybrid draft chain", &e);
            return Err(false);
        }
    };
    if plan.pstride < 2 * hidden || plan.layers.iter().any(|l| !matches!(l.mix, Mix::Attn(_))) {
        decline("hybrid draft chain", "planes narrower than the block input, or a GDN layer");
        return Err(false);
    }
    // the table, the norms and eh_proj's job tables (built once)
    let key = (model.uid(), embed.idx, rows, crate::gpu::fp_f32(enorm), crate::gpu::fp_f32(hnorm));
    let prep = |st: &mut State| -> Res<()> {
        if st.hyb.chain.as_ref().is_none_or(|r| r.key != key) {
            if let Some(old) = st.hyb.chain.take() {
                c.resident.fetch_sub(old.emb.held, Ordering::Relaxed);
            }
            use cortiq_core::quant::q4tp_sections;
            let entry = model.tensors.get(embed.idx).ok_or("embedding index out of range")?;
            let erows = *entry.shape.first().ok_or("embedding without a shape")?;
            if entry.shape.get(1) != Some(&hidden) || rows > erows || rows % 16 != 0 {
                return Err(format!("embedding {:?} against a {rows}-row shortlist of width {hidden}", entry.shape));
            }
            // the first `rows` rows: each section's leading rows, from the file
            let bytes = model.entry_bytes(entry);
            let (po, co, stride) = q4tp_sections(erows, hidden);
            if bytes.len() < co + erows * stride {
                return Err("embedding payload shorter than its planes".into());
            }
            let fx = st.hyb.funcs.values().next().cloned().ok_or("repack before the hybrid kernels")?;
            let t0 = std::time::Instant::now();
            let (w, aux0, aux1) = repack_rows_on_card(c, &fx, &bytes[..po], &bytes[po..co], &bytes[co..], stride, rows, hidden)?;
            if debug() {
                eprintln!("cuda hybrid draft chain: {rows} embedding rows on the card in {:.1} ms", t0.elapsed().as_secs_f64() * 1e3);
            }
            let held = (rows * (hidden / 32) * 16 + rows * (4 + stride)) as u64;
            let emb = DevMat { kind: 6, rows, cols: hidden, w: Arc::new(w), aux0: Some(aux0), aux1: Some(aux1), held };
            c.resident.fetch_add(emb.held, Ordering::Relaxed);
            st.hyb.chain = Some(ChainRes { key, emb, enorm: Buf::from_slice(enorm)?, hnorm: Buf::from_slice(hnorm)? });
        }
        if !st.hyb.chain_sets.contains_key(&(plan.id, eh.idx)) {
            let m = hmat(c, st, model, eh, hidden, 2 * hidden)?;
            let ms = mmaset(&[(&m, plan.h.ptr, 0, hidden, hidden)])?;
            st.hyb.chain_sets.insert((plan.id, eh.idx), ms);
        }
        Ok(())
    };
    if let Err(e) = prep(st) {
        decline("hybrid draft chain", &e);
        return Err(false);
    }
    if let Err((hard, e)) = admit(c, st, &plan, layers, kv_id, layer_base, position, k, 0, &[false]) {
        decline("hybrid draft chain", &e);
        return Err(hard);
    }
    // from here the mirror moves: a failure is terminal
    let run = |st: &mut State| -> Res<Vec<u32>> {
        let mut books = Vec::with_capacity(k);
        for j in 0..k {
            books.push(book(c, st, &plan, kv_id, layer_base, position + j, 1, false, cap, &[false])?);
        }
        let cilen = books[0].ci.len();
        let tabrow = plan.invf.len() * 2;
        let ci_b = st.scr(c, "hy_chain_ci", NMAX * cilen.max(1) * 4)?;
        let tab_b = st.scr(c, "hy_chain_tab", NMAX * tabrow.max(1) * 4)?;
        let e_b = st.scr(c, "hy_chain_e", NMAX * hidden * 4)?;
        let h0_b = st.scr(c, "hy_chain_h0", hidden * 4)?;
        let tok_b = st.scr(c, "hy_chain_tok", (NMAX + 1) * 4)?;
        let ci: Vec<i32> = books.iter().flat_map(|b| b.ci.iter().copied()).collect();
        let tab = rope_rows(&plan.invf, position..position + k);
        let ch = st.hyb.chain.as_ref().unwrap();
        let (etab, ecod, epar, enp, hnp) = (ch.emb.w.ptr, ch.emb.aux0(), ch.emb.aux1.as_ref().map_or(0, |b| b.ptr), ch.enorm.ptr, ch.hnorm.ptr);
        let tok0 = [t0 as i32];
        unsafe {
            if !ci.is_empty() {
                cu::memcpy_htod_async(ci_b, &ci, c.stream).map_err(derr)?;
            }
            if !tab.is_empty() {
                cu::memcpy_htod_async(tab_b, &tab, c.stream).map_err(derr)?;
            }
            cu::memcpy_htod_async(e_b, e0, c.stream).map_err(derr)?;
            cu::memcpy_htod_async(h0_b, h0, c.stream).map_err(derr)?;
            cu::memcpy_htod_async(tok_b, &tok0, c.stream).map_err(derr)?;
        }
        let ms = &st.hyb.chain_sets[&(plan.id, eh.idx)];
        let fx = &plan.fx;
        let (hid_i, gemma_i, rows_i) = (hidden as i32, gemma as i32, rows as i32);
        let ps_i = plan.pstride as i32;
        let (p0, p1, p2) = (plan.pp[0].ptr, plan.pp[1].ptr, plan.pp[2].ptr);
        let off = (hidden * 2) as u64;
        let planes = Planes { p0, p1, p2, stride: plan.pstride };
        let body = || -> Res<()> {
            let nullp = 0u64;
            for (j, bk) in books.iter().enumerate() {
                let ej = e_b + (j * hidden * 4) as u64;
                let hsrc = if j == 0 { h0_b } else { plan.h.ptr };
                let (j_i, j1_i) = (j as i32, j as i32 + 1);
                let (q0, q1, q2) = (p0 + off, p1 + off, p2 + off);
                unsafe {
                    if j > 0 {
                        c.launch(fx.gather, (hidden.div_ceil(256) as u32, 1), 256, 0, kargs!(etab, ecod, epar, rows_i, hid_i, tok_b, j_i, ej))?;
                    }
                    c.launch(fx.rmsnorm2, (hidden.div_ceil(1024) as u32, 1), 256, 0, kargs!(ej, enp, nullp, hid_i, eps, gemma_i, p0, p1, p2, ps_i))?;
                    c.launch(fx.rmsnorm2, (hidden.div_ceil(1024) as u32, 1), 256, 0, kargs!(hsrc, hnp, nullp, hid_i, eps, gemma_i, q0, q1, q2, ps_i))?;
                    ms.run(c, fx, &planes, 2 * hidden, 1)?;
                    launch(c, &plan, bk, 1, false, ci_b + (j * cilen * 4) as u64, tab_b + (j * tabrow * 4) as u64, &[])?;
                    head_launch(c, &plan, 1)?;
                    c.launch(fx.argmax, (1, 1), 1024, 0, kargs!(plan.logits.ptr, rows_i, rows_i, tok_b, j1_i))?;
                }
            }
            Ok(())
        };
        unsafe {
            if graphs_on() {
                let gkey = (plan.id, kv_id, layer_base, k, true, true, 0xC4A1_0000u64);
                let mut sig: Vec<u64> = books[0].layers.iter().flatten().copied().collect();
                sig.extend(books[0].scratch);
                sig.extend([etab, ecod, epar, enp, hnp, ci_b, tab_b, e_b, h0_b, tok_b, ms.bufs[0].ptr, ms.bufs[1].ptr]);
                if st.hyb.graphs.get(&gkey).is_none_or(|g| g.sig != sig) {
                    GRAPH_REC.fetch_add(1, Ordering::Relaxed);
                    let exec = capture(c, body)?;
                    st.hyb.graphs.insert(gkey, GraphRec { exec, sig });
                }
                GRAPH_RUN.fetch_add(1, Ordering::Relaxed);
                curesult(cs::cuGraphLaunch(st.hyb.graphs[&gkey].exec, c.stream), "graph launch")?;
            } else {
                body()?;
            }
            let mut ids = vec![0i32; k + 1];
            cu::memcpy_dtoh_async(ids.as_mut_slice(), tok_b, c.stream).map_err(derr)?;
            c.sync()?;
            Ok(ids[1..].iter().map(|&t| t as u32).collect())
        }
    };
    match run(st) {
        Ok(ids) => Ok(ids),
        Err(e) => {
            tracing::error!("cuda hybrid draft chain failed: {e}");
            let _ = c.sync();
            Err(true)
        }
    }
}

#[allow(clippy::too_many_arguments)]
pub(super) fn batch(
    c: &'static Ctx,
    model: &Arc<CmfModel>,
    kv_id: u64,
    layer_base: usize,
    layers: &[GraphLayer],
    invf: &[f32],
    h: &mut [f32],
    nh: usize,
    nkv: usize,
    hd: usize,
    rd: usize,
    hidden: usize,
    positions: &[usize],
    cap: usize,
    gemma: bool,
    eps: f32,
    attn_scale: f32,
    k: usize,
    o1: &[Option<Vec<crate::nystrom::O1DeviceView<'_>>>],
    o1_epoch: u64,
    spec: Option<SpecTail<'_>>,
    layers_run: Option<&mut usize>,
    defer: bool,
) -> BatchGraphOutcome {
    let b = k;
    // a sealed layer takes a verify window over its far field (the state on
    // the card at this position), or a prompt chunk read past the seal
    let sealed: Vec<bool> = (0..layers.len()).map(|li| o1.get(li).is_some_and(Option::is_some)).collect();
    let any_sealed = sealed.iter().any(|&s| s);
    let prompt_far = any_sealed && spec.is_none();
    if any_sealed && spec.is_some() && (b > NMAX || !far_spec_on()) {
        decline("hybrid batch", "a sealed far field under a verify window longer than the small path");
        return BatchGraphOutcome::Declined;
    }
    if prompt_far && !far_chunks_on() {
        decline("hybrid batch", "a sealed far field under a prompt chunk (CMF_O1_CHUNKS=0)");
        return BatchGraphOutcome::Declined;
    }
    if b == 0
        || positions.len() != b
        || h.len() != b * hidden
        || positions.windows(2).any(|w| w[1] != w[0] + 1)
        || positions[b - 1] >= cap
    {
        decline("hybrid batch", "rows not contiguous or past the context cap");
        return BatchGraphOutcome::Declined;
    }
    if spec.as_ref().is_some_and(|s| s.pick.is_some()) || (spec.is_some() && b > NMAX) {
        decline("hybrid batch", "a verify window longer than the small path, or a device pick");
        return BatchGraphOutcome::Declined;
    }
    let pos0 = positions[0];
    let t_all = std::time::Instant::now();
    let mut st = c.st.lock().unwrap();
    if c.current().is_err() {
        return BatchGraphOutcome::Declined;
    }
    let lm = spec.as_ref().map(|s| (&s.lm, s.lm_rows));
    let final_norm: &[f32] = spec.as_ref().map_or(&[], |s| s.final_norm);
    let plan = match plan_for(c, &mut st, model, layers, layer_base, lm, final_norm, nh, nkv, hd, rd, hidden, gemma, eps, attn_scale, invf, pos0) {
        Ok(p) => p,
        Err(e) => {
            decline("hybrid batch", &e);
            return BatchGraphOutcome::Declined;
        }
    };
    if prompt_far {
        // a prompt chunk past a seal: the epoch's far field goes up when it
        // is new (its rows are this call's first position), and the layer's
        // exact mirror goes
        let moved = sealed.iter().enumerate().any(|(li, &s)| s && st.hyb.o1.get(&(kv_id, layer_base + li)).is_some_and(|d| d.epoch == o1_epoch && d.next_pos.is_some()));
        for (li, views) in o1.iter().enumerate() {
            let Some(views) = views else { continue };
            let key = (kv_id, layer_base + li);
            let at = st.hyb.o1.get(&key).filter(|d| d.epoch == o1_epoch).map(|d| d.next_pos);
            let rows = views.first().map_or(0, |v| v.sink_len + v.win_len + v.far_len);
            let res = match at {
                Some(Some(next)) if next != pos0 => Err(format!("far field: the card is at row {next}, the call at {pos0}")),
                Some(Some(_)) => Ok(()),
                _ if rows != pos0 => Err(format!("far field: the seal holds {rows} rows, the call is at {pos0}")),
                _ => o1_ensure(c, &mut st, key, views, o1_epoch, plan.hpk, nkv, hd),
            };
            if let Err(e) = res {
                decline("hybrid batch", &e);
                return if moved { BatchGraphOutcome::Failed } else { BatchGraphOutcome::Declined };
            }
            if let Some(m) = st.kv.remove(&key) {
                let _ = c.sync();
                c.kv_bytes.fetch_sub((m.cap * m.row * 8) as u64, Ordering::Relaxed);
            }
        }
    } else if any_sealed {
        for (li, &s) in sealed.iter().enumerate() {
            let at = st.hyb.o1.get(&(kv_id, layer_base + li)).and_then(|d| d.next_pos);
            if s && at != Some(pos0) {
                decline("hybrid batch", "a far field the card has not moved to this row");
                return BatchGraphOutcome::Declined;
            }
        }
    }
    let hint = st.kv_hint;
    if let Err((hard, e)) = admit(c, &mut st, &plan, layers, kv_id, layer_base, pos0, b, hint, &sealed) {
        decline("hybrid batch", &e);
        return if hard { BatchGraphOutcome::Failed } else { BatchGraphOutcome::Declined };
    }
    let mut spec = spec;
    let mut run = || -> Res<()> {
        if b <= NMAX && !prompt_far {
            unsafe {
                cu::memcpy_htod_async(plan.h.ptr, h, c.stream).map_err(derr)?;
                let tab = rope_rows(&plan.invf, positions.iter().copied());
                if !tab.is_empty() {
                    cu::memcpy_htod_async(plan.tab.ptr, &tab, c.stream).map_err(derr)?;
                }
                let taps = tap_slots(c, &mut st, layers.len(), layer_base, hidden)?;
                let tl = taps.as_ref().map_or(&[][..], |t| &t.0[..]);
                let rows = run_small(c, &mut st, &plan, kv_id, layer_base, pos0, b, spec.is_some(), spec.is_some(), cap, &sealed, tl)?;
                if taps.is_some() {
                    super::taps_put(taps_read(c, tl, b, hidden)?);
                }
                // a greedy verify the pipeline armed: each row's argmax on
                // the card, the ids back instead of the logits
                let pick = spec.is_some() && super::verify_argmax_armed();
                if pick {
                    let ids_b = st.scr(c, "hy_vids", NMAX * 4)?;
                    let (rows_i, zero_i) = (rows as i32, 0i32);
                    c.launch(plan.fx.argmax, (b as u32, 1), 1024, 0, kargs!(plan.logits.ptr, rows_i, rows_i, ids_b, zero_i))?;
                    let mut ids = vec![0i32; b];
                    cu::memcpy_dtoh_async(ids.as_mut_slice(), ids_b, c.stream).map_err(derr)?;
                    c.sync()?;
                    super::verify_argmax_put(ids.iter().map(|&t| t as u32).collect());
                } else if let Some(sp) = spec.as_mut() {
                    sp.logits_out.resize(b * rows, 0.0);
                    cu::memcpy_dtoh_async(sp.logits_out.as_mut_slice(), plan.logits.ptr, c.stream).map_err(derr)?;
                }
                if !defer || spec.is_some() {
                    cu::memcpy_dtoh_async(h, plan.h.ptr, c.stream).map_err(derr)?;
                } else {
                    // nothing comes back: the stream runs on
                    return Ok(());
                }
            }
            c.sync()
        } else {
            if prompt_far && b > SEAL_ROWS {
                // past a seal the far field's mirror has room for SEAL_ROWS rows
                // after its near ones: the chunk in pieces of that many
                let mut at = 0usize;
                while at < b {
                    let n = SEAL_ROWS.min(b - at);
                    let last = at + n == b;
                    unsafe { big(c, &mut st, &plan, kv_id, layer_base, &positions[at..at + n], &mut h[at * hidden..(at + n) * hidden], defer && last, &sealed)? };
                    at += n;
                }
                return Ok(());
            }
            unsafe { big(c, &mut st, &plan, kv_id, layer_base, positions, h, defer, &sealed) }
        }
    };
    match run() {
        Ok(()) => {
            if let Some(lr) = layers_run {
                *lr = layers.len();
            }
            super::BATCH_ROWS.fetch_add(b as u64, Ordering::Relaxed);
            super::BATCH_NS.fetch_add(t_all.elapsed().as_nanos() as u64, Ordering::Relaxed);
            BatchGraphOutcome::Completed
        }
        Err(e) => {
            tracing::error!("cuda hybrid batch graph failed: {e}");
            let _ = c.sync();
            BatchGraphOutcome::Failed
        }
    }
}

/// A sealed layer's far sums on the card (`kv_cache`'s seal without the
/// host's replay): `hy_far_insert_n` over the far rows a tile at a time, the
/// running maxima, masses and numerators staying on the card between tiles.
/// Row r of group g is `ks[g][r * stride..][..hd]`: `stride` is `hd` for a
/// group's own rows, the whole row's width for rows that hold every KV head
/// (Dory's, `ks[g]` then starting at the group's first value).
#[allow(clippy::too_many_arguments)]
pub(super) fn far_sums(c: &Ctx, nkv: usize, hpk: usize, hd: usize, m: usize, scale: f32, qt: &[f32], idx: &[u32], ks: &[&[f32]], vs: &[&[f32]], stride: usize) -> Res<(Vec<f32>, Vec<f32>)> {
    let nh = nkv * hpk;
    let n = idx.len();
    let top = idx.iter().max().map_or(0, |&r| r as usize + 1);
    if m == 0 || m > 64 || stride < hd || ks.len() != nkv || vs.len() != nkv || qt.len() != nh * m * hd || ks.iter().chain(vs).any(|x| top > 0 && x.len() < (top - 1) * stride + hd) {
        return Err("far sums: unexpected geometry".into());
    }
    let t0 = std::time::Instant::now();
    let mut st = c.st.lock().unwrap();
    c.current()?;
    let fx = match st.hyb.funcs.get(&(hpk, hd)) {
        Some(f) => f.clone(),
        None => {
            let f = Arc::new(load_funcs(c, hpk, hd)?);
            st.hyb.funcs.insert((hpk, hd), f.clone());
            f
        }
    };
    let mut mz0 = vec![0f32; nh * 2 * m];
    for h in 0..nh {
        mz0[h * 2 * m..h * 2 * m + m].fill(f32::NEG_INFINITY);
    }
    let qtb = Buf::from_slice(qt)?;
    let mzb = Buf::from_slice(&mz0)?;
    let thb = Buf::new(nh * m * hd * 4)?;
    unsafe { cu::memset_d8_async(thb.ptr, 0, nh * m * hd * 4, c.stream).map_err(derr)? };
    // a tile of rows [row][kv head][hd] in f16, as the far field's mirror keeps them
    const TILE: usize = 8192;
    let tile = TILE.min(n.max(1));
    let row = nkv * hd;
    let kb = st.scr(c, "fs_k", tile * row * 2)?;
    let vb = st.scr(c, "fs_v", tile * row * 2)?;
    // two host tiles: one fills on the threads while the card sums the other
    let mut hk = [vec![0u16; tile * row], vec![0u16; tile * row]];
    let mut hv = [vec![0u16; tile * row], vec![0u16; tile * row]];
    let h16 = cortiq_core::quant::f32_to_f16;
    let threads = std::thread::available_parallelism().map_or(4, |p| p.get()).clamp(1, 16);
    let mut r0 = 0usize;
    let mut slot = 0usize;
    while r0 < n {
        let nr = tile.min(n - r0);
        let (kh, vh) = (&mut hk[slot], &mut hv[slot]);
        let per = nr.div_ceil(threads).max(1);
        std::thread::scope(|sc| {
            for (ti, (kc, vc)) in kh[..nr * row].chunks_mut(per * row).zip(vh[..nr * row].chunks_mut(per * row)).enumerate() {
                sc.spawn(move || {
                    for (j, (kd, vd)) in kc.chunks_mut(row).zip(vc.chunks_mut(row)).enumerate() {
                        let at = idx[r0 + ti * per + j] as usize * stride;
                        for g in 0..nkv {
                            let (ksrc, vsrc) = (&ks[g][at..at + hd], &vs[g][at..at + hd]);
                            for d in 0..hd {
                                kd[g * hd + d] = h16(ksrc[d]);
                                vd[g * hd + d] = h16(vsrc[d]);
                            }
                        }
                    }
                });
            }
        });
        let (m_i, nkv_i, zero_i, w_i, e_i) = (m as i32, nkv as i32, 0i32, nr as i32, nr as i32);
        let warps = nh * m;
        unsafe {
            // stream order: this tile's copy waits for the last tile's sums
            cu::memcpy_htod_async(kb, &kh[..nr * row], c.stream).map_err(derr)?;
            cu::memcpy_htod_async(vb, &vh[..nr * row], c.stream).map_err(derr)?;
            c.launch(fx.far_insert_n, ((warps * 32).div_ceil(256) as u32, 1), 256, 0, kargs!(kb, vb, qtb.ptr, mzb.ptr, thb.ptr, m_i, nkv_i, scale, zero_i, w_i, zero_i, e_i))?;
        }
        r0 += nr;
        slot ^= 1;
        if slot == 0 {
            // both host tiles are queued: the older must be on the card before it is written again
            c.sync()?;
        }
    }
    let mut mz = vec![0f32; nh * 2 * m];
    let mut th = vec![0f32; nh * m * hd];
    unsafe {
        cu::memcpy_dtoh_async(mz.as_mut_slice(), mzb.ptr, c.stream).map_err(derr)?;
        cu::memcpy_dtoh_async(th.as_mut_slice(), thb.ptr, c.stream).map_err(derr)?;
    }
    c.sync()?;
    if debug() {
        eprintln!("cuda hybrid: far sums of {n} rows x {nkv} groups on the card in {:.1} ms", t0.elapsed().as_secs_f64() * 1e3);
    }
    Ok((mz, th))
}

/// `CMF_CUDA_MEM=1`: what the card holds, on stderr, after `what`.
pub(super) fn mem_report(c: &Ctx, st: &State, what: &str) {
    static ON: std::sync::OnceLock<bool> = std::sync::OnceLock::new();
    if !*ON.get_or_init(|| std::env::var("CMF_CUDA_MEM").as_deref() == Ok("1")) {
        return;
    }
    let mb = |b: u64| b as f64 / (1u64 << 20) as f64;
    let (free, total) = cu::mem_get_info().unwrap_or((0, 0));
    let scratch: u64 = st.scratch.values().map(|b| b.bytes as u64).sum();
    let mut big: Vec<(&str, usize)> = st.scratch.iter().map(|(k, b)| (*k, b.bytes)).collect();
    big.sort_by_key(|x| std::cmp::Reverse(x.1));
    let gdn: u64 = st
        .hyb
        .gdn
        .values()
        .map(|g| {
            let snap = g.snap.as_ref().map_or(0, |s| s.1.bytes + s.2.bytes);
            (g.s.bytes + g.ring.bytes + g.t_qkv.bytes + g.t_cq.bytes + g.t_gb.bytes + g.s_save.bytes + g.ring_save.bytes + snap) as u64
        })
        .sum();
    let o1: u64 = st.hyb.o1.values().map(|d| d.bytes).sum();
    let mirrors: Vec<(usize, usize)> = st.kv.values().map(|m| (m.cap, m.stored - m.base)).collect();
    let mcap: usize = mirrors.iter().map(|m| m.0).max().unwrap_or(0);
    eprintln!(
        "cuda mem after {what}: used {:.0} of {:.0} MiB; weights {:.0}, K/V {:.0} (far fields {:.0}, {} mirrors of up to {mcap} rows), GDN states {:.0} in {}, scratch {:.0} (largest {}), graphs {}",
        mb((total - free) as u64),
        mb(total as u64),
        mb(c.resident.load(Ordering::Relaxed)),
        mb(c.kv_bytes.load(Ordering::Relaxed)),
        mb(o1),
        mirrors.len(),
        mb(gdn),
        st.hyb.gdn.len(),
        mb(scratch),
        big.iter().take(6).map(|(k, b)| format!("{k} {:.0}", mb(*b as u64))).collect::<Vec<_>>().join(", "),
        st.hyb.graphs.len()
    );
}

/// The prompt path: a chunk of `b > NMAX` rows on cuBLAS.
#[allow(clippy::too_many_arguments)]
unsafe fn big(
    c: &Ctx,
    st: &mut State,
    plan: &HPlan,
    kv_id: u64,
    layer_base: usize,
    positions: &[usize],
    h: &mut [f32],
    defer: bool,
    sealed: &[bool],
) -> Res<()> {
    let fx = &plan.fx;
    let b = positions.len();
    let pos0 = positions[0];
    let (hidden, nh, nkv, hd, hpk) = (plan.hidden, plan.nh, plan.nkv, plan.hd, plan.hpk);
    let (gw, _) = prec();
    if gw != Gemm::Half {
        return Err("the hybrid prompt path runs f16 weight GEMMs only".into());
    }
    let (eps, gemma_i, b_i, one) = (plan.eps, plan.gemma as i32, b as i32, 1i32);
    let hid_i = hidden as i32;
    let (nul, zero_i) = (0u64, 0i32);
    let bh = st.scr(c, "hb_h", b * hidden * 4)?;
    let bxn = st.scr(c, "hb_xn", b * hidden * 2)?;
    let wmax = plan
        .layers
        .iter()
        .flat_map(|l| {
            let mut v = vec![(l.gu.rows, l.gu.cols), (l.down.rows, l.down.cols)];
            match &l.mix {
                Mix::Gdn(g) => v.extend([(g.qkv.rows, g.qkv.cols), (g.z.rows, g.z.cols), (g.out.rows, g.out.cols)]),
                Mix::Attn(a) => v.extend([(a.q.rows, a.q.cols), (a.k.rows, a.k.cols), (a.v.rows, a.v.cols), (a.o.rows, a.o.cols)]),
            }
            v
        })
        .map(|(r, c)| r * c)
        .max()
        .unwrap_or(0);
    // a large weight expands a band of rows at a time
    let wd = st.scr(c, "hb_wd", (wmax * 2).min(wd_cap().max(1 << 20)))?;
    // a DFlash drafter's taps: the chunk's rows after each asked layer
    let want = super::taps_wanted();
    let tapping = !want.is_empty() && layer_base == 0 && want.iter().all(|&l| l < plan.layers.len());
    let mut tapped = if tapping { vec![0f32; want.len() * b * hidden] } else { Vec::new() };
    let tab = rope_rows(&plan.invf, positions.iter().copied());
    let btab = st.scr(c, "hb_tab", tab.len().max(1) * 4)?;
    unsafe {
        cu::memcpy_htod_async(bh, h, c.stream).map_err(derr)?;
        if !tab.is_empty() {
            cu::memcpy_htod_async(btab, &tab, c.stream).map_err(derr)?;
        }
    }
    let tm_ = std::cell::Cell::new(std::time::Instant::now());
    let tm = &tm_;
    BROWS.fetch_add(b as u64, Ordering::Relaxed);
    bst(c, 7, tm)?;
    // y [b][ldy] (+)= alpha W x: the weight expanded to f16, a GEMM a band
    // of rows (the whole matrix when it fits the band)
    let wd_rows = (wd_cap().max(1 << 20) / 2).max(1);
    let mul = |m: &DevMat, x: u64, y: u64, ldy: usize, alpha: f32, beta: f32| -> Res<()> {
        let (rows_i, cols_i) = (m.rows as i32, m.cols as i32);
        if m.kind != 6 {
            return Err("a weight other than repacked q4tp on the prompt path".into());
        }
        if m.rows * m.cols <= wd_rows {
            let threads = (m.rows.div_ceil(16) * (m.cols / 32) * 32) as u64;
            unsafe {
                c.launch(fx.expand_p, (threads.div_ceil(256) as u32, 1), 256, 0, kargs!(m.w.ptr, m.aux0(), m.aux1(), wd, rows_i, cols_i))?;
                bst(c, 1, tm)?;
                c.gemm(true, false, m.rows, b, m.cols, alpha, wd, m.cols, x, m.cols, beta, y, ldy, Gemm::Half)?;
                return bst(c, 2, tm);
            }
        }
        let tiles = m.rows.div_ceil(16);
        let band = (wd_rows / m.cols / 16).max(1);
        let mut t0 = 0usize;
        while t0 < tiles {
            let nt = band.min(tiles - t0);
            let r0 = t0 * 16;
            let nr = (nt * 16).min(m.rows - r0);
            let threads = (nt * (m.cols / 32) * 32) as u64;
            let (t0_i, nt_i) = (t0 as i32, nt as i32);
            unsafe {
                c.launch(fx.expand_pr, (threads.div_ceil(256) as u32, 1), 256, 0, kargs!(m.w.ptr, m.aux0(), m.aux1(), wd, rows_i, cols_i, t0_i, nt_i))?;
                bst(c, 1, tm)?;
                c.gemm(true, false, nr, b, m.cols, alpha, wd, m.cols, x, m.cols, beta, y + (r0 * 4) as u64, ldy, Gemm::Half)?;
                bst(c, 2, tm)?;
            }
            t0 += nt;
        }
        Ok(())
    };
    for (li, pl) in plan.layers.iter().enumerate() {
        let key = (kv_id, layer_base + li);
        unsafe {
            c.launch(fx.rmsnorm, (b as u32, 1), 256, 0, kargs!(bh, pl.in_norm.ptr, bxn, hid_i, eps, gemma_i, one, nul, nul, nul, zero_i))?;
        }
        bst(c, 0, tm)?;
        match &pl.mix {
            Mix::Gdn(g) => {
                let (cd_i, kk_i, nv_i, nk_i, dk_i, dv_i) =
                    (g.cdim as i32, g.kk as i32, g.nv as i32, g.nk as i32, g.dk as i32, g.dv as i32);
                let vd = g.nv * g.dv;
                let yq = st.scr(c, "hb_qkv", b * g.cdim * 4)?;
                let yz = st.scr(c, "hb_z", b * vd * 4)?;
                let yab = st.scr(c, "hb_ab", b * 2 * g.nv * 4)?;
                let cq = st.scr(c, "hb_cq", b * g.cdim * 4)?;
                let gbp = st.scr(c, "hb_gb", b * 2 * g.nv * 4)?;
                let yo = st.scr(c, "hb_o", b * vd * 4)?;
                let yof = st.scr(c, "hb_of", b * vd * 2)?;
                mul(&g.qkv, bxn, yq, g.cdim, 1.0, 0.0)?;
                mul(&g.z, bxn, yz, vd, 1.0, 0.0)?;
                unsafe {
                    c.gemm(true, false, g.nv, b, hidden, 1.0, g.a_h.ptr, hidden, bxn, hidden, 0.0, yab, 2 * g.nv, Gemm::Half)?;
                    c.gemm(true, false, g.nv, b, hidden, 1.0, g.b_h.ptr, hidden, bxn, hidden, 0.0, yab + (g.nv * 4) as u64, 2 * g.nv, Gemm::Half)?;
                }
                let gs = st.hyb.gdn.get_mut(&key).ok_or("GDN state missing")?;
                unsafe {
                    let tot = (b * g.cdim) as u64;
                    c.launch(fx.conv32, (tot.div_ceil(256) as u32, 1), 256, 0, kargs!(yq, cd_i, gs.ring.ptr, g.taps.ptr, cq, b_i, cd_i, kk_i))?;
                    if g.kk > 1 {
                        c.launch(fx.ring, ((g.cdim as u32).div_ceil(256), 1), 256, 0, kargs!(gs.ring.ptr, yq, cd_i, b_i, cd_i, kk_i))?;
                    }
                    c.launch(fx.prep32, ((b * 2 * g.nk) as u32, 1), g.dk as u32, 0, kargs!(cq, cd_i, nk_i, dk_i))?;
                    let nvt = (b * g.nv) as u64;
                    c.launch(fx.gb32, (nvt.div_ceil(256) as u32, 1), 256, 0, kargs!(yab, g.a_log.ptr, g.dt_bias.ptr, gbp, b_i, nv_i))?;
                    bst(c, 3, tm)?;
                    c.launch(fx.scan, ((vd as u32).div_ceil(8), 1), 256, 0, kargs!(cq, gbp, gs.s.ptr, yo, b_i, cd_i, nv_i, nk_i, dk_i, dv_i, nul, nul, zero_i, kk_i))?;
                    bst(c, 4, tm)?;
                    let (vd_i, sig0) = (vd as i32, 0i32);
                    c.launch(fx.out32, ((b * g.nv) as u32, 1), g.dv as u32, 0, kargs!(yo, yz, vd_i, g.norm.ptr, yof, nv_i, dv_i, eps, sig0, one, nul, nul, nul, zero_i))?;
                }
                bst(c, 3, tm)?;
                gs.next_pos = pos0 + b;
                gs.tape = None;
                mul(&g.out, yof, bh, hidden, 1.0, 1.0)?;
            }
            Mix::Attn(a) if sealed.get(li).copied().unwrap_or(false) => {
                // past the seal: the chunk's rows after the far field's near
                // rows, their attention over those and the landmarks, then
                // the rows into the window (and the window's oldest into the
                // far sums)
                let qr = a.q.rows;
                let yq = st.scr(c, "hb_q", b * qr * 4)?;
                let yk = st.scr(c, "hb_k", b * nkv * hd * 4)?;
                let yv = st.scr(c, "hb_v", b * nkv * hd * 4)?;
                let qg = st.scr(c, "hb_qg", b * nh * hd * 4)?;
                let qgh = st.scr(c, "hb_qgh", b * nh * hd * 2)?;
                let gt = st.scr(c, "hb_gate", b * nh * hd * 4)?;
                let aog = st.scr(c, "hb_aog", b * nh * hd * 4)?;
                let bao = st.scr(c, "hb_ao", b * nh * hd * 2)?;
                let bis = st.scr(c, "hb_is", b * nh * 4)?;
                let bfa = st.scr(c, "hb_fa", b * nh * 4)?;
                let bfc = st.scr(c, "hb_fc", b * nh * 4)?;
                let bfm = st.scr(c, "hb_fm", b * nh * 4)?;
                let bfz = st.scr(c, "hb_fz", b * nh * 4)?;
                let bfn = st.scr(c, "hb_fn", b * nh * hd * 4)?;
                mul(&a.q, bxn, yq, qr, 1.0, 0.0)?;
                mul(&a.k, bxn, yk, nkv * hd, 1.0, 0.0)?;
                mul(&a.v, bxn, yv, nkv * hd, 1.0, 0.0)?;
                let (kp, vp, kt, qt, mu, mz, th, ns, w, m, win_len, win_head, far_len, fscale, rect_fm) = {
                    let d = st.hyb.o1.get(&key).ok_or("far field missing on the prompt path")?;
                    if (d.nkv, d.hpk, d.hd) != (nkv, hpk, hd) {
                        return Err("far field of another geometry".into());
                    }
                    if d.next_pos.is_some_and(|p| p != pos0) {
                        return Err(format!("far field at row {:?}, the chunk at {pos0}", d.next_pos));
                    }
                    (d.nk.ptr, d.nv.ptr, d.kt.ptr, d.qt.ptr, d.mu.ptr, d.mz.ptr, d.th.ptr, d.ns, d.w, d.m, d.win_len, d.win_head, d.far_len, d.scale, d.rect_fm)
                };
                if b > SEAL_ROWS.max(NMAX) || b > w {
                    return Err(format!("a chunk of {b} rows past a seal (window {w})"));
                }
                // the near rows before the chunk: the sinks with the kept rows,
                // then the window (unwrapped until it is full)
                let n0 = ns + win_len;
                let (nh_i, nkv_i, half_i, gated_i) = (nh as i32, nkv as i32, plan.half as i32, a.gated as i32);
                let (qs_i, kvs_i, n0_i) = (qr as i32, (nkv * hd) as i32, n0 as i32);
                let (qn, kn) = (a.qn.as_ref().map_or(0, |b| b.ptr), a.kn.as_ref().map_or(0, |b| b.ptr));
                unsafe {
                    c.launch(
                        fx.qk,
                        ((nh + nkv) as u32, b as u32),
                        hd as u32,
                        0,
                        kargs!(yq, qs_i, yk, yv, kvs_i, qn, kn, qg, gt, kp, vp, btab, nh_i, nkv_i, half_i, eps, gemma_i, gated_i, n0_i, b_i, nul, qgh),
                    )?;
                }
                // the archive of a prompt past its first seal: the chunk's rows
                // and one query row in the trace's stride off the card
                super::acap_note(c, layer_base + li, kp, vp, n0, b, nkv * hd, qg, nh, nkv, hd, pos0)?;
                let far_on = far_len > 0;
                if far_on {
                    let (m_i, sc, rf, bb_i) = (m as i32, fscale, rect_fm as i32, b as i32);
                    unsafe {
                        c.launch(fx.far_rows, (nh as u32, b as u32), 256, 0, kargs!(qg, kt, mu, mz, th, bfm, bfz, bfn, m_i, nh_i, sc, rf, bb_i))?;
                    }
                }
                let lda = nkv * hd;
                let grow = hpk * hd;
                // a key r of the near rows sits at position col_base + r for the mask:
                // all the rows before the chunk are visible, the chunk's causally
                let col_base = pos0 - n0;
                let mut b0 = 0usize;
                while b0 < b {
                    let bn = (b - b0).min(att_rows());
                    let p0 = pos0 + b0;
                    let l_rows = n0 + b0 + bn;
                    let cols = bn * hpk;
                    let bs = st.scr(c, "hb_s", nkv * l_rows * cols * 4)?;
                    let bp = st.scr(c, "hb_p", nkv * l_rows * cols * 2)?;
                    let (l_i, p0_i, cb_i, hpk_i) = (l_rows as i32, p0 as i32, col_base as i32, hpk as i32);
                    let (cols_i, istride) = (cols as i32, (b * hpk) as i32);
                    let off = (b0 * hpk * 4) as u64;
                    let (fmp, fzp) = if far_on { (bfm + (b0 * nh * 4) as u64, bfz + (b0 * nh * 4) as u64) } else { (0u64, 0u64) };
                    unsafe {
                        gemm_batched_h(c, true, l_rows, cols, hd, plan.scale, kp, lda, hd, qgh + (b0 * grow * 2) as u64, hd, b * grow, bs, l_rows, l_rows * cols, nkv)?;
                        c.launch(fx.softmax_hf, ((nkv * cols) as u32, 1), 256, 0, kargs!(bs, bp, bis + off, bfa + off, bfc + off, l_i, p0_i, cb_i, hpk_i, cols_i, istride, fmp, fzp, nh_i))?;
                        gemm_batched_h(c, false, hd, cols, l_rows, 1.0, vp, lda, hd, bp, l_rows, l_rows * cols, aog + (b0 * grow * 4) as u64, hd, b * grow, nkv)?;
                    }
                    b0 += bn;
                }
                let n_ao = (b * nh * hd) as u64;
                let gate_p = if a.gated { gt } else { 0u64 };
                let fnp = if far_on { bfn } else { 0u64 };
                unsafe {
                    c.launch(fx.attn_out_f, (n_ao.div_ceil(256) as u32, 1), 256, 0, kargs!(aog, bis, bfa, bfc, fnp, gate_p, bao, nh_i, b_i, one))?;
                }
                // the commit: the first rows fill the window where they are; past
                // a full window each row takes the oldest's slot, which goes into
                // the far sums first
                let fill = b.min(w - win_len);
                let rest = b - fill;
                let head0 = if win_len == w { win_head } else { 0 };
                if rest > 0 {
                    let (m_i, ns_i, w_i, h0_i, e_i) = (m as i32, ns as i32, w as i32, head0 as i32, rest as i32);
                    let warps = nkv * hpk * m;
                    let row_b = (nkv * hd * 2) as u64;
                    unsafe {
                        c.launch(fx.far_insert_n, ((warps * 32).div_ceil(256) as u32, 1), 256, 0, kargs!(kp, vp, qt, mz, th, m_i, nkv_i, fscale, ns_i, w_i, h0_i, e_i))?;
                        // the rows past the window's room into the freed slots: two runs at most
                        let src = (n0 + fill) as u64;
                        let r1 = rest.min(w - head0);
                        for (dst, from, n) in [((ns + head0) as u64, src, r1), (ns as u64, src + r1 as u64, rest - r1)] {
                            if n == 0 {
                                continue;
                            }
                            cu::memcpy_dtod_async(kp + dst * row_b, kp + from * row_b, n * row_b as usize, c.stream).map_err(derr)?;
                            cu::memcpy_dtod_async(vp + dst * row_b, vp + from * row_b, n * row_b as usize, c.stream).map_err(derr)?;
                        }
                    }
                }
                {
                    let d = st.hyb.o1.get_mut(&key).ok_or("far field missing on the prompt path")?;
                    d.win_len = (win_len + b).min(w);
                    if rest > 0 {
                        d.win_head = (head0 + rest) % w;
                        d.far_len += rest;
                    }
                    d.next_pos = Some(pos0 + b);
                    d.spec = None;
                }
                bst(c, 5, tm)?;
                mul(&a.o, bao, bh, hidden, 1.0, 1.0)?;
            }
            Mix::Attn(a) => {
                let qr = a.q.rows;
                let yq = st.scr(c, "hb_q", b * qr * 4)?;
                let yk = st.scr(c, "hb_k", b * nkv * hd * 4)?;
                let yv = st.scr(c, "hb_v", b * nkv * hd * 4)?;
                let qg = st.scr(c, "hb_qg", b * nh * hd * 4)?;
                let qgh = st.scr(c, "hb_qgh", b * nh * hd * 2)?;
                let gt = st.scr(c, "hb_gate", b * nh * hd * 4)?;
                let aog = st.scr(c, "hb_aog", b * nh * hd * 4)?;
                let bao = st.scr(c, "hb_ao", b * nh * hd * 2)?;
                let bis = st.scr(c, "hb_is", b * nh * 4)?;
                mul(&a.q, bxn, yq, qr, 1.0, 0.0)?;
                mul(&a.k, bxn, yk, nkv * hd, 1.0, 0.0)?;
                mul(&a.v, bxn, yv, nkv * hd, 1.0, 0.0)?;
                let hint = st.kv_hint;
                let (slot0, kp, vp, m_base) = {
                    let m = st.kv.get_mut(&key).ok_or("K/V mirror missing")?;
                    let out = ((m.stored - m.base) as i32, m.kptr(), m.vptr(), m.base);
                    m.stored += b;
                    out
                };
                let (nh_i, nkv_i, half_i, gated_i) = (nh as i32, nkv as i32, plan.half as i32, a.gated as i32);
                let (qs_i, kvs_i) = (qr as i32, (nkv * hd) as i32);
                let (qn, kn) = (a.qn.as_ref().map_or(0, |b| b.ptr), a.kn.as_ref().map_or(0, |b| b.ptr));
                unsafe {
                    c.launch(
                        fx.qk,
                        ((nh + nkv) as u32, b as u32),
                        hd as u32,
                        0,
                        kargs!(yq, qs_i, yk, yv, kvs_i, qn, kn, qg, gt, kp, vp, btab, nh_i, nkv_i, half_i, eps, gemma_i, gated_i, slot0, b_i, nul, qgh),
                    )?;
                }
                // a far-field layer of a prompt read for its seal: the rotated
                // queries its trace keeps wait for `o1_qcap_drain`
                let cap_stride = super::QCAP.with(|q| {
                    q.borrow().as_ref().filter(|q| q.layers.contains(&(layer_base + li))).map(|q| q.stride.max(1))
                });
                if let Some(stride) = cap_stride {
                    let first = (stride - pos0 % stride) % stride;
                    let kept = if first < b { (b - first).div_ceil(stride) } else { 0 };
                    if kept > 0 {
                        let floats = kept * nh * hd;
                        let buf = Buf::new(floats * 4)?;
                        let (first_i, stride_i, hd_i) = (first as i32, stride as i32, hd as i32);
                        unsafe {
                            c.launch(c.f.q_rows_strided, (kept as u32, 1), 256, 0, kargs!(qg, buf.ptr, nh_i, nkv_i, hd_i, b_i, first_i, stride_i))?;
                        }
                        super::QCAP.with(|q| {
                            if let Some(q) = q.borrow_mut().as_mut() {
                                q.bufs.push((layer_base + li, floats, buf));
                            }
                        });
                    }
                }
                let lda = nkv * hd;
                let grow = hpk * hd;
                let mut b0 = 0usize;
                while b0 < b {
                    let p0 = pos0 + b0;
                    let col_base = m_base;
                    let fit = |n: usize| (p0 + n - col_base) * n * nh * 4 <= 384 << 20;
                    let mut bn = (b - b0).min(att_rows());
                    while bn > 64 && !fit(bn) {
                        bn = bn.div_ceil(2);
                    }
                    let l_rows = p0 + bn - col_base;
                    let cols = bn * hpk;
                    let whole = (hint.max(pos0 + b) * b.min(att_rows()) * nh * 4).min(400 << 20);
                    let bs = st.scr(c, "hb_s", (nkv * l_rows * cols * 4).max(whole))?;
                    let bp = st.scr(c, "hb_p", (nkv * l_rows * cols * 2).max(whole / 2))?;
                    let (l_i, p0_i, cb_i, hpk_i) = (l_rows as i32, p0 as i32, col_base as i32, hpk as i32);
                    let (cols_i, istride) = (cols as i32, (b * hpk) as i32);
                    let isum = bis + (b0 * hpk * 4) as u64;
                    // the mirrors hold f16 rows: both GEMMs on f16 operands
                    // (the query copy and the exponentials), f32 out
                    unsafe {
                        gemm_batched_h(c, true, l_rows, cols, hd, plan.scale, kp, lda, hd, qgh + (b0 * grow * 2) as u64, hd, b * grow, bs, l_rows, l_rows * cols, nkv)?;
                        c.launch(fx.softmax_h, ((nkv * cols) as u32, 1), 256, 0, kargs!(bs, bp, isum, l_i, p0_i, cb_i, hpk_i, cols_i, istride))?;
                        gemm_batched_h(c, false, hd, cols, l_rows, 1.0, vp, lda, hd, bp, l_rows, l_rows * cols, aog + (b0 * grow * 4) as u64, hd, b * grow, nkv)?;
                    }
                    b0 += bn;
                }
                let n_ao = (b * nh * hd) as u64;
                let gate_p = if a.gated { gt } else { 0u64 };
                unsafe {
                    c.launch(fx.attn_out, (n_ao.div_ceil(256) as u32, 1), 256, 0, kargs!(aog, bis, gate_p, bao, nh_i, b_i, one))?;
                }
                bst(c, 5, tm)?;
                mul(&a.o, bao, bh, hidden, 1.0, 1.0)?;
            }
        }
        let inter = pl.inter;
        let bgu = st.scr(c, "hb_gu", b * 2 * inter * 4)?;
        let bmh = st.scr(c, "hb_mh", b * inter * 2)?;
        unsafe {
            c.launch(fx.rmsnorm, (b as u32, 1), 256, 0, kargs!(bh, pl.post_norm.ptr, bxn, hid_i, eps, gemma_i, one, nul, nul, nul, zero_i))?;
        }
        bst(c, 0, tm)?;
        // the gate/up pair as one GEMM, its rows in the pair's order
        mul(&pl.gu, bxn, bgu, 2 * inter, 1.0, 0.0)?;
        let tot = (b * inter) as u64;
        let (in_i, bb_i) = (inter as i32, b as i32);
        unsafe {
            c.launch(fx.silu_pair, (tot.div_ceil(256) as u32, 1), 256, 0, kargs!(bgu, bmh, in_i, bb_i))?;
        }
        bst(c, 6, tm)?;
        mul(&pl.down, bmh, bh, hidden, 16.0, 1.0)?;
        if tapping {
            if let Some(s) = want.iter().position(|&l| l == li) {
                // pageable: done when it returns
                unsafe { cu::memcpy_dtoh_async(&mut tapped[s * b * hidden..(s + 1) * b * hidden], bh, c.stream).map_err(derr)? };
            }
        }
    }
    if tapping {
        super::taps_put(tapped);
    }
    if !defer {
        unsafe { cu::memcpy_dtoh_async(h, bh, c.stream).map_err(derr)? };
    }
    let r = c.sync();
    mem_report(c, st, &format!("a prompt chunk of {b} rows at {pos0}"));
    r
}

/// After a verify window accepted rows `..= slot`: every GDN layer of the
/// sequence back to the window's start, the accepted rows replayed.
pub(super) fn spec_restore(kv_id: u64, slot: usize, base_pos: usize, expected: usize) -> bool {
    let Some(c) = super::ctx() else { return false };
    let mut st = c.st.lock().unwrap();
    if c.current().is_err() {
        return false;
    }
    let st = &mut *st;
    let mut keys: Vec<(u64, usize)> = st.hyb.gdn.keys().filter(|k| k.0 == kv_id).copied().collect();
    keys.sort_unstable();
    let run = |st: &mut State| -> Res<usize> {
        let Some(fx) = st.hyb.funcs.values().next().cloned() else { return Ok(0) };
        // every layer with this window's tape, of one geometry
        let mut dims = None;
        let mut live = Vec::with_capacity(keys.len());
        for k in &keys {
            let gs = &st.hyb.gdn[k];
            let Some((p0, rows)) = gs.tape else { continue };
            if p0 != base_pos || slot >= rows {
                continue;
            }
            if dims.is_some_and(|d| d != (gs.dims, gs.t_stride)) {
                return Err("GDN layers of different geometry in one restore".into());
            }
            dims = Some((gs.dims, gs.t_stride));
            live.push(*k);
        }
        let Some(((nv, nk, dk, dv, kk), tstride)) = dims else { return Ok(0) };
        let cdim = 2 * nk * dk + nv * dv;
        let keep = slot + 1;
        // the table of the layers' pointers, made again when they change
        let fresh = st.hyb.rtabs.get(&kv_id).is_none_or(|(_, ks)| *ks != live);
        if fresh {
            let mut t = Vec::with_capacity(live.len() * 7);
            for k in &live {
                let gs = &st.hyb.gdn[k];
                t.extend([gs.s_save.ptr, gs.s.ptr, gs.t_cq.ptr, gs.t_gb.ptr, gs.ring_save.ptr, gs.ring.ptr, gs.t_qkv.ptr]);
            }
            st.hyb.rtabs.insert(kv_id, (Buf::from_slice(&t)?, live.clone()));
        }
        let tab = st.hyb.rtabs[&kv_id].0.ptr;
        let nl = live.len() as u32;
        let (keep_i, cd_i, nv_i, nk_i, dk_i, dv_i, kk_i, qs_i) =
            (keep as i32, cdim as i32, nv as i32, nk as i32, dk as i32, dv as i32, kk as i32, tstride as i32);
        unsafe {
            if dk == 128 && dv % 32 == 0 {
                c.launch_3d(fx.rs_scan4, ((nv * dv / 32) as u32, nl, 1), 128, 0, kargs!(tab, keep_i, cd_i, nv_i, nk_i, dv_i))?;
            } else {
                c.launch_3d(fx.rs_scan, (((nv * dv) as u32).div_ceil(8), nl, 1), 256, 0, kargs!(tab, keep_i, cd_i, nv_i, nk_i, dk_i, dv_i))?;
            }
            if kk > 1 {
                c.launch_3d(fx.rs_ring, ((cdim as u32).div_ceil(256), nl, 1), 256, 0, kargs!(tab, keep_i, qs_i, cd_i, kk_i))?;
            }
        }
        for k in &live {
            let gs = st.hyb.gdn.get_mut(k).unwrap();
            gs.next_pos = base_pos + keep;
            gs.tape = None;
        }
        Ok(live.len())
    };
    match run(st) {
        Ok(done) => done == expected,
        Err(e) => {
            tracing::error!("cuda GDN restore failed: {e}");
            false
        }
    }
}

/// The arm's kernels are compiled (a trunk plan was built).
/// A verify window over the far fields of `kv_id` kept `rows` rows: they move
/// into the window in order, each pushing the ring's oldest row into the far
/// sums once the ring is full (`nystrom.rs` advance, as a decode step does).
/// Drafts over a far field (`CMF_O1_SPEC=0` turns them off).
pub(crate) fn far_spec_on() -> bool {
    std::env::var("CMF_O1_SPEC").as_deref() != Ok("0")
}

/// A prompt read past a seal in chunks on the card (`CMF_O1_CHUNKS=0`: none).
pub(crate) fn far_chunks_on() -> bool {
    std::env::var("CMF_O1_CHUNKS").as_deref() != Ok("0")
}

pub(super) fn spec_commit(kv_id: u64, pos0: usize, rows: usize) -> bool {
    let Some(c) = super::ctx() else { return false };
    let mut st = c.st.lock().unwrap();
    if c.current().is_err() {
        return false;
    }
    let r = (|| -> Res<()> {
        let mut recs: Vec<i32> = Vec::new();
        // (layer key, record offset, the row to move, the ring slot it takes)
        let mut steps: Vec<((u64, usize), usize, usize, usize)> = Vec::new();
        let mut keys: Vec<(u64, usize)> = st.hyb.o1.keys().filter(|k| k.0 == kv_id).copied().collect();
        keys.sort_unstable();
        for key in keys {
            let d = st.hyb.o1.get_mut(&key).unwrap();
            let Some((base, n)) = d.spec.take() else { continue };
            for j in 0..rows.min(n) {
                let row = base + j;
                if d.win_len < d.w {
                    // the row is where the window's next slot is
                    debug_assert_eq!(row, d.ns + d.win_len);
                    d.win_len += 1;
                    continue;
                }
                let slot = d.ns + d.win_head;
                let keys_n = d.ns + d.w;
                let rb = attn_rb(keys_n);
                let ncap = keys_n.next_multiple_of(4096);
                let at = recs.len();
                recs.extend([(keys_n - 1) as i32, slot as i32, rb as i32, ncap as i32, (ncap.div_ceil(rb) + 1) as i32, 1, slot as i32, 1]);
                steps.push((key, at, row, slot));
                d.far_len += 1;
                d.win_head = (d.win_head + 1) % d.w;
            }
            d.next_pos = Some(pos0 + rows);
        }
        if steps.is_empty() {
            return Ok(());
        }
        let rb = st.scr(c, "hy_spec_ci", recs.len() * 4)?;
        unsafe { cu::memcpy_htod_async(rb, &recs, c.stream).map_err(derr)? };
        for (key, at, row, slot) in steps {
            let d = &st.hyb.o1[&key];
            let fx = st.hyb.funcs.get(&(d.hpk, d.hd)).ok_or("far field commit before the arm's kernels")?;
            let row_b = (d.nkv * d.hd * 2) as u64;
            let cip = rb + (at * 4) as u64;
            let (fm_i, nkv_i) = (d.m as i32, d.nkv as i32);
            unsafe {
                // the oldest row into the far sums, then the kept row in its slot
                c.launch(fx.far_insert, ((d.nkv * d.hpk * d.m).div_ceil(8) as u32, 1), 256, 0, kargs!(d.nk.ptr, d.nv.ptr, d.qt.ptr, d.mz.ptr, d.th.ptr, cip, fm_i, nkv_i, d.scale))?;
                cu::memcpy_dtod_async(d.nk.ptr + slot as u64 * row_b, d.nk.ptr + row as u64 * row_b, row_b as usize, c.stream).map_err(derr)?;
                cu::memcpy_dtod_async(d.nv.ptr + slot as u64 * row_b, d.nv.ptr + row as u64 * row_b, row_b as usize, c.stream).map_err(derr)?;
            }
        }
        Ok(())
    })();
    match r {
        Ok(()) => true,
        Err(e) => {
            tracing::error!("cuda hybrid: far field commit: {e}");
            false
        }
    }
}

pub(super) fn kernels_ready(st: &State) -> bool {
    !st.hyb.funcs.is_empty()
}

/// A recurrent state of this sequence lives on the card.
pub(super) fn state_resident(st: &State, kv_id: u64, layer: usize) -> bool {
    st.hyb.gdn.contains_key(&(kv_id, layer))
}

/// A GDN layer's state back to the host, `[ring | S]` as the host keeps it.
pub(super) fn gdn_state_read(c: &Ctx, st: &State, kv_id: u64, layer: usize) -> Option<Vec<f32>> {
    let gs = st.hyb.gdn.get(&(kv_id, layer))?;
    let (nv, nk, dk, dv, kk) = gs.dims;
    let cdim = 2 * nk * dk + nv * dv;
    let ring_n = (kk - 1) * cdim;
    let mut out = vec![0f32; ring_n + nv * dk * dv];
    unsafe {
        if ring_n > 0 {
            cu::memcpy_dtoh_async(&mut out[..ring_n], gs.ring.ptr, c.stream).ok()?;
        }
        cu::memcpy_dtoh_async(&mut out[ring_n..], gs.s.ptr, c.stream).ok()?;
    }
    c.sync().ok()?;
    Some(out)
}

/// Drop a sequence's recurrent states.
pub(super) fn reset(c: &Ctx, st: &mut State, kv_id: u64) {
    st.hyb.rtabs.remove(&kv_id);
    st.hyb.o1.retain(|&(id, _), d| {
        if id == kv_id {
            c.kv_bytes.fetch_sub(d.bytes, Ordering::Relaxed);
        }
        id != kv_id
    });
    st.hyb.gdn.retain(|&(id, _), g| {
        if id == kv_id {
            let (nv, nk, dk, dv, kk) = g.dims;
            let cdim = 2 * nk * dk + nv * dv;
            let ring_n = (kk - 1) * cdim;
            let s_n = nv * dk * dv;
            c.kv_bytes.fetch_sub(((s_n * 2 + ring_n * 2 + NMAX * (2 * cdim + 2 * nv)) * 4) as u64, Ordering::Relaxed);
            if g.snap.is_some() {
                c.kv_bytes.fetch_sub(((s_n + ring_n.max(1)) * 4) as u64, Ordering::Relaxed);
            }
        }
        id != kv_id
    });
}

/// Copy every GDN state of the sequence, all at prompt position `pos`, to
/// its snapshot (a device-to-device copy on the stream, after the prompt
/// rows before it). False when a layer is elsewhere or none is resident.
pub(super) fn gdn_snapshot(kv_id: u64, pos: usize) -> bool {
    let Some(c) = super::ctx() else { return false };
    let mut st = c.st.lock().unwrap();
    if c.current().is_err() {
        return false;
    }
    let run = |st: &mut State| -> Res<bool> {
        let keys: Vec<(u64, usize)> = st.hyb.gdn.keys().filter(|k| k.0 == kv_id).copied().collect();
        if keys.is_empty() || keys.iter().any(|k| st.hyb.gdn[k].next_pos != pos) {
            return Ok(false);
        }
        for k in keys {
            let gs = st.hyb.gdn.get_mut(&k).unwrap();
            let (nv, nk, dk, dv, kk) = gs.dims;
            let ring_n = (kk - 1) * (2 * nk * dk + nv * dv);
            let s_n = nv * dk * dv;
            if gs.snap.is_none() {
                let (bs, br) = (Buf::new(s_n * 4)?, Buf::new(ring_n.max(1) * 4)?);
                c.kv_bytes.fetch_add(((s_n + ring_n.max(1)) * 4) as u64, Ordering::Relaxed);
                gs.snap = Some((0, bs, br));
            }
            let (at, bs, br) = gs.snap.as_mut().unwrap();
            unsafe {
                cu::memcpy_dtod_async(bs.ptr, gs.s.ptr, s_n * 4, c.stream).map_err(derr)?;
                if ring_n > 0 {
                    cu::memcpy_dtod_async(br.ptr, gs.ring.ptr, ring_n * 4, c.stream).map_err(derr)?;
                }
            }
            *at = pos;
        }
        Ok(true)
    };
    run(&mut st).unwrap_or_else(|e| {
        tracing::warn!("cuda GDN snapshot failed: {e}");
        false
    })
}

/// Every GDN state of the sequence back to its snapshot at `pos`; the
/// attention mirrors rewind themselves when the next call starts there.
/// False (nothing changed) unless every layer holds a snapshot at `pos`.
pub(super) fn gdn_snapshot_restore(kv_id: u64, pos: usize) -> bool {
    let Some(c) = super::ctx() else { return false };
    let mut st = c.st.lock().unwrap();
    if c.current().is_err() {
        return false;
    }
    let run = |st: &mut State| -> Res<bool> {
        let keys: Vec<(u64, usize)> = st.hyb.gdn.keys().filter(|k| k.0 == kv_id).copied().collect();
        if keys.is_empty() || keys.iter().any(|k| st.hyb.gdn[k].snap.as_ref().is_none_or(|s| s.0 != pos)) {
            return Ok(false);
        }
        for k in keys {
            let gs = st.hyb.gdn.get_mut(&k).unwrap();
            let (nv, nk, dk, dv, kk) = gs.dims;
            let ring_n = (kk - 1) * (2 * nk * dk + nv * dv);
            let s_n = nv * dk * dv;
            let (_, bs, br) = gs.snap.as_ref().unwrap();
            unsafe {
                cu::memcpy_dtod_async(gs.s.ptr, bs.ptr, s_n * 4, c.stream).map_err(derr)?;
                if ring_n > 0 {
                    cu::memcpy_dtod_async(gs.ring.ptr, br.ptr, ring_n * 4, c.stream).map_err(derr)?;
                }
            }
            gs.next_pos = pos;
            gs.tape = None;
        }
        Ok(true)
    };
    run(&mut st).unwrap_or_else(|e| {
        tracing::warn!("cuda GDN snapshot restore failed: {e}");
        false
    })
}

/// y [n][rows] = W x [n][cols] on the card for a q4tp graph weight (the
/// tensor-core kernel, NMAX rows a launch): an MTP draft block's input
/// projection, done where the rest of the draft runs.
pub(super) fn rows_matvec(model: &Arc<CmfModel>, gw: &GraphW, rows: usize, cols: usize, x: &[f32], n: usize, y: &mut [f32]) -> Res<()> {
    let c = super::ctx().ok_or("no CUDA context")?;
    if gw.kind != 6 || x.len() < n * cols || y.len() < n * rows || cols % 32 != 0 {
        return Err("rows_matvec: a q4tp weight and rows of its width".into());
    }
    let mut st = c.st.lock().unwrap();
    c.current()?;
    let st = &mut *st;
    let m = hmat(c, st, model, gw, rows, cols)?;
    let fx = match st.hyb.funcs.values().next() {
        Some(f) => f.clone(),
        None => return Err("rows_matvec before the first hybrid plan".into()),
    };
    let xin = st.scr(c, "hy_rx", NMAX * cols * 4)?;
    let yout = st.scr(c, "hy_ry", NMAX * rows * 4)?;
    let p0 = st.scr(c, "hy_rp0", NMAX * cols * 2)?;
    let p1 = st.scr(c, "hy_rp1", NMAX * cols * 2)?;
    let p2 = st.scr(c, "hy_rp2", NMAX * cols * 2)?;
    let key = (model.uid(), gw.idx);
    if st.hyb.rowsets.get(&key).is_none_or(|(_, yp)| *yp != yout) {
        let ms = mmaset(&[(&m, yout, 0, rows, rows)])?;
        st.hyb.rowsets.insert(key, (ms, yout));
    }
    let (ms, _) = &st.hyb.rowsets[&key];
    let planes = Planes { p0, p1, p2, stride: cols };
    let mut done = 0;
    while done < n {
        let k = (n - done).min(NMAX);
        unsafe {
            cu::memcpy_htod_async(xin, &x[done * cols..(done + k) * cols], c.stream).map_err(derr)?;
            split(c, &fx, &planes, xin, cols, cols, k)?;
            ms.run(c, &fx, &planes, cols, k)?;
            cu::memcpy_dtoh_async(&mut y[done * rows..(done + k) * rows], yout, c.stream).map_err(derr)?;
        }
        c.sync()?;
        done += k;
    }
    Ok(())
}


#[cfg(test)]
mod tests {
    use super::*;

    fn lcg(seed: &mut u64) -> u64 {
        *seed = seed.wrapping_mul(6364136223846793005).wrapping_add(1442695040888963407);
        *seed >> 33
    }

    /// A q4tp payload of random nibbles, codes and row params.
    fn payload(rows: usize, cols: usize, seed: &mut u64) -> Vec<u8> {
        let (_, codes_off, stride) = cortiq_core::quant::q4tp_sections(rows, cols);
        let mut b: Vec<u8> = (0..codes_off + rows * stride).map(|_| lcg(seed) as u8).collect();
        let (params_off, _, _) = cortiq_core::quant::q4tp_sections(rows, cols);
        for r in 0..rows {
            // finite params: f16 lo in [-12, -4], step in [0, 0.25]
            let lo = cortiq_core::quant::f32_to_f16(-12.0 + 8.0 * (lcg(seed) % 1000) as f32 / 1000.0);
            let st = cortiq_core::quant::f32_to_f16(0.25 * (lcg(seed) % 1000) as f32 / 1000.0);
            b[params_off + r * 4..params_off + r * 4 + 2].copy_from_slice(&lo.to_le_bytes());
            b[params_off + r * 4 + 2..params_off + r * 4 + 4].copy_from_slice(&st.to_le_bytes());
        }
        b
    }

    fn down<T: Copy + Default>(c: &Ctx, b: &Buf, n: usize) -> Vec<T> {
        let mut v = vec![T::default(); n];
        unsafe { cu::memcpy_dtoh_sync(&mut v, b.ptr) }.unwrap();
        let _ = c;
        v
    }

    /// A seal's far sums on the card (`far_sums`, several tiles of rows) are
    /// the host's replay of the same rows: the keys and values exact in f16,
    /// so only the dot products' order differs.
    #[test]
    fn card_far_sums_match_the_host_replay() {
        let Some(c) = super::super::ctx() else { return };
        let (nkv, hpk, hd, m, w, sink) = (2usize, 2usize, 64usize, 8usize, 32usize, 4usize);
        let t = 9000usize;
        let nh = nkv * hpk;
        let mut seed = 5u64;
        let mut f = || cortiq_core::quant::f16_to_f32(cortiq_core::quant::f32_to_f16((lcg(&mut seed) % 2000) as f32 / 1000.0 - 1.0));
        let ks: Vec<Vec<f32>> = (0..nkv).map(|_| (0..t * hd).map(|_| f()).collect()).collect();
        let vs: Vec<Vec<f32>> = (0..nkv).map(|_| (0..t * hd).map(|_| f()).collect()).collect();
        let qtr: Vec<f32> = (0..t * nh * hd).map(|_| f()).collect();
        let order: Vec<u32> = (0..t as u32).collect();
        let mut host = Vec::new();
        let mut qt = Vec::new();
        let mut range = 0..0;
        for g in 0..nkv {
            let mut st = crate::nystrom::NystromState::new_group(m, w, sink, hpk);
            range = st.seal_gathered(&ks[g], &vs[g], &qtr, 1, nh, g * hpk, &order, t, hd, hd).unwrap();
            for h in 0..hpk {
                qt.extend_from_slice(st.far_q_tilde(h));
            }
            st.replay_far_idx(&ks[g], &vs[g], &order[range.clone()]);
            host.push(st);
        }
        let (kr, vr): (Vec<&[f32]>, Vec<&[f32]>) = (ks.iter().map(|k| &k[..]).collect(), vs.iter().map(|v| &v[..]).collect());
        let scale = 1.0 / (hd as f32).sqrt();
        let (mz, th) = far_sums(c, nkv, hpk, hd, m, scale, &qt, &order[range], &kr, &vr, hd).unwrap();
        let mut worst = 0f32;
        for (g, st) in host.iter().enumerate() {
            let view = st.device_view();
            for (j, hv) in view.heads.iter().enumerate() {
                let h = g * hpk + j;
                for i in 0..m {
                    let (mm, z) = (mz[h * 2 * m + i], mz[h * 2 * m + m + i]);
                    assert!((mm - hv.m_max[i]).abs() < 1e-4, "largest logit {mm} against {}", hv.m_max[i]);
                    // the sums at the two scales, brought to one
                    let r = (hv.m_max[i] - mm).exp();
                    worst = worst.max((z - hv.z_hat[i] * r).abs() / z);
                    for d in 0..hd {
                        let (a, b) = (th[(h * m + i) * hd + d], hv.t_hat[i * hd + d] * r);
                        worst = worst.max((a - b).abs() / z);
                    }
                }
            }
        }
        assert!(worst < 1e-4, "far sums off the host's by {worst}");
    }

    /// The card's repack (one matrix and a gate/up pair) is the host's,
    /// word for word: the layout the tensor-core kernels read.
    #[test]
    fn card_repack_matches_the_host_repack() {
        let Some(c) = super::super::ctx() else { return };
        c.current().unwrap();
        let fx = load_funcs(c, 4, 64).unwrap();
        let mut seed = 7u64;
        for (rows, cols) in [(48usize, 96usize), (40, 64)] {
            let a = payload(rows, cols, &mut seed);
            let (hn, hc, hl) = repack_q4tp(&a, rows, cols).unwrap();
            let (dn, dc, dl) = repack_on_card(c, &fx, &[&a], rows, rows, cols).unwrap();
            assert_eq!(down::<u32>(c, &dn, hn.len()), hn, "nibbles {rows}x{cols}");
            assert_eq!(down::<u8>(c, &dc, hc.len()), hc, "codes {rows}x{cols}");
            assert_eq!(down::<f32>(c, &dl, hl.len()), hl, "params {rows}x{cols}");
        }
        let (inter, cols) = (24usize, 64usize);
        let g = payload(inter, cols, &mut seed);
        let u = payload(inter, cols, &mut seed);
        let map = |r: usize| (((r & 15) >= 8) as usize, (r >> 4) * 8 + (r & 7));
        let (hn, hc, hl) = repack_rows(&[(&g, inter), (&u, inter)], 2 * inter, cols, &map).unwrap();
        let (dn, dc, dl) = repack_on_card(c, &fx, &[&g, &u], inter, 2 * inter, cols).unwrap();
        assert_eq!(down::<u32>(c, &dn, hn.len()), hn, "pair nibbles");
        assert_eq!(down::<u8>(c, &dc, hc.len()), hc, "pair codes");
        assert_eq!(down::<f32>(c, &dl, hl.len()), hl, "pair params");
    }
}