memra_engine/gemma_spec.rs
1//! gemma4 MTP spec-decode: the "gemma4-assistant" drafter (4-layer, Q-only attention over the
2//! MAIN model's KV cache — no draft KV, no trims) + the greedy draft/verify loop.
3//!
4//! Wiring verified from llama gemma4-assistant.cpp + llama-model.cpp:2162 (HANDOVER "GEMMA4 MTP
5//! DRAFTER — VERIFIED WIRING"): per draft token, x = MAIN tok_embd(token) * sqrt(2816);
6//! xh = concat(x, h[2816]) -> pre_proj [5632->1024]; 4 gemma-style blocks whose attention
7//! projects Q ONLY and attends the main cache (SWA layers 0..2 -> main layer n-2 = 28 windowed;
8//! global layer 3 -> main layer n-1 = 29 full); dense GELU_PAR ffn; final output_norm ->
9//! TIED 1024-dim head (no softcap); h_next = post_proj [1024->2816].
10
11use crate::Engine;
12use crate::cache::Cache;
13use crate::hybrid::HybridModel;
14use crate::model::GpuTensor;
15use cudarc::driver::CudaSlice;
16use memra_gguf::GgufFile;
17use memra_gguf::source::{GgufSource, TensorSource};
18
19pub struct GemmaDraftLayer {
20 pub attn_norm: GpuTensor,
21 pub wq: GpuTensor,
22 pub wo: GpuTensor,
23 pub q_norm: GpuTensor,
24 pub post_attn_norm: GpuTensor,
25 pub ffn_norm: GpuTensor,
26 pub ffn_gate: GpuTensor,
27 pub ffn_up: GpuTensor,
28 pub ffn_down: GpuTensor,
29 pub ffn_post_norm: GpuTensor,
30 pub out_scale: f32,
31 pub swa: bool,
32 pub hd: usize,
33 pub nh: usize,
34}
35
36pub struct GemmaDraft {
37 pub layers: Vec<GemmaDraftLayer>,
38 pub pre_proj: GpuTensor, // [5632 -> 1024]
39 pub post_proj: GpuTensor, // [1024 -> 2816]
40 pub output_norm: GpuTensor,
41 pub head: GpuTensor, // tied drafter token_embd [1024, n_vocab] (or FR-trimmed rows)
42 /// FR-Spec trim map: draft-row index -> target token id (None = full head, identity).
43 pub d2t: Option<Vec<u32>>,
44 /// Device copy of `d2t` — the async round translates each drafted trim-idx in place
45 /// (u32_map_k) before it seeds the next draft step or meets the verify argmax.
46 pub d2t_dev: Option<CudaSlice<u32>>,
47 /// Adaptive trim (coverage escapes are the entire trim cost — oracle-proven +2% on the
48 /// cell the static trim lost by 17%, jsonl 2026-07-19): spare head slots learned at
49 /// serve time from the prompt's own ids and verify-correction tokens.
50 pub trim_adapt: Option<TrimAdapt>,
51 pub rope_freqs: CudaSlice<f32>,
52 pub ones: CudaSlice<f32>, // weightless-norm weight (max hd 512)
53 pub n_embd: usize, // 1024
54 pub n_backbone: usize, // 2816
55 pub rope_base_global: f32,
56 pub rope_base_swa: f32,
57 pub sliding_window: usize,
58}
59
60/// Serve-time adaptive trim (MEMRA_GEMMA_TRIM_ADAPT=<spare slots>): the static FR trim's whole
61/// loss is coverage escapes — tokens the base emits that the trim can't propose (guaranteed
62/// rejections; the oracle control that injected the exact escapees flipped a -17% cell to +2%
63/// at identical acceptance, jsonl 2026-07-19). Every escape self-identifies at serve time: it
64/// arrives as a verify CORRECTION token (and its cousins ride in with the prompt), so the head
65/// keeps `n_spare` extra rows and learns them — prompt ids up front, corrections as they land.
66/// First miss pays one rejected round; every recurrence after is proposable. Rows are written
67/// into the existing device buffers (no realloc — captured graphs keep their baked addresses).
68pub struct TrimAdapt {
69 /// full-vocab head rows (host copy) — the gather source for learned rows.
70 src_rows: Vec<u8>,
71 row_bytes: usize,
72 n_vocab: usize,
73 /// trim-set membership by token id (ranked + learned).
74 present: Vec<bool>,
75 /// spare slots live at [spare_base, spare_base + n_spare) in the gathered head.
76 spare_base: usize,
77 n_spare: usize,
78 used: usize,
79 logged_full: bool,
80}
81
82impl TrimAdapt {
83 /// Add `tok`'s head row to the trim set if absent and a spare slot is free.
84 fn maybe_add(
85 &mut self,
86 e: &Engine,
87 tok: u32,
88 head: &mut GpuTensor,
89 d2t: &mut [u32],
90 d2t_dev: &mut CudaSlice<u32>,
91 ) -> Result<bool, Box<dyn std::error::Error>> {
92 let t = tok as usize;
93 if t >= self.n_vocab || self.present[t] {
94 return Ok(false);
95 }
96 if self.used == self.n_spare {
97 if !self.logged_full {
98 self.logged_full = true;
99 eprintln!(
100 "[trim-adapt] spare slots exhausted ({}) — later escapes stay unproposable",
101 self.n_spare
102 );
103 }
104 return Ok(false);
105 }
106 let slot = self.spare_base + self.used;
107 self.used += 1;
108 self.present[t] = true;
109 if let GpuTensor::Quant { bytes, .. } = head {
110 e.htod_u8_into(
111 bytes,
112 slot * self.row_bytes,
113 &self.src_rows[t * self.row_bytes..(t + 1) * self.row_bytes],
114 )?;
115 }
116 d2t[slot] = tok;
117 e.u32_set_k(d2t_dev, tok, slot)?;
118 Ok(true)
119 }
120}
121
122/// Union `toks` into the adaptive trim set (no-op when the draft has no adaptive state).
123/// Split-borrow helper: the fields move together or not at all.
124fn trim_adapt_learn(
125 e: &Engine,
126 d: &mut GemmaDraft,
127 toks: &[u32],
128) -> Result<(), Box<dyn std::error::Error>> {
129 let GemmaDraft {
130 trim_adapt,
131 head,
132 d2t,
133 d2t_dev,
134 ..
135 } = d;
136 let (Some(ta), Some(d2t), Some(d2t_dev)) =
137 (trim_adapt.as_mut(), d2t.as_mut(), d2t_dev.as_mut())
138 else {
139 return Ok(());
140 };
141 for &tok in toks {
142 ta.maybe_add(e, tok, head, d2t, d2t_dev)?;
143 }
144 Ok(())
145}
146
147impl GemmaDraft {
148 /// Adaptive-trim stats: (slots used, slot budget). None when adaptation is off.
149 pub fn trim_adapt_stats(&self) -> Option<(usize, usize)> {
150 self.trim_adapt.as_ref().map(|ta| (ta.used, ta.n_spare))
151 }
152
153 /// Persist the learned trim rows: append ids not yet in the sidecar to
154 /// `<ranks>.learned` (the load path pre-fills spare slots from it, so a distribution's
155 /// escapes pay their first-miss round ONCE across the serve lifetime, not per request).
156 pub fn trim_adapt_save(&self) -> std::io::Result<usize> {
157 let (Some(ta), Some(d2t), Some(path)) = (
158 self.trim_adapt.as_ref(),
159 self.d2t.as_ref(),
160 self.trim_learned_path(),
161 ) else {
162 return Ok(0);
163 };
164 let prior: std::collections::HashSet<u32> = std::fs::read_to_string(&path)
165 .map(|t| t.lines().filter_map(|l| l.trim().parse().ok()).collect())
166 .unwrap_or_default();
167 let fresh: Vec<u32> = d2t[ta.spare_base..ta.spare_base + ta.used]
168 .iter()
169 .copied()
170 .filter(|id| !prior.contains(id))
171 .collect();
172 if !fresh.is_empty() {
173 use std::io::Write;
174 let mut f = std::fs::OpenOptions::new()
175 .create(true)
176 .append(true)
177 .open(&path)?;
178 for id in &fresh {
179 writeln!(f, "{id}")?;
180 }
181 }
182 Ok(fresh.len())
183 }
184
185 fn trim_learned_path(&self) -> Option<String> {
186 std::env::var("MEMRA_GEMMA_DRAFT_RANKS")
187 .ok()
188 .map(|p| format!("{p}.learned"))
189 }
190}
191
192fn load_t(
193 e: &Engine,
194 src: &dyn TensorSource,
195 name: &str,
196) -> Result<GpuTensor, Box<dyn std::error::Error>> {
197 GpuTensor::load_from_source(e, src, name)
198}
199
200impl GemmaDraft {
201 pub fn load(e: &Engine, g: &GgufFile) -> Result<Self, Box<dyn std::error::Error>> {
202 // two published spellings of the same arch: the 26B/31B drafters ship
203 // "gemma4-assistant", the E4B assistant ships "gemma4_assistant" — the metadata
204 // key prefix follows the arch string verbatim.
205 let arch = match g.arch() {
206 Some(a @ ("gemma4-assistant" | "gemma4_assistant")) => a.to_string(),
207 other => panic!("not a gemma4-assistant drafter (arch {other:?})"),
208 };
209 let src = GgufSource(g);
210 let meta_u = |k: &str| -> u32 {
211 g.metadata
212 .get(&format!("{arch}.{k}"))
213 .and_then(|v| v.as_u64())
214 .unwrap_or(0) as u32
215 };
216 let meta_f = |k: &str, d: f32| -> f32 {
217 match g.metadata.get(&format!("{arch}.{k}")) {
218 Some(memra_gguf::MetaValue::F32(v)) => *v,
219 Some(memra_gguf::MetaValue::F64(v)) => *v as f32,
220 _ => d,
221 }
222 };
223 let n_layer = meta_u("block_count") as usize;
224 let n_embd = meta_u("embedding_length") as usize;
225 // 26B/31B carry the target width as embedding_length_out; the E4B assistant as
226 // n_embd_backbone.
227 let n_backbone = match meta_u("embedding_length_out") as usize {
228 0 => meta_u("n_embd_backbone") as usize,
229 v => v,
230 };
231 let hd_g = meta_u("attention.key_length") as usize;
232 let hd_s = meta_u("attention.key_length_swa") as usize;
233 let swa_pat: Vec<bool> = match g
234 .metadata
235 .get(&format!("{arch}.attention.sliding_window_pattern"))
236 {
237 Some(memra_gguf::MetaValue::Array(a)) => a
238 .iter()
239 .filter_map(|v| v.as_u64().map(|x| x != 0))
240 .collect(),
241 _ => return Err("drafter missing sliding_window_pattern".into()),
242 };
243
244 let mut layers = Vec::with_capacity(n_layer);
245 #[allow(clippy::needless_range_loop)]
246 // allow: the explicit index loop keeps the offset arithmetic visible and aligned with the device-side indexing
247 for il in 0..n_layer {
248 let p = |n: &str| format!("blk.{il}.{n}");
249 let swa = swa_pat[il];
250 let out_scale = {
251 let t = src
252 .find(&p("layer_output_scale.weight"))
253 .ok_or("missing layer_output_scale")?;
254 memra_gguf::dequant::dequantize(t.ggml_type, &t.bytes, 1)[0]
255 };
256 let hd = if swa { hd_s } else { hd_g };
257 let wq = load_t(e, &src, &p("attn_q.weight"))?;
258 // heads per layer from the projection shape (the E4B assistant keeps 4 heads on
259 // BOTH classes — hd differs — while 26B/31B are uniform; the shape is the truth).
260 let nh = wq.out_features() / hd;
261 layers.push(GemmaDraftLayer {
262 attn_norm: load_t(e, &src, &p("attn_norm.weight"))?,
263 wq,
264 wo: load_t(e, &src, &p("attn_output.weight"))?,
265 q_norm: load_t(e, &src, &p("attn_q_norm.weight"))?,
266 post_attn_norm: load_t(e, &src, &p("post_attention_norm.weight"))?,
267 ffn_norm: load_t(e, &src, &p("ffn_norm.weight"))?,
268 ffn_gate: load_t(e, &src, &p("ffn_gate.weight"))?,
269 ffn_up: load_t(e, &src, &p("ffn_up.weight"))?,
270 ffn_down: load_t(e, &src, &p("ffn_down.weight"))?,
271 ffn_post_norm: load_t(e, &src, &p("post_ffw_norm.weight"))?,
272 out_scale,
273 swa,
274 hd,
275 nh,
276 });
277 }
278 let rope_freqs = {
279 let t = src
280 .find("rope_freqs.weight")
281 .ok_or("drafter missing rope_freqs")?;
282 e.htod(&memra_gguf::dequant::dequantize(
283 t.ggml_type,
284 &t.bytes,
285 t.ne.iter().product::<u64>() as usize,
286 ))?
287 };
288 // FR-Spec head trim (MEMRA_GEMMA_DRAFT_RANKS=<ids file, rank order>): gather the ranked
289 // rows of the drafter head + d2t map. (Top-N-IDS truncation measured NEGATIVE — id
290 // order is not frequency; the CORPUS-ranked gather is the real FR-Spec.)
291 // MEMRA_GEMMA_TRIM_ADAPT=<n> (default 512 when ranks are set, 0 = off) appends n spare
292 // rows the serve loop fills from prompt ids + verify corrections (see TrimAdapt).
293 let (head, d2t, trim_adapt) = {
294 let t = src
295 .find("token_embd.weight")
296 .ok_or("drafter missing token_embd")?;
297 let in_f = t.ne[0] as usize;
298 let n_vocab = t.ne[1] as usize;
299 match std::env::var("MEMRA_GEMMA_DRAFT_RANKS").ok() {
300 Some(path) => {
301 // row gather is layout-agnostic given the per-row byte stride: Q4_0 (26B
302 // drafter) and Q8_0 (31B drafter) both ship 32-elem blocks row-major.
303 // (qtype, elems/block, bytes/block) — the gather is stride-agnostic.
304 let (qtype, blk_e, blk_b) = match t.ggml_type {
305 memra_gguf::GgmlType::Q4_0 => (crate::QT_Q4_0, 32, 18),
306 memra_gguf::GgmlType::Q8_0 => (crate::QT_Q8_0, 32, 34),
307 memra_gguf::GgmlType::Q6_K => (crate::QT_Q6_K, 256, 210),
308 other => panic!("drafter head trim: unsupported head type {other:?}"),
309 };
310 let ids: Vec<u32> = std::fs::read_to_string(&path)?
311 .lines()
312 .filter_map(|l| l.trim().parse().ok())
313 .filter(|&id| (id as usize) < n_vocab)
314 .collect();
315 let n_spare: usize = std::env::var("MEMRA_GEMMA_TRIM_ADAPT")
316 .ok()
317 .and_then(|v| v.parse().ok())
318 .unwrap_or(512);
319 let row_bytes = in_f / blk_e * blk_b;
320 let mut gathered = Vec::with_capacity((ids.len() + n_spare) * row_bytes);
321 for &id in &ids {
322 let off = id as usize * row_bytes;
323 gathered.extend_from_slice(&t.bytes[off..off + row_bytes]);
324 }
325 // spare slots start as copies of row ids[0] mapping to ids[0] — a real,
326 // already-present token, so however the argmax resolves the duplicate-
327 // logit tie, the d2t translation lands on the same token id.
328 for _ in 0..n_spare {
329 let off = ids[0] as usize * row_bytes;
330 gathered.extend_from_slice(&t.bytes[off..off + row_bytes]);
331 }
332 eprintln!(
333 "[gemma-draft] FR head trim: {} rows + {} adaptive ({} MB vs {} MB full)",
334 ids.len(),
335 n_spare,
336 (ids.len() + n_spare) * row_bytes / 1_000_000,
337 n_vocab * row_bytes / 1_000_000
338 );
339 let mut trim_adapt = (n_spare > 0).then(|| {
340 let mut present = vec![false; n_vocab];
341 for &id in &ids {
342 present[id as usize] = true;
343 }
344 TrimAdapt {
345 src_rows: t.bytes.to_vec(),
346 row_bytes,
347 n_vocab,
348 present,
349 spare_base: ids.len(),
350 n_spare,
351 used: 0,
352 logged_full: false,
353 }
354 });
355 let mut d2t = ids;
356 let spare_fill = d2t[0];
357 d2t.extend(std::iter::repeat_n(spare_fill, n_spare));
358 // pre-fill spare slots from the learned sidecar (trim_adapt_save):
359 // prior serves' escapes are proposable from round 1 of THIS serve.
360 if let Some(ta) = trim_adapt.as_mut() {
361 let learned: Vec<u32> = std::fs::read_to_string(format!("{path}.learned"))
362 .map(|t| t.lines().filter_map(|l| l.trim().parse().ok()).collect())
363 .unwrap_or_default();
364 let mut n_pre = 0usize;
365 for id in learned {
366 let i = id as usize;
367 if i < n_vocab && !ta.present[i] && ta.used < ta.n_spare {
368 let slot = ta.spare_base + ta.used;
369 ta.used += 1;
370 ta.present[i] = true;
371 let off = i * row_bytes;
372 gathered[slot * row_bytes..(slot + 1) * row_bytes]
373 .copy_from_slice(&t.bytes[off..off + row_bytes]);
374 d2t[slot] = id;
375 n_pre += 1;
376 }
377 }
378 if n_pre > 0 {
379 eprintln!(
380 "[trim-adapt] {n_pre} learned rows pre-filled from {path}.learned"
381 );
382 }
383 }
384 // upload AFTER the sidecar pre-fill wrote its rows into `gathered`.
385 let bytes = e.htod_bytes(&gathered)?;
386 (
387 GpuTensor::Quant {
388 bytes,
389 qtype,
390 row_bytes,
391 ne: vec![in_f as u64, d2t.len() as u64],
392 scale: 1.0,
393 rp: false,
394 #[cfg(memra_cutlass)]
395 cutlass: None,
396 fp8: None,
397 blk: None,
398 rp4: None,
399 f16: None,
400 },
401 Some(d2t),
402 trim_adapt,
403 )
404 }
405 None => (load_t(e, &src, "token_embd.weight")?, None, None),
406 }
407 };
408 // Q4_0 split-plane decode mirrors (MEMRA_Q4RP, same as the main trunk — see hybrid.rs):
409 // the draft chain is 3 serial mmvq trips/round; the head alone is ~137MB/draft.
410 // projection tensor prefix: 26B/31B "nextn.", the E4B assistant "mtp.".
411 let proj_prefix = if src.find("nextn.pre_projection.weight").is_some() {
412 "nextn"
413 } else {
414 "mtp"
415 };
416 let (mut pre_proj, mut post_proj) = (
417 load_t(e, &src, &format!("{proj_prefix}.pre_projection.weight"))?,
418 load_t(e, &src, &format!("{proj_prefix}.post_projection.weight"))?,
419 );
420 let mut head = head;
421 let mut layers = layers;
422 if crate::Engine::q4rp_enabled() {
423 // adaptive-trim heads skip the split-plane mirror: the mmvq _rp twins read the
424 // MIRROR, so an in-place row learn on `bytes` would be invisible to the matmul.
425 let head_ws: &mut [&mut GpuTensor] = if trim_adapt.is_some() {
426 &mut [&mut pre_proj, &mut post_proj]
427 } else {
428 &mut [&mut pre_proj, &mut post_proj, &mut head]
429 };
430 for w in head_ws.iter_mut() {
431 e.build_q4_rp4(w)?;
432 }
433 for l in layers.iter_mut() {
434 for w in [
435 &mut l.wq,
436 &mut l.wo,
437 &mut l.ffn_gate,
438 &mut l.ffn_up,
439 &mut l.ffn_down,
440 ] {
441 e.build_q4_rp4(w)?;
442 }
443 }
444 }
445 let d2t_dev = match &d2t {
446 Some(m) => Some(e.stream().clone_htod(&m[..])?),
447 None => None,
448 };
449 Ok(GemmaDraft {
450 layers,
451 pre_proj,
452 post_proj,
453 output_norm: load_t(e, &src, "output_norm.weight")?,
454 head,
455 d2t,
456 d2t_dev,
457 trim_adapt,
458 rope_freqs,
459 ones: e.htod(&[1.0f32; 512])?,
460 n_embd,
461 n_backbone,
462 rope_base_global: meta_f("rope.freq_base", 1e6),
463 rope_base_swa: meta_f("rope.freq_base_swa", 1e4),
464 sliding_window: meta_u("attention.sliding_window") as usize,
465 })
466 }
467}
468
469impl HybridModel {
470 /// The MAIN layer whose KV cache a drafter layer attends (llama-model.cpp:2139):
471 /// the last OWN-KV layer of the class — `boundary - 2` windowed / `boundary - 1`
472 /// global, where boundary = n_layer - shared_kv_layers. Shared across every
473 /// gemma4-assistant drafter (26B/31B: boundary = n_layer; E4B: 24).
474 pub(crate) fn gemma4_draft_kv_target(&self, swa: bool) -> usize {
475 let shared = self
476 .cfg
477 .gemma4
478 .as_ref()
479 .map(|g| g.shared_kv_layers as usize)
480 .unwrap_or(0);
481 let boundary = self.layers.len() - shared;
482 boundary - if swa { 2 } else { 1 }
483 }
484
485 /// One drafter step: (token, h[2816 device]) at absolute position `pos` over the FROZEN main
486 /// cache. Returns (draft logits host [n_vocab], h_next [2816 device]).
487 pub fn gemma4_draft_step(
488 &self,
489 e: &Engine,
490 d: &GemmaDraft,
491 token: u32,
492 h: &CudaSlice<f32>,
493 pos: usize,
494 cache: &Cache,
495 ) -> Result<(Vec<f32>, CudaSlice<f32>), Box<dyn std::error::Error>> {
496 let (hn, h_next) = self.gemma4_draft_trunk(e, d, token, h, pos, cache)?;
497 let logits = e.dtoh(&e.matmul(&d.head, &hn, 1)?)?;
498 Ok((logits, h_next))
499 }
500
501 /// Drafter trunk with the token in DEVICE memory (a 1-elem view of the round's batch
502 /// buffer) — zero host traffic.
503 #[allow(clippy::too_many_arguments)] // allow: the parameter list mirrors the kernel/FFI/call contract; bundling into a struct is a refactor, not a lint fix
504 fn gemma4_draft_trunk_dev(
505 &self,
506 e: &Engine,
507 d: &GemmaDraft,
508 tok_v: &cudarc::driver::CudaView<u32>,
509 h: &CudaSlice<f32>,
510 pos_d: &CudaSlice<i32>,
511 cache: &Cache,
512 dc_bucket: Option<usize>,
513 ) -> Result<(CudaSlice<f32>, CudaSlice<f32>), Box<dyn std::error::Error>> {
514 let nb = d.n_backbone;
515 let embd_gpu = self
516 .embd_gpu
517 .get_or_init(|| e.upload_u8(&self.embd.raw).expect("embed table upload"));
518 let (qt, rb) = self.embd.qt_and_row_bytes(nb);
519 let mut xs = e.embed_gather_device_tv(embd_gpu, tok_v, 1, nb, qt, rb)?;
520 e.scale_inplace(&mut xs, (nb as f32).sqrt(), nb)?;
521 self.gemma4_draft_trunk_from_x(e, d, &xs, h, pos_d, cache, dc_bucket)
522 }
523
524 /// Drafter trunk: returns (post-output_norm hidden [1024], h_next [2816]).
525 fn gemma4_draft_trunk(
526 &self,
527 e: &Engine,
528 d: &GemmaDraft,
529 token: u32,
530 h: &CudaSlice<f32>,
531 pos: usize,
532 cache: &Cache,
533 ) -> Result<(CudaSlice<f32>, CudaSlice<f32>), Box<dyn std::error::Error>> {
534 let nb = d.n_backbone;
535 let mut xs = e.htod(&self.embd.gather(nb, &[token]))?;
536 e.scale_inplace(&mut xs, (nb as f32).sqrt(), nb)?;
537 let pos_d = e.htod_i32(&[pos as i32])?;
538 self.gemma4_draft_trunk_from_x(e, d, &xs, h, &pos_d, cache, None)
539 }
540
541 /// Trunk body from the pre-scaled main-embed row.
542 #[allow(clippy::too_many_arguments)] // allow: the parameter list mirrors the kernel/FFI/call contract; bundling into a struct is a refactor, not a lint fix
543 fn gemma4_draft_trunk_from_x(
544 &self,
545 e: &Engine,
546 d: &GemmaDraft,
547 xs: &CudaSlice<f32>,
548 h: &CudaSlice<f32>,
549 pos_d: &CudaSlice<i32>,
550 cache: &Cache,
551 dc_bucket: Option<usize>,
552 ) -> Result<(CudaSlice<f32>, CudaSlice<f32>), Box<dyn std::error::Error>> {
553 // pos rides a DEVICE slot (burst-arc step a, 2026-07-12): the round fills persistent
554 // slots via set_i32_one (kernel-arg stores — no per-step htod/alloc) and the chain
555 // becomes graph-capturable (an in-graph i32_copy_add can feed the slots later).
556 let eps = self.cfg.rms_eps;
557 let ne = d.n_embd;
558
559 // xh = concat(x, h) [2*n_backbone]
560 let nb = d.n_backbone;
561 let mut xh = e.uninit(2 * nb)?;
562 e.copy_into(&mut xh, 0, xs, nb)?;
563 e.copy_into(&mut xh, nb, h, nb)?;
564
565 let mut cur = e.matmul(&d.pre_proj, &xh, 1)?; // [1024]
566
567 for dl in d.layers.iter() {
568 // attention over the shared MAIN KV: swa -> the last OWN-KV windowed layer,
569 // global -> the last OWN-KV global layer (llama-model.cpp:2139 rule). Plain
570 // 26B/31B trunks have no shared tail, so this is n-2 / n-1 there; E4B's 18
571 // KV-shared tail layers move the boundary to 24 -> targets 22 (swa) / 23.
572 let main_il = self.gemma4_draft_kv_target(dl.swa);
573 let kvl = cache.kv[main_il].as_ref().unwrap();
574 let (hd, nhh) = (dl.hd, dl.nh);
575 let nkv = kvl.kv_dim_k / hd;
576 let base = if dl.swa {
577 d.rope_base_swa
578 } else {
579 d.rope_base_global
580 };
581
582 let mut hn = e.uninit(ne)?;
583 e.rms_norm(&cur, dl.attn_norm.float_data(), &mut hn, ne, 1, eps)?;
584 let q0 = e.matmul(&dl.wq, &hn, 1)?;
585 let mut q = e.uninit(nhh * hd)?;
586 e.rms_norm(&q0, dl.q_norm.float_data(), &mut q, hd, nhh, eps)?;
587 if dl.swa {
588 e.rope_neox(&mut q, pos_d, hd, hd, nhh, 1, base, 1.0)?;
589 } else {
590 e.rope_neox_ff(&mut q, pos_d, hd, hd, nhh, 1, base, 1.0, &d.rope_freqs)?;
591 }
592 let avail = kvl.len;
593 let win = d.sliding_window;
594 let mut attn = e.uninit(nhh * hd)?;
595 // drafter attends the MAIN cache — its format follows the main layer's class
596 // (windowed L28 = wkv arm, global L29 = gkv arm; gkv routing is hd-keyed inside).
597 // DEVICE-LEN arms (burst arc): the length rides the main layer's len_d counter
598 // so the chain is replay-correct across rounds. dc_bucket = the RUNG the round
599 // derived (power-of-2, shared by eager and captured replays — same n_splits,
600 // same combine order; the main graph arc's bucket lesson). None = host-len arm.
601 if let Some(bucket) = dc_bucket {
602 let k_view = e.view_u8(&kvl.k, kvl.k.len());
603 let v_view = e.view_u8(&kvl.v, kvl.v.len());
604 if dl.swa && avail > win {
605 e.fa_decode_rows_w(
606 &q,
607 &k_view,
608 &v_view,
609 &mut attn,
610 hd,
611 nhh,
612 nkv,
613 &kvl.len_d,
614 -1,
615 1,
616 1.0,
617 win,
618 kvl.k_tok_bytes,
619 kvl.v_tok_bytes,
620 None,
621 )?;
622 } else {
623 e.fa_decode_dc(
624 &q,
625 &k_view,
626 &v_view,
627 &mut attn,
628 hd,
629 nhh,
630 nkv,
631 &kvl.len_d,
632 bucket,
633 1.0,
634 kvl.k_tok_bytes,
635 kvl.v_tok_bytes,
636 dl.swa && crate::Engine::wkv_on(),
637 )?;
638 }
639 } else {
640 let (off_tok, t_kv) = if dl.swa && avail > win {
641 (avail - win, win)
642 } else {
643 (0, avail)
644 };
645 let k_view = e.view_u8_range(
646 &kvl.k,
647 off_tok * kvl.k_tok_bytes,
648 (off_tok + t_kv) * kvl.k_tok_bytes,
649 );
650 let v_view = e.view_u8_range(
651 &kvl.v,
652 off_tok * kvl.v_tok_bytes,
653 (off_tok + t_kv) * kvl.v_tok_bytes,
654 );
655 e.fa_decode_kvmod(
656 &q,
657 &k_view,
658 &v_view,
659 &mut attn,
660 hd,
661 nhh,
662 nkv,
663 t_kv,
664 1.0,
665 kvl.k_tok_bytes,
666 kvl.v_tok_bytes,
667 dl.swa && crate::Engine::wkv_on(),
668 )?;
669 }
670 let o = e.matmul(&dl.wo, &attn, 1)?;
671
672 let mut post = e.uninit(ne)?;
673 e.rms_norm(&o, dl.post_attn_norm.float_data(), &mut post, ne, 1, eps)?;
674 let mut attn_out = e.uninit(ne)?;
675 e.add(&post, &cur, &mut attn_out, ne)?;
676
677 let mut z = e.uninit(ne)?;
678 e.rms_norm(&attn_out, dl.ffn_norm.float_data(), &mut z, ne, 1, eps)?;
679 let n_ff = dl.ffn_gate.out_features();
680 let gate = e.matmul(&dl.ffn_gate, &z, 1)?;
681 let up = e.matmul(&dl.ffn_up, &z, 1)?;
682 let mut act = e.uninit(n_ff)?;
683 e.gelu_tanh_mul(&gate, &up, &mut act, n_ff)?;
684 let f0 = e.matmul(&dl.ffn_down, &act, 1)?;
685 let mut fpost = e.uninit(ne)?;
686 e.rms_norm(&f0, dl.ffn_post_norm.float_data(), &mut fpost, ne, 1, eps)?;
687 let mut xn = e.uninit(ne)?;
688 e.add_scale(&fpost, &attn_out, dl.out_scale, &mut xn, ne)?;
689 cur = xn;
690 }
691
692 let mut hn = e.uninit(ne)?;
693 e.rms_norm(&cur, d.output_norm.float_data(), &mut hn, ne, 1, eps)?;
694 let h_next = e.matmul(&d.post_proj, &hn, 1)?; // [2816]; head applied by callers (NO softcap)
695 Ok((hn, h_next))
696 }
697
698 /// Greedy draft step: like gemma4_draft_step but the token argmax stays on device —
699 /// host sees 4 bytes (no 1MB logits dtoh per draft). Returns (token, h_next).
700 pub fn gemma4_draft_step_greedy(
701 &self,
702 e: &Engine,
703 d: &GemmaDraft,
704 token: u32,
705 h: &CudaSlice<f32>,
706 pos: usize,
707 cache: &Cache,
708 ) -> Result<(u32, CudaSlice<f32>), Box<dyn std::error::Error>> {
709 let (hn, h_next) = self.gemma4_draft_trunk(e, d, token, h, pos, cache)?;
710 let ld = e.matmul(&d.head, &hn, 1)?;
711 let tok_d = e.argmax_token_device(&ld, d.head.out_features())?;
712 let idx = e.dtoh_u32(&tok_d)?[0];
713 let tok = match &d.d2t {
714 Some(map) => map[idx as usize],
715 None => idx,
716 };
717 Ok((tok, h_next))
718 }
719}
720
721impl HybridModel {
722 /// gemma4 MTP greedy spec loop: prime the prompt, then rounds of (chained K-token draft
723 /// over the frozen main cache) + (ONE batched verify) + longest-prefix accept + KV rollback.
724 /// Returns generated tokens; prints acceptance stats.
725 #[allow(clippy::too_many_arguments)]
726 #[allow(clippy::unnecessary_unwrap)] // allow: the Some-guards sit in multi-clause regime gates; if-let would reshape the arm structure
727 #[allow(clippy::map_entry)] // allow: the init bodies are fallible (`?`); Entry::or_insert_with cannot propagate errors
728 pub fn generate_spec_gemma(
729 &self,
730 e: &Engine,
731 d: &mut GemmaDraft,
732 prompt: &[u32],
733 max_new: usize,
734 k: usize,
735 eos: &[u32],
736 ) -> Result<Vec<u32>, Box<dyn std::error::Error>> {
737 let n_embd = self.cfg.n_embd as usize;
738 let eps = self.cfg.rms_eps;
739 let mut cache = Cache::new(e, &self.cfg, prompt.len() + max_new + k + 8)?;
740
741 // Adaptive trim, learn point 1: the PROMPT's own ids — the measured escapees are the
742 // prompt's domain content words echoed back (▁oceans, clouds, Explain...), so the
743 // prompt is the cheapest predictor of what the trim is about to miss.
744 trim_adapt_learn(e, d, prompt)?;
745
746 let t_prime = std::time::Instant::now();
747 // short prompts fall below prime_cache's T floor — the batched verify IS a prime.
748 let (pl, h_seed) = if prompt.len() >= crate::hybrid_forward::PRIME_MIN_T {
749 let (l, hs, _hh) = self.prime_cache(e, prompt, &mut cache, 0)?;
750 (l, hs)
751 } else if self.is_gemma4_e4b() {
752 // E4B short-prompt prime: TOKENWISE — the batched e4b trunk at base_len==0
753 // rides the PRIME-FA f32 arm (a different numerics class from the plain arm's
754 // tokenwise prime), and the class skew flipped near-tie streams (3/64,
755 // 2026-07-13). decode_step_h is the same chain the plain arm primes with.
756 let n_embd_ = self.cfg.n_embd as usize;
757 let mut ll = Vec::new();
758 let mut hx = e.zeros(n_embd_)?;
759 for &tok in prompt {
760 let (l, hh) = self.gemma4_e4b_decode_step_h(e, tok, &mut cache)?;
761 ll = l;
762 hx = hh;
763 }
764 // decode_step_h returns the PRE-output_norm hidden; the short-prompt arm's
765 // h convention below is POST-norm — norm here.
766 let mut hp = e.uninit(n_embd_)?;
767 e.rms_norm(&hx, self.output_norm.float_data(), &mut hp, n_embd_, 1, eps)?;
768 (ll, hp)
769 } else {
770 let n_vocab = self.output.out_features();
771 let (lv, hv) = self.gemma4_decode_step_t_h(e, prompt, 0, &mut cache)?;
772 let t = prompt.len();
773 let last = lv[(t - 1) * n_vocab..t * n_vocab].to_vec();
774 // NOTE hv rows are POST-output_norm; h_seed convention below expects PRE-norm and
775 // re-norms — so recover a pre-norm-free path: use the post-norm row DIRECTLY.
776 let hvv = e.view(&hv, t * n_embd);
777 let row = hvv.slice((t - 1) * n_embd..t * n_embd);
778 let mut hrow = e.uninit(n_embd)?;
779 e.copy_view_into(&mut hrow, 0, &row, n_embd)?;
780 // mark: already post-norm — skip the re-norm below via the flag
781 (last, hrow)
782 };
783 e.stream().synchronize()?;
784 crate::PRIME_NANOS.store(
785 t_prime.elapsed().as_nanos() as u64,
786 std::sync::atomic::Ordering::Relaxed,
787 );
788 // drafter h = POST-output_norm hidden (llama h_nextn); prime returns PRE-norm h_seed,
789 // the short-prompt verify path already returns post-norm rows.
790 let mut h = if prompt.len() >= crate::hybrid_forward::PRIME_MIN_T {
791 let mut hh = e.uninit(n_embd)?;
792 e.rms_norm(
793 &h_seed,
794 self.output_norm.float_data(),
795 &mut hh,
796 n_embd,
797 1,
798 eps,
799 )?;
800 hh
801 } else {
802 h_seed
803 };
804
805 let mut last = crate::forward::argmax(&pl) as u32;
806 // MEMRA_PROFILE_SPEC=2: capture starts at the ROUND LOOP (prime excluded) — pair
807 // with `nsys -c cudaProfilerApi` (the qwen loop's pattern, spec.rs).
808 if std::env::var("MEMRA_PROFILE_SPEC").as_deref() == Ok("2") {
809 unsafe extern "C" {
810 fn cudaProfilerStart() -> i32;
811 }
812 unsafe {
813 cudaProfilerStart();
814 }
815 }
816 let mut out: Vec<u32> = Vec::with_capacity(max_new);
817 let (mut drafted, mut accepted, mut rounds) = (0usize, 0usize, 0usize);
818 // per-position accept histogram (MEMRA_SPEC_STATS): [attempted, accepted] per slot —
819 // the depth-K policy statistic (deep slots' marginal accept decides fixed-cap vs deep).
820 let mut pos_att = [0usize; 16];
821 let mut pos_acc = [0usize; 16];
822
823 // ASYNC ROUND v2 (dc class): the whole draft chain + verify enqueue with ZERO host
824 // syncs — token seeds via kernel-arg store (u32_set_k, no host-memory transfer), draft
825 // argmaxes land in the batch buffer, verify argmaxes in vam_d; ONE pack + ONE dtoh of
826 // (k drafts + k+1 vam) closes the round. (v1 with memcpy_htod seeding measured
827 // NEGATIVE — the pageable-copy sync; this is the retry with the sync removed.)
828 let mut batch_d = e.stream().alloc_zeros::<u32>(k + 1)?;
829 let mut packed = e.stream().alloc_zeros::<u32>(2 * k + 1)?;
830 // confidence-adaptive depth (MEMRA_SPEC_PMIN, default 0 = off): per-draft probs.
831 let pmin: f32 = std::env::var("MEMRA_SPEC_PMIN")
832 .ok()
833 .and_then(|v| v.parse().ok())
834 .unwrap_or(0.0);
835 // IN-ROUND confidence cut (2026-07-28): llama's draft-mtp stops drafting the
836 // moment a draft's top-1 prob falls below p-min; our MEMRA_SPEC_PMIN is one round
837 // LATE by design (zero-sync round). This arm pays one small dtoh sync per draft
838 // step (steps ~150µs; sync ~15µs) to cut the chain mid-round and verify at the
839 // shrunk width. Eager arm only — burst/graph arms draft fixed depth.
840 // DEFAULT is SELF-KEYED: active at depth (pos >= floor_ctx) and only in rounds
841 // following a MISS — measured: depth cells with sub-0.9 acceptance win (26B
842 // +1.4-3.2% @ 0.868-0.882 accept, 31B +2% @ 0.845-0.883), chat cells and the
843 // 0.95-accept 12B depth lose under an ALWAYS-on cut (-0.9 to -6%) but their
844 // rounds are mostly full-accept so the self-key idles there. Explicit
845 // MEMRA_SPEC_PMIN_INROUND pins the cut at every position/round; =0 disables.
846 let pmin_ir_env: Option<f32> = std::env::var("MEMRA_SPEC_PMIN_INROUND")
847 .ok()
848 .and_then(|v| v.parse().ok());
849 const PMIN_IR_DEFAULT: f32 = 0.7;
850 let mut prev_full = true; // round 1: no miss evidence yet — draft at full depth
851 let mut p_d = e.stream().alloc_zeros::<f32>(k.max(1))?;
852
853 // ADAPTIVE DRAFT LENGTH (default ON 2026-07-10; MEMRA_SPEC_ADAPT=0 reverts): llama's
854 // draft-mtp reaches 0.64-0.70 acceptance on the SAME drafter (ours fixed-K: 0.52) by
855 // drafting fewer tokens when unconfident (p-min gate). Zero-sync host proxy: next
856 // round's depth = last round's accepted run + 1, clamped to [floor=1, k] — rounds
857 // after a miss shrink, streaks re-deepen. The round's ONE dtoh already carries the
858 // acceptance; no new syncs. Policy sweep (short chat, N=1 each): floor1/cap3 239.2
859 // vs fixed-K3 231.1 (+3.5%, accept .52->.58); floor2 and cap4/5 all worse.
860 let adapt = std::env::var("MEMRA_SPEC_ADAPT").as_deref() != Ok("0");
861 // ADAPTIVE FLOOR default is per-model (MEMRA_SPEC_ADAPT_FLOOR overrides): the floor-1
862 // policy collapses to shallow drafts after any miss and pays a slow re-deepen; on
863 // models with an expensive verify step the deep-draft upside dwarfs the wasted-draft
864 // cost. Measured 2026-07-25 (chat cell, own-gen trim; peak grids both models):
865 // 31B K=5 floor=4 120.2 vs floor=1 103.8 (+15.7%, N=3; floor 5-6 falls off);
866 // 12B K=4-5 floor=4 240.5-240.8 vs floor=1 200.6 (+20%, floor 5+ falls off).
867 // The floor clamps to k_cap, so shallow-K callers are unaffected.
868 // 26B tier (2026-07-26 re-sweep under the f16pv spec flip): floor=2 wins BOTH its
869 // cells — short 329.5 vs 307.0 floor1 (+7%, best at every K), depth 329.7 vs ~318
870 // (the 2026-07-10 "floor2 worse" verdict predates the flip and is superseded).
871 // E4B (n_embd < 2500) keeps floor=1 — unmeasured, cheap verify.
872 let adapt_floor_default: usize = if self.cfg.n_embd >= 3500 {
873 4
874 } else if self.cfg.n_embd >= 2500 {
875 2
876 } else {
877 1
878 };
879 // (stream-k spec key lives in HybridModel::load_from_source_impl — it must be set
880 // before the PRIME's GEMMs autotune, not here.)
881 let adapt_floor_env: Option<usize> = std::env::var("MEMRA_SPEC_ADAPT_FLOOR")
882 .ok()
883 .and_then(|v| v.parse().ok());
884 let adapt_floor: usize = adapt_floor_env.unwrap_or(adapt_floor_default);
885 // POSITION KEY (2026-07-26): the HIGH floor is a SHORT-CTX win. At depth the
886 // per-position acceptance is lower and FORCED-DEEP drafts turn net-negative:
887 // 31B d1736 floor4 99-101 and floor2 97.4-99.8 @ 0.758-0.778 vs floor1
888 // 103.8-104.2 @ 0.817 (two perf-ci batteries + flip-tree N=2 — floor2 is a REAL
889 // small loss there, not noise), while its chat cell holds +15-20% under floor4.
890 // The 26B is the opposite at depth: its mild floor2 WINS (304-305 vs ~297).
891 // Default: full floor while pos < floor_ctx; past it HIGH-floor models (>=4)
892 // relax to 1, MILD-floor models keep their floor. MEMRA_SPEC_FLOOR_CTX overrides
893 // the boundary; an explicit MEMRA_SPEC_ADAPT_FLOOR pins the floor everywhere.
894 let floor_ctx: usize = std::env::var("MEMRA_SPEC_FLOOR_CTX")
895 .ok()
896 .and_then(|v| v.parse().ok())
897 .unwrap_or(1024);
898 let floor_at = |pos: usize| -> usize {
899 if adapt_floor_env.is_some() || pos < floor_ctx {
900 adapt_floor
901 } else if adapt_floor >= 4 {
902 1
903 } else {
904 adapt_floor
905 }
906 };
907 // cap ceiling 7 by default; MEMRA_SPEC_CAPMAX opens the b16 verify tier (t=9..16).
908 // The historical cap>=8 "crash" was two host bugs, both fixed 2026-07-12: round 1
909 // ran UNCLAMPED (`kc = k` — verify t=K+1 entered the b16 tier while it was gated)
910 // and the b16 dispatch requested _r2 twins that were never compiled (mcols==16 now
911 // forces the base variant). Stream gates arbitrate any raised cap.
912 let cap_max: usize = std::env::var("MEMRA_SPEC_CAPMAX")
913 .ok()
914 .and_then(|v| v.parse().ok())
915 .unwrap_or(7);
916 let k_cap = k.min(cap_max).max(1);
917 // DRAFT-CHAIN GRAPHS (burst-arc step c, MEMRA_GEMMA_DRAFT_GRAPH=1): the whole k-step
918 // draft chain replays as ONE captured graph — position slots fill in-graph,
919 // the seed hidden rides the persistent g_seed buffer, KV lengths ride len_d (step b).
920 // Keyed on (kr, rung, over_win): a new depth/rung/window regime captures lazily.
921 let graph_on = std::env::var("MEMRA_GEMMA_DRAFT_GRAPH").as_deref() == Ok("1");
922 #[allow(clippy::type_complexity)]
923 // allow: one-shot composite type; naming it would hide the shape that matters at the call site
924 let mut draft_graphs: std::collections::HashMap<
925 (usize, usize, bool),
926 (
927 cudarc::driver::CudaGraph,
928 Vec<Box<dyn std::any::Any + Send>>,
929 ),
930 > = Default::default();
931 let mut g_seed = e.zeros(n_embd)?;
932 // seed len_d before round 1 (prime went through the host-len path).
933 for kvl in cache.kv.iter_mut().flatten() {
934 e.set_i32_one(&mut kvl.len_d, kvl.len as i32)?;
935 }
936 // persistent per-step rope-pos slots (device; filled by set_i32_one kernel-arg stores).
937 let mut pos_slots: Vec<CudaSlice<i32>> = (0..k_cap.max(1))
938 .map(|_| e.htod_i32(&[0]))
939 .collect::<Result<_, _>>()?;
940 // clamp round 1 too (the leak above).
941 let mut kc = k_cap;
942 // BURST (MEMRA_GEMMA_SPEC_BURST=M, default off): pre-issue M full rounds — draft-graph
943 // replay + verify-stream + device accept/seed/rollback/ring-commit — with ONE host
944 // sync per M rounds (the ring drain). The draft(N+1)-overlapping-verify(N) window this
945 // opens is the burst arc's whole prize (~14% of a round; launch tax alone is hidden
946 // at 96.7% busy). Requires the draft graphs (step c) and a regime-stable horizon.
947 let burst_m: usize = std::env::var("MEMRA_GEMMA_SPEC_BURST")
948 .ok()
949 .and_then(|v| v.parse().ok())
950 .unwrap_or(0);
951 let mut burst_state: Option<(
952 crate::round_stream::StreamBufs,
953 CudaSlice<f32>,
954 CudaSlice<u64>,
955 crate::hybrid_forward::VerifyStreamScratch,
956 )> = None;
957 let win_main = self
958 .cfg
959 .gemma4
960 .as_ref()
961 .map(|g| g.sliding_window as usize)
962 .unwrap_or(0);
963 let g4_shared = self
964 .cfg
965 .gemma4
966 .as_ref()
967 .map(|g| g.shared_kv_layers)
968 .unwrap_or(0);
969 'outer: while out.len() < max_new {
970 // burst gate first (see the BURST ARM below): a burst round drafts at FULL depth
971 // (kr = k_cap — the captured chain replays a fixed K; adaptation is host logic).
972 let horizon = burst_m * (k_cap + 1);
973 let burst_ok = burst_m >= 1 && pmin == 0.0 && g4_shared == 0
974 && (cache.pos + horizon + k_cap + 4 < win_main || cache.pos > win_main)
975 // fa512 crossover: the whole horizon on one side (the stream verify's global
976 // arm picks per-row-dc vs rows by hint; straddling rounds stay eager).
977 && (cache.pos + horizon + k_cap + 4 < crate::fa512_min_tkv()
978 || cache.pos + 1 >= crate::fa512_min_tkv())
979 && e.fa_rows_eligible(cache.pos, 256)
980 && cache.pos + horizon + k_cap + 2 <= cache.max_ctx
981 && out.len() + horizon <= max_new;
982 let mut kr = if burst_ok {
983 k_cap
984 } else if adapt {
985 kc
986 } else {
987 k_cap
988 };
989 // power-of-2 rung bucket for the dc arms (shared by eager and captured replays);
990 // MEMRA_GEMMA_DRAFT_DC=0 reverts to the host-len kvmod arm.
991 let dc_bucket: Option<usize> = {
992 static DC: std::sync::OnceLock<bool> = std::sync::OnceLock::new();
993 if *DC.get_or_init(|| std::env::var("MEMRA_GEMMA_DRAFT_DC").as_deref() != Ok("0")) {
994 let ml = cache
995 .kv
996 .iter()
997 .flatten()
998 .map(|kv| kv.len)
999 .max()
1000 .unwrap_or(1);
1001 // burst rounds size the rung for the WHOLE horizon: the captured chain
1002 // replays M rounds between host looks, so the grid must cover the last
1003 // round's len too (a per-round rung undersizes past its pow2 boundary).
1004 let slack = if burst_ok { horizon } else { 0 };
1005 Some((ml + slack + k_cap + 2).next_power_of_two().max(512))
1006 } else {
1007 None
1008 }
1009 };
1010 e.u32_set_k(&mut batch_d, last, 0)?;
1011 e.copy_into(&mut g_seed, 0, &h, n_embd)?;
1012 // the draft chain, step j: reads g_seed via the hc chain, pos from pos_slots[j]
1013 // (eager: host-filled; graph: filled in-graph).
1014 let run_chain = |e: &Engine,
1015 d: &GemmaDraft,
1016 batch_d: &mut CudaSlice<u32>,
1017 p_d: &mut CudaSlice<f32>,
1018 g_seed: &CudaSlice<f32>,
1019 pos_slots: &Vec<CudaSlice<i32>>,
1020 inround: f32|
1021 -> Result<usize, Box<dyn std::error::Error>> {
1022 // uninit+copy (NOT clone_dtod): clone_dtod's internal alloc bypasses the
1023 // capture-retain hooks — its address got pool-reused between replays and the
1024 // replayed chain read a corrupted seed (accept 0.52 vs 0.76).
1025 let mut hc = e.uninit(n_embd)?;
1026 e.copy_into(&mut hc, 0, g_seed, n_embd)?;
1027 for j in 0..kr {
1028 let tv = batch_d.slice(j..j + 1);
1029 let (hn, h_next) = self.gemma4_draft_trunk_dev(
1030 e,
1031 d,
1032 &tv,
1033 &hc,
1034 &pos_slots[j],
1035 &cache,
1036 dc_bucket,
1037 )?;
1038 let ld = e.matmul(&d.head, &hn, 1)?;
1039 e.argmax_token_device_col(&ld, 0, d.head.out_features(), batch_d, j + 1)?;
1040 // confidence-adaptive depth (MEMRA_SPEC_PMIN): TRIM-space prob before d2t.
1041 if pmin > 0.0 || inround > 0.0 {
1042 e.prob_of_token_device_col(
1043 &ld,
1044 batch_d,
1045 j + 1,
1046 p_d,
1047 j,
1048 d.head.out_features(),
1049 )?;
1050 }
1051 // FR-trimmed head: translate the trim-space argmax to the vocab id.
1052 if let Some(map) = &d.d2t_dev {
1053 e.u32_map_k(batch_d, map, j + 1)?;
1054 }
1055 hc = h_next;
1056 // IN-ROUND cut: one small dtoh sync per step; stop drafting the moment
1057 // confidence falls below the gate and verify at the shrunk width.
1058 // (A DSpark-class marginal-rate window — S_{j+1}*T(j) > E[tok](j)*t_d
1059 // with profiled t_draft/t_verify EMAs — measured FLAT here 2026-07-30:
1060 // never cuts at accept >= 0.8, par-to-noise on 26B/31B depth x3
1061 // interleaved; arm removed per flags doctrine, jsonl row is the record.)
1062 if inround > 0.0 && j + 1 < kr {
1063 let ph = e.dtoh(p_d)?;
1064 if ph[j] < inround {
1065 return Ok(j + 1);
1066 }
1067 }
1068 }
1069 Ok(kr)
1070 };
1071 let over_win = {
1072 let win = d.sliding_window;
1073 d.layers.iter().any(|dl| {
1074 dl.swa
1075 && cache.kv[self.gemma4_draft_kv_target(true)]
1076 .as_ref()
1077 .is_some_and(|kv| kv.len > win)
1078 })
1079 };
1080 // ---- ROUND-GRAPH ARM ---- (MEMRA_GEMMA_ROUND_GRAPH=1): the WHOLE round —
1081 // draft chain + stream verify + device accept/seed/rollback/commit + the
1082 // device adaptive-depth update — captured ONCE per (k_cap, rung, over_win)
1083 // regime and replayed as ONE graph launch per round (the llama round-cost
1084 // mechanism: ~600 per-round enqueues collapse to 1). The round is SELF-FEEDING
1085 // (pos_ctr/pend/brk/g_seed all advance in-graph), so the capture warmups are
1086 // simply two SERVED rounds — their tokens land in the ring and drain normally
1087 // (no snapshot/rollback needed, unlike the E4B token door).
1088 // Adaptive K rides brk[0] via spec_adapt_k: drafts always run k_cap deep (the
1089 // drafter is cheap) but the accept walk depth follows the host policy exactly.
1090 let round_graph_on = std::env::var("MEMRA_GEMMA_ROUND_GRAPH").as_deref() == Ok("1");
1091 if round_graph_on
1092 && burst_m == 0
1093 && dc_bucket.is_some()
1094 && pmin == 0.0
1095 && g4_shared == 0
1096 && !self.is_gemma4_e4b()
1097 && (cache.pos + 2 * (k_cap + 1) + k_cap + 4 < win_main || cache.pos > win_main)
1098 && (cache.pos + 2 * (k_cap + 1) + k_cap + 4 < crate::fa512_min_tkv()
1099 || cache.pos + 1 >= crate::fa512_min_tkv())
1100 && e.fa_rows_eligible(cache.pos, 256)
1101 && cache.pos + 2 * (k_cap + 1) + k_cap + 2 <= cache.max_ctx
1102 {
1103 if burst_state.is_none() {
1104 // ring sized for the capture warmups (2 rounds) + the live round.
1105 let bufs = crate::round_stream::StreamBufs::new(e, k_cap, 3)?;
1106 let fill_dummy = e.zeros(n_embd)?;
1107 let ptrs =
1108 crate::round_stream::kv_len_ptr_table(e, &cache, Some(&bufs.pos_ctr))?;
1109 let scr = self.verify_stream_scratch(e, k_cap + 1)?;
1110 burst_state = Some((bufs, fill_dummy, ptrs, scr));
1111 }
1112 // entry: `last` is the pending token (emitted at drain), h is the seed.
1113 let (bufs, fill_dummy, ptrs, scr) = burst_state.as_mut().unwrap();
1114 let n_rows = cache.kv.len() + 1;
1115 e.set_i32_one(&mut bufs.pos_ctr, cache.pos as i32)?;
1116 e.u32_set_k(&mut bufs.ring_d, 0, 0)?;
1117 e.u32_set_k(&mut bufs.pend_d, last, 0)?;
1118 e.u32_set_k(&mut bufs.brk_d, (if adapt { kc } else { k_cap }) as u32, 0)?;
1119 e.u32_set_k(&mut bufs.brk_d, 1, 1)?;
1120 e.copy_into(&mut g_seed, 0, &h, n_embd)?;
1121 // entry pend is emitted host-side (the ring only carries accepted drafts
1122 // + bonuses — the burst-arm contract).
1123 out.push(last);
1124 if eos.contains(&last) {
1125 break 'outer;
1126 }
1127 if out.len() >= max_new {
1128 break 'outer;
1129 }
1130 #[allow(clippy::unnecessary_unwrap)]
1131 // allow: the Some-guard sits in a multi-clause regime gate; if-let would reshape the arm structure
1132 let key = (usize::MAX - k_cap, dc_bucket.unwrap(), over_win);
1133 let mut fresh_rounds = 1usize; // rounds executed by this iteration
1134 // `hint` is the verify stream's ARM-GATING upper bound — it must sit on
1135 // the SAME side of every crossover as the live lengths this capture
1136 // serves, INCLUDING the arms' own margins (`hint + t < f512` gates the
1137 // global scalar arm; `hint + 1 >= win` gates rows_w), or the captured
1138 // verify bakes a different kernel class than the eager reference
1139 // (107-vs-106 / 4-64 drifts; the regime gate above guarantees the live
1140 // side with the same margins).
1141 let hint = if cache.pos > win_main {
1142 dc_bucket.unwrap() + k_cap + 2 // over-window: rows_w regime
1143 } else if cache.pos + 1 >= crate::fa512_min_tkv() {
1144 win_main - 2 // above f512, under window
1145 } else {
1146 crate::fa512_min_tkv().saturating_sub(k_cap + 5) // under both
1147 };
1148 let bufs_ptr: *mut crate::round_stream::StreamBufs = &mut *bufs;
1149 let scr_ptr: *mut crate::hybrid_forward::VerifyStreamScratch = &mut *scr;
1150 let cache_ptr: *mut Cache = &mut cache;
1151 let batch_ptr: *mut CudaSlice<u32> = &mut batch_d;
1152 let seed_ptr: *mut CudaSlice<f32> = &mut g_seed;
1153 let slots_ptr: *mut Vec<CudaSlice<i32>> = &mut pos_slots;
1154 let mut round_body = |e: &Engine| -> Result<(), Box<dyn std::error::Error>> {
1155 // SAFETY: single-threaded round body; the raw pointers alias the outer
1156 // &mut only within this closure (no overlapping borrows).
1157 let (bufs, scr, cache, batch_d, g_seed, pos_slots) = unsafe {
1158 (
1159 &mut *bufs_ptr,
1160 &mut *scr_ptr,
1161 &mut *cache_ptr,
1162 &mut *batch_ptr,
1163 &mut *seed_ptr,
1164 &mut *slots_ptr,
1165 )
1166 };
1167 e.i32_copy_add(&bufs.pos_ctr, &mut bufs.pos_start_d, 0)?;
1168 e.u32_copy(&bufs.pend_d, batch_d)?;
1169 for (j, slot) in pos_slots.iter_mut().take(k_cap).enumerate() {
1170 e.i32_copy_add(&bufs.pos_ctr, slot, j as i32)?;
1171 }
1172 let mut hc = e.uninit(n_embd)?;
1173 e.copy_into(&mut hc, 0, g_seed, n_embd)?;
1174 #[allow(clippy::needless_range_loop)]
1175 // allow: the explicit index loop keeps the offset arithmetic visible and aligned with the device-side indexing
1176 for j in 0..k_cap {
1177 let tv = batch_d.slice(j..j + 1);
1178 let (hn, h_next) = self.gemma4_draft_trunk_dev(
1179 e,
1180 d,
1181 &tv,
1182 &hc,
1183 &pos_slots[j],
1184 cache,
1185 dc_bucket,
1186 )?;
1187 let ld = e.matmul(&d.head, &hn, 1)?;
1188 e.argmax_token_device_col(&ld, 0, d.head.out_features(), batch_d, j + 1)?;
1189 if let Some(map) = &d.d2t_dev {
1190 e.u32_map_k(batch_d, map, j + 1)?;
1191 }
1192 hc = h_next;
1193 }
1194 let (vam_d, vh) = self.gemma4_verify_t_am_stream(
1195 e,
1196 batch_d,
1197 k_cap + 1,
1198 &bufs.pos_ctr,
1199 hint,
1200 cache,
1201 scr,
1202 )?;
1203 e.spec_accept_greedy_dc(
1204 &vam_d,
1205 batch_d,
1206 &bufs.last_pred_d,
1207 &bufs.brk_d,
1208 &mut bufs.acc_d,
1209 )?;
1210 if std::env::var("MEMRA_DEBUG_SPEC").as_deref() == Ok("1")
1211 && std::env::var("MEMRA_ROUND_GRAPH_CHECK").as_deref() == Ok("1")
1212 {
1213 let vhh = e.dtoh(&vh)?;
1214 let nrm = |r: usize| {
1215 vhh[r * n_embd..(r + 1) * n_embd]
1216 .iter()
1217 .map(|x| x * x)
1218 .sum::<f32>()
1219 .sqrt()
1220 };
1221 let vamh = e.dtoh_u32(&vam_d)?;
1222 eprintln!(
1223 "[rg-vh] |row0|={:.3} |row1|={:.3} |row2|={:.3} vam={:?}",
1224 nrm(0),
1225 nrm(1),
1226 nrm(2),
1227 &vamh[..(k_cap + 1).min(7)]
1228 );
1229 }
1230 e.spec_seed_gather(&vh, fill_dummy, &bufs.acc_d, g_seed, 1, n_embd)?;
1231 e.spec_rollback_stream(ptrs, &bufs.pos_start_d, &bufs.acc_d, 1, n_rows)?;
1232 e.spec_ring_commit(
1233 batch_d,
1234 &bufs.acc_d,
1235 &bufs.brk_d,
1236 &mut bufs.ring_d,
1237 &mut bufs.pend_d,
1238 )?;
1239 e.spec_adapt_k(&bufs.acc_d, &mut bufs.brk_d, floor_at(cache.pos), k_cap)?;
1240 Ok(())
1241 };
1242 // MEMRA_ROUND_GRAPH_CHECK=1: run the body EAGERLY (no capture/replay) —
1243 // splits "body semantics wrong" from "replay mechanics wrong".
1244 let body_check = std::env::var("MEMRA_ROUND_GRAPH_CHECK").as_deref() == Ok("1");
1245 if body_check {
1246 round_body(e)?;
1247 if std::env::var("MEMRA_DEBUG_SPEC").as_deref() == Ok("1") {
1248 let acc = e.dtoh_u32(&bufs.acc_d)?;
1249 let brk = e.dtoh_u32(&bufs.brk_d)?;
1250 let bt = e.dtoh_u32(&batch_d)?;
1251 let tgt = self.gemma4_draft_kv_target(true);
1252 let ld = e.dtoh_i32(&cache.kv[tgt].as_ref().unwrap().len_d)?[0];
1253 let gs = e.dtoh(&g_seed)?;
1254 let gn: f32 = gs.iter().map(|x| x * x).sum::<f32>().sqrt();
1255 eprintln!(
1256 "[rg-check] pos0={} batch={bt:?} n_acc={} bonus={} brk_next={:?} len_d[L{tgt}]={ld} |g_seed|={gn:.3}",
1257 cache.pos, acc[0], acc[1], brk
1258 );
1259 }
1260 } else {
1261 if !draft_graphs.contains_key(&key) {
1262 let g = e.capture_graph_retained(&mut round_body)?;
1263 draft_graphs.insert(key, g);
1264 fresh_rounds += 2; // the capture warmups were served rounds
1265 }
1266 draft_graphs.get(&key).unwrap().0.launch()?;
1267 }
1268 // drain: ONE host sync per iteration (warmup rounds included on capture).
1269 let toks = bufs.drain_ring(e)?;
1270 let posh = e.dtoh_i32(&bufs.pos_ctr)?[0] as usize;
1271 if std::env::var("MEMRA_DEBUG_SPEC").as_deref() == Ok("1") {
1272 eprintln!(
1273 "[round-graph] fresh={fresh_rounds} drained={} posh={posh} toks={:?}",
1274 toks.len(),
1275 &toks[..toks.len().min(12)]
1276 );
1277 }
1278 drafted += fresh_rounds * k_cap;
1279 rounds += fresh_rounds;
1280 accepted += toks.len().saturating_sub(fresh_rounds);
1281 let mut ended = false;
1282 for &tk in &toks[..toks.len() - 1] {
1283 out.push(tk);
1284 if eos.contains(&tk) || out.len() >= max_new {
1285 ended = true;
1286 break;
1287 }
1288 }
1289 last = *toks.last().unwrap();
1290 cache.pos = posh;
1291 for kvl in cache.kv.iter_mut().flatten() {
1292 kvl.len = posh;
1293 }
1294 // NO allocation between replays: a pool alloc here can land on a baked
1295 // transient address and corrupt the next replay (the draft-graph lesson).
1296 // g_seed already holds the next seed (in-graph gather); copy INTO the
1297 // existing h buffer for the (possible) eager-arm handoff.
1298 e.copy_into(&mut h, 0, &g_seed, n_embd)?;
1299 kc = k_cap; // device brk owns the walk depth; host kc only seeds entry
1300 // learn point 2 (round-graph drain): ring = accepted drafts + bonuses; only
1301 // bonuses can be escapes, and the present-bitmap check skips the rest cheap.
1302 trim_adapt_learn(e, d, &toks)?;
1303 if ended {
1304 break 'outer;
1305 }
1306 continue 'outer;
1307 }
1308 // ---- BURST ARM ---- (gate computed at the loop top; needs dc arms too)
1309 if burst_ok && dc_bucket.is_some() {
1310 if burst_state.is_none() {
1311 let bufs = crate::round_stream::StreamBufs::new(e, k_cap, burst_m)?;
1312 let fill_dummy = e.zeros(n_embd)?; // spec_seed_gather j>=1 always: unread
1313 let ptrs =
1314 crate::round_stream::kv_len_ptr_table(e, &cache, Some(&bufs.pos_ctr))?;
1315 let scr = self.verify_stream_scratch(e, k_cap + 1)?;
1316 burst_state = Some((bufs, fill_dummy, ptrs, scr));
1317 }
1318 // the loop-top dc_bucket already carries the horizon slack on burst rounds,
1319 // so the key below matches the rung the captured chain actually launches with.
1320 #[allow(clippy::unnecessary_unwrap)]
1321 // allow: the Some-guard sits in a multi-clause regime gate; if-let would reshape the arm structure
1322 let key = (k_cap, dc_bucket.unwrap(), over_win);
1323 if std::env::var("MEMRA_GEMMA_BURST_GRAPH").as_deref() == Ok("1")
1324 && !draft_graphs.contains_key(&key)
1325 {
1326 let g = e.capture_graph_retained(|e| {
1327 run_chain(e, d, &mut batch_d, &mut p_d, &g_seed, &pos_slots, 0.0)
1328 .map(|_| ())
1329 })?;
1330 draft_graphs.insert(key, g);
1331 }
1332 // entry: `last` is the not-yet-emitted pending token (the ring only ever
1333 // carries accepted drafts + bonuses; the entry pend is emitted host-side).
1334 out.push(last);
1335 if eos.contains(&last) {
1336 break 'outer;
1337 }
1338 if out.len() >= max_new {
1339 break 'outer;
1340 }
1341 let (bufs, fill_dummy, ptrs, scr) = burst_state.as_mut().unwrap();
1342 let n_rows = cache.kv.len() + 1; // + the pos counter row
1343 e.set_i32_one(&mut bufs.pos_ctr, cache.pos as i32)?;
1344 e.u32_set_k(&mut bufs.ring_d, 0, 0)?;
1345 e.u32_set_k(&mut bufs.pend_d, last, 0)?;
1346 e.u32_set_k(&mut bufs.brk_d, k_cap as u32, 0)?; // k_used = K (no p-min cut)
1347 e.u32_set_k(&mut bufs.brk_d, 1, 1)?; // base = 1 (pend always set)
1348 e.copy_into(&mut g_seed, 0, &h, n_embd)?;
1349 let pos0 = cache.pos;
1350 for r in 0..burst_m {
1351 // every op below is ENQUEUED; nothing reads back until the drain.
1352 e.i32_copy_add(&bufs.pos_ctr, &mut bufs.pos_start_d, 0)?;
1353 e.u32_copy(&bufs.pend_d, &mut batch_d)?; // batch_d[0] <- pend
1354 for (j, slot) in pos_slots.iter_mut().take(k_cap).enumerate() {
1355 e.i32_copy_add(&bufs.pos_ctr, slot, j as i32)?;
1356 }
1357 // the chain enqueues ZERO-SYNC with device pos slots — the captured-graph
1358 // replay is measured EXPENSIVE (26B eager 379 -> 253 with replay), so the
1359 // burst runs the chain eagerly by default; MEMRA_GEMMA_BURST_GRAPH=1 keeps
1360 // the replay door for A/B.
1361 if std::env::var("MEMRA_GEMMA_BURST_GRAPH").as_deref() == Ok("1") {
1362 draft_graphs.get(&key).unwrap().0.launch()?;
1363 } else {
1364 // run_chain's body inlined: the closure holds &cache for the loop's
1365 // lifetime and collides with the verify's &mut cache borrow.
1366 let mut hc = e.uninit(n_embd)?;
1367 e.copy_into(&mut hc, 0, &g_seed, n_embd)?;
1368 #[allow(clippy::needless_range_loop)]
1369 // allow: the explicit index loop keeps the offset arithmetic visible and aligned with the device-side indexing
1370 for j in 0..k_cap {
1371 let tv = batch_d.slice(j..j + 1);
1372 let (hn, h_next) = self.gemma4_draft_trunk_dev(
1373 e,
1374 d,
1375 &tv,
1376 &hc,
1377 &pos_slots[j],
1378 &cache,
1379 dc_bucket,
1380 )?;
1381 let ld = e.matmul(&d.head, &hn, 1)?;
1382 e.argmax_token_device_col(
1383 &ld,
1384 0,
1385 d.head.out_features(),
1386 &mut batch_d,
1387 j + 1,
1388 )?;
1389 if let Some(map) = &d.d2t_dev {
1390 e.u32_map_k(&mut batch_d, map, j + 1)?;
1391 }
1392 hc = h_next;
1393 }
1394 }
1395 // host UPPER bound on this round's base (full-accept growth): sizes the
1396 // stream verify's splits + window-arm gate; device len is the true bound.
1397 let hint = pos0 + (r + 1) * (k_cap + 1) + 2;
1398 let (vam_d, vh) = self.gemma4_verify_t_am_stream(
1399 e,
1400 &batch_d,
1401 k_cap + 1,
1402 &bufs.pos_ctr,
1403 hint,
1404 &mut cache,
1405 scr,
1406 )?;
1407 e.spec_accept_greedy_dc(
1408 &vam_d,
1409 &batch_d,
1410 &bufs.last_pred_d,
1411 &bufs.brk_d,
1412 &mut bufs.acc_d,
1413 )?;
1414 e.spec_seed_gather(&vh, fill_dummy, &bufs.acc_d, &mut g_seed, 1, n_embd)?;
1415 e.spec_rollback_stream(ptrs, &bufs.pos_start_d, &bufs.acc_d, 1, n_rows)?;
1416 e.spec_ring_commit(
1417 &batch_d,
1418 &bufs.acc_d,
1419 &bufs.brk_d,
1420 &mut bufs.ring_d,
1421 &mut bufs.pend_d,
1422 )?;
1423 }
1424 // drain: THE one sync per M rounds. Ring = [acc..., bonus] per round; the
1425 // final element is the next pending token (eager pushes it next round).
1426 let toks = bufs.drain_ring(e)?;
1427 let posh = e.dtoh_i32(&bufs.pos_ctr)?[0] as usize;
1428 drafted += burst_m * k_cap;
1429 rounds += burst_m;
1430 accepted += toks.len().saturating_sub(burst_m); // each round adds n_acc + 1
1431 let mut ended = false;
1432 for &tk in &toks[..toks.len() - 1] {
1433 out.push(tk);
1434 if eos.contains(&tk) || out.len() >= max_new {
1435 ended = true;
1436 break;
1437 }
1438 }
1439 last = *toks.last().unwrap();
1440 // host mirrors re-sync (device counters are already correct from rollback).
1441 cache.pos = posh;
1442 for kvl in cache.kv.iter_mut().flatten() {
1443 kvl.len = posh;
1444 }
1445 // next seed hidden = g_seed (the final round's device gather).
1446 let mut hrow = e.uninit(n_embd)?;
1447 e.copy_into(&mut hrow, 0, &g_seed, n_embd)?;
1448 h = hrow;
1449 kc = k_cap;
1450 // learn point 2 (burst drain): same contract as the round-graph drain.
1451 trim_adapt_learn(e, d, &toks)?;
1452 if ended {
1453 break 'outer;
1454 }
1455 continue 'outer;
1456 }
1457 if graph_on && dc_bucket.is_some() {
1458 #[allow(clippy::unnecessary_unwrap)]
1459 // allow: the Some-guard sits in a multi-clause regime gate; if-let would reshape the arm structure
1460 let key = (kr, dc_bucket.unwrap(), over_win);
1461 if !draft_graphs.contains_key(&key) {
1462 // chain-only capture; pos slots are graph INPUTS (filled eagerly before
1463 // each launch, like g_seed — the in-graph copy_add fills replayed one
1464 // round stale, see jsonl).
1465 let g = e.capture_graph_retained(|e| {
1466 run_chain(e, d, &mut batch_d, &mut p_d, &g_seed, &pos_slots, 0.0)
1467 .map(|_| ())
1468 })?;
1469 draft_graphs.insert(key, g);
1470 }
1471 for (j, slot) in pos_slots.iter_mut().take(kr).enumerate() {
1472 e.set_i32_one(slot, (cache.pos + j) as i32)?;
1473 }
1474 draft_graphs.get(&key).unwrap().0.launch()?;
1475 // MEMRA_DRAFT_GRAPH_CHECK=1: re-run the chain eagerly from the same state and
1476 // diff the drafted slots (replay-vs-eager divergence bisect).
1477 if std::env::var("MEMRA_DRAFT_GRAPH_CHECK").as_deref() == Ok("1") {
1478 // NON-DESTRUCTIVE: compare, then restore the graph's tokens so the round
1479 // proceeds exactly as it would without the check.
1480 let gtoks = e.dtoh_u32(&batch_d)?;
1481 for (j, slot) in pos_slots.iter_mut().take(kr).enumerate() {
1482 e.set_i32_one(slot, (cache.pos + j) as i32)?;
1483 }
1484 run_chain(e, d, &mut batch_d, &mut p_d, &g_seed, &pos_slots, 0.0)?;
1485 let etoks = e.dtoh_u32(&batch_d)?;
1486 if gtoks[..=kr] != etoks[..=kr] {
1487 eprintln!(
1488 "[draft-graph] DIVERGE round={rounds} graph={:?} eager={:?}",
1489 >oks[..=kr],
1490 &etoks[..=kr]
1491 );
1492 }
1493 for (j, &t) in gtoks.iter().enumerate().take(kr + 1) {
1494 e.u32_set_k(&mut batch_d, t, j)?;
1495 }
1496 }
1497 } else {
1498 for (j, slot) in pos_slots.iter_mut().take(kr).enumerate() {
1499 e.set_i32_one(slot, (cache.pos + j) as i32)?;
1500 }
1501 let ir_now = match pmin_ir_env {
1502 Some(p) => p, // explicit pin (0 disables)
1503 None if cache.pos >= floor_ctx && !prev_full => PMIN_IR_DEFAULT,
1504 None => 0.0,
1505 };
1506 kr = run_chain(e, d, &mut batch_d, &mut p_d, &g_seed, &pos_slots, ir_now)?;
1507 }
1508 drafted += kr;
1509 rounds += 1;
1510 let pos0 = cache.pos;
1511 // MEMRA_BURST_VCHECK=1: run the STREAM verify first on the same batch/state and
1512 // diff its argmaxes against the eager verify (bisect harness — the stream append
1513 // writes the same rows the eager append then overwrites, so state is untouched).
1514 let vcheck = std::env::var("MEMRA_BURST_VCHECK").as_deref() == Ok("1");
1515 let kvsum = |e: &Engine,
1516 cache: &Cache|
1517 -> Result<Vec<(u64, u64)>, Box<dyn std::error::Error>> {
1518 let mut out = Vec::new();
1519 for kvl in cache.kv.iter().flatten() {
1520 let kb = e.dtoh_u8(&kvl.k)?;
1521 let vb = e.dtoh_u8(&kvl.v)?;
1522 let lo = pos0 * kvl.k_tok_bytes;
1523 let hi = (pos0 + kr + 1) * kvl.k_tok_bytes;
1524 let lov = pos0 * kvl.v_tok_bytes;
1525 let hiv = (pos0 + kr + 1) * kvl.v_tok_bytes;
1526 out.push((
1527 kb[lo..hi].iter().map(|&b| b as u64).sum(),
1528 vb[lov..hiv].iter().map(|&b| b as u64).sum(),
1529 ));
1530 }
1531 Ok(out)
1532 };
1533 let vam_s = if vcheck && !self.is_gemma4_e4b() {
1534 let mut ctr = e.htod_i32(&[pos0 as i32])?;
1535 e.set_i32_one(&mut ctr, pos0 as i32)?;
1536 let mut scr0 = self.verify_stream_scratch(e, kr + 1)?;
1537 let (vs, vhs) = self.gemma4_verify_t_am_stream(
1538 e,
1539 &batch_d,
1540 kr + 1,
1541 &ctr,
1542 pos0 + kr + 3,
1543 &mut cache,
1544 &mut scr0,
1545 )?;
1546 let ss = kvsum(e, &cache)?;
1547 Some((e.dtoh_u32(&vs)?, ss, e.dtoh(&vhs)?))
1548 } else {
1549 None
1550 };
1551 let (vam_d, vh) = if self.is_gemma4_e4b() {
1552 self.gemma4_e4b_decode_step_t_am_dev(e, &batch_d, kr + 1, pos0, &mut cache)?
1553 } else {
1554 self.gemma4_decode_step_t_am_dev(e, &batch_d, kr + 1, pos0, &mut cache)?
1555 };
1556 if let Some((vs, ss, vhs)) = vam_s {
1557 let vhe = e.dtoh(&vh)?;
1558 for r in 0..kr + 1 {
1559 let md = vhs[r * n_embd..(r + 1) * n_embd]
1560 .iter()
1561 .zip(&vhe[r * n_embd..(r + 1) * n_embd])
1562 .map(|(a, b)| (a - b).abs())
1563 .fold(0.0f32, f32::max);
1564 if md > 1e-3 {
1565 eprintln!("[vcheck-vh] round={rounds} row={r} maxdiff={md:.3e}");
1566 }
1567 }
1568 let se = kvsum(e, &cache)?;
1569 for (il, (a, b)) in ss.iter().zip(&se).enumerate() {
1570 if a != b {
1571 eprintln!("[vcheck-kv] round={rounds} il={il} stream={a:?} eager={b:?}");
1572 }
1573 }
1574 let ve = e.dtoh_u32(&vam_d)?;
1575 if vs[..kr + 1] != ve[..kr + 1] {
1576 eprintln!(
1577 "[vcheck] DIVERGE round={rounds} pos0={pos0} stream={:?} eager={:?}",
1578 &vs[..kr + 1],
1579 &ve[..kr + 1]
1580 );
1581 } else {
1582 eprintln!("[vcheck] match round={rounds} pos0={pos0}");
1583 }
1584 }
1585 e.u32_pack2(&batch_d, 1, kr, &vam_d, kr + 1, &mut packed)?;
1586 let host = e.dtoh_u32(&packed)?; // the round's ONE sync
1587 let k = kr;
1588 let dtoks: Vec<u32> = host[..k].to_vec();
1589 let vam: Vec<u32> = host[k..2 * k + 1].to_vec();
1590 // longest accepted prefix: d_i accepted iff d_i == argmax(verify[i-1])
1591 // (trimmed heads: batch_d slots were d2t-translated in the draft loop, so dtoks
1592 // are full-vocab ids here — the 2026-07-10 async rewrite silently dropped this
1593 // and the trim probes read accept=0.000 through it.)
1594 let mut m = 0usize;
1595 while m < k {
1596 if dtoks[m] == vam[m] {
1597 m += 1;
1598 } else {
1599 break;
1600 }
1601 }
1602 prev_full = m == k; // feeds the self-keyed in-round cut (miss → next round cuts)
1603 if std::env::var("MEMRA_DEBUG_SPEC").as_deref() == Ok("1") {
1604 let l0 = cache
1605 .kv
1606 .iter()
1607 .flatten()
1608 .next()
1609 .map(|kv| kv.len)
1610 .unwrap_or(0);
1611 let hh = e.dtoh(&h)?;
1612 let hn: f32 = hh.iter().map(|x| x * x).sum::<f32>().sqrt();
1613 eprintln!(
1614 "[round {rounds}] pos0={pos0} post_pos={} kv0_len={l0} last={last} dtoks={dtoks:?} vam={vam:?} m={m} |h_in|={hn:.3}",
1615 cache.pos
1616 );
1617 }
1618 accepted += m;
1619 for j in 0..k.min(16) {
1620 pos_att[j] += 1;
1621 if j < m {
1622 pos_acc[j] += 1;
1623 }
1624 }
1625 // emit last + accepted drafts; the correction token comes from verify row m.
1626 out.push(last);
1627 if eos.contains(&last) {
1628 break 'outer;
1629 }
1630 for &dt in &dtoks[..m] {
1631 out.push(dt);
1632 if eos.contains(&dt) {
1633 break 'outer;
1634 }
1635 if out.len() >= max_new {
1636 break 'outer;
1637 }
1638 }
1639 let next = vam[m];
1640 // roll back rejected rows: batch appended k+1 rows; keep m+1 (positions of
1641 // last + accepted drafts). SWA layers cap t_kv by the window view, so a plain
1642 // len rewind is safe for every layer.
1643 let keep = m + 1;
1644 for kvl in cache.kv.iter_mut().flatten() {
1645 kvl.len -= (k + 1) - keep;
1646 // keep len_d in lockstep: the drafter's device-len attention arms read it
1647 // (the gemma round appends via the HOST-len path, which doesn't maintain
1648 // the counter — stale len_d gutted acceptance to 0.059 on the dc probe).
1649 e.set_i32_one(&mut kvl.len_d, kvl.len as i32)?;
1650 }
1651 cache.pos -= (k + 1) - keep;
1652 // h for the next round = main hidden at the LAST KEPT position (verify row m).
1653 let hv = e.view(&vh, (k + 1) * n_embd);
1654 let row = hv.slice(m * n_embd..(m + 1) * n_embd);
1655 let mut hrow = e.uninit(n_embd)?;
1656 e.copy_view_into(&mut hrow, 0, &row, n_embd)?;
1657 h = hrow;
1658 last = next;
1659 // Adaptive trim, learn point 2: ALL verify argmaxes — vam[m] is the emitted
1660 // correction (the only emitted token that can sit outside the trim set; accepted
1661 // drafts are trim members by construction), and vam[i>m] are main-model
1662 // predictions for positions never reached this round: next round usually wants
1663 // exactly those tokens, so learning them here lets the draft propose them
1664 // BEFORE any miss is paid (prose escapes are first-occurrence-dominated —
1665 // corrections-only learning measured +0.5 acceptance pts, jsonl 2026-07-19).
1666 trim_adapt_learn(e, d, &vam)?;
1667 if adapt {
1668 let fl_now = floor_at(cache.pos);
1669 kc = (m + 1).clamp(fl_now.min(k_cap), k_cap);
1670 // confidence cut (MEMRA_SPEC_PMIN > 0): next round drafts no deeper than one
1671 // past the first low-confidence draft of THIS round (llama's p-min class,
1672 // one round late — the zero-sync enqueue stays intact). One extra tiny dtoh.
1673 if pmin > 0.0 {
1674 let ph = e.dtoh(&p_d)?;
1675 if let Some(fl) = ph[..kr].iter().position(|&p| p < pmin) {
1676 kc = kc.min((fl + 1).max(fl_now.min(k_cap)));
1677 }
1678 }
1679 }
1680 }
1681 eprintln!(
1682 "[gemma-spec] rounds={rounds} drafted={drafted} accepted={accepted} accept-rate={:.3} tok/round={:.2}",
1683 accepted as f64 / drafted.max(1) as f64,
1684 out.len() as f64 / rounds.max(1) as f64
1685 );
1686 if let Some((used, budget)) = d.trim_adapt_stats() {
1687 eprintln!("[trim-adapt] {used}/{budget} spare slots learned");
1688 match d.trim_adapt_save() {
1689 Ok(n) if n > 0 => {
1690 eprintln!("[trim-adapt] {n} new ids appended to the .learned sidecar")
1691 }
1692 Ok(_) => {}
1693 Err(err) => eprintln!("[trim-adapt] sidecar save failed: {err}"),
1694 }
1695 }
1696 if std::env::var("MEMRA_SPEC_STATS").as_deref() == Ok("1") {
1697 let hist: Vec<String> = (0..16)
1698 .filter(|&j| pos_att[j] > 0)
1699 .map(|j| format!("p{j}:{}/{}", pos_acc[j], pos_att[j]))
1700 .collect();
1701 eprintln!("[gemma-spec] per-position accept: {}", hist.join(" "));
1702 }
1703 Ok(out)
1704 }
1705}
1706
1707/// BURST-SCOPED gemma4 spec session (lane/gemma-batched stage 1, 2026-08-16): the serve
1708/// twin of `generate_spec_gemma`. That function is GENERATION-scoped — it builds its own
1709/// cache, primes, loops to completion, and its `break 'outer` exits deliberately skip the
1710/// final round's rollback/h/pending updates (safe only because the cache dies with the
1711/// call). A served session must instead stop and RESUME across scheduler ticks, so this
1712/// type carries the exact cross-round state the eager loop threads through its locals:
1713///
1714/// * `cache` — the trunk Cache; rows = `committed` (prompt + emitted, INCL. overshoot).
1715/// * `h` — post-output_norm hidden of the LAST committed row (device; draft seed).
1716/// * `pending`— the `last` local: the predicted next token. Emitted as the FIRST token
1717/// of the next round and appended as verify col 0 there; it has NO cache
1718/// row while parked here (the Q38 `next_pred` convention).
1719/// * `kc_next`/`prev_full` — the adaptive-depth + self-keyed in-round-cut carries.
1720///
1721/// BOUNDARY LAW (the Q38 pending-carry/empty-suffix bug class, banked as gate cases in
1722/// gemma-spec-session-gate before this was written): a burst NEVER exits mid-round.
1723/// Every round runs to completion — emission, rollback to the accepted prefix, h/pending
1724/// update, trim-adapt learn — and only then does the burst-target check run. Overshoot
1725/// past `target` is committed and returned (the caller clamps VISIBLE emission; state
1726/// counts every row, exactly like Q38's `SpecSession::committed`). EOS ends the burst at
1727/// its round boundary with the same complete-state guarantee.
1728///
1729/// V1 scope (greedy serve): EAGER round arm only — the round-graph / burst-ring arms are
1730/// generation-scoped perf doors (their ring/pos-counter state does not checkpoint at
1731/// round boundaries) and the shipping bench receipts (154.9/176-179, ASSISTANT-ARM-
1732/// RESULTS.md) were measured on this same eager arm. Dense gemma4 only (E4B refused).
1733/// Fresh session per request: no prefix reuse, no multi-turn suffix — continuation
1734/// bursts are always empty-suffix by construction.
1735pub struct GemmaSpecSession {
1736 pub cache: Cache,
1737 /// Every token whose rows the cache holds, in order (prompt + emitted, incl. overshoot).
1738 pub committed: Vec<u32>,
1739 h: CudaSlice<f32>,
1740 pending: u32,
1741 kc_next: usize,
1742 prev_full: bool,
1743 pub prompt_len: usize,
1744 /// Session-lifetime spec telemetry (rounds / drafted / accepted).
1745 pub rounds: usize,
1746 pub drafted: usize,
1747 pub accepted: usize,
1748}
1749
1750impl GemmaSpecSession {
1751 /// Tokens the session has emitted (committed past the prompt). The pending token is
1752 /// NOT included — it has no cache row and the next burst emits it first.
1753 pub fn emitted_len(&self) -> usize {
1754 self.committed.len() - self.prompt_len
1755 }
1756 /// Context capacity of the session's cache (the server's ContextFull guard).
1757 pub fn cache_max_ctx(&self) -> usize {
1758 self.cache.max_ctx
1759 }
1760 /// DEMOTE HANDOFF (stage-2 seam, gated by the session gate's demote case): hand the
1761 /// trunk cache to the plain path. The cache rows are exactly `committed` (boundary
1762 /// law), and the pending token is returned as the plain path's device_next-equivalent
1763 /// — the plain loop feeds it as its first decode input. The draft side holds no
1764 /// per-session state (the assistant drafter reads the TRUNK's KV; trim-adapt is
1765 /// model-lifetime, not session), so dropping self is the whole handoff.
1766 pub fn into_demoted(self) -> (Cache, u32, Vec<u32>) {
1767 (self.cache, self.pending, self.committed)
1768 }
1769}
1770
1771impl HybridModel {
1772 /// Open a burst-scoped gemma spec session: prime the prompt, park the first predicted
1773 /// token as `pending`. Mirrors `generate_spec_gemma`'s entry verbatim (trim-adapt
1774 /// learn point 1, the PRIME_MIN_T split, the post-norm h convention).
1775 pub fn gemma_spec_session_new(
1776 &self,
1777 e: &Engine,
1778 d: &mut GemmaDraft,
1779 prompt: &[u32],
1780 max_ctx: usize,
1781 ) -> Result<GemmaSpecSession, Box<dyn std::error::Error>> {
1782 if self.is_gemma4_e4b() || !self.uses_gemma_program() {
1783 return Err(
1784 "gemma_spec_session_new: dense gemma4 only (E4B keeps its own arms)".into(),
1785 );
1786 }
1787 let n_embd = self.cfg.n_embd as usize;
1788 let eps = self.cfg.rms_eps;
1789 let mut cache = Cache::new(e, &self.cfg, max_ctx)?;
1790 trim_adapt_learn(e, d, prompt)?;
1791 let (pl, h_seed, post_norm) = if prompt.len() >= crate::hybrid_forward::PRIME_MIN_T {
1792 let (l, hs, _hh) = self.prime_cache(e, prompt, &mut cache, 0)?;
1793 (l, hs, false)
1794 } else {
1795 let n_vocab = self.output.out_features();
1796 let (lv, hv) = self.gemma4_decode_step_t_h(e, prompt, 0, &mut cache)?;
1797 let t = prompt.len();
1798 let last = lv[(t - 1) * n_vocab..t * n_vocab].to_vec();
1799 let hvv = e.view(&hv, t * n_embd);
1800 let row = hvv.slice((t - 1) * n_embd..t * n_embd);
1801 let mut hrow = e.uninit(n_embd)?;
1802 e.copy_view_into(&mut hrow, 0, &row, n_embd)?;
1803 (last, hrow, true)
1804 };
1805 // drafter h = POST-output_norm hidden; the prime returns PRE-norm h_seed.
1806 let h = if post_norm {
1807 h_seed
1808 } else {
1809 let mut hh = e.uninit(n_embd)?;
1810 e.rms_norm(
1811 &h_seed,
1812 self.output_norm.float_data(),
1813 &mut hh,
1814 n_embd,
1815 1,
1816 eps,
1817 )?;
1818 hh
1819 };
1820 let pending = crate::forward::argmax(&pl) as u32;
1821 Ok(GemmaSpecSession {
1822 cache,
1823 committed: prompt.to_vec(),
1824 h,
1825 pending,
1826 kc_next: usize::MAX, // clamped to the burst's k_cap at entry (one-shot: kc = k_cap)
1827 prev_full: true, // round 1: no miss evidence yet
1828 prompt_len: prompt.len(),
1829 rounds: 0,
1830 drafted: 0,
1831 accepted: 0,
1832 })
1833 }
1834
1835 /// SPEC-ON-CACHE-HIT restore (lane/spec-on-cache-hit, 2026-08-18): open a gemma spec
1836 /// session over a trunk cache the worker already restored from a WHOLE prefix-cache
1837 /// entry (rows `[0..prefix.len())` == `prefix`, `cache.pos == prefix.len()`), feeding
1838 /// only the prompt SUFFIX. The assistant drafter holds no per-session KV of its own —
1839 /// it attends the TRUNK's cache — so the restored rows already ARE the draft state;
1840 /// the only products a fresh prime supplied were the boundary logits (-> `pending`)
1841 /// and the post-norm hidden of the last prompt row (-> `h`, the drafter seed), and a
1842 /// non-empty suffix feed regenerates both. An empty suffix therefore REFUSES: there is
1843 /// no drafter seed hidden without feeding at least one row (the plain path serves that
1844 /// shape from the entry's boundary logits, as before).
1845 ///
1846 /// PROGRAM CHOICE (the splitiso two-programs law, 0b0ffa13c6): gemma4's monolithic
1847 /// prime refuses pos > 0, so the suffix rides `gemma4_decode_step_t_h` — the SAME
1848 /// verify-trunk program every spec round runs and the same arm the cold
1849 /// sub-PRIME_MIN_T session prime uses (banked byte-identical by
1850 /// gemma-spec-session-gate). The restored bytes below `prefix.len()` are never
1851 /// recomputed by construction.
1852 pub fn gemma_spec_session_from_restored(
1853 &self,
1854 e: &Engine,
1855 d: &mut GemmaDraft,
1856 mut cache: Cache,
1857 prefix: &[u32],
1858 suffix: &[u32],
1859 ) -> Result<GemmaSpecSession, Box<dyn std::error::Error>> {
1860 if self.is_gemma4_e4b() || !self.uses_gemma_program() {
1861 return Err(
1862 "gemma_spec_session_from_restored: dense gemma4 only (E4B keeps its own arms)"
1863 .into(),
1864 );
1865 }
1866 if prefix.is_empty() {
1867 return Err("gemma_spec_session_from_restored: empty restored prefix".into());
1868 }
1869 if suffix.is_empty() {
1870 return Err(
1871 "gemma_spec_session_from_restored: empty suffix — the drafter seed \
1872 hidden only exists after feeding at least one row (plain path owns \
1873 the whole-prompt hit)"
1874 .into(),
1875 );
1876 }
1877 if cache.pos != prefix.len() {
1878 return Err(format!(
1879 "gemma_spec_session_from_restored: restored cache pos {} != prefix len {}",
1880 cache.pos,
1881 prefix.len(),
1882 )
1883 .into());
1884 }
1885 let n_embd = self.cfg.n_embd as usize;
1886 let n_vocab = self.output.out_features();
1887 // trim-adapt learning is model-lifetime (not session state); feed the full logical
1888 // prompt so restored traffic teaches the head trim exactly what cold traffic does.
1889 let full: Vec<u32> = prefix.iter().chain(suffix.iter()).copied().collect();
1890 trim_adapt_learn(e, d, &full)?;
1891 let base = cache.pos;
1892 let (lv, hv) = self.gemma4_decode_step_t_h(e, suffix, base, &mut cache)?;
1893 let t = suffix.len();
1894 let last = lv[(t - 1) * n_vocab..t * n_vocab].to_vec();
1895 // gemma4_decode_step_t_h returns POST-output_norm hiddens (the drafter's h
1896 // convention — same arm gemma_spec_session_new uses below PRIME_MIN_T).
1897 let hvv = e.view(&hv, t * n_embd);
1898 let row = hvv.slice((t - 1) * n_embd..t * n_embd);
1899 let mut h = e.uninit(n_embd)?;
1900 e.copy_view_into(&mut h, 0, &row, n_embd)?;
1901 let pending = crate::forward::argmax(&last) as u32;
1902 let prompt_len = full.len();
1903 Ok(GemmaSpecSession {
1904 cache,
1905 committed: full,
1906 h,
1907 pending,
1908 kc_next: usize::MAX,
1909 prev_full: true,
1910 prompt_len,
1911 rounds: 0,
1912 drafted: 0,
1913 accepted: 0,
1914 })
1915 }
1916
1917 /// One serve burst: run complete spec rounds until >= `target` NEW tokens have been
1918 /// emitted this burst (overshoot committed and returned) or EOS lands. Returns
1919 /// (tokens emitted this burst in order, drafted, accepted). The round body is the
1920 /// EAGER arm of `generate_spec_gemma`, kept behaviorally identical under default env
1921 /// (adapt/floor/pmin/in-round-cut logic verbatim) — gemma-spec-session-gate enforces
1922 /// byte-equality of the emitted stream against the one-shot at every burst width.
1923 pub fn gemma_spec_session_burst(
1924 &self,
1925 e: &Engine,
1926 d: &mut GemmaDraft,
1927 sess: &mut GemmaSpecSession,
1928 target: usize,
1929 k: usize,
1930 eos: &[u32],
1931 ) -> Result<(Vec<u32>, usize, usize), Box<dyn std::error::Error>> {
1932 let n_embd = self.cfg.n_embd as usize;
1933 if target == 0 {
1934 return Ok((Vec::new(), 0, 0));
1935 }
1936 let pmin: f32 = std::env::var("MEMRA_SPEC_PMIN")
1937 .ok()
1938 .and_then(|v| v.parse().ok())
1939 .unwrap_or(0.0);
1940 let pmin_ir_env: Option<f32> = std::env::var("MEMRA_SPEC_PMIN_INROUND")
1941 .ok()
1942 .and_then(|v| v.parse().ok());
1943 const PMIN_IR_DEFAULT: f32 = 0.7;
1944 let adapt = std::env::var("MEMRA_SPEC_ADAPT").as_deref() != Ok("0");
1945 let adapt_floor_default: usize = if self.cfg.n_embd >= 3500 {
1946 4
1947 } else if self.cfg.n_embd >= 2500 {
1948 2
1949 } else {
1950 1
1951 };
1952 let adapt_floor_env: Option<usize> = std::env::var("MEMRA_SPEC_ADAPT_FLOOR")
1953 .ok()
1954 .and_then(|v| v.parse().ok());
1955 let adapt_floor: usize = adapt_floor_env.unwrap_or(adapt_floor_default);
1956 let floor_ctx: usize = std::env::var("MEMRA_SPEC_FLOOR_CTX")
1957 .ok()
1958 .and_then(|v| v.parse().ok())
1959 .unwrap_or(1024);
1960 let floor_at = |pos: usize| -> usize {
1961 if adapt_floor_env.is_some() || pos < floor_ctx {
1962 adapt_floor
1963 } else if adapt_floor >= 4 {
1964 1
1965 } else {
1966 adapt_floor
1967 }
1968 };
1969 let cap_max: usize = std::env::var("MEMRA_SPEC_CAPMAX")
1970 .ok()
1971 .and_then(|v| v.parse().ok())
1972 .unwrap_or(7);
1973 let k_cap = k.min(cap_max).max(1);
1974 let mut kc = sess.kc_next.min(k_cap);
1975 let mut prev_full = sess.prev_full;
1976
1977 // per-burst device scratch (the one-shot allocates these per generation; per-burst
1978 // re-allocation is micro against a >= (K+1)-token round).
1979 let mut batch_d = e.stream().alloc_zeros::<u32>(k_cap + 1)?;
1980 let mut packed = e.stream().alloc_zeros::<u32>(2 * k_cap + 1)?;
1981 let mut p_d = e.stream().alloc_zeros::<f32>(k_cap.max(1))?;
1982 let mut pos_slots: Vec<CudaSlice<i32>> = (0..k_cap.max(1))
1983 .map(|_| e.htod_i32(&[0]))
1984 .collect::<Result<_, _>>()?;
1985 let mut g_seed = e.zeros(n_embd)?;
1986 // len_d lockstep at burst entry: the drafter's device-len arms read it, and the
1987 // previous burst's rollback set it — a fresh session's prime went host-len.
1988 for kvl in sess.cache.kv.iter_mut().flatten() {
1989 e.set_i32_one(&mut kvl.len_d, kvl.len as i32)?;
1990 }
1991
1992 let mut burst_out: Vec<u32> = Vec::with_capacity(target + k_cap + 1);
1993 let (mut drafted, mut accepted) = (0usize, 0usize);
1994 let mut ended = false;
1995 while burst_out.len() < target && !ended {
1996 let mut kr = if adapt { kc } else { k_cap };
1997 let dc_bucket: Option<usize> = {
1998 static DC: std::sync::OnceLock<bool> = std::sync::OnceLock::new();
1999 if *DC.get_or_init(|| std::env::var("MEMRA_GEMMA_DRAFT_DC").as_deref() != Ok("0")) {
2000 let ml = sess
2001 .cache
2002 .kv
2003 .iter()
2004 .flatten()
2005 .map(|kv| kv.len)
2006 .max()
2007 .unwrap_or(1);
2008 Some((ml + k_cap + 2).next_power_of_two().max(512))
2009 } else {
2010 None
2011 }
2012 };
2013 e.u32_set_k(&mut batch_d, sess.pending, 0)?;
2014 e.copy_into(&mut g_seed, 0, &sess.h, n_embd)?;
2015 for (j, slot) in pos_slots.iter_mut().take(kr).enumerate() {
2016 e.set_i32_one(slot, (sess.cache.pos + j) as i32)?;
2017 }
2018 let ir_now = match pmin_ir_env {
2019 Some(p) => p,
2020 None if sess.cache.pos >= floor_ctx && !prev_full => PMIN_IR_DEFAULT,
2021 None => 0.0,
2022 };
2023 // draft chain (the one-shot's run_chain, eager): reads g_seed, seeds batch_d.
2024 {
2025 let mut hc = e.uninit(n_embd)?;
2026 e.copy_into(&mut hc, 0, &g_seed, n_embd)?;
2027 let mut j = 0usize;
2028 while j < kr {
2029 let tv = batch_d.slice(j..j + 1);
2030 let (hn, h_next) = self.gemma4_draft_trunk_dev(
2031 e,
2032 d,
2033 &tv,
2034 &hc,
2035 &pos_slots[j],
2036 &sess.cache,
2037 dc_bucket,
2038 )?;
2039 let ld = e.matmul(&d.head, &hn, 1)?;
2040 e.argmax_token_device_col(&ld, 0, d.head.out_features(), &mut batch_d, j + 1)?;
2041 if pmin > 0.0 || ir_now > 0.0 {
2042 e.prob_of_token_device_col(
2043 &ld,
2044 &batch_d,
2045 j + 1,
2046 &mut p_d,
2047 j,
2048 d.head.out_features(),
2049 )?;
2050 }
2051 if let Some(map) = &d.d2t_dev {
2052 e.u32_map_k(&mut batch_d, map, j + 1)?;
2053 }
2054 hc = h_next;
2055 if ir_now > 0.0 && j + 1 < kr {
2056 let ph = e.dtoh(&p_d)?;
2057 if ph[j] < ir_now {
2058 kr = j + 1;
2059 break;
2060 }
2061 }
2062 j += 1;
2063 }
2064 }
2065 drafted += kr;
2066 sess.rounds += 1;
2067 let pos0 = sess.cache.pos;
2068 let (vam_d, vh) =
2069 self.gemma4_decode_step_t_am_dev(e, &batch_d, kr + 1, pos0, &mut sess.cache)?;
2070 e.u32_pack2(&batch_d, 1, kr, &vam_d, kr + 1, &mut packed)?;
2071 let host = e.dtoh_u32(&packed)?; // the round's ONE sync
2072 let dtoks: Vec<u32> = host[..kr].to_vec();
2073 let vam: Vec<u32> = host[kr..2 * kr + 1].to_vec();
2074 let mut m = 0usize;
2075 while m < kr {
2076 if dtoks[m] == vam[m] {
2077 m += 1;
2078 } else {
2079 break;
2080 }
2081 }
2082 prev_full = m == kr;
2083 accepted += m;
2084 // ---- ROUND COMPLETES UNCONDITIONALLY (the boundary law) ----
2085 // rollback rejected rows FIRST, then emit — an EOS mid-emission must still
2086 // leave cache rows == committed tokens.
2087 let keep = m + 1;
2088 for kvl in sess.cache.kv.iter_mut().flatten() {
2089 kvl.len -= (kr + 1) - keep;
2090 e.set_i32_one(&mut kvl.len_d, kvl.len as i32)?;
2091 }
2092 sess.cache.pos -= (kr + 1) - keep;
2093 // h for the next round = main hidden at the LAST KEPT position (verify row m).
2094 let hv2 = e.view(&vh, (kr + 1) * n_embd);
2095 let row = hv2.slice(m * n_embd..(m + 1) * n_embd);
2096 let mut hrow = e.uninit(n_embd)?;
2097 e.copy_view_into(&mut hrow, 0, &row, n_embd)?;
2098 sess.h = hrow;
2099 // emit: pending + accepted drafts. VISIBLE emission stops at the first EOS
2100 // (the one-shot's exact stream); COMMIT accounting continues — every kept row
2101 // must appear in `committed` or the cache-rows == committed invariant breaks.
2102 sess.committed.push(sess.pending);
2103 burst_out.push(sess.pending);
2104 if eos.contains(&sess.pending) {
2105 ended = true;
2106 }
2107 for &dt in &dtoks[..m] {
2108 sess.committed.push(dt);
2109 if !ended {
2110 burst_out.push(dt);
2111 if eos.contains(&dt) {
2112 ended = true;
2113 }
2114 }
2115 }
2116 sess.pending = vam[m];
2117 trim_adapt_learn(e, d, &vam)?;
2118 if adapt {
2119 let fl_now = floor_at(sess.cache.pos);
2120 kc = (m + 1).clamp(fl_now.min(k_cap), k_cap);
2121 if pmin > 0.0 {
2122 let ph = e.dtoh(&p_d)?;
2123 if let Some(fl) = ph[..kr].iter().position(|&p| p < pmin) {
2124 kc = kc.min((fl + 1).max(fl_now.min(k_cap)));
2125 }
2126 }
2127 }
2128 }
2129 sess.kc_next = kc;
2130 sess.prev_full = prev_full;
2131 sess.drafted += drafted;
2132 sess.accepted += accepted;
2133 Ok((burst_out, drafted, accepted))
2134 }
2135}
2136
2137impl HybridModel {
2138 /// PLAIN-DECODE CUDA-GRAPH loop (gemma4, greedy): one captured verify-trunk step
2139 /// (t=1, device tokens/pos/lens) replayed per token — the launch-gap eraser the
2140 /// decode decomposition demanded (2026-07-23: ~2.3ms/token idle at 128 launches).
2141 /// Self-feeding: argmax -> tok_d -> next embed; counters advance in-graph via
2142 /// spec_rollback_stream(base=1, acc=0). Tokens land in a device ring; ONE host sync
2143 /// per drain window. Captures are keyed on the (rung, window-side, f512-side) regime
2144 /// (the round-graph hint law); regime-crossing stretches run the same body eagerly.
2145 /// Caller guarantees: gemma4, greedy, shared_kv_layers == 0, prompt already primed
2146 /// (cache.pos = prompt len, host kvl.len mirrors set).
2147 pub fn gemma4_generate_plain_graph(
2148 &self,
2149 e: &Engine,
2150 cache: &mut Cache,
2151 last: u32,
2152 max_new: usize,
2153 eos: &[u32],
2154 ) -> Result<Vec<u32>, Box<dyn std::error::Error>> {
2155 const RING: usize = 64;
2156 const DRAIN: usize = 32; // replays per host sync
2157 let win_main = self
2158 .cfg
2159 .gemma4
2160 .as_ref()
2161 .map(|g| g.sliding_window as usize)
2162 .unwrap_or(0);
2163 let n_rows = cache.kv.len() + 1;
2164
2165 let was_tracking = e.ctx().is_event_tracking();
2166 if was_tracking {
2167 unsafe {
2168 e.ctx().disable_event_tracking();
2169 }
2170 }
2171 let r = self
2172 .gemma4_plain_graph_inner(e, cache, last, max_new, eos, RING, DRAIN, win_main, n_rows);
2173 if was_tracking {
2174 unsafe {
2175 e.ctx().enable_event_tracking();
2176 }
2177 }
2178 r
2179 }
2180
2181 #[allow(clippy::too_many_arguments)]
2182 #[allow(clippy::map_entry)] // allow: the init body is fallible (`?`); Entry::or_insert_with cannot propagate errors
2183 fn gemma4_plain_graph_inner(
2184 &self,
2185 e: &Engine,
2186 cache: &mut Cache,
2187 last: u32,
2188 max_new: usize,
2189 eos: &[u32],
2190 ring_cap: usize,
2191 drain: usize,
2192 win_main: usize,
2193 n_rows: usize,
2194 ) -> Result<Vec<u32>, Box<dyn std::error::Error>> {
2195 let mut scr = self.verify_stream_scratch(e, 1)?;
2196 let mut tok_d = e.stream().alloc_zeros::<u32>(1)?;
2197 e.u32_set_k(&mut tok_d, last, 0)?;
2198 let pos_ctr = e.htod_i32(&[cache.pos as i32])?;
2199 let mut pos_start_d = e.htod_i32(&[cache.pos as i32])?;
2200 let acc0 = e.stream().alloc_zeros::<u32>(2)?; // acc[0] = 0 -> counters +1
2201 let mut ring = e.stream().alloc_zeros::<u32>(ring_cap)?;
2202 let ptrs = crate::round_stream::kv_len_ptr_table(e, cache, Some(&pos_ctr))?;
2203 for kvl in cache.kv.iter_mut().flatten() {
2204 e.set_i32_one(&mut kvl.len_d, kvl.len as i32)?;
2205 }
2206 let ring_base = cache.pos; // baked into every capture
2207
2208 #[allow(clippy::type_complexity)]
2209 // allow: one-shot composite type; naming it would hide the shape that matters at the call site
2210 let mut graphs: std::collections::HashMap<
2211 (usize, bool, bool),
2212 (
2213 cudarc::driver::CudaGraph,
2214 Vec<Box<dyn std::any::Any + Send>>,
2215 ),
2216 > = Default::default();
2217
2218 let mut out: Vec<u32> = Vec::with_capacity(max_new);
2219 let mut drained = 0usize; // tokens read off the ring
2220
2221 // hint law (round-graph): the arm-gating bound must sit on the SAME side of every
2222 // crossover as the live lengths this capture serves, with the arms' own margins.
2223 let hint_for = |pos: usize| -> usize {
2224 if pos > win_main {
2225 pos + drain + 2
2226 } else if pos + 1 >= crate::fa512_min_tkv() {
2227 win_main.saturating_sub(2)
2228 } else {
2229 crate::fa512_min_tkv().saturating_sub(5)
2230 }
2231 };
2232 let regime_key = |pos: usize| -> (usize, bool, bool) {
2233 let rung = (pos + drain + 2).next_power_of_two().max(512);
2234 (rung, pos > win_main, pos + 1 >= crate::fa512_min_tkv())
2235 };
2236 // the whole [pos, pos+n) stretch must share one regime for a captured replay run.
2237 let stable_for = |pos: usize, n: usize| -> bool {
2238 regime_key(pos) == regime_key(pos + n)
2239 && (pos > win_main || pos + n + 2 < win_main)
2240 && (pos + 1 >= crate::fa512_min_tkv() || pos + n + 2 < crate::fa512_min_tkv())
2241 };
2242
2243 while out.len() < max_new {
2244 let pos = cache.pos;
2245 let hint = hint_for(pos);
2246 let scr_ptr: *mut crate::hybrid_forward::VerifyStreamScratch = &mut scr;
2247 let cache_ptr: *mut Cache = cache as *mut Cache;
2248 let tok_ptr: *mut CudaSlice<u32> = &mut tok_d;
2249 let ring_ptr: *mut CudaSlice<u32> = &mut ring;
2250 let start_ptr: *mut CudaSlice<i32> = &mut pos_start_d;
2251 let step = |e: &Engine| -> Result<(), Box<dyn std::error::Error>> {
2252 // SAFETY: single-threaded body; raw pointers alias the outer &mut only here.
2253 let (scr, cache, tok_d, ring, pos_start_d) = unsafe {
2254 (
2255 &mut *scr_ptr,
2256 &mut *cache_ptr,
2257 &mut *tok_ptr,
2258 &mut *ring_ptr,
2259 &mut *start_ptr,
2260 )
2261 };
2262 e.i32_copy_add(&pos_ctr, pos_start_d, 0)?;
2263 let (vam, _hn) =
2264 self.gemma4_verify_t_am_stream(e, tok_d, 1, &pos_ctr, hint, cache, scr)?;
2265 e.u32_copy(&vam, tok_d)?;
2266 e.plain_tok_ring(&vam, pos_start_d, ring_base, ring)?;
2267 e.spec_rollback_stream(&ptrs, pos_start_d, &acc0, 1, n_rows)?;
2268 Ok(())
2269 };
2270
2271 let n_left = max_new - out.len();
2272 let burst = drain.min(n_left);
2273 // MEMRA_G4PLAIN_EAGER=1: run the body eagerly every step (no capture/replay) —
2274 // splits "body semantics wrong" from "replay mechanics wrong" (round-graph law).
2275 let force_eager = std::env::var("MEMRA_G4PLAIN_EAGER").as_deref() == Ok("1");
2276 let steps_done = if !force_eager && burst >= 4 && stable_for(pos, burst + 3) {
2277 let key = regime_key(pos);
2278 if !graphs.contains_key(&key) {
2279 // capture cost = 3 SERVED steps (2 warmups + the captured run itself):
2280 // the loop is self-feeding, so they are real tokens in the ring.
2281 let g = e.capture_graph_retained(step)?;
2282 graphs.insert(key, g);
2283 3
2284 } else {
2285 let (g, _keep) = graphs.get(&key).unwrap();
2286 for _ in 0..burst {
2287 g.launch()?;
2288 }
2289 burst
2290 }
2291 } else {
2292 step(e)?; // eager fallback (same body)
2293 1
2294 };
2295
2296 // host mirrors + drain
2297 cache.pos += steps_done;
2298 for kvl in cache.kv.iter_mut().flatten() {
2299 kvl.len = cache.pos;
2300 }
2301 e.stream().synchronize()?;
2302 let ringh = e.dtoh_u32(&ring)?;
2303 let total = cache.pos - ring_base;
2304 while drained < total && out.len() < max_new {
2305 let t = ringh[drained % ring_cap];
2306 out.push(t);
2307 drained += 1;
2308 if eos.contains(&t) {
2309 return Ok(out);
2310 }
2311 }
2312 }
2313 Ok(out)
2314 }
2315}