memra_engine/decode_batch.rs
1//! Batched decode step — B sequences share one fused pass (ARCHITECTURE-H100.md §3 B2').
2//!
3//! The bandwidth thesis: decode is weight-stream-bound, so every projection at m=B rows
4//! amortizes one weight read across B sequences. Row-parallel ops (norm/rope/quantize/
5//! activation) batch trivially — they are the SAME kernels prefill already runs at T rows.
6//! Only truly per-sequence state stays in a loop: KV append + fa_decode over each cache,
7//! and the GDN/conv recurrent step (v1: per-seq loop via the existing single-seq path;
8//! a blockIdx.z-batched GDN state kernel is the v2 fusion).
9//!
10//! EXACTNESS CONTRACT (the law this module lives under):
11//! - B == 1 must be BIT-IDENTICAL to `decode_step_h` (gate: decode-batch-gate).
12//! - 2 <= B <= 8: each row rides the m=2..9 verify-tier mmvq kernels, which are per-row
13//! bit-identical to m=1 (the spec-exactness machinery decode_step_t relies on). Each
14//! sequence's token stream must equal its isolated single-seq run (worker.rs contract:
15//! "byte-identical to isolated").
16//! - 9 <= B <= 16 (the EXACT-16 tier, inc3 2026-08-01): admitted iff
17//! `decode_batch_exact16_ok` — every matmul rides the b16 batched-mmvq class
18//! (bit-identical per (token,row) to m=1; Q8_0 needs the q8rp mirror) under a
19//! verify_exact scope that disables the m>=16 GEMM/MMQ arms. gate2 bit-strength
20//! PASS at B=12/16 (research/batched-tick-inc3-20260801). Refused otherwise.
21//! - B > 16 crosses into GEMM/dp4a-tail numeric configs with NO exact kernel class —
22//! refused (MEMRA_DECODE_BATCH_CAP stays a measurement door).
23//!
24//! v1 scope: the hybrid (Qwen3.5-class) non-gemma4 trunk. Fused m=1 micro-launches
25//! (fused3 QKV, cross-layer add+norm+q8 chain) are NOT used — the unfused sequence is
26//! bit-identical (kernel_check: add_rms_norm == add;rms_norm; _q8_1 == +quantize_q8_1)
27//! and keeps the batched path simple. Batched fusions are tuning work, not correctness.
28
29use crate::Engine;
30use crate::cache::Cache;
31use crate::hybrid::{HybridModel, Mixer};
32use cudarc::driver::{CudaEvent, CudaSlice};
33use memra_gguf::config::Arch;
34
35type DualPpCudaSpan = Option<(CudaEvent, CudaEvent)>;
36
37fn dual_pp_timing_event(e: &Engine, context: &str) -> Option<CudaEvent> {
38 if !crate::pp::dual_pp_timing_on() {
39 return None;
40 }
41 match e
42 .stream()
43 .record_event(Some(cudarc::driver::sys::CUevent_flags::CU_EVENT_DEFAULT))
44 {
45 Ok(event) => Some(event),
46 Err(err) => {
47 crate::pp::record_dual_pp_timing_drop(context, &err);
48 None
49 }
50 }
51}
52
53/// Per-step, per-LAYER-RANGE invariants the batched trunk needs: the device state-pointer
54/// table for the range's layers, the arm picks, and the per-row `t_kv` snapshot. Built once
55/// per step per range by `HybridModel::batch_layer_ctx`, consumed by `decode_batch_layers`.
56///
57/// WHY IT IS RANGE-SCOPED AND NOT STEP-SCOPED (this is the whole point of the struct):
58/// `ptr_table` is a `CudaSlice<u64>` of DEVICE ADDRESSES, uploaded through `e` — so it lives
59/// on `e`'s device, and its entries are pointers into caches that live on the device that
60/// OWNS those layers. Under a pp stage split, stage s runs layers [fence[s], fence[s+1])
61/// whose cache state was allocated by stage s's engine (`pp::new_cache` -> `Cache::new_ppn`),
62/// so stage s must build its OWN table through its OWN engine. One step-wide table built on
63/// the primary would put every stage's kernel arguments in stage-0's HBM — a peer read per
64/// pointer fetch, which is the exact cliff `pp::refuse_unsplit_if_remote` exists to stop.
65/// `lo`/`hi` are recorded so the consumer can assert the ctx it was handed matches the range
66/// it was asked to run (the offsets in `lin_base`/`attn_base` are only valid for that range).
67pub(crate) struct BatchLayerCtx {
68 /// Offset into `ptr_table` of layer il's [conv x B][ssm_in x B][ssm_out x B] block
69 /// (linear-attn layers only). Indexed by ABSOLUTE layer id; `None` off-range.
70 lin_base: Vec<Option<usize>>,
71 /// Offset into `ptr_table` of layer il's [k0,v0,k1,v1,..] block (full-attn layers only).
72 /// Indexed by ABSOLUTE layer id; `None` off-range.
73 attn_base: Vec<Option<usize>>,
74 ptr_table: Option<CudaSlice<u64>>,
75 /// Per-row `pos + 1` — the t_kv each sequence attends at this step. Layer-invariant
76 /// within a step, so the arm picks below are decided once.
77 t_kvs: Vec<usize>,
78 t_kv_max: usize,
79 /// The single `fa_split_keys` rung every row shares (the rows-twins straddle law).
80 sp0: usize,
81 seqs_append: bool,
82 seqs_fa: bool,
83 lo: usize,
84 hi: usize,
85}
86
87// ---- MEMRA_BATCH_PHASE=1 (diagnostics): sync-bounded per-phase accumulators for the batched
88// tick. Each boundary syncs the stream, so the TOTAL inflates (launch pipelining is destroyed);
89// the value is the RANKING/shares, not absolute ms. Read via `batch_phase_report()`.
90pub(crate) static BATCH_PHASE: std::sync::Mutex<[f64; 12]> = std::sync::Mutex::new([0.0; 12]);
91/// Device-sample request for one batched row: (temp, seed, ctr, top_k, top_p, min_p).
92/// `top_k=0` / `top_p>=1.0` / `min_p<=0.0` = that filter off. Greedy = temp<=0 (device
93/// argmax); pure temperature = seeded gumbel; any filter on = filter_stats floor + the
94/// filtered gumbel draw. Penalty configs never reach device sampling (worker eligibility).
95pub type DevSamp = (f32, u64, u32, i32, f32, f32);
96
97pub const BATCH_PHASE_NAMES: [&str; 12] = [
98 "setup(ptrs+embed H2D)",
99 "attn batched pre (norm/qkv/rope)",
100 "attn per-seq: kv append",
101 "attn per-seq: q/a dtod copies",
102 "attn per-seq: fa_decode",
103 "attn post (gate+o-proj)",
104 "gdn batched projections",
105 "gdn state ops (conv/prep/scan)",
106 "gdn out (gated norm+proj)",
107 "ffn (add/norm/gate/up/act/down)",
108 "lm_head (norm+matmul)",
109 "logits D2H + host split",
110];
111pub fn batch_phase_on() -> bool {
112 static ON: std::sync::OnceLock<bool> = std::sync::OnceLock::new();
113 *ON.get_or_init(|| std::env::var("MEMRA_BATCH_PHASE").as_deref() == Ok("1"))
114}
115/// Accumulate the elapsed time since `last` into phase slot `slot` and re-stamp `last`.
116/// No-op unless `MEMRA_BATCH_PHASE=1`. Syncs the ambient stream first, so under a pp stage
117/// scope this bounds the STAGE's stream, which is what the caller is timing.
118///
119/// A free fn rather than the closure it replaced: `decode_batch_layers` (the pp stage seam)
120/// runs the instrumented layer loop, so the marker has to be callable from both the seam
121/// and its caller's epilogue. `batch_phase_on()` is a `OnceLock` memo, so per-call cost is
122/// the same atomic load the hoisted `ph_on` local was.
123fn ph_mark(
124 e: &Engine,
125 slot: usize,
126 last: &mut std::time::Instant,
127) -> Result<(), Box<dyn std::error::Error>> {
128 if batch_phase_on() {
129 e.stream().synchronize()?;
130 let now = std::time::Instant::now();
131 BATCH_PHASE.lock().unwrap()[slot] += (now - *last).as_secs_f64();
132 *last = now;
133 }
134 Ok(())
135}
136
137pub fn batch_phase_report() -> String {
138 let ph = BATCH_PHASE.lock().unwrap();
139 let tot: f64 = ph.iter().sum();
140 let mut rows: Vec<(usize, f64)> = ph.iter().copied().enumerate().collect();
141 rows.sort_by(|a, b| b.1.total_cmp(&a.1));
142 let mut s = format!(
143 "[batch-phase] total {:.1} ms (sync-bounded; shares rank, not walltime)\n",
144 tot * 1e3
145 );
146 for (i, v) in rows {
147 s += &format!(
148 " {:>6.1} ms {:>5.1}% {}\n",
149 v * 1e3,
150 v / tot * 100.0,
151 BATCH_PHASE_NAMES[i]
152 );
153 }
154 s
155}
156
157impl HybridModel {
158 /// Batched-decode width cap. 8 = the exactness-tier default (see the assert below);
159 /// MEMRA_DECODE_BATCH_CAP overrides for tier-probe measurement, clamped to 32.
160 pub fn decode_batch_cap() -> usize {
161 use std::sync::OnceLock;
162 static CAP: OnceLock<usize> = OnceLock::new();
163 *CAP.get_or_init(|| {
164 std::env::var("MEMRA_DECODE_BATCH_CAP")
165 .ok()
166 .and_then(|v| v.parse().ok())
167 .map(|c: usize| c.clamp(1, 32))
168 .unwrap_or(8)
169 })
170 }
171
172 /// EXACT-16 TIER admission (increment 3a, 2026-08-01, 5090 receipts
173 /// research/batched-tick-inc3-20260801): true iff EVERY matmul the batched decode step
174 /// runs has a per-(token,row) bit-exact kernel class at m=9..16 under the verify_exact
175 /// scope — i.e. the batched-mmvq b16 family (32-thread warp reduce, the exact m=1 mmvq
176 /// program per column) or the e4m3 grid.y=m mmvq catch-all. Q8_0 qualifies only with
177 /// the split-plane mirror (rp4, MEMRA_Q8RP): its b16 kernel exists only as the _rp twin.
178 /// Float matmuls (cuBLASLt, n-dependent reductions) and MoE FFNs disqualify the model.
179 /// Measured attribution for WHY the naked m=16 tier is not exact: the m>=16 arms
180 /// (MMQ int8-MMA `mul_mat_q` — MEMRA_PP_Q8MMQ default-on — and `qmatvec_gemm`, both
181 /// block-scale f32) and the m=9..15 dp4a tail (128-thread two-level reduce) all break
182 /// per-row bit-identity vs isolated decode (gate2 step-0 bit-diffs, maxdiff ~1.3-2.3e-1).
183 pub fn decode_batch_exact16_ok(&self) -> bool {
184 fn ok(w: &crate::model::GpuTensor) -> bool {
185 match w {
186 crate::model::GpuTensor::Quant { qtype, .. } => {
187 *qtype == crate::QT_Q4_0 || *qtype == crate::QT_Q6_K
188 || *qtype == crate::QT_F8_E4M3
189 // BLOCK-128 FP8-ST (lane/rp-on-st, 2026-08-06): admitted now that the class
190 // has a b16 batched kernel (`qmatvec_e4m3_blk_mmvq_b16`), bit-identical per
191 // (token,row) to its m=1 launch. Before that kernel existed this class fell to
192 // the grid.y=m form at every width — still EXACT, so the tier's correctness
193 // bar was met, but it re-read the weight m times, which is why admitting it
194 // without the kernel would have been a throughput trap rather than a win.
195 || *qtype == crate::QT_F8_E4M3_BLK
196 // NVFP4 (lane/rp-on-st, 2026-08-06) — THE blocker this lane measured. The
197 // mixed FP8-ST 27B is 193 NVFP4 dense-MLP tensors, and this predicate is an
198 // ALL over every matmul, so NVFP4's missing b16 refused the whole checkpoint
199 // (`B=16 > cap 8 with no exact tier ... refused`) even with both e4m3 classes
200 // admitted. It now has base + _rp b16 twins off its existing batched template
201 // (bit-identical per (token,row) to the m=1 mmvq: same nibble decode, dp4a
202 // order, ue4m3 scale, warp reduce). This also opens the tier for pure-NVFP4
203 // GGUF models, which is a behavior change on the primary format — hence the
204 // full decode-batch config+strict battery on both.
205 || *qtype == crate::QT_NVFP4
206 // Q4_K (lane/rp-on-st): named by MEMRA_EXACT16_WHY as the 9B NVFP4 GGUF's
207 // refusing class (`L0.wqkv qtype=1`) — mixed NVFP4 checkpoints keep Q4_K
208 // attention. Now has base + _rp b16.
209 || *qtype == crate::QT_Q4_K
210 // Q5_K (lane/rp-on-st): the FOURTH class the diagnostic named on the same 9B
211 // GGUF (`L0.wqkv_gate qtype=3`). A shipped mixed checkpoint spreads ~500
212 // matmuls over four/five classes, and this predicate is an ALL — so chunk 16
213 // was unreachable for every real artifact until every class had a b16.
214 || *qtype == crate::QT_Q5_K
215 // Q8_0 NO LONGER requires the mirror (rp4): it has a base b16 too, so the
216 // tier is reachable at zero VRAM. Named by the diagnostic as the FP8-ST
217 // refusal — `L0.ssm_beta qtype=0 rp4=false`, a 23.9 MiB residual class that
218 // was gating chunk 16 for a 16.4 GiB checkpoint.
219 || *qtype == crate::QT_Q8_0
220 }
221 _ => false,
222 }
223 }
224 // WHY-NOT DIAGNOSTIC (lane/rp-on-st, 2026-08-06): this predicate is a bare bool over
225 // ~500 tensors, so a refusal produced only `B=16 > cap 8 with no exact tier ... refused`
226 // with no way to tell WHICH class refused. That cost this lane two wrong hypotheses (the
227 // rp mirror, then e4m3-only) before the NVFP4 gap was found. MEMRA_EXACT16_WHY=1 names
228 // the first refusing tensor + its qtype. Diagnostic-only per flags doctrine; default off,
229 // zero cost when unread.
230 let why = std::env::var("MEMRA_EXACT16_WHY").is_ok();
231 macro_rules! chk {
232 ($t:expr, $label:expr) => {{
233 let r = ok($t);
234 if !r && why {
235 // qtype = -1 means the tensor is NOT Quant at all (a float/BF16/F16
236 // container), which the tier can never admit — a distinct diagnosis from
237 // "quantized, but in a class with no b16 kernel".
238 let (qt, rp4) = match $t {
239 crate::model::GpuTensor::Quant { qtype, rp4, .. } => {
240 (*qtype, rp4.is_some())
241 }
242 _ => (-1, false),
243 };
244 eprintln!("[exact16] REFUSED by {} qtype={qt} rp4={rp4}", $label);
245 }
246 r
247 }};
248 }
249 if self.cfg.m3.is_some() || self.is_gemma4_e4b() || self.cfg.gemma4.is_some() {
250 if why {
251 eprintln!("[exact16] REFUSED by architecture (m3/gemma4)");
252 }
253 return false;
254 }
255 self.layers.iter().enumerate().all(|(li, l)| {
256 let mix_ok = match &l.mixer {
257 Mixer::Full(fa) => {
258 chk!(&fa.wq, format!("L{li}.wq"))
259 && chk!(&fa.wk, format!("L{li}.wk"))
260 && chk!(&fa.wv, format!("L{li}.wv"))
261 && chk!(&fa.wo, format!("L{li}.wo"))
262 }
263 Mixer::Linear(la) => {
264 chk!(&la.wqkv, format!("L{li}.wqkv"))
265 && chk!(&la.wqkv_gate, format!("L{li}.wqkv_gate"))
266 && chk!(&la.ssm_beta, format!("L{li}.ssm_beta"))
267 && chk!(&la.ssm_alpha, format!("L{li}.ssm_alpha"))
268 && chk!(&la.ssm_out, format!("L{li}.ssm_out"))
269 }
270 // MLA rides its own increment-4 arm; never admitted to the exact-16 tier here.
271 Mixer::Mla(_) => {
272 if why {
273 eprintln!("[exact16] REFUSED by L{li} MLA mixer");
274 }
275 false
276 }
277 };
278 let ffn_ok = match &l.ffn {
279 crate::hybrid::Ffn::Dense {
280 ffn_gate,
281 ffn_up,
282 ffn_down,
283 } => {
284 chk!(ffn_gate, format!("L{li}.ffn_gate"))
285 && chk!(ffn_up, format!("L{li}.ffn_up"))
286 && chk!(ffn_down, format!("L{li}.ffn_down"))
287 }
288 crate::hybrid::Ffn::Moe(_) => {
289 if why {
290 eprintln!("[exact16] REFUSED by L{li} MoE ffn");
291 }
292 false
293 }
294 };
295 mix_ok && ffn_ok
296 }) && chk!(&self.output, "output".to_string())
297 }
298
299 /// Opt-in/A-B seam for the eager B=1 fusion program. `MEMRA_SERVE_B1FAST=1` sends an
300 /// eligible solo tick through that program; unset/other values keep B=1 on the generic
301 /// batched body, the same numeric class used at B>=2.
302 ///
303 /// EXACTNESS, stated precisely (measured on-box 2026-08-05, sm_120 q9 NVFP4-MTP):
304 /// the fast path is BIT-IDENTICAL TO `decode_step_h` — decode-batch-gate's STRICT
305 /// gate1 (`--mode strict`) PASSes with it ON and FAILs with it OFF at maxdiff
306 /// 1.591e-1. It is deliberately NOT bit-identical to the batched body: the two
307 /// carry a decode-config FP-composition gap (same class gate1's config mode measures).
308 /// That gap became correctness-visible under live load: Step35, Q35-MoE, and finally
309 /// dense Q27 all produced load-history-dependent token streams, including early EOS,
310 /// when a request crossed between the two programs. The generic body is therefore the
311 /// correctness default; the eager program remains available only for fixed-solo A/Bs.
312 /// Historical token-stream/performance receipts:
313 /// research/servepath-p2-20260805 (greedy 150 ids + seeded-sampled identical to the
314 /// run-gen oracle AND cross-arm, so the gap is sub-token here as designed).
315 ///
316 /// Read fresh (an `AtomicU8` memo, not a `OnceLock`): decode-batch-gate flips this
317 /// seam BETWEEN gates in-process — gate1 needs the fast path ON to prove bit-identity,
318 /// gate2 needs it pinned OFF to keep testing the batched body. A latch-once read would
319 /// bake whichever gate ran first, so the gate could never test both sides. The memo
320 /// caches the parse but `set_b1_fast` invalidates it.
321 pub fn b1_fast_on() -> bool {
322 // 0 = unknown/invalidated, 1 = off, 2 = on
323 match Self::b1_fast_memo().load(std::sync::atomic::Ordering::Relaxed) {
324 1 => false,
325 2 => true,
326 _ => {
327 let value = std::env::var("MEMRA_SERVE_B1FAST").ok();
328 let on = b1_fast_env_on(value.as_deref());
329 Self::b1_fast_memo()
330 .store(if on { 2 } else { 1 }, std::sync::atomic::Ordering::Relaxed);
331 on
332 }
333 }
334 }
335
336 fn b1_fast_memo() -> &'static std::sync::atomic::AtomicU8 {
337 static MEMO: std::sync::atomic::AtomicU8 = std::sync::atomic::AtomicU8::new(0);
338 &MEMO
339 }
340
341 /// Test/gate seam: force the B=1 fast path on or off for the rest of the process,
342 /// overriding the env. Used by decode-batch-gate to exercise the opt-in eager arm and
343 /// pin gate2's default reference arm.
344 pub fn set_b1_fast(on: bool) {
345 Self::b1_fast_memo().store(if on { 2 } else { 1 }, std::sync::atomic::Ordering::Relaxed);
346 }
347
348 /// Whether this architecture may switch a live serving row onto the eager B=1 fusion
349 /// class. Qwen35-MoE must stay on the batched trunk at every width: its eager and batched
350 /// hybrid/MoE walks are each deterministic, but crossing B=1 -> B>=2 changes greedy token
351 /// ids and can introduce an early EOS (Q35 sellgate, 2026-08-12).
352 pub fn b1_fast_arch_eligible(&self) -> bool {
353 b1_fast_arch_eligible(&self.cfg.arch)
354 }
355
356 /// H3 body: the m=1 FUSED trunk (`decode_layers_eager` — shared verbatim with
357 /// `decode_step_h`/the ppN stages) plus the batched path's own serving epilogue
358 /// (grammar mask, device sample, lean-logits park). See the call-site comment in
359 /// `decode_step_batch_sampled_lean_masked` for why this is bit-identical.
360 fn decode_step_b1_fast(
361 &self,
362 e: &Engine,
363 token: u32,
364 caches: &mut [&mut Cache],
365 samp: &[Option<DevSamp>],
366 masks: &[Option<(&CudaSlice<u32>, usize)>],
367 lean: bool,
368 ) -> Result<(Vec<Vec<f32>>, Vec<Option<u32>>), Box<dyn std::error::Error>> {
369 let n_embd = self.cfg.n_embd as usize;
370 let eps = self.cfg.rms_eps;
371 let pos = caches[0].pos;
372 let pos_d = e.htod_i32(&[pos as i32])?;
373 let x = e.htod(&self.embd.gather(n_embd, &[token]))?;
374 // the SHARED m=1 trunk: same function decode_step_h runs, so every m=1 fusion
375 // (cross-layer add+norm+q8_1, fused SwiGLU, lever 1's gate+up dual) fires here.
376 let x = self.decode_layers_eager(e, x, 0, self.layers.len(), &pos_d, pos, caches[0])?;
377 let mut hn = e.uninit(n_embd)?;
378 e.rms_norm(&x, self.output_norm.float_data(), &mut hn, n_embd, 1, eps)?;
379 let logits = e.matmul(&self.output, &hn, 1)?;
380
381 // ---- epilogue: byte-for-byte the batched path's, at b_n=1 ----
382 let n_vocab = self.output.out_features();
383 let mut logits = logits;
384 let mut pristine: Option<CudaSlice<f32>> = None;
385 if let Some((mask, words)) = masks.first().copied().flatten() {
386 assert!(
387 samp.first().copied().flatten().is_some(),
388 "grammar-masked row 0 must request a device sample"
389 );
390 if lean {
391 let cache = &mut caches[0];
392 if cache
393 .last_logits_dev
394 .as_ref()
395 .map(|d| d.len() < n_vocab)
396 .unwrap_or(true)
397 {
398 cache.last_logits_dev = Some(e.uninit(n_vocab)?);
399 }
400 let dst = cache.last_logits_dev.as_mut().unwrap();
401 e.dtod_copy_view(&logits.slice(0..n_vocab), dst)?;
402 } else {
403 let mut p = e.uninit(n_vocab)?;
404 e.dtod_copy_view(&logits.slice(0..n_vocab), &mut p)?;
405 pristine = Some(p);
406 }
407 e.mask_logits_col(&mut logits, mask, 0, n_vocab, words)?;
408 }
409
410 let mut next: Vec<Option<u32>> = vec![None; 1];
411 if let Some((temp, seed, ctr, top_k, top_p, min_p)) = samp.first().copied().flatten() {
412 let mut toks = e.alloc_u32_zeroed(1)?;
413 // Filtered-greedy degenerates to plain argmax (the max always survives every
414 // truncation filter), so temp<=0 short-circuits regardless of filters.
415 let filtered = temp > 0.0 && (top_k > 0 || top_p < 1.0 || min_p > 0.0);
416 if temp <= 0.0 {
417 e.argmax_token_device_col(&logits, 0, n_vocab, &mut toks, 0)?;
418 } else if filtered {
419 let mut pb = e.zeros(n_vocab)?;
420 self.devsample_filtered_col(
421 e, &logits, 0, n_vocab, temp, seed, ctr, top_k, top_p, min_p, &mut pb,
422 &mut toks, 0,
423 )?;
424 } else {
425 let mut pb = e.zeros(n_vocab)?;
426 e.gumbel_perturb_col(&logits, 0, &mut pb, n_vocab, seed, ctr, temp)?;
427 e.argmax_token_device_col(&pb, 0, n_vocab, &mut toks, 0)?;
428 }
429 next[0] = Some(e.dtoh_u32(&toks)?[0]);
430 }
431
432 let sampled = samp.first().copied().flatten().is_some();
433 let rows: Vec<Vec<f32>> = if lean && sampled {
434 if masks.first().copied().flatten().is_none() {
435 let cache = &mut caches[0];
436 if cache
437 .last_logits_dev
438 .as_ref()
439 .map(|d| d.len() < n_vocab)
440 .unwrap_or(true)
441 {
442 cache.last_logits_dev = Some(e.uninit(n_vocab)?);
443 }
444 let dst = cache.last_logits_dev.as_mut().unwrap();
445 e.dtod_copy_view(&logits.slice(0..n_vocab), dst)?;
446 }
447 vec![Vec::new()]
448 } else if let Some(p) = pristine.as_ref() {
449 vec![e.dtoh(p)?]
450 } else {
451 vec![e.dtoh(&logits)?]
452 };
453 // decode_layers_eager does NOT advance cache.pos (decode_step_h advances it after
454 // the head); the batched path advances every cache at the tail — same here.
455 caches[0].pos += 1;
456 Ok((rows, next))
457 }
458
459 /// One filtered device draw for stacked-logits row `col`: `filter_stats` solves the
460 /// single unnormalized-prob floor that encodes top-k AND top-p AND min-p (block-internal
461 /// binary search, bit-stable), then the filtered gumbel perturb + argmax draws one token
462 /// from the truncated softmax into `toks[slot]`. All device-side — no stat D2H, no row
463 /// copy; the only host traffic stays the caller's one [B]-u32 token readback.
464 #[allow(clippy::too_many_arguments)]
465 fn devsample_filtered_col(
466 &self,
467 e: &Engine,
468 logits: &CudaSlice<f32>,
469 col: usize,
470 n_vocab: usize,
471 temp: f32,
472 seed: u64,
473 ctr: u32,
474 top_k: i32,
475 top_p: f32,
476 min_p: f32,
477 pb: &mut CudaSlice<f32>,
478 toks: &mut CudaSlice<u32>,
479 slot: usize,
480 ) -> Result<(), Box<dyn std::error::Error>> {
481 let rows = e.htod_i32(&[col as i32])?;
482 let mut th = e.zeros(1)?;
483 let mut z = e.zeros(1)?;
484 let mut mx = e.zeros(1)?;
485 e.filter_stats(
486 logits, n_vocab, &rows, &mut th, &mut z, &mut mx, n_vocab, 1, temp, top_k, top_p, min_p,
487 )?;
488 e.gumbel_perturb_filtered_col(logits, col, pb, n_vocab, seed, ctr, temp, &mx, &th, 0)?;
489 e.argmax_token_device_col(pb, 0, n_vocab, toks, slot)?;
490 Ok(())
491 }
492
493 /// One batched greedy-decode step over B independent sequences.
494 /// `tokens[b]` is sequence b's input token; `caches[b]` its private cache (position,
495 /// quantized KV, GDN/conv state). Returns the B logits rows (host, [n_vocab] each).
496 /// Each cache's pos/len advance exactly as `decode_step_h` would.
497 pub fn decode_step_batch(
498 &self,
499 e: &Engine,
500 tokens: &[u32],
501 caches: &mut [&mut Cache],
502 ) -> Result<Vec<Vec<f32>>, Box<dyn std::error::Error>> {
503 let (rows, _) = self.decode_step_batch_sampled(e, tokens, caches, &[])?;
504 Ok(rows)
505 }
506
507 /// `decode_step_batch` + DEVICE-SIDE SAMPLING for eligible rows (the batched-tick lever,
508 /// 2026-08-01): the host sampler's temp-path is O(n_vocab) with a full-vocab exp per row
509 /// (measured 1.36 ms/row at the 9B's 248320 vocab = 10.9 ms/tick at B=8 — the single
510 /// largest component of the serving tick). Here each requested row samples ON DEVICE
511 /// between the lm_head matmul and the logits D2H:
512 /// temp <= 0 (greedy): the 2-pass device argmax — bit-identical to host argmax
513 /// (argmax-gate contract, same kernels as the dc serving path).
514 /// temp > 0: gumbel_perturb(seed, ctr, temp) + the same argmax = ONE categorical draw
515 /// from softmax(logits/temp) — the sampled-spec Philox machinery. Deterministic per
516 /// (seed, ctr) and INDEPENDENT of batch composition (the isolation contract;
517 /// decode-batch-gate gate3). NOTE: the draw stream differs from the host sampler's
518 /// SplitMix64 (distribution-equal, seed-deterministic, NOT byte-equal to the old
519 /// host draws) — greedy rows are unchanged bit-exact.
520 /// `samp[bi] = Some((temp, seed, ctr))` requests a device sample for row bi; the full
521 /// logits rows are still returned (worker keeps last_logits semantics + fallback rows).
522 pub fn decode_step_batch_sampled(
523 &self,
524 e: &Engine,
525 tokens: &[u32],
526 caches: &mut [&mut Cache],
527 samp: &[Option<DevSamp>],
528 ) -> Result<(Vec<Vec<f32>>, Vec<Option<u32>>), Box<dyn std::error::Error>> {
529 self.decode_step_batch_sampled_lean(e, tokens, caches, samp, false)
530 }
531
532 /// `decode_step_batch_sampled` + LEAN LOGITS (increment 2 component 3, 2026-08-01):
533 /// with `lean`, device-sampled rows SKIP the [n_vocab] logits D2H (9.4%/32.5% of the
534 /// pre-/post-inc2 tick profile) — their returned row is EMPTY. The audit-mapped
535 /// consumers: (a) the next tick's host sample — never fires, `device_next` carries the
536 /// token; (b) the graph-promotion argmax — reads only prefill logits (generated empty);
537 /// (c) the KV-reuse pool park at retire — the REAL consumer, served by a per-cache
538 /// device park: the row is dtod-copied into `cache.last_logits_dev` (device bandwidth)
539 /// and D2H'd ONCE at retire by the worker. Rows without a device sample keep a per-row
540 /// D2H. `lean=false` is bit-for-bit the previous method (gates + non-serving callers).
541 pub fn decode_step_batch_sampled_lean(
542 &self,
543 e: &Engine,
544 tokens: &[u32],
545 caches: &mut [&mut Cache],
546 samp: &[Option<DevSamp>],
547 lean: bool,
548 ) -> Result<(Vec<Vec<f32>>, Vec<Option<u32>>), Box<dyn std::error::Error>> {
549 self.decode_step_batch_sampled_lean_masked(e, tokens, caches, samp, &[], lean)
550 }
551
552 /// `decode_step_batch_sampled_lean` + GRAMMAR MASKS (constrained decoding, 2026-08-03):
553 /// `masks[bi] = Some((packed_bitset, words))` bans every unset-bit vocab id on row bi
554 /// (mask_logits_f32, -FLT_MAX) BETWEEN the lm_head matmul and the device sampler, so a
555 /// constrained row rides the SAME device-sample/lean-logits tick as everyone else — no
556 /// full-row D2H, no host O(n_vocab) sample. Contract: a masked row must also request a
557 /// device sample. The row's PRISTINE logits are preserved for their consumers before the
558 /// in-place ban: lean rows park the unmasked row into `cache.last_logits_dev` (the
559 /// retire-time reuse-pool park stays unmasked — continuations resume grammar-free, the
560 /// v1 host-path contract), non-lean rows D2H the unmasked row. `masks = &[]` is
561 /// bit-for-bit the unmasked method.
562 pub fn decode_step_batch_sampled_lean_masked(
563 &self,
564 e: &Engine,
565 tokens: &[u32],
566 caches: &mut [&mut Cache],
567 samp: &[Option<DevSamp>],
568 masks: &[Option<(&CudaSlice<u32>, usize)>],
569 lean: bool,
570 ) -> Result<(Vec<Vec<f32>>, Vec<Option<u32>>), Box<dyn std::error::Error>> {
571 self.decode_step_batch_sampled_lean_masked_schedule(
572 e, tokens, caches, samp, masks, lean, None,
573 )
574 }
575
576 /// Worker-scheduled twin of [`Self::decode_step_batch_sampled_lean_masked`]. The worker
577 /// supplies the balanced dual-wave boundary it used when forming this tick. Direct engine
578 /// callers keep the automatic midpoint above; the explicit seam makes scheduler chunking and
579 /// engine execution one checked contract instead of two coincident width calculations.
580 pub fn decode_step_batch_sampled_lean_masked_scheduled(
581 &self,
582 e: &Engine,
583 tokens: &[u32],
584 caches: &mut [&mut Cache],
585 samp: &[Option<DevSamp>],
586 masks: &[Option<(&CudaSlice<u32>, usize)>],
587 lean: bool,
588 dual_wave_mid: usize,
589 ) -> Result<(Vec<Vec<f32>>, Vec<Option<u32>>), Box<dyn std::error::Error>> {
590 self.decode_step_batch_sampled_lean_masked_schedule(
591 e,
592 tokens,
593 caches,
594 samp,
595 masks,
596 lean,
597 Some(dual_wave_mid),
598 )
599 }
600
601 #[allow(clippy::too_many_arguments)]
602 fn decode_step_batch_sampled_lean_masked_schedule(
603 &self,
604 e: &Engine,
605 tokens: &[u32],
606 caches: &mut [&mut Cache],
607 samp: &[Option<DevSamp>],
608 masks: &[Option<(&CudaSlice<u32>, usize)>],
609 lean: bool,
610 scheduled_dual_mid: Option<usize>,
611 ) -> Result<(Vec<Vec<f32>>, Vec<Option<u32>>), Box<dyn std::error::Error>> {
612 // NOTE (inc3 3c, 2026-08-01, KILLED ARM): a deferred-token-readback variant (all
613 // chunks of a tick writing device-sampled tokens into one shared buffer, ONE
614 // dtoh_u32 after the last chunk instead of one per chunk) measured FLAT at serve
615 // level on the 5090 (N=4 medians within +-0.7% at c=8/16/32 — 3 saved syncs
616 // against a ~100 ms weight-bound tick is ~0.1%, below resolution). Killed per the
617 // flags doctrine; receipts research/batched-tick-inc3-20260801 (serve-points.jsonl
618 // base vs defer arms) are the record. The per-chunk [B]-u32 readback below IS the
619 // tick's only steady-state D2H — one per chunk, none per seq.
620 let b_n = tokens.len();
621 assert!(
622 b_n >= 1 && b_n == caches.len(),
623 "tokens/caches length mismatch"
624 );
625 // ---- PP DOOR: THE BATCHED STAGE SPLIT (pp2-batch 2026-08-06) ----------------------
626 // Until this increment this body had NO pp arm: it walked lo=0..n_layers on the
627 // primary engine's stream, with no stage split, no boundary, and no `rt.enter()`. With
628 // the door open and a sharded cross-device placement, every projection for the remote
629 // stages' layers was read over PCIe, per step, silently — measured 7.4 vs 208.9 tok/s
630 // at B=1 (28x), 47.4 vs 657.0 at B=8 (13.9x) on a PRO 6000 pair over Gen5 x16 P2P.
631 // Nothing failed or warned, because peer reads return identical bytes and all three
632 // `decode-batch-gate` gates PASS on that config — the failure mode was performance,
633 // and a green exactness battery hid it. `pp2-hardening` made that regime FAIL CLOSED
634 // (research/pp2-hardening-20260806); this lane makes it legitimately split, so the
635 // refusal lifts for the batched path.
636 //
637 // `decode_step_batch_ppn` runs each stage's layer range through that stage's engine
638 // and stream with a [B, n_embd] boundary transfer between them, i.e. every stage
639 // touches only LOCAL weights and LOCAL cache state. The refusal below still guards
640 // the residue: the door open with `MEMRA_PP_STREAMS=0` (the same-stream rollback,
641 // which also disables the sharded loader, so nothing is remote — `pp_shard_off` and
642 // `pp2_streams_off` both make `pp_sharded_cross_device()` false) or a placement whose
643 // PpNRt fails to build. Keeping the call means a future path that reaches here in a
644 // remote regime still refuses instead of regressing 28x.
645 if let Some(fence) = crate::pp::pp_cuts(self.layers.len()) {
646 if !crate::pp::pp2_streams_off() && crate::pp::batch_pp_on() {
647 // Auto (flipped default) routes dual only in the re-gated regime and
648 // degrades serially elsewhere; Forced keeps every ineligible placement on
649 // the refusing dual body so the binding negative cells stay reachable.
650 let route_dual = crate::pp::dual_pp_route(
651 crate::pp::dual_pp_mode(),
652 b_n,
653 fence.len() - 1,
654 crate::pp::pp2_overlap(),
655 crate::pp::pp_host_bounce_active(),
656 );
657 if route_dual {
658 let mid = scheduled_dual_mid
659 .or_else(|| crate::pp::dual_pp_wave_mid(b_n))
660 .expect("dual PP B>=2 must have a wave midpoint");
661 return self
662 .decode_step_batch_dual(e, tokens, caches, samp, masks, lean, &fence, mid);
663 }
664 return self.decode_step_batch_ppn(e, tokens, caches, samp, masks, lean, &fence);
665 }
666 }
667 if scheduled_dual_mid.is_some() {
668 return Err(
669 "decode_step_batch: worker supplied a dual-wave schedule but the PP-2 dual path is unavailable"
670 .into(),
671 );
672 }
673 crate::pp::refuse_unsplit_if_remote(
674 "decode_step_batch",
675 "drop MEMRA_PP_STREAMS=0 / MEMRA_BATCH_PP=0 so the batched path takes its OWN \
676 stage split (decode_step_batch_ppn), or serve single-stream over the eager pp \
677 arm (decode_step_h), which is also split",
678 )?;
679 // ---- H3: B=1 FAST-PATH (serve-path phase 2, 2026-08-05) ----------------------------
680 // At b_n==1 every projection below calls `matmul_pre(.., b_n)` with m=1, which is
681 // ALREADY the m=1 mmvq dispatch — so the m=1 *kernel family* was never the gap. What
682 // this body does NOT have is the m=1 *fusion chain* that `decode_step_h` carries:
683 // - the cross-layer add+norm+quantize fusion (`add_rms_norm_q8_1`: 3 launches -> 1),
684 // - the fused SwiGLU epilogue (`silu_mul_scaled_q8_1`: folds ffn_down's quantize
685 // into its producer) and, with it, `matmul_pre_dual_noscale`'s gate+up pair
686 // fusion — i.e. phase-1 LEVER 1.
687 // Routing b_n==1 through `decode_layers_eager` (the SHARED trunk `decode_step_h` and
688 // the ppN stages already use, lifted verbatim — not a copy) makes every present and
689 // future m=1 lever fire on the opt-in path automatically. The epilogue (grammar mask ->
690 // device sample -> lean logits park) stays exactly as the batched path runs it; the trunk's
691 // different FP composition is why this path cannot be a load-changing default.
692 // BIT-IDENTITY: the trunk is the same function `decode_step_h` calls, and every
693 // fusion it enables is kernel-check-pinned bit-identical to its unfused sequence
694 // (add_rms_norm == add;rms_norm | _q8_1 == +quantize_q8_1 | dual_noscale == two
695 // matmul_pre_noscale). Gate: decode-batch-gate B=1 vs decode_step_h + serve stream
696 // identity. MEMRA_SERVE_B1FAST=1 is the fixed-solo opt-in/A-B seam; the default
697 // stays on this function's generic body so batch-width changes cannot change the
698 // FP program mid-request.
699 if b_n == 1
700 && Self::b1_fast_on()
701 && self.b1_fast_arch_eligible()
702 && !self.is_gemma4_e4b()
703 && self.cfg.gemma4.is_none()
704 && self.cfg.m3.is_none()
705 && crate::pp::pp_cuts(self.layers.len()).is_none()
706 && !e.verify_exact_on()
707 {
708 return self.decode_step_b1_fast(e, tokens[0], caches, samp, masks, lean);
709 }
710 // MEMRA_DECODE_BATCH_CAP (experimental door, serving-lane tier probe 2026-08-01):
711 // default 8 keeps the v1 exactness policy — B=2..8 rides the verify-tier batched
712 // mmvq arms, per-row bit-identical to isolated m=1 decode. Values >8 are a
713 // MEASUREMENT DOOR ONLY: m=9..15 falls to the grid.y=m dp4a tail (m weight
714 // re-reads + a different reduce shape) and m>=16 crosses into the GEMM tier
715 // (block-scale f32 rounding) — BOTH break the "byte-identical to isolated"
716 // serving contract. Never default this above 8 without the batched-tier
717 // exactness policy landing.
718 let cap = Self::decode_batch_cap();
719 // EXACT-16 TIER (increment 3a): chunks of 9..=16 are admitted WITHOUT the env door
720 // when every matmul has a bit-exact b16-class kernel (see decode_batch_exact16_ok).
721 // The verify_exact scope below pins that dispatch for the whole step: it turns off
722 // the m>=16 GEMM arms (qmatvec_gemm + MMQ + fp8/f16/fp4 — all block-scale/foreign
723 // numeric configs) so every projection rides the batched-mmvq b16 tier, which is
724 // per-(token,row) bit-identical to isolated m=1 decode (gate2 bit-strength PASS at
725 // B=12/16, s32+s160, 5090 receipts research/batched-tick-inc3-20260801). Without
726 // the exact tier, B>cap stays refused; the env door (MEMRA_DECODE_BATCH_CAP) keeps
727 // its old meaning as the non-exact measurement probe.
728 let exact16 = b_n > 8 && b_n <= 16 && self.decode_batch_exact16_ok();
729 assert!(
730 b_n <= cap || exact16,
731 "decode_step_batch: B={b_n} > cap {cap} with no exact tier — refused. Either \
732 B>16 (there is NO exact kernel class above 16: m>16 crosses GEMM/dp4a numeric \
733 configs; the serve scheduler chunks wider concurrency into <=16 groups instead), \
734 or some matmul in this checkpoint has no bit-exact b16 kernel — run with \
735 MEMRA_EXACT16_WHY=1 to see which tensor and qtype refuses"
736 );
737 struct ExactScope<'a>(&'a Engine, bool);
738 impl Drop for ExactScope<'_> {
739 fn drop(&mut self) {
740 if self.1 {
741 self.0.set_verify_exact(false);
742 }
743 }
744 }
745 let _exact_scope = ExactScope(e, exact16);
746 if exact16 {
747 e.set_verify_exact(true);
748 }
749 // gemma4: NO batched arm at any B (per-layer SWA/global geometry, hd-512 MQA globals,
750 // weightless V-norm, softcapped head — none of it in the generic body below). This was
751 // an assert until 2026-08-07: one serve request panicked the worker, the respawn
752 // re-panicked on the queued request, and the process FATALed
753 // (research/gemma4-serve-20260807/raw/repro-panic-server-*.log). The worker now routes
754 // gemma4 sessions to the per-session eager loop and never calls here; this Err is the
755 // defense-in-depth backstop — a future path that reaches it refuses PER-REQUEST
756 // instead of killing the process. The eager arm (gemma4_decode_step_h) is the
757 // supported decode.
758 if self.is_gemma4_e4b() || self.cfg.gemma4.is_some() {
759 return Err(
760 "decode_step_batch has no gemma4 arm (per-layer swa/global geometry, \
761 softcapped head) — serve gemma4 on the eager per-session path"
762 .into(),
763 );
764 }
765 // step35 (lane/step35-batched-decode, 2026-08-08): its OWN batched walk. The generic
766 // body below is the uniform Full arm — global n_head, 128-dim rope on every layer, no
767 // SWA window, no head-wise gate — which on step35 produced HTTP-200 GARBAGE at c>1
768 // (research/step-sku-20260807/raw/b2ab-pre-*.log), so step35 NEVER enters it at any B.
769 // `step35_decode_batch_layers` carries the real geometry: per-layer n_head (64/96),
770 // partial rope (64 full / 128 SWA, dual base, rope_freqs on FULL only), per-SESSION
771 // SWA view offsets from each session's own kvl.len, the separate head-wise gate at
772 // m=B, and the sigmoid-router MoE via the same moe_ffn_il_zq8 the eager path uses.
773 // MEMRA_STEP35_BATCH=0 = the fail-closed rollback seam. The server caps chunks at
774 // B=1; on PP-N the B=1 correctness default also refuses the eager numeric class, while
775 // an unsplit deployment can still use its existing eager B=1 route.
776 if self.cfg.step35.is_some() {
777 if !Self::step35_batch_on() {
778 return Err(
779 "step35 batched decode is disabled (MEMRA_STEP35_BATCH=0) — \
780 only a non-PP eager B=1 route remains available"
781 .into(),
782 );
783 }
784 let n_embd = self.cfg.n_embd as usize;
785 let eps = self.cfg.rms_eps;
786 let mut ph_last = std::time::Instant::now();
787 let pos_v: Vec<i32> = caches.iter().map(|c| c.pos as i32).collect();
788 let pos_d = e.htod_i32(&pos_v)?;
789 let x = e.htod(&self.embd.gather(n_embd, tokens))?;
790 ph_mark(e, 0, &mut ph_last)?;
791 let x = self.step35_decode_batch_layers(
792 e,
793 x,
794 caches,
795 &pos_d,
796 0,
797 self.layers.len(),
798 &mut ph_last,
799 )?;
800 let mut hn = e.uninit(b_n * n_embd)?;
801 e.rms_norm(&x, self.output_norm.float_data(), &mut hn, n_embd, b_n, eps)?;
802 let logits = e.matmul(&self.output, &hn, b_n)?;
803 ph_mark(e, 10, &mut ph_last)?;
804 return self.decode_batch_epilogue(
805 e,
806 caches,
807 samp,
808 masks,
809 lean,
810 logits,
811 b_n,
812 &mut ph_last,
813 );
814 }
815 let n_embd = self.cfg.n_embd as usize;
816 let eps = self.cfg.rms_eps;
817
818 // MEMRA_BATCH_PHASE=1: sync-bounded phase accumulation (diagnostics — see header note).
819 // Initialized BEFORE the tick-input assembly below so slot 0 covers the HOST side of
820 // setup (pos_v/ptr-table builds, embed gather) as well as the H2D sync — the audit-fix
821 // lane's Q6 instrumentation gap (research/audit-fixes2-20260805): the old placement
822 // started the clock after the assembly, so slot 0 under-reported setup.
823 let mut ph_last = std::time::Instant::now();
824
825 // Per-row rope positions (each sequence at its own depth).
826 let pos_v: Vec<i32> = caches.iter().map(|c| c.pos as i32).collect();
827 let pos_d = e.htod_i32(&pos_v)?;
828
829 // Per-step, whole-trunk layer context: state pointer table + arm picks. Under a pp
830 // split this call is made once PER STAGE with that stage's engine and range instead
831 // (see `batch_layer_ctx`'s doc for why the table cannot be shared across devices).
832 let n_layers = self.layers.len();
833 let ctx = self.batch_layer_ctx(e, caches, 0, n_layers)?;
834
835 // Embed all B tokens -> x [B, n_embd] (host gather, one H2D).
836 let x = e.htod(&self.embd.gather(n_embd, tokens))?;
837 ph_mark(e, 0, &mut ph_last)?;
838
839 let x = self.decode_batch_layers(e, x, caches, &ctx, &pos_d, &mut ph_last)?;
840
841 // ---- output norm + lm_head at m=B, one D2H ----
842 let mut hn = e.uninit(b_n * n_embd)?;
843 e.rms_norm(&x, self.output_norm.float_data(), &mut hn, n_embd, b_n, eps)?;
844 let logits = e.matmul(&self.output, &hn, b_n)?;
845 ph_mark(e, 10, &mut ph_last)?;
846
847 self.decode_batch_epilogue(e, caches, samp, masks, lean, logits, b_n, &mut ph_last)
848 }
849
850 /// DUAL-ACTIVE PP-2 DECODE (increment 0): split one batch into wave A/B and drive
851 /// stage 0(B) from a scoped host walker while this thread drives stage 1(A). Step's
852 /// per-layer router readback synchronizes the host, so two CUDA streams issued by one
853 /// host thread would remain serial; this mirrors the proven prime PP-2 host schedule.
854 ///
855 /// This arm is the naked PP-2 default since the 2026-08-11 owner flip (`MEMRA_DUAL_PP`
856 /// unset = Auto; `0` is the serial rollback seam). It is fail-closed unless the
857 /// double-slot door is open, prewarms both slots, and uses `tx_pipelined` exclusively.
858 #[allow(clippy::too_many_arguments)]
859 fn decode_step_batch_dual(
860 &self,
861 e: &Engine,
862 tokens: &[u32],
863 caches: &mut [&mut Cache],
864 samp: &[Option<DevSamp>],
865 masks: &[Option<(&CudaSlice<u32>, usize)>],
866 lean: bool,
867 fence: &[usize],
868 mid: usize,
869 ) -> Result<(Vec<Vec<f32>>, Vec<Option<u32>>), Box<dyn std::error::Error>> {
870 let b_n = tokens.len();
871 assert!(
872 b_n >= 1 && b_n == caches.len(),
873 "tokens/caches length mismatch"
874 );
875 let Some(expected_mid) = crate::pp::dual_pp_wave_mid(b_n) else {
876 return self.decode_step_batch_ppn(e, tokens, caches, samp, masks, lean, fence);
877 };
878 if mid != expected_mid {
879 return Err(format!(
880 "decode_step_batch_dual: worker midpoint {mid} is not the balanced midpoint {expected_mid} for B={b_n}"
881 ).into());
882 }
883 if self.is_gemma4_e4b() || self.cfg.gemma4.is_some() {
884 return Err(
885 "decode_step_batch_dual has no gemma4 arm — serve gemma4 on the eager \
886 per-session path"
887 .into(),
888 );
889 }
890 assert!(
891 samp.is_empty() || samp.len() == b_n,
892 "decode_step_batch_dual: samp must be empty or have one entry per row"
893 );
894 assert!(
895 masks.is_empty() || masks.len() == b_n,
896 "decode_step_batch_dual: masks must be empty or have one entry per row"
897 );
898
899 let cap = Self::decode_batch_cap();
900 let max_wave = mid.max(b_n - mid);
901 let exact16 = max_wave > 8 && max_wave <= 16 && self.decode_batch_exact16_ok();
902 if max_wave > cap && !exact16 {
903 return Err(format!(
904 "decode_step_batch_dual: B={b_n} waves {mid}+{} exceed per-wave cap {cap} with no exact tier — refused",
905 b_n - mid,
906 ).into());
907 }
908 let n_st = fence.len() - 1;
909 crate::pp::dual_pp_eligibility(
910 n_st,
911 crate::pp::pp2_overlap(),
912 crate::pp::pp_host_bounce_active(),
913 )
914 .map_err(|msg| -> Box<dyn std::error::Error> { msg.into() })?;
915 let rt = crate::pp::PpNRt::get(e)?;
916 assert_eq!(
917 rt.n_stages(),
918 n_st,
919 "PpNRt stage count {} != fence stages {n_st}",
920 rt.n_stages()
921 );
922 let caller_stream = e.stream();
923 rt.fence_stages_behind(&caller_stream)?;
924
925 let n_embd = self.cfg.n_embd as usize;
926 let wave_cap = mid.max(b_n - mid) * n_embd;
927 rt.prepare_overlap_slots(0, wave_cap)?;
928
929 // EXACT-16 is a property of either scheduled wave, not the combined live width. Keep
930 // the scope live across both host walkers and set it on both stage-owned Engines.
931 struct ExactScopeN<'a>(Vec<&'a Engine>);
932 impl Drop for ExactScopeN<'_> {
933 fn drop(&mut self) {
934 for eng in &self.0 {
935 eng.set_verify_exact(false);
936 }
937 }
938 }
939 let _exact_scope = if exact16 {
940 let engines: Vec<&Engine> = (0..n_st).map(|s| rt.engine(s, e)).collect();
941 for eng in &engines {
942 eng.set_verify_exact(true);
943 }
944 Some(ExactScopeN(engines))
945 } else {
946 None
947 };
948
949 let step35_batched = self.cfg.step35.is_some();
950 if step35_batched && !Self::step35_batch_on() {
951 return Err(
952 "step35 batched decode is disabled (MEMRA_STEP35_BATCH=0) — \
953 dual-active PP-2 decode has no correct fallback trunk"
954 .into(),
955 );
956 }
957
958 let (tokens_a, tokens_b) = tokens.split_at(mid);
959 let (caches_a, caches_b) = caches.split_at_mut(mid);
960 let (samp_a, samp_b) = if samp.is_empty() {
961 (&[][..], &[][..])
962 } else {
963 samp.split_at(mid)
964 };
965 let (masks_a, masks_b) = if masks.is_empty() {
966 (&[][..], &[][..])
967 } else {
968 masks.split_at(mid)
969 };
970
971 let (slot_a, ph_a, span_a0) = self.decode_step_batch_dual_stage0(
972 e,
973 rt,
974 tokens_a,
975 caches_a,
976 fence,
977 step35_batched,
978 false,
979 )?;
980
981 static LOGGED: std::sync::Once = std::sync::Once::new();
982 LOGGED.call_once(|| {
983 eprintln!("[dual-pp] dual-active PP-2 decode engaged (naked default since 2026-08-11; two waves)");
984 });
985
986 let (out_a, out_b, span_b0, span_b1) = std::thread::scope(
987 |scope| -> Result<_, Box<dyn std::error::Error>> {
988 let stage0_b = scope.spawn(move || {
989 let staged = self
990 .decode_step_batch_dual_stage0(
991 e,
992 rt,
993 tokens_b,
994 caches_b,
995 fence,
996 step35_batched,
997 true,
998 )
999 .map_err(|err| err.to_string())?;
1000 Ok::<_, String>((staged, caches_b))
1001 });
1002
1003 let out_a = self.decode_step_batch_dual_stage1(
1004 e,
1005 rt,
1006 slot_a,
1007 caches_a,
1008 samp_a,
1009 masks_a,
1010 lean,
1011 fence,
1012 step35_batched,
1013 ph_a,
1014 true,
1015 )?;
1016 let ((slot_b, ph_b, span_b0), caches_b) = stage0_b
1017 .join()
1018 .map_err(|_| "dual PP stage-0 wave-B host walker panicked")?
1019 .map_err(|err| -> Box<dyn std::error::Error> { err.into() })?;
1020 if !crate::pp::record_dual_pp_slot_pair(slot_a, slot_b) {
1021 return Err(format!(
1022 "decode_step_batch_dual: refused: wave A and B both selected boundary slot {slot_a}"
1023 ).into());
1024 }
1025 let (out_b, span_b1) = self.decode_step_batch_dual_stage1(
1026 e,
1027 rt,
1028 slot_b,
1029 caches_b,
1030 samp_b,
1031 masks_b,
1032 lean,
1033 fence,
1034 step35_batched,
1035 ph_b,
1036 false,
1037 )?;
1038 Ok((out_a, out_b, span_b0, span_b1))
1039 },
1040 )?;
1041
1042 // Wave B is the final producer. One event publishes all last-stage work back to the
1043 // caller after both epilogues, preserving the ordinary PP-N exit law.
1044 rt.publish_to(1, &caller_stream)?;
1045 let (out_a, span_a1) = out_a;
1046 for (stage, span) in [span_a0, span_a1, span_b0, span_b1].into_iter().enumerate() {
1047 if let Some((start, end)) = span {
1048 crate::pp::record_dual_pp_stage_result(stage, start.elapsed_ms(&end));
1049 }
1050 }
1051 let (mut rows, mut next) = out_a;
1052 rows.extend(out_b.0);
1053 next.extend(out_b.1);
1054 Ok((rows, next))
1055 }
1056
1057 #[allow(clippy::too_many_arguments)]
1058 fn decode_step_batch_dual_stage0(
1059 &self,
1060 e: &Engine,
1061 rt: &crate::pp::PpNRt,
1062 tokens: &[u32],
1063 caches: &mut [&mut Cache],
1064 fence: &[usize],
1065 step35_batched: bool,
1066 track_overlap: bool,
1067 ) -> Result<(usize, std::time::Instant, DualPpCudaSpan), Box<dyn std::error::Error>> {
1068 let b_n = tokens.len();
1069 let n_embd = self.cfg.n_embd as usize;
1070 let mut ph_last = std::time::Instant::now();
1071 rt.bind_stage(0)?;
1072 let _st0 = rt.enter(0);
1073 let e0 = rt.engine(0, e);
1074 let pos_v: Vec<i32> = caches.iter().map(|c| c.pos as i32).collect();
1075 let pos_d = e0.htod_i32(&pos_v)?;
1076 let x = e0.htod(&self.embd.gather(n_embd, tokens))?;
1077 ph_mark(e0, 0, &mut ph_last)?;
1078 let timing_start = dual_pp_timing_event(e0, "stage0 start event");
1079 let x = {
1080 let _overlap = track_overlap.then(crate::pp::enter_dual_pp_stage);
1081 if step35_batched {
1082 self.step35_decode_batch_layers(
1083 e0,
1084 x,
1085 caches,
1086 &pos_d,
1087 fence[0],
1088 fence[1],
1089 &mut ph_last,
1090 )?
1091 } else {
1092 let ctx = self.batch_layer_ctx(e0, caches, fence[0], fence[1])?;
1093 self.decode_batch_layers(e0, x, caches, &ctx, &pos_d, &mut ph_last)?
1094 }
1095 };
1096 let timing = timing_start
1097 .and_then(|start| dual_pp_timing_event(e0, "stage0 end event").map(|end| (start, end)));
1098 let slot = rt.tx_pipelined(0, &x, b_n * n_embd)?;
1099 Ok((slot, ph_last, timing))
1100 }
1101
1102 #[allow(clippy::too_many_arguments)]
1103 fn decode_step_batch_dual_stage1(
1104 &self,
1105 e: &Engine,
1106 rt: &crate::pp::PpNRt,
1107 slot: usize,
1108 caches: &mut [&mut Cache],
1109 samp: &[Option<DevSamp>],
1110 masks: &[Option<(&CudaSlice<u32>, usize)>],
1111 lean: bool,
1112 fence: &[usize],
1113 step35_batched: bool,
1114 mut ph_last: std::time::Instant,
1115 track_overlap: bool,
1116 ) -> Result<((Vec<Vec<f32>>, Vec<Option<u32>>), DualPpCudaSpan), Box<dyn std::error::Error>>
1117 {
1118 let b_n = caches.len();
1119 let n_embd = self.cfg.n_embd as usize;
1120 let eps = self.cfg.rms_eps;
1121 rt.bind_stage(1)?;
1122 let _st1 = rt.enter(1);
1123 let e1 = rt.engine(1, e);
1124 let pos_v: Vec<i32> = caches.iter().map(|c| c.pos as i32).collect();
1125 let pos_d = e1.htod_i32(&pos_v)?;
1126 let x = rt.rx(0, slot, b_n * n_embd)?;
1127 let timing_start = dual_pp_timing_event(e1, "stage1 start event");
1128 let x = {
1129 let _overlap = track_overlap.then(crate::pp::enter_dual_pp_stage);
1130 if step35_batched {
1131 self.step35_decode_batch_layers(
1132 e1,
1133 x,
1134 caches,
1135 &pos_d,
1136 fence[1],
1137 fence[2],
1138 &mut ph_last,
1139 )?
1140 } else {
1141 let ctx = self.batch_layer_ctx(e1, caches, fence[1], fence[2])?;
1142 self.decode_batch_layers(e1, x, caches, &ctx, &pos_d, &mut ph_last)?
1143 }
1144 };
1145 let timing = timing_start
1146 .and_then(|start| dual_pp_timing_event(e1, "stage1 end event").map(|end| (start, end)));
1147 let mut hn = e1.uninit(b_n * n_embd)?;
1148 e1.rms_norm(&x, self.output_norm.float_data(), &mut hn, n_embd, b_n, eps)?;
1149 let logits = e1.matmul(&self.output, &hn, b_n)?;
1150 ph_mark(e1, 10, &mut ph_last)?;
1151 Ok((
1152 self.decode_batch_epilogue(e1, caches, samp, masks, lean, logits, b_n, &mut ph_last)?,
1153 timing,
1154 ))
1155 }
1156
1157 /// THE BATCHED PP-N STEP (pp2-batch increment 2, 2026-08-06): the batched tick split
1158 /// across `fence.len()-1` stages, each stage running ONLY its own layer range through
1159 /// ITS OWN engine and stream, with a `[B, n_embd]` boundary activation between them.
1160 /// The batched twin of `decode_step_h_ppn`, and the #1 item on the PP-2 serving bill —
1161 /// without it a >VRAM SKU (Step-3.7-Flash: 105 GB, fits only across two cards) serves
1162 /// SINGLE-STREAM only, because the batched path was the one loop with no stage split.
1163 ///
1164 /// STRUCTURE (mirrors the eager arm exactly, so the two stay comparable):
1165 /// stage 0 `rt.enter(0)` -> per-stage pos_d + embed -> range -> `rt.tx`
1166 /// middle stages `rt.rx` -> per-stage pos_d -> range -> `rt.tx`
1167 /// last stage `rt.rx` -> per-stage pos_d -> range -> output_norm + lm_head ->
1168 /// the batched serving epilogue (masks, device sample, lean park)
1169 ///
1170 /// FOUR THINGS ARE PER-STAGE, and each is per-stage for a measured reason:
1171 ///
1172 /// 1. THE ENGINE (`rt.engine(s, e)`). Not just for the remote device: `Engine` owns
1173 /// lazily-grown stable-pointer scratch pools (`fa_part_pool`, `fa_vf16_scratch`,
1174 /// `argmax_partials`) that are single-stream-safe BY DESIGN. Two stage streams
1175 /// through one Engine is the shared-scratch race the pp2 lane hit (2026-08-02
1176 /// nondeterministic all-logits divergence, 35% flake). `PpNRt::build` already gives
1177 /// every stage s>0 its own Engine even on the primary device, so honouring
1178 /// `rt.engine(s, e)` here is what scopes the pools per stage — the batched path
1179 /// allocates MORE of that scratch than the eager one (fa at m=B), so this is the
1180 /// load-bearing half of the trap's mitigation, not an inherited nicety.
1181 ///
1182 /// 2. THE POINTER TABLE (`batch_layer_ctx(es, caches, lo, hi)`). See [`BatchLayerCtx`]:
1183 /// it holds DEVICE ADDRESSES of that range's cache state, uploaded through that
1184 /// stage's engine. One step-wide table on the primary would put every stage's kernel
1185 /// arguments in stage-0's HBM — a peer read per pointer fetch, the exact cliff this
1186 /// whole lane exists to remove.
1187 ///
1188 /// 3. `pos_d` (the M2 pipelining law, learned on the eager arm): each stage uploads its
1189 /// own copy of the step's per-row positions on ITS stream, so the buffer is
1190 /// allocated, consumed and freed on one stream. A shared stage-0 `pos_d` freed at fn
1191 /// return breaks under deferred readback — the free enqueues on stream 0 while later
1192 /// stages still dereference it.
1193 ///
1194 /// 4. THE HEAD + EPILOGUE run on the LAST stage: `output_norm`/`output` were uploaded
1195 /// through the last stage's engine by the sharded loader (`hybrid.rs`: `e_head =
1196 /// layer_engine(e, n_trunk, n_trunk-1)`), and `cache.last_logits_dev` must be
1197 /// allocated where the logits are.
1198 ///
1199 /// EXACTNESS: PP-N adds ZERO deviation. Each stage runs the SAME kernels on the SAME
1200 /// bytes in the same order — the split only moves where the residual is materialized,
1201 /// and the boundary is a straight f32 copy (dtod same-device / `cudaMemcpyPeerAsync`
1202 /// cross-device, no conversion). So batched PP-N must be BIT-IDENTICAL to single-device
1203 /// batched at the same B, in both placement orders. Gate: `decode-batch-gate --mode
1204 /// pp` (logit-dump, both orders) — the batched analogue of the eager arm's 48 steps x
1205 /// 248,320 f32 logits with zero differing bits.
1206 ///
1207 /// The B=1 fast path is NOT taken here (its condition already excludes an open door):
1208 /// it routes through `decode_layers_eager` whole-trunk on one engine, which is exactly
1209 /// the unsplit walk. B=1 under the door rides this function's B=1 case instead — the
1210 /// same trade the eager arm's own ppn step makes, and the reason the pp2 lane measured
1211 /// B=1 door-open at 0.854x (the lost fusion chain), not a cliff.
1212 #[allow(clippy::too_many_arguments)]
1213 fn decode_step_batch_ppn(
1214 &self,
1215 e: &Engine,
1216 tokens: &[u32],
1217 caches: &mut [&mut Cache],
1218 samp: &[Option<DevSamp>],
1219 masks: &[Option<(&CudaSlice<u32>, usize)>],
1220 lean: bool,
1221 fence: &[usize],
1222 ) -> Result<(Vec<Vec<f32>>, Vec<Option<u32>>), Box<dyn std::error::Error>> {
1223 let b_n = tokens.len();
1224 assert!(
1225 b_n >= 1 && b_n == caches.len(),
1226 "tokens/caches length mismatch"
1227 );
1228 // gemma4: same no-arm refusal as the unsplit body (see decode_step_batch), Err not
1229 // assert — a request must never kill the worker process.
1230 if self.is_gemma4_e4b() || self.cfg.gemma4.is_some() {
1231 return Err(
1232 "decode_step_batch_ppn has no gemma4 arm — serve gemma4 on the eager \
1233 per-session path"
1234 .into(),
1235 );
1236 }
1237 // Same width policy as the unsplit body — the stage split changes WHERE kernels run,
1238 // never WHICH tier admits the width. Duplicated deliberately rather than hoisted:
1239 // the exact-16 scope must wrap the whole multi-stage walk (`set_verify_exact` is
1240 // per-Engine state read at dispatch on every stage), so it has to be established
1241 // here, and a shared helper returning a guard would have to own `e` plus the flag.
1242 let cap = Self::decode_batch_cap();
1243 let exact16 = b_n > 8 && b_n <= 16 && self.decode_batch_exact16_ok();
1244 assert!(
1245 b_n <= cap || exact16,
1246 "decode_step_batch_ppn: B={b_n} > cap {cap} with no exact tier — refused"
1247 );
1248 let rt = crate::pp::PpNRt::get(e)?;
1249 let n_st = fence.len() - 1;
1250 assert_eq!(
1251 rt.n_stages(),
1252 n_st,
1253 "PpNRt stage count {} != fence stages {n_st}",
1254 rt.n_stages()
1255 );
1256 // #87 REVERSE PUBLICATION (lane/pp2spec-crash): order every stage stream behind
1257 // the caller before this body's first stage allocation can reuse a pool block
1258 // whose queued primary-stream consumer has not read it yet. Anatomy:
1259 // `PpNRt::fence_stages_behind`. (This body dtoh+syncs its own logits, but its
1260 // PP-mode callers interleave with the spec verify's device-resident outputs in
1261 // the same worker, so the entry fence is the uniform law, not an optimization.)
1262 rt.fence_stages_behind(&e.stream())?;
1263 let n_embd = self.cfg.n_embd as usize;
1264 let eps = self.cfg.rms_eps;
1265 let payload = b_n * n_embd;
1266
1267 // EXACT-16 SCOPE, PER STAGE ENGINE: `verify_exact` is per-Engine state (an AtomicBool
1268 // on the Engine the dispatch reads), and each stage runs through a DIFFERENT Engine —
1269 // so setting it on the primary alone would leave stages 1..N-1 dispatching the m>=16
1270 // GEMM/MMQ arms while stage 0 used the exact b16 tier. That is a silent per-stage
1271 // numeric split (the failure this tier exists to prevent), so the flag is set on
1272 // every stage engine and cleared on all of them at scope exit.
1273 struct ExactScopeN<'a>(Vec<&'a Engine>);
1274 impl Drop for ExactScopeN<'_> {
1275 fn drop(&mut self) {
1276 for eng in &self.0 {
1277 eng.set_verify_exact(false);
1278 }
1279 }
1280 }
1281 let _exact_scope = if exact16 {
1282 let engines: Vec<&Engine> = (0..n_st).map(|s| rt.engine(s, e)).collect();
1283 for eng in &engines {
1284 eng.set_verify_exact(true);
1285 }
1286 Some(ExactScopeN(engines))
1287 } else {
1288 None
1289 };
1290
1291 let mut ph_last = std::time::Instant::now();
1292
1293 // B=1 PER-STAGE FAST PATH (measured 2026-08-06, PRO 6000 pair). The unsplit body's
1294 // b1_fast guard includes `pp_cuts().is_none()`, so opening the pp door dropped every
1295 // solo session off the m=1 FUSION chain (cross-layer add+norm+q8_1, fused SwiGLU,
1296 // lever 1's gate+up dual) and onto the batched m=1 walk. Cost, arm A vs arm C at B=1:
1297 // 208.5 vs 177.3 tok/s = -15.0% — and NOT a split cost, since arm B (stages=2 on ONE
1298 // card) pays the same 177, and the prior lane's `MEMRA_PP_SHARD=0` batched-body B=1
1299 // was 178.5. It was the fusion chain going missing, on the config the Step SKU serves
1300 // solo requests from.
1301 //
1302 // `decode_layers_eager(lo, hi)` is ALREADY range-scoped and is exactly what the eager
1303 // ppn arm (`decode_step_h_ppn`) calls per stage, so B=1 rides the same per-stage
1304 // structure: same engines, same streams, same [1, n_embd] boundary slots, same
1305 // stage-owned caches. Only the trunk kernels differ, and they differ identically to
1306 // how they differ off-door. Exactness is therefore the SAME accepted decode-config FP
1307 // class the unsplit b1_fast lever already carries (strict gate1 PASSes with it on,
1308 // FAILs with it off at maxdiff 1.591e-1) — which is why the pp gate pins
1309 // `set_b1_fast(false)`: with it on, the B=1 reference and the split arm would
1310 // legitimately sit on opposite sides of that gap and the bit-identity arm would
1311 // report a fake stage-split failure.
1312 //
1313 // Step3.5/Step3.7 are an exception (lane/cx-b1fix, 2026-08-10): their B>1 route is
1314 // `step35_decode_batch_layers`, and the live scheduler may move a session from B=1
1315 // to B>1. The eager/fused class and that batched class produce different greedy bytes,
1316 // so selecting the eager arm at B=1 made output depend on load history. Keep one
1317 // numeric class for this model family: Step35 always takes its stage-scoped batched
1318 // trunk at every width. The live transition gate in step35-b2-geometry-gate pins it.
1319 // Qwen35-MoE is the second exception (lane/cx-q35bug, 2026-08-12): on the Q35
1320 // sellgate workload the eager-B1 -> batched-B2 transition changed emitted token ids and
1321 // selected EOS at tokens 15/17/25. Keep that family on this generic batched trunk at B=1
1322 // too; dense Qwen35 retains the measured eager fast path.
1323 let b1_stage_fast = b_n == 1
1324 && Self::b1_fast_on()
1325 && self.b1_fast_arch_eligible()
1326 && !self.is_gemma4_e4b()
1327 && self.cfg.gemma4.is_none()
1328 && self.cfg.m3.is_none()
1329 && self.cfg.step35.is_none()
1330 && !e.verify_exact_on();
1331 // step35 (lane/step35-batched-decode, 2026-08-08): B>1 rides its OWN stage-scoped
1332 // batched walk (`step35_decode_batch_layers`) — the generic `decode_batch_layers`
1333 // remains OFF-LIMITS for this arch at every B (its uniform geometry produced the
1334 // b2ab HTTP-200 garbage: research/step-sku-20260807/raw/b2ab-pre-*.log). Since
1335 // lane/cx-b1fix, B=1 also takes this walk: a Step35 PP-N session must not change
1336 // numeric class when live decode width changes. The refusal below guards the
1337 // rollback residue; under PP-N, disabling the only correct trunk makes Step35
1338 // requests fail closed instead of falling back to the eager class.
1339 let step35_batched = self.cfg.step35.is_some();
1340 if step35_batched && !Self::step35_batch_on() {
1341 return Err(
1342 "step35 batched decode is disabled (MEMRA_STEP35_BATCH=0) — \
1343 PP-N Step35 decode is unavailable because eager B=1 is a different \
1344 numeric class"
1345 .into(),
1346 );
1347 }
1348 // Hoisted: `caches[0].pos` as a value argument alongside `caches[0]` as `&mut` in one
1349 // call is a borrow conflict; `pos` is Copy and the epilogue is what advances it.
1350 let pos0 = if b1_stage_fast { caches[0].pos } else { 0 };
1351
1352 // ---- STAGE 0: embed (the table lives with stage 0) + layers [0, fence[1]) + TX ----
1353 let mut slot = {
1354 let _st0 = rt.enter(0);
1355 let e0 = rt.engine(0, e);
1356 let pos_v: Vec<i32> = caches.iter().map(|c| c.pos as i32).collect();
1357 let pos_d = e0.htod_i32(&pos_v)?;
1358 let x = e0.htod(&self.embd.gather(n_embd, tokens))?;
1359 ph_mark(e0, 0, &mut ph_last)?;
1360 let x = if b1_stage_fast {
1361 self.decode_layers_eager(e0, x, fence[0], fence[1], &pos_d, pos0, caches[0])?
1362 } else if step35_batched {
1363 self.step35_decode_batch_layers(
1364 e0,
1365 x,
1366 caches,
1367 &pos_d,
1368 fence[0],
1369 fence[1],
1370 &mut ph_last,
1371 )?
1372 } else {
1373 let ctx = self.batch_layer_ctx(e0, caches, fence[0], fence[1])?;
1374 self.decode_batch_layers(e0, x, caches, &ctx, &pos_d, &mut ph_last)?
1375 };
1376 rt.tx(0, &x, payload)?
1377 // x + pos_d + ctx.ptr_table drop here: freed stream-ordered on stage-0's stream.
1378 };
1379
1380 // ---- MIDDLE STAGES: RX boundary s-1 -> range -> TX boundary s ----
1381 for s in 1..n_st - 1 {
1382 let _st = rt.enter(s);
1383 let es = rt.engine(s, e);
1384 let pos_v: Vec<i32> = caches.iter().map(|c| c.pos as i32).collect();
1385 let pos_d = es.htod_i32(&pos_v)?;
1386 let x = rt.rx(s - 1, slot, payload)?;
1387 let x = if b1_stage_fast {
1388 self.decode_layers_eager(es, x, fence[s], fence[s + 1], &pos_d, pos0, caches[0])?
1389 } else if step35_batched {
1390 self.step35_decode_batch_layers(
1391 es,
1392 x,
1393 caches,
1394 &pos_d,
1395 fence[s],
1396 fence[s + 1],
1397 &mut ph_last,
1398 )?
1399 } else {
1400 let ctx = self.batch_layer_ctx(es, caches, fence[s], fence[s + 1])?;
1401 self.decode_batch_layers(es, x, caches, &ctx, &pos_d, &mut ph_last)?
1402 };
1403 slot = rt.tx(s, &x, payload)?;
1404 }
1405
1406 // ---- LAST STAGE: RX + final range + head + the batched serving epilogue ----
1407 let _stl = rt.enter(n_st - 1);
1408 let el = rt.engine(n_st - 1, e);
1409 let pos_v: Vec<i32> = caches.iter().map(|c| c.pos as i32).collect();
1410 let pos_d = el.htod_i32(&pos_v)?;
1411 let x = rt.rx(n_st - 2, slot, payload)?;
1412 let x = if b1_stage_fast {
1413 self.decode_layers_eager(el, x, fence[n_st - 1], fence[n_st], &pos_d, pos0, caches[0])?
1414 } else if step35_batched {
1415 self.step35_decode_batch_layers(
1416 el,
1417 x,
1418 caches,
1419 &pos_d,
1420 fence[n_st - 1],
1421 fence[n_st],
1422 &mut ph_last,
1423 )?
1424 } else {
1425 let ctx = self.batch_layer_ctx(el, caches, fence[n_st - 1], fence[n_st])?;
1426 self.decode_batch_layers(el, x, caches, &ctx, &pos_d, &mut ph_last)?
1427 };
1428
1429 let mut hn = el.uninit(payload)?;
1430 el.rms_norm(&x, self.output_norm.float_data(), &mut hn, n_embd, b_n, eps)?;
1431 let logits = el.matmul(&self.output, &hn, b_n)?;
1432 ph_mark(el, 10, &mut ph_last)?;
1433
1434 self.decode_batch_epilogue(el, caches, samp, masks, lean, logits, b_n, &mut ph_last)
1435 }
1436
1437 /// Build the per-step layer context for layers `[lo, hi)`: the device state-pointer
1438 /// table plus the step's arm picks. See [`BatchLayerCtx`] for why this is RANGE-scoped
1439 /// (the table holds device addresses and must be uploaded through the engine whose
1440 /// device runs those layers).
1441 ///
1442 /// Table layout is unchanged from the whole-trunk version — `lin_base`/`attn_base` are
1443 /// still indexed by ABSOLUTE layer id, so `decode_batch_layers`' body indexes them
1444 /// exactly as the old inline loop did. Only layers in `[lo, hi)` contribute entries; the
1445 /// rest stay `None`, which is a loud `expect` if a range ever reads outside its own.
1446 pub(crate) fn batch_layer_ctx(
1447 &self,
1448 e: &Engine,
1449 caches: &[&mut Cache],
1450 lo: usize,
1451 hi: usize,
1452 ) -> Result<BatchLayerCtx, Box<dyn std::error::Error>> {
1453 let cfg = &self.cfg;
1454 let head_dim = cfg.head_dim_k as usize;
1455 // Per-step STATE POINTER TABLE (one H2D): for every linear layer, [conv x B]
1456 // [ssm_in x B][ssm_out x B] device addresses. The batched state kernels read their
1457 // sequence's pointer from these arrays — states stay per-cache (no pooling refactor),
1458 // yet conv/prep/scan collapse from 3xB launches per layer to 3. Rebuilt every step
1459 // because the ssm ping-pong swaps pointers host-side after each scan.
1460 // INCREMENT 2 (2026-08-01): the SAME table now also carries, for every FULL-attn
1461 // layer, [k0,v0,k1,v1,...] cache base addresses — the z-batched seqs append and
1462 // seqs fa_decode kernels read their sequence's cache through it (the MoE
1463 // expert-table pattern), collapsing 2xB launches per attn layer to 2.
1464 let mut lin_base: Vec<Option<usize>> = vec![None; self.layers.len()];
1465 let mut attn_base: Vec<Option<usize>> = vec![None; self.layers.len()];
1466 let mut ptrs: Vec<u64> = Vec::new();
1467 {
1468 use cudarc::driver::DevicePtr;
1469 let s = &e.gpu.stream();
1470 for il in lo..hi {
1471 match &self.layers[il].mixer {
1472 Mixer::Linear(_) => {
1473 lin_base[il] = Some(ptrs.len());
1474 for c in caches.iter() {
1475 let rl = c.recur[il].as_ref().unwrap();
1476 let (p, _g) = rl.conv_state.device_ptr(s);
1477 ptrs.push(p as u64);
1478 }
1479 for c in caches.iter() {
1480 let rl = c.recur[il].as_ref().unwrap();
1481 let (p, _g) = rl.ssm_state.device_ptr(s);
1482 ptrs.push(p as u64);
1483 }
1484 for c in caches.iter() {
1485 let rl = c.recur[il].as_ref().unwrap();
1486 let (p, _g) = rl.ssm_state_alt.device_ptr(s);
1487 ptrs.push(p as u64);
1488 }
1489 }
1490 Mixer::Full(_) => {
1491 attn_base[il] = Some(ptrs.len());
1492 for c in caches.iter() {
1493 let kvl = c.kv[il].as_ref().unwrap();
1494 let (pk, _g) = kvl.k.device_ptr(s);
1495 let (pv, _g2) = kvl.v.device_ptr(s);
1496 ptrs.push(pk as u64);
1497 ptrs.push(pv as u64);
1498 }
1499 }
1500 Mixer::Mla(_) => crate::hybrid::mla_forward_unimplemented(),
1501 }
1502 }
1503 }
1504 let ptr_table = if ptrs.is_empty() {
1505 None
1506 } else {
1507 Some(e.htod_u64(&ptrs)?)
1508 };
1509
1510 // INCREMENT 2 arm picks (per STEP — t_kv is layer-invariant within a tick):
1511 // - seqs APPEND: format-only condition (per-row program is t_kv-independent);
1512 // default flash module only (fp8-KV rides the per-seq g-module path).
1513 // - seqs FA: every row must take the v4 eager arm at ITS OWN t_kv AND all rows
1514 // must share ONE fa_split_keys rung (the rows-twins' straddle law) — a rung
1515 // crossing inside the batch keeps the per-seq loop for that step, so each
1516 // sequence always executes the exact program its isolated run would.
1517 // MEMRA_BATCH_APPEND=0 / MEMRA_BATCH_FA=0 are the rollback/A-B seams.
1518 //
1519 // The picks are t_kv-driven, and t_kv is layer-INVARIANT within a step, so every
1520 // stage of a pp split independently computes the SAME arms from the same `caches`
1521 // — a stage cannot silently take a different program than its unsplit self.
1522 let t_kvs: Vec<usize> = caches.iter().map(|c| c.pos + 1).collect();
1523 let t_kv_max = *t_kvs.iter().max().unwrap();
1524 let seqs_append = {
1525 static ON: std::sync::OnceLock<bool> = std::sync::OnceLock::new();
1526 *ON.get_or_init(|| std::env::var("MEMRA_BATCH_APPEND").as_deref() != Ok("0"))
1527 } && !Engine::kv_fp8_on();
1528 let sp0 = crate::fa_split_keys(t_kvs[0], cfg.n_head_kv as usize);
1529 let seqs_fa = {
1530 static ON: std::sync::OnceLock<bool> = std::sync::OnceLock::new();
1531 *ON.get_or_init(|| std::env::var("MEMRA_BATCH_FA").as_deref() != Ok("0"))
1532 } && t_kvs.iter().all(|&t| crate::fa_seqs_eligible(t, head_dim))
1533 && t_kvs
1534 .iter()
1535 .all(|&t| crate::fa_split_keys(t, cfg.n_head_kv as usize) == sp0);
1536
1537 Ok(BatchLayerCtx {
1538 lin_base,
1539 attn_base,
1540 ptr_table,
1541 t_kvs,
1542 t_kv_max,
1543 sp0,
1544 seqs_append,
1545 seqs_fa,
1546 lo,
1547 hi,
1548 })
1549 }
1550
1551 /// THE PP SEAM (pp2-batch increment 1, 2026-08-06): run the batched trunk over layers
1552 /// `[ctx.lo, ctx.hi)`, entering with a materialized `[B, n_embd]` residual and exiting
1553 /// with the range's final residual materialized. The batched twin of
1554 /// `decode_layers_eager` — the eager arm has had this seam since M1-PP2 and every ppN
1555 /// stage calls it; the batched body had no equivalent, which is why every later PP-2
1556 /// increment (and spec-over-PP2, whose verify is a batched T=K+1 forward) waited on this
1557 /// extraction (`research/pp2-hardening-20260806/PROGRESS.md` bill item 1).
1558 ///
1559 /// SINGLE-DEVICE SEMANTICS ARE UNCHANGED BY CONSTRUCTION: the body is the old
1560 /// `for (il, layer) in self.layers.iter().enumerate()` loop moved verbatim, with `for il
1561 /// in ctx.lo..ctx.hi` as the header and the per-step invariants (`ptr_table`, arm picks,
1562 /// `t_kv`) read from `ctx` instead of enclosing locals. At `lo=0, hi=n_layers` — every
1563 /// call today — the launch sequence is identical, so the exactness contract in this
1564 /// module's header carries over untouched rather than needing a re-proof.
1565 ///
1566 /// UNLIKE the eager seam, this one is NOT yet stage-callable: `caches` is `&mut [&mut
1567 /// Cache]` mutated in place (KV `len` bumps, ssm ping-pong swaps), and `pos_d`/`x` come
1568 /// from the caller's device. Wiring a stage split means per-stage `pos_d` + a boundary
1569 /// `[B, n_embd]` transfer around this call, which is the NEXT increment. The seam exists
1570 /// so that increment is a call-site change, not a 250-line surgery.
1571 #[allow(clippy::too_many_arguments)]
1572 pub(crate) fn decode_batch_layers(
1573 &self,
1574 e: &Engine,
1575 mut x: CudaSlice<f32>,
1576 caches: &mut [&mut Cache],
1577 ctx: &BatchLayerCtx,
1578 pos_d: &CudaSlice<i32>,
1579 ph_last: &mut std::time::Instant,
1580 ) -> Result<CudaSlice<f32>, Box<dyn std::error::Error>> {
1581 let b_n = caches.len();
1582 let cfg = &self.cfg;
1583 let n_embd = cfg.n_embd as usize;
1584 let eps = cfg.rms_eps;
1585 let (lin_base, attn_base) = (&ctx.lin_base, &ctx.attn_base);
1586 let ptr_table = &ctx.ptr_table;
1587 let (seqs_append, seqs_fa, sp0, t_kv_max) =
1588 (ctx.seqs_append, ctx.seqs_fa, ctx.sp0, ctx.t_kv_max);
1589 debug_assert_eq!(
1590 ctx.t_kvs.len(),
1591 b_n,
1592 "ctx built for a different batch width"
1593 );
1594
1595 for il in ctx.lo..ctx.hi {
1596 let layer = &self.layers[il];
1597 // ---- attn_norm + q8_1 quantize, batched (B rows) ----
1598 let anorm = layer.attn_norm.float_data();
1599 let mut xn = e.uninit(b_n * n_embd)?;
1600 e.rms_norm(&x, anorm, &mut xn, n_embd, b_n, eps)?;
1601 let (hq, hd) = e.quantize_q8_1(&xn, b_n, n_embd)?;
1602
1603 // ---- mixer ----
1604 let mixed: CudaSlice<f32> = match &layer.mixer {
1605 Mixer::Mla(_) => crate::hybrid::mla_forward_unimplemented(),
1606 Mixer::Full(fa) => {
1607 let geometry = cfg.full_attention_geometry_at(il as u32);
1608 let n_head = geometry.n_head as usize;
1609 let n_head_kv = geometry.n_head_kv as usize;
1610 let head_dim = geometry.head_dim_k as usize;
1611 let rope_dims = geometry.n_rot as usize;
1612 let rope_base = geometry.rope_base;
1613 let scale = geometry.attention_scale();
1614 // Batched projections: one weight read serves all B rows. At B=1 the
1615 // QKV triple fuses into ONE launch (rig-native decode increment 1 —
1616 // bit-identical per (tensor,row), RIG-NATIVE-DECODE.md); B>1 and
1617 // non-NVFP4 trunks keep the three singles.
1618 let (qf, mut k, v) =
1619 match e.matmul_nvfp4_fused3(&fa.wq, &fa.wk, &fa.wv, &hq, &hd, b_n)? {
1620 Some(t) => t,
1621 None => (
1622 e.matmul_pre(&fa.wq, &hq, &hd, &xn, b_n)?,
1623 e.matmul_pre(&fa.wk, &hq, &hd, &xn, b_n)?,
1624 e.matmul_pre(&fa.wv, &hq, &hd, &xn, b_n)?,
1625 ),
1626 };
1627
1628 let gated =
1629 geometry.attention_gate == memra_gguf::config::AttentionGateKind::FusedQ;
1630 let (mut q, gate) = if gated {
1631 let mut qs = e.uninit(b_n * n_head * head_dim)?;
1632 let mut gs = e.uninit(b_n * n_head * head_dim)?;
1633 e.q_gate_split(&qf, &mut qs, &mut gs, head_dim, n_head, b_n)?;
1634 (qs, Some(gs))
1635 } else {
1636 (qf, None)
1637 };
1638
1639 // QK-norm over B*n_head rows, rope with per-row positions.
1640 let mut qn = e.uninit(b_n * n_head * head_dim)?;
1641 e.rms_norm(
1642 &q,
1643 fa.q_norm.float_data(),
1644 &mut qn,
1645 head_dim,
1646 b_n * n_head,
1647 eps,
1648 )?;
1649 q = qn;
1650 let mut kn = e.uninit(b_n * n_head_kv * head_dim)?;
1651 e.rms_norm(
1652 &k,
1653 fa.k_norm.float_data(),
1654 &mut kn,
1655 head_dim,
1656 b_n * n_head_kv,
1657 eps,
1658 )?;
1659 k = kn;
1660 e.rope_neox(
1661 &mut q, &pos_d, head_dim, rope_dims, n_head, b_n, rope_base, 1.0,
1662 )?;
1663 e.rope_neox(
1664 &mut k, &pos_d, head_dim, rope_dims, n_head_kv, b_n, rope_base, 1.0,
1665 )?;
1666 ph_mark(e, 1, ph_last)?;
1667
1668 // INCREMENT 2 (2026-08-01): the per-seq (append, attend) launch train
1669 // becomes two phases. Phase A appends all B rows (one z-batched launch,
1670 // or the per-seq loop on the seam/fp8 path); phase B attends all B
1671 // sequences (one blockIdx.z launch + one combine on the batched arm —
1672 // which also reads q / writes attn at row offsets, killing the per-seq
1673 // q/a dtod copies — or the per-seq loop when any row is outside the v4
1674 // arm / a split rung crosses inside the batch). Caches are disjoint per
1675 // sequence, so the phase split leaves every row's math untouched.
1676 let q_dim = n_head * head_dim;
1677 let kv_dim = n_head_kv * head_dim;
1678 let mut attn = e.uninit(b_n * q_dim)?;
1679 // ---- phase A: KV append (all B rows) ----
1680 if seqs_append {
1681 let (kdk, kdv, ktb, vtb) = {
1682 let kvl = caches[0].kv[il].as_ref().unwrap();
1683 (kvl.kv_dim_k, kvl.kv_dim_v, kvl.k_tok_bytes, kvl.v_tok_bytes)
1684 };
1685 let base = attn_base[il].expect("full layer missing from pointer table");
1686 let table = ptr_table.as_ref().expect("pointer table missing");
1687 let kv_view = table.slice(base..base + 2 * b_n);
1688 e.append_kv_quantized_seqs(
1689 &k, &v, &kv_view, &pos_d, b_n, kdk, kdv, ktb, vtb,
1690 )?;
1691 for cache in caches.iter_mut() {
1692 let kvl = cache.kv[il].as_mut().unwrap();
1693 debug_assert_eq!(kvl.len, cache.pos, "kv len / pos out of lockstep");
1694 kvl.len += 1;
1695 }
1696 } else {
1697 for (bi, cache) in caches.iter_mut().enumerate() {
1698 let kvl = cache.kv[il].as_mut().unwrap();
1699 let k_row = k.slice(bi * kv_dim..(bi + 1) * kv_dim);
1700 let v_row = v.slice(bi * kv_dim..(bi + 1) * kv_dim);
1701 e.append_kv_quantized_view(
1702 &k_row,
1703 &v_row,
1704 &mut kvl.k,
1705 &mut kvl.v,
1706 kvl.len,
1707 kvl.kv_dim_k,
1708 kvl.kv_dim_v,
1709 kvl.k_tok_bytes,
1710 kvl.v_tok_bytes,
1711 Engine::kv_fp8_on(),
1712 )?;
1713 kvl.len += 1;
1714 }
1715 }
1716 ph_mark(e, 2, ph_last)?;
1717 // ---- phase B: attention (all B sequences) ----
1718 if seqs_fa {
1719 let (ktb, vtb) = {
1720 let kvl = caches[0].kv[il].as_ref().unwrap();
1721 (kvl.k_tok_bytes, kvl.v_tok_bytes)
1722 };
1723 let base = attn_base[il].expect("full layer missing from pointer table");
1724 let table = ptr_table.as_ref().expect("pointer table missing");
1725 let kv_view = table.slice(base..base + 2 * b_n);
1726 e.fa_decode_batch_seqs_v4(
1727 &q, &kv_view, &pos_d, &mut attn, head_dim, n_head, n_head_kv, b_n,
1728 t_kv_max, scale, sp0, ktb, vtb,
1729 )?;
1730 ph_mark(e, 4, ph_last)?;
1731 } else {
1732 for (bi, cache) in caches.iter_mut().enumerate() {
1733 let kvl = cache.kv[il].as_mut().unwrap();
1734 let t_kv = kvl.len;
1735 let k_view = e.view_u8(&kvl.k, t_kv * kvl.k_tok_bytes);
1736 let v_view = e.view_u8(&kvl.v, t_kv * kvl.v_tok_bytes);
1737 // The fallback keeps one FA launch per distinct KV view, but Q and
1738 // attention already live in packed row-major buffers. Pass those row
1739 // views directly; only the arithmetic-free materialization copies go.
1740 let q_row = q.slice(bi * q_dim..(bi + 1) * q_dim);
1741 let mut a_row = attn.slice_mut(bi * q_dim..(bi + 1) * q_dim);
1742 e.fa_decode_kvmod_view(
1743 &q_row,
1744 &k_view,
1745 &v_view,
1746 &mut a_row,
1747 head_dim,
1748 n_head,
1749 n_head_kv,
1750 t_kv,
1751 scale,
1752 kvl.k_tok_bytes,
1753 kvl.v_tok_bytes,
1754 Engine::kv_fp8_on(),
1755 )?;
1756 ph_mark(e, 4, ph_last)?;
1757 }
1758 }
1759
1760 // Output gate (element-wise — batches whole) + o-proj at m=B.
1761 let attn_g = match &gate {
1762 Some(g) => {
1763 let n = b_n * q_dim;
1764 let mut gsig = e.uninit(n)?;
1765 e.sigmoid(g, &mut gsig, n)?;
1766 let mut ag = e.uninit(n)?;
1767 e.mul(&attn, &gsig, &mut ag, n)?;
1768 ag
1769 }
1770 None => attn,
1771 };
1772 let o = e.matmul(&fa.wo, &attn_g, b_n)?;
1773 ph_mark(e, 5, ph_last)?;
1774 o
1775 }
1776 Mixer::Linear(la) => {
1777 // v2 (the B-scaling fix): the GDN mixer's PROJECTIONS carry the layer's
1778 // weight mass — batch them at m=B so wqkv/gate/beta/alpha/ssm_out stream
1779 // ONCE per step instead of once per sequence. Only the recurrent state ops
1780 // (fused conv ring, gdn prep, gdn scan) stay per-seq — they are state-bound
1781 // micro-kernels, not weight readers. Composition unchanged vs v1 (matmul_pre
1782 // == fused2 per (tensor,row); _bN mmvq per-row == m=1): same numeric config.
1783 let ssm = cfg.ssm.as_ref().expect("linear mixer requires ssm cfg");
1784 let d_state = ssm.state_size as usize;
1785 let num_k = ssm.group_count as usize;
1786 let num_v = ssm.time_step_rank as usize;
1787 let d_conv = ssm.conv_kernel as usize;
1788 let key_dim = d_state * num_k;
1789 let value_dim = d_state * num_v;
1790 let conv_dim = key_dim * 2 + value_dim;
1791 let gdn_scale = 1.0 / (d_state as f32).sqrt();
1792
1793 // ---- batched projections (the weight win) ----
1794 let qkv_mixed = e.matmul_pre(&la.wqkv, &hq, &hd, &xn, b_n)?;
1795 let z = e.matmul_pre(&la.wqkv_gate, &hq, &hd, &xn, b_n)?;
1796 let beta_raw = e.matmul_pre(&la.ssm_beta, &hq, &hd, &xn, b_n)?;
1797 let alpha = e.matmul_pre(&la.ssm_alpha, &hq, &hd, &xn, b_n)?;
1798 ph_mark(e, 6, ph_last)?;
1799
1800 // ---- batched recurrent state ops (3 launches for all B sequences) ----
1801 let base = lin_base[il].expect("linear layer missing from pointer table");
1802 let table = ptr_table.as_ref().expect("pointer table missing");
1803 let conv_view = table.slice(base..base + b_n);
1804 let in_view = table.slice(base + b_n..base + 2 * b_n);
1805 let out_view = table.slice(base + 2 * b_n..base + 3 * b_n);
1806 let mut conv_outs = e.uninit(b_n * conv_dim)?;
1807 e.ssm_conv1d_fused_decode_b(
1808 &qkv_mixed,
1809 &conv_view,
1810 la.ssm_conv1d.float_data(),
1811 &mut conv_outs,
1812 conv_dim,
1813 d_conv,
1814 b_n,
1815 )?;
1816 let mut q_l2 = e.uninit(b_n * value_dim)?;
1817 let mut k_l2 = e.uninit(b_n * value_dim)?;
1818 let mut v_gd = e.uninit(b_n * value_dim)?;
1819 let mut beta_b = e.uninit(b_n * num_v)?;
1820 let mut g_log = e.uninit(b_n * num_v)?;
1821 e.gdn_prep_decode_b(
1822 &conv_outs,
1823 &beta_raw,
1824 &alpha,
1825 la.ssm_dt.float_data(),
1826 la.ssm_a.float_data(),
1827 &mut q_l2,
1828 &mut k_l2,
1829 &mut v_gd,
1830 &mut beta_b,
1831 &mut g_log,
1832 d_state,
1833 num_v,
1834 num_k,
1835 key_dim,
1836 eps,
1837 conv_dim,
1838 b_n,
1839 )?;
1840 let mut o_all = e.uninit(b_n * value_dim)?;
1841 e.gdn_scan_s128_batched(
1842 &q_l2, &k_l2, &v_gd, &g_log, &beta_b, &in_view, &out_view, &mut o_all,
1843 num_v, b_n, gdn_scale,
1844 )?;
1845 // ping-pong: scan wrote each seq's alt buffer; swap host handles (the
1846 // NEXT step's table rebuild picks up the new canonical pointers).
1847 for cache in caches.iter_mut() {
1848 let rl = cache.recur[il].as_mut().unwrap();
1849 std::mem::swap(&mut rl.ssm_state, &mut rl.ssm_state_alt);
1850 }
1851 ph_mark(e, 7, ph_last)?;
1852
1853 // ---- batched gated norm + out-projection ----
1854 let o = if e.uses_q8_1_fast(&la.ssm_out) {
1855 let (gq, gd) = e.gated_rmsnorm_q8_1(
1856 &o_all,
1857 la.ssm_norm.float_data(),
1858 &z,
1859 d_state,
1860 b_n * num_v,
1861 eps,
1862 )?;
1863 let g0 = e.zeros(0)?;
1864 e.matmul_pre(&la.ssm_out, &gq, &gd, &g0, b_n)?
1865 } else {
1866 let mut gn = e.uninit(b_n * value_dim)?;
1867 e.gated_rmsnorm(
1868 &o_all,
1869 la.ssm_norm.float_data(),
1870 &z,
1871 &mut gn,
1872 d_state,
1873 b_n * num_v,
1874 eps,
1875 )?;
1876 e.matmul(&la.ssm_out, &gn, b_n)?
1877 };
1878 ph_mark(e, 8, ph_last)?;
1879 o
1880 }
1881 };
1882
1883 // ---- residual add + post_attn_norm + FFN, batched ----
1884 let pnorm = layer.post_attn_norm.float_data();
1885 let mut x1 = e.uninit(b_n * n_embd)?;
1886 let mut z = e.uninit(b_n * n_embd)?;
1887 e.add_rms_norm(&x, &mixed, pnorm, &mut x1, &mut z, n_embd, b_n, eps)?;
1888 let ffn_out = match &layer.ffn {
1889 crate::hybrid::Ffn::Dense {
1890 ffn_gate,
1891 ffn_up,
1892 ffn_down,
1893 } => {
1894 // v1 covers the SiLU family; M3's swigluoai clamp rides a scaled epilogue
1895 // (m=1 fused tier) — batched M3 lands with the batched-fusion pass.
1896 assert!(
1897 self.cfg.m3.is_none(),
1898 "decode_step_batch v1: M3 swigluoai FFN not yet batched"
1899 );
1900 let n_ff = ffn_gate.out_features();
1901 let (zq, zd) = e.quantize_q8_1(&z, b_n, n_embd)?;
1902 // REFUTED ARM (lane/q27-deepdive, 2026-08-05): fusing this gate+up pair
1903 // into `matmul_q8_fused2_t` (the fused2_b8 tier) measured FLAT-TO-NEGATIVE
1904 // at the serving tick — bench c=8 213.1/213.8, 213.9/214.4, 214.4/213.5
1905 // (sign flips) and serve c=8 paired mean −0.20% over 3 passes. Mechanism:
1906 // unlike m=1 (where the pair is 128 of 1015 launches in a 7.67%-gap tick),
1907 // the c=8 tick is 73.2% one weight-bound kernel class with launch cost
1908 // already hidden — halving 128 launches of ~28k buys nothing. The m=1 arm
1909 // in `matmul_pre_dual_noscale` (+0.94%) stays; this call site keeps the two
1910 // launches. Kernel + fused2_b8 wrapper retained: kernel-check gates it at
1911 // m=5/8 and matmul_q8_fused2_t serves the verify tier. Receipts:
1912 // research/q27-deepdive-20260805/ (lever3-bench-*, serve-points.jsonl).
1913 let g = e.matmul_pre(ffn_gate, &zq, &zd, &z, b_n)?;
1914 let u = e.matmul_pre(ffn_up, &zq, &zd, &z, b_n)?;
1915 let mut act = e.uninit(b_n * n_ff)?;
1916 e.silu_mul(&g, &u, &mut act, b_n * n_ff)?;
1917 let (aq, ad) = e.quantize_q8_1(&act, b_n, n_ff)?;
1918 e.matmul_pre(ffn_down, &aq, &ad, &act, b_n)?
1919 }
1920 crate::hybrid::Ffn::Moe(m) => {
1921 self.moe_ffn_il_zq8(e, m, &z, None, b_n, il as u16)?
1922 }
1923 };
1924 // next-layer input x = x1 + ffn_out (batched element-wise add)
1925 let mut x2 = e.uninit(b_n * n_embd)?;
1926 e.add(&x1, &ffn_out, &mut x2, b_n * n_embd)?;
1927 x = x2;
1928 ph_mark(e, 9, ph_last)?;
1929 }
1930 Ok(x)
1931 }
1932
1933 /// Rollback seam for the step35 batched decode arm (lane/step35-batched-decode,
1934 /// 2026-08-08). Default ON; `MEMRA_STEP35_BATCH=0` caps serving at B=1 and makes the
1935 /// batched bodies return Err. Since lane/cx-b1fix, PP-N also refuses the eager B=1
1936 /// numeric class, so the seam disables PP-N Step35 decode rather than serving unstable
1937 /// bytes. Also the b2geo35 gate's CANARY seam — the live assertions must fail under it.
1938 pub fn step35_batch_on() -> bool {
1939 static ON: std::sync::OnceLock<bool> = std::sync::OnceLock::new();
1940 *ON.get_or_init(|| std::env::var("MEMRA_STEP35_BATCH").as_deref() != Ok("0"))
1941 }
1942
1943 /// THE step35 BATCHED LAYER WALK (lane/step35-batched-decode, 2026-08-08): B sequences
1944 /// share one pass over layers `[lo, hi)` with the REAL step35 geometry — the arm that
1945 /// kills the B=1 pin (34 tok/s aggregate FLAT across c=1..8, round-robin serialized;
1946 /// research/step-sku-20260807 §4) without re-opening the b2ab garbage hole (the generic
1947 /// `decode_batch_layers` ran uniform n_head/full-width rope/no window/no gate over
1948 /// step35 weights and returned HTTP-200 garbage at c>1).
1949 ///
1950 /// SHAPE — batched where the weights are, per-session where the state is:
1951 /// * attn_norm + quantize + wq/wk/wv/attn_gate projections + q/k norms + rope + head
1952 /// gate + wo + residual/post-norm + FFN all run at m=B: ONE weight stream serves B
1953 /// rows (decode is weight-BW-bound; this is the entire win).
1954 /// * KV append + fa_decode stay a per-session loop — the SWA window makes each
1955 /// session's KV view a function of ITS OWN `kvl.len` (`off = len-win` when past the
1956 /// window), and the z-batched seqs kernels take one shared t_kv/rung, not per-row
1957 /// offsets. This is the same shape as `decode_batch_layers`' per-seq fallback arm,
1958 /// and it costs launches, not weight bandwidth (KV is per-session state either way).
1959 ///
1960 /// PER-LAYER GEOMETRY (the five mechanisms that make the generic body wrong here, all
1961 /// from `step35_geom`/cfg): n_head 64 full / 96 SWA (wq/wo/attn_gate widths per layer),
1962 /// partial rope (n_rot 64 full / 128 SWA), dual base (5e6/1e4) + `rope_freqs` factors
1963 /// on FULL layers only, SWA window 512 with per-SESSION view offsets, and the separate
1964 /// head-wise `attn_gate` (one pre-sigmoid scalar per (token, head), input = the
1965 /// post-attn_norm hidden, applied before wo).
1966 ///
1967 /// EXACTNESS (the isolation contract, decode-batch-gate gate2's bar): every kernel here
1968 /// is row-independent at m=B or per-session:
1969 /// * `rms_norm`/`add_rms_norm`/`quantize_q8_1`/`attn_head_gate`/activations: per-row
1970 /// programs, grid over rows — row bi's bytes are the 1-row call's bytes.
1971 /// * projections via `matmul_pre` at m=2..8: Q8_0/Q6_K-class rides the b2/b4/b8
1972 /// batched-mmvq tier (bit-identical per (token,row) to m=1 mmvq); IQ4_XS — this
1973 /// SKU's trunk class — has no mmvq/batched kernel, so BOTH m=1 decode and the m=B
1974 /// walk ride `qmatvec_iq4_XS_dp4a` (grid (out_f, m): each column IS the m=1 dp4a
1975 /// program). Same class at every width = the decode-parity law by construction.
1976 /// * `rope_neox2` takes per-row positions (tok = row / n_heads) — row bi rotates at
1977 /// ITS pos with the layer's (n_rot, base, ff), same bits as its solo call.
1978 /// * per-session append/fa_decode_kvmod: literally the eager arm's calls on that
1979 /// session's own cache and views.
1980 /// * MoE (`moe_ffn_il_zq8` at t=B): the router is per-column decode-exact at
1981 /// t < PRIME_MIN_T (m=1 program per column), sigmoid routing + expert dispatch are
1982 /// per-token — a session's experts are a function of its own row only.
1983 /// The known eager-vs-batched FP gap is why PP-N Step35 deliberately serves THIS walk at
1984 /// B=1 too: the scheduler can change width during a session, so one numeric class must
1985 /// cover every live width. `b2geo35` pins static widths and an explicit B=1 -> B>1
1986 /// transition under live defaults.
1987 ///
1988 /// STAGE-SCOPED FROM BIRTH: `[lo, hi)` + caller-supplied engine/pos_d, so
1989 /// `decode_step_batch_ppn` calls it per stage (per-stage engine, per-stage pos_d, the
1990 /// #87 entry fence and boundary slots unchanged) — the pp2-batch seam lesson.
1991 #[allow(clippy::too_many_arguments)]
1992 pub(crate) fn step35_decode_batch_layers(
1993 &self,
1994 e: &Engine,
1995 x: CudaSlice<f32>,
1996 caches: &mut [&mut Cache],
1997 pos_d: &CudaSlice<i32>,
1998 lo: usize,
1999 hi: usize,
2000 ph_last: &mut std::time::Instant,
2001 ) -> Result<CudaSlice<f32>, Box<dyn std::error::Error>> {
2002 self.step35_decode_rows_layers(e, x, caches, pos_d, None, lo, hi, ph_last)
2003 }
2004
2005 /// Diagnostic generalization of the serving walk: `row_to_cache[r]` names the session
2006 /// whose KV row is consumed by hidden row `r`. Serving passes `None`, preserving the
2007 /// identity mapping and its launch sequence. The MoESD harness passes B groups of gamma
2008 /// consecutive rows so each session's verify columns append causally while projections and
2009 /// MoE dispatch see the full B*gamma target width.
2010 #[allow(clippy::too_many_arguments)]
2011 fn step35_decode_rows_layers(
2012 &self,
2013 e: &Engine,
2014 mut x: CudaSlice<f32>,
2015 caches: &mut [&mut Cache],
2016 pos_d: &CudaSlice<i32>,
2017 row_to_cache: Option<&[usize]>,
2018 lo: usize,
2019 hi: usize,
2020 ph_last: &mut std::time::Instant,
2021 ) -> Result<CudaSlice<f32>, Box<dyn std::error::Error>> {
2022 let b_n = row_to_cache.map_or(caches.len(), |rows| rows.len());
2023 let cfg = &self.cfg;
2024 let n_embd = cfg.n_embd as usize;
2025 let eps = cfg.rms_eps;
2026 cfg.step35
2027 .as_ref()
2028 .ok_or("step35_decode_batch_layers requires step35 cfg")?;
2029 if b_n == 0 || x.len() != b_n * n_embd || pos_d.len() != b_n {
2030 return Err(format!(
2031 "step35 row mapping shape mismatch: rows={b_n} x={} pos={} n_embd={n_embd}",
2032 x.len(),
2033 pos_d.len(),
2034 )
2035 .into());
2036 }
2037 if row_to_cache.is_some_and(|rows| rows.iter().any(|&ci| ci >= caches.len())) {
2038 return Err("step35 row mapping names a missing cache".into());
2039 }
2040 let cache_index = |row: usize| row_to_cache.map_or(row, |rows| rows[row]);
2041 // b2geo35 gate evidence: one line, first B>1 walk only (grep-stable prefix).
2042 if b_n > 1 {
2043 static ONCE: std::sync::Once = std::sync::Once::new();
2044 ONCE.call_once(|| {
2045 eprintln!(
2046 "[step35-batch] first B>1 batched step35 walk: B={b_n} layers=[{lo},{hi})"
2047 );
2048 });
2049 }
2050
2051 for il in lo..hi {
2052 let layer = &self.layers[il];
2053 let Mixer::Full(fa) = &layer.mixer else {
2054 return Err(format!("step35 layer {il} is not full-attn — corrupt config").into());
2055 };
2056 let geometry = self.step35_geom(il);
2057 let hd = geometry.head_dim_k as usize;
2058 let nkv = geometry.n_head_kv as usize;
2059 let nh = geometry.n_head as usize;
2060 let rbase = geometry.rope_base;
2061 let scale = geometry.attention_scale();
2062 let swa = geometry.window.is_some();
2063 let win = geometry.window.unwrap_or(0) as usize;
2064 let n_rot = geometry.n_rot as usize;
2065 let q_dim = nh * hd;
2066 let kv_dim = nkv * hd;
2067
2068 // ---- attn_norm + q8_1 quantize, batched (B rows) ----
2069 let anorm = layer.attn_norm.float_data();
2070 let mut xn = e.uninit(b_n * n_embd)?;
2071 e.rms_norm(&x, anorm, &mut xn, n_embd, b_n, eps)?;
2072 let (hq, hdq) = e.quantize_q8_1(&xn, b_n, n_embd)?;
2073
2074 // ---- batched projections: q/k/v + the separate head-wise gate (one weight
2075 // stream for B rows; xn is the live f32 fallback for non-q8_1-fast classes) ----
2076 let q0 = e.matmul_pre(&fa.wq, &hq, &hdq, &xn, b_n)?;
2077 let k0 = e.matmul_pre(&fa.wk, &hq, &hdq, &xn, b_n)?;
2078 let v0 = e.matmul_pre(&fa.wv, &hq, &hdq, &xn, b_n)?;
2079 let gw = fa
2080 .attn_gate
2081 .as_ref()
2082 .ok_or("step35 layer is missing attn_gate.weight (head-wise attention gate)")?;
2083 // gate input = the post-attn_norm hidden (upstream `cur`) — same xn/q8 pair.
2084 let gt = e.matmul_pre(gw, &hq, &hdq, &xn, b_n)?;
2085
2086 // ---- q/k RMSNorm over head_dim rows + the per-layer PARTIAL rope ----
2087 let mut q = e.uninit(b_n * q_dim)?;
2088 e.rms_norm(&q0, fa.q_norm.float_data(), &mut q, hd, b_n * nh, eps)?;
2089 let mut k = e.uninit(b_n * kv_dim)?;
2090 e.rms_norm(&k0, fa.k_norm.float_data(), &mut k, hd, b_n * nkv, eps)?;
2091 let ff = if geometry.rope_factors {
2092 self.step35_aux.as_ref().and_then(|a| a.rope_freqs(e))
2093 } else {
2094 None
2095 };
2096 e.rope_neox2(
2097 &mut q, &mut k, pos_d, hd, n_rot, nh, nkv, b_n, rbase, 1.0, ff,
2098 )?;
2099 ph_mark(e, 1, ph_last)?;
2100
2101 // ---- per-session: KV append + windowed/global fa_decode (each session's OWN
2102 // len drives its view offset — the iso-gap law, no cross-session term) ----
2103 let mut attn = e.uninit(b_n * q_dim)?;
2104 if b_n == 1 {
2105 // B=1 SPECIALIZED ENTRY (lane/cx-eagerpar): the general row loop below
2106 // materializes q_row and a_row because a B>1 FA call consumes/produces one
2107 // contiguous row at a time. At B=1, q and attn already ARE those whole rows.
2108 // Pass them directly to the same fa_decode_kvmod call: this removes two
2109 // arithmetic-free D2D copies (90 launches/token on Step3.7's 45 layers)
2110 // without changing any arithmetic kernel, shape, argument value, or order.
2111 // Keep the B>1 body verbatim below; b1fix's one-class/transition gates are
2112 // the promotion bar, not an FP-similarity tolerance.
2113 let kvl = caches[cache_index(0)].kv[il].as_mut().unwrap();
2114 let k_row = k.slice(0..kv_dim);
2115 let v_row = v0.slice(0..kv_dim);
2116 let next_len = kvl.len + 1;
2117 let (off, t_kv) = if swa && next_len > win {
2118 (next_len - win, win)
2119 } else {
2120 (0, next_len)
2121 };
2122 let write_row = e.prepare_kv_append(kvl, off & !31usize, 1)?;
2123 e.append_kv_quantized_view(
2124 &k_row,
2125 &v_row,
2126 &mut kvl.k,
2127 &mut kvl.v,
2128 write_row,
2129 kvl.kv_dim_k,
2130 kvl.kv_dim_v,
2131 kvl.k_tok_bytes,
2132 kvl.v_tok_bytes,
2133 Engine::kv_fp8_on(),
2134 )?;
2135 kvl.len = next_len;
2136 ph_mark(e, 2, ph_last)?;
2137 let physical = kvl.physical_rows(off, off + t_kv)?;
2138 let k_view = e.view_u8_range(
2139 &kvl.k,
2140 physical.start * kvl.k_tok_bytes,
2141 physical.end * kvl.k_tok_bytes,
2142 );
2143 let v_view = e.view_u8_range(
2144 &kvl.v,
2145 physical.start * kvl.v_tok_bytes,
2146 physical.end * kvl.v_tok_bytes,
2147 );
2148 e.fa_decode_kvmod(
2149 &q,
2150 &k_view,
2151 &v_view,
2152 &mut attn,
2153 hd,
2154 nh,
2155 nkv,
2156 t_kv,
2157 scale,
2158 kvl.k_tok_bytes,
2159 kvl.v_tok_bytes,
2160 Engine::kv_fp8_on(),
2161 )?;
2162 ph_mark(e, 4, ph_last)?;
2163 } else {
2164 for bi in 0..b_n {
2165 let cache = &mut caches[cache_index(bi)];
2166 let kvl = cache.kv[il].as_mut().unwrap();
2167 let k_row = k.slice(bi * kv_dim..(bi + 1) * kv_dim);
2168 let v_row = v0.slice(bi * kv_dim..(bi + 1) * kv_dim);
2169 let next_len = kvl.len + 1;
2170 let (off, t_kv) = if swa && next_len > win {
2171 (next_len - win, win)
2172 } else {
2173 (0, next_len)
2174 };
2175 let write_row = e.prepare_kv_append(kvl, off & !31usize, 1)?;
2176 e.append_kv_quantized_view(
2177 &k_row,
2178 &v_row,
2179 &mut kvl.k,
2180 &mut kvl.v,
2181 write_row,
2182 kvl.kv_dim_k,
2183 kvl.kv_dim_v,
2184 kvl.k_tok_bytes,
2185 kvl.v_tok_bytes,
2186 Engine::kv_fp8_on(),
2187 )?;
2188 kvl.len = next_len;
2189 ph_mark(e, 2, ph_last)?;
2190 // the eager arm's SWA view arithmetic, verbatim (step35_decode_attn):
2191 // token-aligned offset, keys carry absolute rope, mask is positional.
2192 let physical = kvl.physical_rows(off, off + t_kv)?;
2193 let k_view = e.view_u8_range(
2194 &kvl.k,
2195 physical.start * kvl.k_tok_bytes,
2196 physical.end * kvl.k_tok_bytes,
2197 );
2198 let v_view = e.view_u8_range(
2199 &kvl.v,
2200 physical.start * kvl.v_tok_bytes,
2201 physical.end * kvl.v_tok_bytes,
2202 );
2203 // The per-session cache view remains authoritative (including SWA's
2204 // physical-row rebase), while Q/O use their existing packed row views.
2205 // This preserves the exact FA program and removes only the two D2D copies.
2206 let q_row = q.slice(bi * q_dim..(bi + 1) * q_dim);
2207 let mut a_row = attn.slice_mut(bi * q_dim..(bi + 1) * q_dim);
2208 e.fa_decode_kvmod_view(
2209 &q_row,
2210 &k_view,
2211 &v_view,
2212 &mut a_row,
2213 hd,
2214 nh,
2215 nkv,
2216 t_kv,
2217 scale,
2218 kvl.k_tok_bytes,
2219 kvl.v_tok_bytes,
2220 Engine::kv_fp8_on(),
2221 )?;
2222 ph_mark(e, 4, ph_last)?;
2223 }
2224 }
2225
2226 // ---- head-wise gate (one sigmoid per (token, head), pre-wo) + o-proj at m=B ----
2227 let mut ag = e.uninit(b_n * q_dim)?;
2228 e.attn_head_gate(&attn, >, &mut ag, None, hd, nh, b_n)?;
2229 let mixed = e.matmul(&fa.wo, &ag, b_n)?;
2230 ph_mark(e, 5, ph_last)?;
2231
2232 // ---- residual add + post_attn_norm + FFN, batched ----
2233 let pnorm = layer.post_attn_norm.float_data();
2234 let mut x1 = e.uninit(b_n * n_embd)?;
2235 let mut z = e.uninit(b_n * n_embd)?;
2236 e.add_rms_norm(&x, &mixed, pnorm, &mut x1, &mut z, n_embd, b_n, eps)?;
2237 let ffn_out = match &layer.ffn {
2238 crate::hybrid::Ffn::Dense {
2239 ffn_gate,
2240 ffn_up,
2241 ffn_down,
2242 } => {
2243 // A dense step35 FFN's clamp is the SHEXP array (upstream's one
2244 // build_ffn serves dense + shared expert, llama-graph.cpp:1751);
2245 // ffn_act_lim dispatches clamped/plain per layer. Layers 0-2 (the
2246 // leading dense) have no live limit on this artifact, but the route
2247 // is correct by construction, not by artifact.
2248 let n_ff = ffn_gate.out_features();
2249 let (zq, zd) = e.quantize_q8_1(&z, b_n, n_embd)?;
2250 let g = e.matmul_pre(ffn_gate, &zq, &zd, &z, b_n)?;
2251 let u = e.matmul_pre(ffn_up, &zq, &zd, &z, b_n)?;
2252 let mut act = e.uninit(b_n * n_ff)?;
2253 Self::ffn_act_lim(
2254 e,
2255 cfg,
2256 &g,
2257 &u,
2258 1.0,
2259 1.0,
2260 cfg.clamp_shexp_at(il as u32),
2261 &mut act,
2262 b_n * n_ff,
2263 )?;
2264 let (aq, ad) = e.quantize_q8_1(&act, b_n, n_ff)?;
2265 e.matmul_pre(ffn_down, &aq, &ad, &act, b_n)?
2266 }
2267 // t=B < PRIME_MIN_T: per-column decode-exact router + host sigmoid routing
2268 // + per-token expert dispatch — the same per-token program as eager t=1,
2269 // including the per-layer SwiGLU clamp (43/44) via the sequential path's
2270 // ffn_act_lim. The sigmoid-router deny on dev/pairs holds by predicate.
2271 crate::hybrid::Ffn::Moe(m) => {
2272 self.moe_ffn_il_zq8(e, m, &z, None, b_n, il as u16)?
2273 }
2274 };
2275 let mut x2 = e.uninit(b_n * n_embd)?;
2276 e.add(&x1, &ffn_out, &mut x2, b_n * n_embd)?;
2277 x = x2;
2278 ph_mark(e, 9, ph_last)?;
2279 }
2280 Ok(x)
2281 }
2282
2283 /// Standalone MoESD target forward. This entrypoint is not used by serving: it widens the
2284 /// existing Step-3.7 batched layer walk to B*gamma rows while preserving one causal KV chain
2285 /// per session. It returns device logits and performs no sampling or logits D2H, matching the
2286 /// target-model term T_T measured by the paper.
2287 pub fn moesd_target_forward(
2288 &self,
2289 e: &Engine,
2290 tokens: &[u32],
2291 batch: usize,
2292 gamma: usize,
2293 caches: &mut [&mut Cache],
2294 ) -> Result<CudaSlice<f32>, Box<dyn std::error::Error>> {
2295 if self.cfg.step35.is_none() {
2296 return Err("MoESD target forward currently requires Step-3.7/Step35 geometry".into());
2297 }
2298 if batch == 0 || gamma == 0 || caches.len() != batch || tokens.len() != batch * gamma {
2299 return Err(format!(
2300 "MoESD shape mismatch: B={batch} gamma={gamma} caches={} tokens={}",
2301 caches.len(),
2302 tokens.len(),
2303 )
2304 .into());
2305 }
2306 let rows = batch * gamma;
2307 if rows > 256 {
2308 return Err(format!("MoESD target width {rows} exceeds the frozen 32*8 matrix").into());
2309 }
2310 let n_embd = self.cfg.n_embd as usize;
2311 let eps = self.cfg.rms_eps;
2312 let payload = rows * n_embd;
2313 let row_to_cache: Vec<usize> = (0..batch)
2314 .flat_map(|session| (0..gamma).map(move |_| session))
2315 .collect();
2316 let positions: Vec<i32> = row_to_cache
2317 .iter()
2318 .enumerate()
2319 .map(|(row, &session)| (caches[session].pos + row % gamma) as i32)
2320 .collect();
2321 let mut ph_last = std::time::Instant::now();
2322
2323 let logits = if let Some(fence) = crate::pp::pp_cuts(self.layers.len()) {
2324 if fence.len() != 3 || crate::pp::pp2_streams_off() {
2325 return Err(
2326 "MoESD PP target forward requires the live two-stage stream split".into(),
2327 );
2328 }
2329 let rt = crate::pp::PpNRt::get(e)?;
2330 if rt.n_stages() != 2 {
2331 return Err(format!("MoESD expected two PP stages, got {}", rt.n_stages()).into());
2332 }
2333 let caller_stream = e.stream();
2334 rt.fence_stages_behind(&caller_stream)?;
2335 let slot = {
2336 let _st0 = rt.enter(0);
2337 let e0 = rt.engine(0, e);
2338 let pos_d = e0.htod_i32(&positions)?;
2339 let x = e0.htod(&self.embd.gather(n_embd, tokens))?;
2340 ph_mark(e0, 0, &mut ph_last)?;
2341 let x = self.step35_decode_rows_layers(
2342 e0,
2343 x,
2344 caches,
2345 &pos_d,
2346 Some(&row_to_cache),
2347 fence[0],
2348 fence[1],
2349 &mut ph_last,
2350 )?;
2351 rt.tx(0, &x, payload)?
2352 };
2353 let logits = {
2354 let _st1 = rt.enter(1);
2355 let e1 = rt.engine(1, e);
2356 let pos_d = e1.htod_i32(&positions)?;
2357 let x = rt.rx(0, slot, payload)?;
2358 let x = self.step35_decode_rows_layers(
2359 e1,
2360 x,
2361 caches,
2362 &pos_d,
2363 Some(&row_to_cache),
2364 fence[1],
2365 fence[2],
2366 &mut ph_last,
2367 )?;
2368 let mut hn = e1.uninit(payload)?;
2369 e1.rms_norm(
2370 &x,
2371 self.output_norm.float_data(),
2372 &mut hn,
2373 n_embd,
2374 rows,
2375 eps,
2376 )?;
2377 let logits = e1.matmul(&self.output, &hn, rows)?;
2378 rt.publish_to(1, &caller_stream)?;
2379 logits
2380 };
2381 logits
2382 } else {
2383 let pos_d = e.htod_i32(&positions)?;
2384 let x = e.htod(&self.embd.gather(n_embd, tokens))?;
2385 ph_mark(e, 0, &mut ph_last)?;
2386 let x = self.step35_decode_rows_layers(
2387 e,
2388 x,
2389 caches,
2390 &pos_d,
2391 Some(&row_to_cache),
2392 0,
2393 self.layers.len(),
2394 &mut ph_last,
2395 )?;
2396 let mut hn = e.uninit(payload)?;
2397 e.rms_norm(
2398 &x,
2399 self.output_norm.float_data(),
2400 &mut hn,
2401 n_embd,
2402 rows,
2403 eps,
2404 )?;
2405 e.matmul(&self.output, &hn, rows)?
2406 };
2407 for cache in caches.iter_mut() {
2408 cache.pos += gamma;
2409 }
2410 Ok(logits)
2411 }
2412
2413 /// The batched tick's TAIL, after the trunk: grammar masks -> device sampling -> lean
2414 /// logits park -> `pos` bump. Split out with the pp seam (`decode_batch_layers`) because
2415 /// under a stage split this runs on the LAST stage's engine and device — the lm_head, the
2416 /// masks, the sampler, and `cache.last_logits_dev` all live where the final residual
2417 /// lands, and the caller must be able to place them there without duplicating 90 lines of
2418 /// serving contract. `logits` is `[b_n, n_vocab]` already computed by the caller (the
2419 /// output_norm + lm_head pair stays at the call site so a stage split can fence around
2420 /// it); everything after it is here, verbatim.
2421 #[allow(clippy::too_many_arguments)]
2422 fn decode_batch_epilogue(
2423 &self,
2424 e: &Engine,
2425 caches: &mut [&mut Cache],
2426 samp: &[Option<DevSamp>],
2427 masks: &[Option<(&CudaSlice<u32>, usize)>],
2428 lean: bool,
2429 logits: CudaSlice<f32>,
2430 b_n: usize,
2431 ph_last: &mut std::time::Instant,
2432 ) -> Result<(Vec<Vec<f32>>, Vec<Option<u32>>), Box<dyn std::error::Error>> {
2433 // GRAMMAR MASKS (constrained decoding): preserve each masked row's PRISTINE logits
2434 // for its consumer (lean park into cache.last_logits_dev — the reuse-pool park stays
2435 // unmasked, the v1 contract — or the non-lean D2H), then ban in place BEFORE the
2436 // device sampler reads the row. All stream-ordered; masks=&[] takes no new branch.
2437 let n_vocab = self.output.out_features();
2438 let mut logits = logits;
2439 let mut pristine: Vec<Option<CudaSlice<f32>>> = Vec::new();
2440 if masks.iter().take(b_n).any(|m| m.is_some()) {
2441 pristine.resize_with(b_n, || None);
2442 for (bi, m) in masks.iter().take(b_n).enumerate() {
2443 let Some((mask, words)) = m else { continue };
2444 assert!(
2445 samp.get(bi).copied().flatten().is_some(),
2446 "grammar-masked row {bi} must request a device sample"
2447 );
2448 if lean {
2449 let cache = &mut caches[bi];
2450 if cache
2451 .last_logits_dev
2452 .as_ref()
2453 .map(|d| d.len() < n_vocab)
2454 .unwrap_or(true)
2455 {
2456 cache.last_logits_dev = Some(e.uninit(n_vocab)?);
2457 }
2458 let dst = cache.last_logits_dev.as_mut().unwrap();
2459 e.dtod_copy_view(&logits.slice(bi * n_vocab..(bi + 1) * n_vocab), dst)?;
2460 } else {
2461 let mut p = e.uninit(n_vocab)?;
2462 e.dtod_copy_view(&logits.slice(bi * n_vocab..(bi + 1) * n_vocab), &mut p)?;
2463 pristine[bi] = Some(p);
2464 }
2465 e.mask_logits_col(&mut logits, mask, bi, n_vocab, *words)?;
2466 }
2467 }
2468
2469 // Device-side sampling for requested rows (see the method doc). Enqueued before the
2470 // big logits D2H so the tiny [B] token readback rides the same sync.
2471 let mut next: Vec<Option<u32>> = vec![None; b_n];
2472 if samp.iter().take(b_n).any(|s| s.is_some()) {
2473 let mut toks = e.alloc_u32_zeroed(b_n)?;
2474 let mut perturb: Option<CudaSlice<f32>> = None;
2475 for (bi, s) in samp.iter().take(b_n).enumerate() {
2476 let Some((temp, seed, ctr, top_k, top_p, min_p)) = s else {
2477 continue;
2478 };
2479 let filtered = *temp > 0.0 && (*top_k > 0 || *top_p < 1.0 || *min_p > 0.0);
2480 if *temp <= 0.0 {
2481 e.argmax_token_device_col(&logits, bi, n_vocab, &mut toks, bi)?;
2482 } else if filtered {
2483 if perturb.is_none() {
2484 perturb = Some(e.zeros(n_vocab)?);
2485 }
2486 let pb = perturb.as_mut().unwrap();
2487 self.devsample_filtered_col(
2488 e, &logits, bi, n_vocab, *temp, *seed, *ctr, *top_k, *top_p, *min_p, pb,
2489 &mut toks, bi,
2490 )?;
2491 } else {
2492 if perturb.is_none() {
2493 perturb = Some(e.zeros(n_vocab)?);
2494 }
2495 let pb = perturb.as_mut().unwrap();
2496 e.gumbel_perturb_col(&logits, bi, pb, n_vocab, *seed, *ctr, *temp)?;
2497 e.argmax_token_device_col(pb, 0, n_vocab, &mut toks, bi)?;
2498 }
2499 }
2500 let host_toks = e.dtoh_u32(&toks)?;
2501 for (bi, s) in samp.iter().take(b_n).enumerate() {
2502 if s.is_some() {
2503 next[bi] = Some(host_toks[bi]);
2504 }
2505 }
2506 }
2507
2508 let lean_any = lean && samp.iter().take(b_n).any(|s| s.is_some());
2509 let rows: Vec<Vec<f32>> = if lean_any {
2510 // LEAN: park device-sampled rows on-device (per-cache buffer, dtod); D2H only
2511 // the rows that still need host logits. No sampled rows + no fallback rows =
2512 // the big D2H disappears (the [B] token readback above already synced).
2513 for (bi, s) in samp.iter().take(b_n).enumerate() {
2514 if s.is_none() {
2515 continue;
2516 }
2517 // grammar-masked rows already parked their PRISTINE copy above — the
2518 // in-place ban has since poisoned this row for the reuse-pool consumer.
2519 if masks.get(bi).copied().flatten().is_some() {
2520 continue;
2521 }
2522 let cache = &mut caches[bi];
2523 if cache
2524 .last_logits_dev
2525 .as_ref()
2526 .map(|d| d.len() < n_vocab)
2527 .unwrap_or(true)
2528 {
2529 cache.last_logits_dev = Some(e.uninit(n_vocab)?);
2530 }
2531 let dst = cache.last_logits_dev.as_mut().unwrap();
2532 e.dtod_copy_view(&logits.slice(bi * n_vocab..(bi + 1) * n_vocab), dst)?;
2533 }
2534 (0..b_n)
2535 .map(|bi| {
2536 if samp.get(bi).copied().flatten().is_some() {
2537 Ok(Vec::new())
2538 } else {
2539 e.dtoh_view(&logits.slice(bi * n_vocab..(bi + 1) * n_vocab))
2540 }
2541 })
2542 .collect::<Result<_, _>>()?
2543 } else {
2544 let host = e.dtoh(&logits)?;
2545 (0..b_n)
2546 .map(|bi| {
2547 // grammar-masked non-lean rows return the PRISTINE copy (the in-place ban
2548 // must never leak into last_logits — reuse-pool/park semantics unchanged).
2549 if let Some(p) = pristine.get(bi).and_then(|p| p.as_ref()) {
2550 return e.dtoh(p);
2551 }
2552 Ok(host[bi * n_vocab..(bi + 1) * n_vocab].to_vec())
2553 })
2554 .collect::<Result<_, _>>()?
2555 };
2556 for c in caches.iter_mut() {
2557 c.pos += 1;
2558 }
2559 ph_mark(e, 11, ph_last)?;
2560 Ok((rows, next))
2561 }
2562}
2563
2564fn b1_fast_arch_eligible(arch: &Arch) -> bool {
2565 // The whole qwen35 family is excluded, not just MoE: spec verify for these archs runs
2566 // the generic batched numeric class (spec.rs qwen35_serving_class), so live B=1 serving
2567 // must stay in that same class. B1FAST's eager program would reopen the near-tie-flip
2568 // divergence the 2026-08-14 exactness fix closed (1 ULP at layer 2 -> 2.3e-1 head
2569 // maxdiff, amplified by the GDN recurrence).
2570 !matches!(arch, Arch::Qwen35 | Arch::Qwen35Moe)
2571}
2572
2573fn b1_fast_env_on(value: Option<&str>) -> bool {
2574 value == Some("1")
2575}
2576
2577#[cfg(test)]
2578mod tests {
2579 use super::{b1_fast_arch_eligible, b1_fast_env_on};
2580 use memra_gguf::config::Arch;
2581
2582 #[test]
2583 fn qwen35_family_stays_in_one_decode_numeric_class_across_widths() {
2584 assert!(!b1_fast_arch_eligible(&Arch::Qwen35Moe));
2585 assert!(!b1_fast_arch_eligible(&Arch::Qwen35));
2586 assert!(b1_fast_arch_eligible(&Arch::Qwen3Moe));
2587 }
2588
2589 #[test]
2590 fn b1_eager_program_requires_explicit_opt_in() {
2591 assert!(!b1_fast_env_on(None));
2592 assert!(!b1_fast_env_on(Some("0")));
2593 assert!(!b1_fast_env_on(Some("true")));
2594 assert!(b1_fast_env_on(Some("1")));
2595 }
2596}