1use crate::Engine;
6use crate::model::{EmbedHost, GpuTensor, HostExps};
7use cudarc::driver::CudaSlice;
8use memra_gguf::config::{ModelConfig, SwigluClamp};
9use memra_gguf::model_plan::{AttentionPlan, MlpPlan};
10use memra_gguf::source::{GgufSource, TensorSource};
11use memra_gguf::{GgmlType, GgufFile};
12use std::collections::HashMap;
13use std::sync::Arc;
14
15fn load_t(
18 e: &Engine,
19 src: &dyn TensorSource,
20 name: &str,
21) -> Result<GpuTensor, Box<dyn std::error::Error>> {
22 GpuTensor::load_from_source(e, src, name)
23}
24fn load_opt(
25 e: &Engine,
26 src: &dyn TensorSource,
27 name: &str,
28) -> Result<Option<GpuTensor>, Box<dyn std::error::Error>> {
29 GpuTensor::load_opt_from_source(e, src, name)
30}
31
32struct ResidencyBytes {
33 experts: HashMap<usize, usize>,
34 rest: usize,
35 saw_experts: bool,
36}
37
38fn block_index(name: &str) -> Option<usize> {
39 name.strip_prefix("blk.")?.split('.').next()?.parse().ok()
40}
41
42fn residency_bytes_by_device<'a>(
43 tensors: impl IntoIterator<Item = (&'a str, usize)>,
44 layer_devices: &[usize],
45 primary_device: usize,
46) -> ResidencyBytes {
47 let mut out = ResidencyBytes {
48 experts: HashMap::new(),
49 rest: 0,
50 saw_experts: false,
51 };
52 for (name, bytes) in tensors {
53 if name.starts_with("blk.") && name.contains("_exps.") {
54 let device = block_index(name)
55 .and_then(|il| layer_devices.get(il).copied())
56 .unwrap_or(primary_device);
57 *out.experts.entry(device).or_default() += bytes;
58 out.saw_experts = true;
59 } else {
60 out.rest += bytes;
61 }
62 }
63 out
64}
65
66pub(crate) struct ResidentPlan {
69 primary_device: usize,
70 layer_devices: Vec<usize>,
71 layer_counts: HashMap<usize, usize>,
72 exact_expert_bytes: Option<HashMap<usize, usize>>,
73 trunk_bytes: usize,
74 decisions: HashMap<usize, bool>,
75 pp: bool,
76}
77
78#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
88enum StepExpertArtifact {
89 #[default]
90 E4m3,
91 Nvfp4,
92}
93
94#[derive(Clone, Debug, Default)]
95struct StepParallelLoadConfig {
96 ep_specs: Vec<crate::tp::StepEpLayerSpec>,
97 tp_specs: Vec<crate::tp::StepTpLayerSpec>,
98 native_p2p: bool,
99 ep_device_arithmetic: bool,
100 f32_mirror: bool,
101 bulk_p2p: bool,
102 nvfp4_device_routes: bool,
103 auto_parallel: bool,
104 expert_artifact: StepExpertArtifact,
105}
106
107#[derive(Default)]
108pub(crate) struct StepParallelRuntimeRegistry {
109 config: StepParallelLoadConfig,
110 runtimes: HashMap<(Vec<usize>, bool, bool, bool), Arc<crate::tp::TpE4m3HostBounce>>,
111}
112
113#[derive(Clone, Copy, Debug, PartialEq, Eq)]
114enum StepExpertLayout {
115 TensorParallel,
116 ExpertParallel,
117}
118
119#[derive(Clone, Debug, PartialEq, Eq)]
120struct StepExpertSelection {
121 spec: crate::tp::StepEpLayerSpec,
122 layout: StepExpertLayout,
123 configured_by_tp: bool,
124}
125
126fn select_step_expert_layout(
127 layer: usize,
128 ep_specs: &[crate::tp::StepEpLayerSpec],
129 tp_specs: &[crate::tp::StepTpLayerSpec],
130) -> Result<Option<StepExpertSelection>, String> {
131 let ep = ep_specs.iter().find(|spec| spec.layer == layer);
132 let tp = tp_specs.iter().find(|spec| spec.layer == layer);
133 if ep.is_some() && tp.is_some() {
134 return Err(format!(
135 "Step layer {layer} cannot enable MEMRA_STEP_EP and MEMRA_STEP_TP together"
136 ));
137 }
138 Ok(match (ep, tp) {
139 (Some(spec), None) => Some(StepExpertSelection {
140 spec: spec.clone(),
141 layout: StepExpertLayout::ExpertParallel,
142 configured_by_tp: false,
143 }),
144 (None, Some(spec)) => Some(StepExpertSelection {
145 spec: spec.clone(),
146 layout: if spec.devices.len() > 2 {
147 StepExpertLayout::ExpertParallel
148 } else {
149 StepExpertLayout::TensorParallel
150 },
151 configured_by_tp: true,
152 }),
153 (None, None) => None,
154 (Some(_), Some(_)) => unreachable!(),
155 })
156}
157
158impl StepParallelRuntimeRegistry {
159 fn with_config(config: StepParallelLoadConfig) -> Self {
160 Self {
161 config,
162 runtimes: HashMap::new(),
163 }
164 }
165
166 fn tp_spec(&self, layer: usize) -> Option<&crate::tp::StepTpLayerSpec> {
167 self.config.tp_specs.iter().find(|spec| spec.layer == layer)
168 }
169
170 fn expert_selection(&self, layer: usize) -> Result<Option<StepExpertSelection>, String> {
171 select_step_expert_layout(layer, &self.config.ep_specs, &self.config.tp_specs)
172 }
173
174 fn runtime(
175 &mut self,
176 devices: &[usize],
177 native_p2p: bool,
178 ep_device_arithmetic: bool,
179 ) -> Result<Arc<crate::tp::TpE4m3HostBounce>, Box<dyn std::error::Error>> {
180 let bulk_p2p = self.config.bulk_p2p && native_p2p;
181 let key = (devices.to_vec(), native_p2p, ep_device_arithmetic, bulk_p2p);
182 if let Some(runtime) = self.runtimes.get(&key) {
183 return Ok(Arc::clone(runtime));
184 }
185 let runtime = Arc::new(crate::tp::TpE4m3HostBounce::new_configured(
186 devices,
187 native_p2p,
188 ep_device_arithmetic,
189 bulk_p2p,
190 )?);
191 let names = runtime.device_names()?;
192 if names
193 .iter()
194 .any(|name| !name.contains("RTX PRO 6000") || !name.contains("Blackwell"))
195 {
196 return Err(format!(
197 "Step distributed execution is qualified only on RTX PRO 6000 Blackwell, \
198 got {names:?}"
199 )
200 .into());
201 }
202 self.runtimes.insert(key, Arc::clone(&runtime));
203 Ok(runtime)
204 }
205}
206
207impl ResidentPlan {
208 fn from_layout(
209 src: &dyn TensorSource,
210 primary_device: usize,
211 layer_devices: Vec<usize>,
212 pp: bool,
213 ) -> Self {
214 let mut layer_counts = HashMap::new();
215 for &device in &layer_devices {
216 *layer_counts.entry(device).or_default() += 1;
217 }
218 let (exact_expert_bytes, trunk_bytes) = match src.gguf() {
219 Some(g) => {
220 let bytes = residency_bytes_by_device(
221 g.tensors
222 .iter()
223 .map(|t| (t.name.as_str(), t.n_bytes as usize)),
224 &layer_devices,
225 primary_device,
226 );
227 if bytes.saw_experts {
228 (Some(bytes.experts), bytes.rest)
229 } else {
230 (None, 0)
231 }
232 }
233 None => (None, 0),
234 };
235 Self {
236 primary_device,
237 layer_devices,
238 layer_counts,
239 exact_expert_bytes,
240 trunk_bytes,
241 decisions: HashMap::new(),
242 pp,
243 }
244 }
245
246 pub(crate) fn unsharded(e: &Engine, src: &dyn TensorSource, cfg: &ModelConfig) -> Self {
247 let device = e.ctx().ordinal();
248 Self::from_layout(src, device, vec![device; cfg.n_layer as usize], false)
249 }
250
251 pub(crate) fn pp(
252 e: &Engine,
253 src: &dyn TensorSource,
254 cfg: &ModelConfig,
255 n_trunk: usize,
256 ) -> Result<Self, Box<dyn std::error::Error>> {
257 let primary = e.ctx().ordinal();
258 let Some(_fence) = crate::pp::pp_cuts(n_trunk) else {
259 return Ok(Self::unsharded(e, src, cfg));
260 };
261 let mut layer_devices = vec![primary; cfg.n_layer as usize];
262 for (il, device) in layer_devices.iter_mut().take(n_trunk).enumerate() {
263 *device = crate::pp::layer_engine(e, n_trunk, il)?.ctx().ordinal();
264 }
265 Ok(Self::from_layout(src, primary, layer_devices, true))
266 }
267
268 fn exclude_distributed_expert_layers(&mut self, specs: impl IntoIterator<Item = usize>) {
272 for layer in specs {
273 let device = self
274 .layer_devices
275 .get(layer)
276 .copied()
277 .unwrap_or(self.primary_device);
278 if let Some(count) = self.layer_counts.get_mut(&device) {
279 *count = count.saturating_sub(1);
280 }
281 }
282 }
283
284 fn should_reside(&mut self, e: &Engine, il: usize, per_layer: usize) -> bool {
285 let device = self
286 .layer_devices
287 .get(il)
288 .copied()
289 .unwrap_or(self.primary_device);
290 debug_assert_eq!(e.ctx().ordinal(), device);
291 if let Some(&decision) = self.decisions.get(&device) {
292 return decision;
293 }
294 if std::env::var("MEMRA_MOE_RESIDENT").as_deref() == Ok("0") {
295 self.decisions.insert(device, false);
296 return false;
297 }
298 let (free, _total) = match e.ctx().mem_get_info() {
299 Ok(v) => v,
300 Err(_) => {
301 self.decisions.insert(device, false);
302 return false;
303 }
304 };
305 let projected = self
306 .exact_expert_bytes
307 .as_ref()
308 .map(|bytes| bytes.get(&device).copied().unwrap_or(0))
309 .unwrap_or(per_layer * self.layer_counts.get(&device).copied().unwrap_or(1));
310 let budget = std::env::var("MEMRA_MOE_RESIDENT_GB")
311 .ok()
312 .and_then(|v| v.parse::<f64>().ok())
313 .map(|gb| (gb * 1e9) as usize)
314 .unwrap_or_else(|| {
315 let reserve = std::env::var("MEMRA_MOE_RESIDENT_HEADROOM_GB")
316 .ok()
317 .and_then(|v| v.parse::<f64>().ok())
318 .map(|gb| (gb * 1e9) as usize)
319 .unwrap_or(2_000_000_000);
320 free.saturating_sub(self.trunk_bytes + reserve)
321 });
322 let ok = projected <= budget;
323 eprintln!(
324 "[moe] resident-experts decision ({}dev{}): experts {:.2}GB + trunk {:.2}GB vs free {:.2}GB (expert budget {:.2}GB) -> {}",
325 if self.pp { "PP " } else { "" },
326 device,
327 projected as f64 / 1e9,
328 self.trunk_bytes as f64 / 1e9,
329 free as f64 / 1e9,
330 budget as f64 / 1e9,
331 if ok { "RESIDENT" } else { "SLRU cache" }
332 );
333 self.decisions.insert(device, ok);
334 ok
335 }
336}
337
338fn load_mixer_kind(
340 e: &Engine,
341 src: &dyn TensorSource,
342 cfg: &ModelConfig,
343 il: u32,
344 attention: &AttentionPlan,
345 step_runtimes: &mut StepParallelRuntimeRegistry,
346) -> Result<Mixer, Box<dyn std::error::Error>> {
347 let p = |s: &str| format!("blk.{il}.{s}");
348 Ok(match attention {
349 AttentionPlan::Mla(mla) => Mixer::Mla(MlaAttnLayer::load(e, src, il, mla)?),
350 AttentionPlan::Full(full)
351 | AttentionPlan::SlidingWindow {
352 attention: full, ..
353 } => {
354 Mixer::Full(FullAttnLayer {
355 wq: load_t(e, src, &p("attn_q.weight"))?,
356 wk: load_t(e, src, &p("attn_k.weight"))?,
357 wv: match load_opt(e, src, &p("attn_v.weight"))? {
362 Some(v) => v,
363 None => load_t(e, src, &p("attn_k.weight"))?,
364 },
365 wo: load_t(e, src, &p("attn_output.weight"))?,
366 q_norm: load_t(e, src, &p("attn_q_norm.weight"))?,
367 k_norm: load_t(e, src, &p("attn_k_norm.weight"))?,
368 attn_gate: if full.output_gate
372 == memra_gguf::config::AttentionGateKind::SeparateHead
373 {
374 Some(load_t(e, src, &p("attn_gate.weight"))?)
375 } else {
376 None
377 },
378 step_tp_qkv: build_step_tp_qkv(e, src, cfg, il as usize, step_runtimes)?,
379 })
380 }
381 AttentionPlan::KimiDeltaNet(kda) => {
384 Mixer::Kda(crate::kda::KdaAttnLayer::load(e, src, il, kda)?)
385 }
386 AttentionPlan::GatedDeltaNet(geometry) => Mixer::Linear(LinearAttnLayer {
387 geometry: *geometry,
388 wqkv: load_t(e, src, &p("attn_qkv.weight"))?,
389 wqkv_gate: load_t(e, src, &p("attn_gate.weight"))?,
390 ssm_beta: load_t(e, src, &p("ssm_beta.weight"))?,
391 ssm_alpha: load_t(e, src, &p("ssm_alpha.weight"))?,
392 ssm_a: load_t(e, src, &p("ssm_a"))?,
393 ssm_dt: load_t(e, src, &p("ssm_dt.bias"))?,
394 ssm_conv1d: load_t(e, src, &p("ssm_conv1d.weight"))?,
395 ssm_norm: load_t(e, src, &p("ssm_norm.weight"))?,
396 ssm_out: load_t(e, src, &p("ssm_out.weight"))?,
397 }),
398 })
399}
400
401#[allow(clippy::too_many_arguments)] pub(crate) fn load_ffn(
409 e: &Engine,
410 src: &dyn TensorSource,
411 cfg: &ModelConfig,
412 mlp: &MlpPlan,
413 il: u32,
414 spill: Option<(&GgufFile, &mut crate::spill::SpillCtx)>,
415 resident: &mut ResidentPlan,
416 step_runtimes: &mut StepParallelRuntimeRegistry,
417) -> Result<Ffn, Box<dyn std::error::Error>> {
418 let p = |s: &str| format!("blk.{il}.{s}");
419 let artifact_dense = matches!(mlp, MlpPlan::Moe(_))
426 && !src.has(&p("ffn_gate_exps.weight"))
427 && !src.has(&p("ffn_gate_up_exps.weight"))
428 && src.has(&p("ffn_gate.weight"));
429 Ok(if artifact_dense {
430 Ffn::Dense {
431 ffn_gate: GpuTensor::load_from_source(e, src, &p("ffn_gate.weight"))?,
432 ffn_up: GpuTensor::load_from_source(e, src, &p("ffn_up.weight"))?,
433 ffn_down: GpuTensor::load_from_source(e, src, &p("ffn_down.weight"))?,
434 }
435 } else if let MlpPlan::Moe(moe) = mlp {
436 let n_expert = moe.expert_count as usize;
437 let (gate_exps, up_exps, down_exps) = match spill {
443 Some((g, ctx)) => (
444 HostExps::load_tiered(e, g, &p("ffn_gate_exps.weight"), ctx)?,
445 HostExps::load_tiered(e, g, &p("ffn_up_exps.weight"), ctx)?,
446 HostExps::load_tiered(e, g, &p("ffn_down_exps.weight"), ctx)?,
447 ),
448 None => {
449 let exps = |e: &Engine, n: &str| -> Result<HostExps, Box<dyn std::error::Error>> {
450 if src.has(n) {
451 HostExps::load_stacked_from_source(e, src, n)
452 } else {
453 HostExps::load_from_source(e, src, n, n_expert)
454 }
455 };
456 let fused = p("ffn_gate_up_exps.weight");
458 if !src.has(&p("ffn_gate_exps.weight")) && src.has(&fused) {
459 let ff = moe.expert_intermediate_size as usize;
460 (
461 HostExps::load_stacked_split_from_source(e, src, &fused, 0, ff)?,
462 HostExps::load_stacked_split_from_source(e, src, &fused, ff, 2 * ff)?,
463 exps(e, &p("ffn_down_exps.weight"))?,
464 )
465 } else {
466 (
467 exps(e, &p("ffn_gate_exps.weight"))?,
468 exps(e, &p("ffn_up_exps.weight"))?,
469 exps(e, &p("ffn_down_exps.weight"))?,
470 )
471 }
472 }
473 };
474 let (step_ep, step_tp) = build_step_distributed_exps(
475 e,
476 cfg,
477 src,
478 il as usize,
479 &gate_exps,
480 &up_exps,
481 &down_exps,
482 step_runtimes,
483 )?;
484 let dev_exps = if step_ep.is_some() || step_tp.is_some() {
490 None
491 } else {
492 build_dev_exps(e, resident, il as usize, &gate_exps, &up_exps, &down_exps)?
493 };
494 let mut macro_row = vec![1.0f32; 3 * n_expert];
496 for (slot, exps) in [(0usize, &gate_exps), (1, &up_exps), (2, &down_exps)] {
497 if let Some(ms) = exps.macros.as_ref() {
498 macro_row[slot * n_expert..(slot + 1) * n_expert].copy_from_slice(ms);
499 }
500 }
501 let has_macros = macro_row.iter().any(|&m| m != 1.0);
502 let dev_macros = e.htod(¯o_row)?;
503 let exp_probs_b = src
506 .find(&p("exp_probs_b.bias"))
507 .map(|v| memra_gguf::dequant::dequantize(v.ggml_type, &v.bytes, n_expert));
508 if exp_probs_b.is_none()
517 && matches!(
518 moe.router,
519 memra_gguf::model_plan::RouterPlan::Sigmoid {
520 selection_bias: true,
521 ..
522 } | memra_gguf::model_plan::RouterPlan::SqrtSoftplus {
523 selection_bias: true,
524 ..
525 }
526 )
527 {
528 return Err(format!(
529 "layer {il}: {} is absent, but the compiled ModelPlan declares a router with a \
530 selection bias ({:?}). Refusing to load: a zero-filled bias would route to \
531 different experts than this model does, silently. Either the checkpoint does \
532 not carry the tensor, or this arch has no `exp_probs_b.bias` entry in \
533 hf_mapping's ggml->HF map",
534 p("exp_probs_b.bias"),
535 moe.router
536 )
537 .into());
538 }
539 let active_experts = src.active_experts(il).map(<[bool]>::to_vec);
540 let route_bias = exp_probs_b.clone().unwrap_or_else(|| vec![0.0; n_expert]);
541 let active_row: Vec<u8> = active_experts
542 .as_ref()
543 .map(|mask| mask.iter().map(|&is_active| u8::from(is_active)).collect())
544 .unwrap_or_else(|| vec![1; n_expert]);
545 let exp_probs_b_dev = e.htod(&route_bias)?;
546 let active_experts_dev = e.htod_bytes(&active_row)?;
547 let gate_shexp = load_opt(e, src, &p("ffn_gate_shexp.weight"))?;
548 let up_shexp = load_opt(e, src, &p("ffn_up_shexp.weight"))?;
549 let down_shexp = load_opt(e, src, &p("ffn_down_shexp.weight"))?;
550 if moe.shared.is_some()
558 && (gate_shexp.is_none() || up_shexp.is_none() || down_shexp.is_none())
559 {
560 return Err(format!(
561 "layer {il}: the compiled ModelPlan declares an always-on shared expert, but \
562 {}{}{} could not be resolved in the checkpoint. Refusing to load: dropping the \
563 shared branch computes a different model, silently. Either the checkpoint does \
564 not carry it, or this arch's shared-expert spelling is missing from \
565 hf_mapping's ggml->HF map",
566 if gate_shexp.is_none() {
567 format!("{} ", p("ffn_gate_shexp.weight"))
568 } else {
569 String::new()
570 },
571 if up_shexp.is_none() {
572 format!("{} ", p("ffn_up_shexp.weight"))
573 } else {
574 String::new()
575 },
576 if down_shexp.is_none() {
577 p("ffn_down_shexp.weight")
578 } else {
579 String::new()
580 },
581 )
582 .into());
583 }
584 Ffn::Moe(MoeWeights {
585 gate_inp: load_t(e, src, &p("ffn_gate_inp.weight"))?,
586 gate_inp_shexp: load_opt(e, src, &p("ffn_gate_inp_shexp.weight"))?,
587 exp_probs_b,
588 exp_probs_b_dev,
589 active_experts,
590 active_experts_dev,
591 gate_exps,
592 up_exps,
593 down_exps,
594 gate_shexp,
595 up_shexp,
596 down_shexp,
597 dev_exps,
598 step_ep,
599 step_tp,
600 glm5_ep: None,
601 dev_macros,
602 has_macros,
603 w4a16_bf16_activations: matches!(
604 src.expert_activation_precision(),
605 memra_gguf::source::ExpertActivationPrecision::Bf16
606 ),
607 })
608 } else {
609 Ffn::Dense {
610 ffn_gate: load_t(e, src, &p("ffn_gate.weight"))?,
611 ffn_up: load_t(e, src, &p("ffn_up.weight"))?,
612 ffn_down: load_t(e, src, &p("ffn_down.weight"))?,
613 }
614 })
615}
616
617fn host_e4m3_bank(
618 exps: &HostExps,
619) -> Result<crate::tp::E4m3ExpertBank<'_>, Box<dyn std::error::Error>> {
620 if exps.qtype != crate::QT_F8_E4M3_BLK {
621 return Err(format!(
622 "Step EP requires native block-E4M3 expert banks, got qtype {}",
623 exps.qtype
624 )
625 .into());
626 }
627 let scales = exps
628 .fp8_blk
629 .as_ref()
630 .ok_or("Step EP native expert bank has no block-E4M3 scale plane")?;
631 Ok(crate::tp::E4m3ExpertBank {
632 codes: exps.bytes.as_bytes(),
633 scales: &scales.scales,
634 expert_count: exps.n_expert,
635 out_features: exps.out_f,
636 in_features: exps.in_f,
637 })
638}
639
640fn validate_step_expert_specs(
641 contract: &crate::parallel::ModelParallelContract,
642 flag: &str,
643 specs: &[crate::tp::StepEpLayerSpec],
644 allow_dense_attention_only: bool,
645) -> Result<(), Box<dyn std::error::Error>> {
646 for candidate in specs {
647 if candidate.layer >= contract.trunk_layers {
648 return Err(format!(
649 "{flag} layer {} is outside Step trunk layers 0..{}",
650 candidate.layer, contract.trunk_layers
651 )
652 .into());
653 }
654 if candidate.layer < contract.dense_prefix_layers {
655 if allow_dense_attention_only {
656 continue;
657 }
658 return Err(format!(
659 "{flag} layer {} is outside Step routed-expert layers {}..{}",
660 candidate.layer, contract.dense_prefix_layers, contract.trunk_layers
661 )
662 .into());
663 }
664 }
665 Ok(())
666}
667
668fn validate_step_expert_activation_layout(
669 cfg: &ModelConfig,
670 flag: &str,
671 selection: &StepExpertSelection,
672) -> Result<(), Box<dyn std::error::Error>> {
673 let _ = (cfg, flag, selection);
679 Ok(())
680}
681
682fn parse_auto_w4a16_bf16_mmv(value: Option<&str>) -> Result<bool, String> {
683 match value {
684 None => Ok(true),
685 Some("0") => Ok(false),
686 Some("1") => Ok(true),
687 Some(value) => Err(format!(
688 "MEMRA_BF16_MMV={value:?} is invalid under MEMRA_PARALLEL=auto; expected 0 or 1"
689 )),
690 }
691}
692
693fn parse_auto_parallel_tp_attention(value: Option<&str>) -> Result<bool, String> {
694 match value {
695 None | Some("") | Some("0") => Ok(false),
696 Some("1") => Ok(true),
697 Some(value) => Err(format!(
698 "MEMRA_PARALLEL_TP_ATTENTION={value:?} is invalid; expected 0 or 1"
699 )),
700 }
701}
702
703fn auto_parallel_tp_attention_enabled() -> Result<bool, String> {
704 parse_auto_parallel_tp_attention(std::env::var("MEMRA_PARALLEL_TP_ATTENTION").ok().as_deref())
705}
706
707fn prepare_auto_parallel(
713 src: &dyn TensorSource,
714 cfg: &ModelConfig,
715 plan: &memra_gguf::model_plan::ModelPlan,
716) -> Result<Option<crate::parallel::AutoParallelPlacement>, Box<dyn std::error::Error>> {
717 let Some(devices) = crate::tp::auto_parallel_devices()? else {
718 return Ok(None);
719 };
720 if std::env::var_os("MEMRA_PP_STAGES").is_some()
721 || std::env::var_os("MEMRA_PP_DEVICES").is_some()
722 || std::env::var_os("MEMRA_PP_SPLITS").is_some()
723 {
724 return Err(
725 "MEMRA_PARALLEL=auto cannot be combined with MEMRA_PP_STAGES, MEMRA_PP_DEVICES, or \
726 MEMRA_PP_SPLITS"
727 .into(),
728 );
729 }
730 let placement = crate::parallel::plan_auto_parallel(src, cfg, plan, &devices)?;
731 let auto_w4a16_bf16 = placement.backend == crate::parallel::AutoParallelBackend::ExpertParallel
732 && matches!(
733 src.expert_activation_precision(),
734 memra_gguf::source::ExpertActivationPrecision::Bf16
735 );
736 let bf16_nonexpert = if auto_w4a16_bf16 {
737 let explicit = match std::env::var("MEMRA_BF16_MMV") {
738 Ok(value) => Some(value),
739 Err(std::env::VarError::NotPresent) => None,
740 Err(error) => return Err(format!("cannot read MEMRA_BF16_MMV: {error}").into()),
741 };
742 let enabled = parse_auto_w4a16_bf16_mmv(explicit.as_deref())?;
743 if enabled && explicit.is_none() {
744 unsafe {
747 std::env::set_var("MEMRA_BF16_MMV", "1");
748 }
749 }
750 match (enabled, explicit.is_some()) {
751 (true, false) => "bf16-resident(auto)",
752 (true, true) => "bf16-resident(explicit)",
753 (false, true) => "f32-expanded(explicit-rollback)",
754 (false, false) => unreachable!("unset auto W4A16 defaults BF16 residency on"),
755 }
756 } else {
757 "placement-default"
758 };
759 if placement.backend == crate::parallel::AutoParallelBackend::Pipeline {
760 let stages = placement.devices.len();
761 let device_list = placement
762 .devices
763 .iter()
764 .map(usize::to_string)
765 .collect::<Vec<_>>()
766 .join(",");
767 let splits = placement
768 .pipeline_splits
769 .iter()
770 .map(usize::to_string)
771 .collect::<Vec<_>>()
772 .join(",");
773 unsafe {
776 std::env::set_var("MEMRA_PP_STAGES", stages.to_string());
777 std::env::set_var("MEMRA_PP_DEVICES", &device_list);
778 std::env::set_var("MEMRA_PP_SPLITS", &splits);
779 }
780 }
781 let family = if placement.routed_layers.is_empty() {
782 "dense-transformer"
783 } else {
784 "routed-moe"
785 };
786 eprintln!(
787 "[parallel-auto] family={family} variant={:?} devices={:?} placement={} \
788 checkpoint_peak={:.2}GB ep_root={:.2}GB ep_peer={:.2}GB reserve={:.2}GB \
789 capacity={:?} splits={:?} bf16_nonexpert={bf16_nonexpert} \
790 wavefront=off(default) performance_claim=false",
791 cfg.name,
792 placement.devices,
793 match placement.backend {
794 crate::parallel::AutoParallelBackend::Pipeline => "pipeline",
795 crate::parallel::AutoParallelBackend::ExpertParallel => "expert-parallel",
796 },
797 placement.checkpoint_peak_bytes as f64 / 1e9,
798 placement.expert_root_bytes as f64 / 1e9,
799 placement.expert_peer_bytes as f64 / 1e9,
800 placement.reserve_bytes as f64 / 1e9,
801 placement.device_capacity_bytes,
802 placement.pipeline_splits,
803 );
804 Ok(Some(placement))
805}
806
807fn prepare_step_parallel_load(
808 e: &Engine,
809 src: &dyn TensorSource,
810 cfg: &ModelConfig,
811 trunk_layers: usize,
812 auto_placement: Option<&crate::parallel::AutoParallelPlacement>,
813) -> Result<StepParallelLoadConfig, Box<dyn std::error::Error>> {
814 let mut tp_specs = crate::tp::step_tp_layer_specs()?;
815 let mut ep_specs = crate::tp::step_ep_layer_specs()?;
816 let device_arithmetic = crate::tp::step_ep_device_arithmetic_enabled()?;
817 let f32_mirror = crate::tp::step_tp_f32_mirror_enabled()?;
818 let bulk_p2p = crate::tp::step_tp_bulk_p2p_enabled()?;
819 let mut native_p2p = crate::tp::step_tp_native_p2p_enabled()?;
820 let mut nvfp4_device_routes = crate::tp::step_nvfp4_dev_routes_enabled()?;
821 let auto_tp_attention = auto_parallel_tp_attention_enabled()?;
822 let mut auto_parallel = false;
823 if auto_tp_attention && auto_placement.is_none() {
824 return Err(
825 "MEMRA_PARALLEL_TP_ATTENTION=1 requires MEMRA_PARALLEL=auto; explicit per-layer \
826 recipes remain under MEMRA_STEP_TP"
827 .into(),
828 );
829 }
830 if let Some(placement) = auto_placement {
831 if !tp_specs.is_empty() || !ep_specs.is_empty() {
832 return Err(
833 "MEMRA_PARALLEL=auto cannot be combined with MEMRA_STEP_EP or MEMRA_STEP_TP".into(),
834 );
835 }
836 if placement.backend == crate::parallel::AutoParallelBackend::Pipeline {
837 if auto_tp_attention {
838 return Err(
839 "MEMRA_PARALLEL_TP_ATTENTION=1 requires automatic whole-expert EP; the \
840 selected checkpoint fits only the pipeline backend"
841 .into(),
842 );
843 }
844 return Ok(StepParallelLoadConfig::default());
845 }
846 if auto_tp_attention {
847 let contract = crate::parallel::ModelParallelContract::from_model(cfg)?;
848 if !contract.tensor_attention_supported {
849 return Err(format!(
850 "MEMRA_PARALLEL_TP_ATTENTION=1 cannot shard attention for {:?}: the \
851 compiled ModelPlan has no generic tensor-attention contract",
852 cfg.name
853 )
854 .into());
855 }
856 tp_specs = (0..trunk_layers)
857 .map(|layer| crate::tp::StepTpLayerSpec {
858 layer,
859 devices: placement.devices.clone(),
860 })
861 .collect();
862 ep_specs.clear();
863 } else {
864 ep_specs = placement
865 .routed_layers
866 .iter()
867 .map(|&layer| crate::tp::StepEpLayerSpec {
868 layer,
869 devices: placement.devices.clone(),
870 })
871 .collect();
872 }
873 auto_parallel = true;
874 native_p2p = true;
875 nvfp4_device_routes = matches!(
876 src.expert_activation_precision(),
877 memra_gguf::source::ExpertActivationPrecision::Bf16
878 );
879 eprintln!(
880 "[parallel-auto-backend] devices={:?} routed_layers={} native_p2p=true \
881 artifact_activation={:?} attention_layout={} expert_layout=expert-parallel \
882 backend={} performance_claim=false",
883 placement.devices,
884 placement.routed_layers.len(),
885 src.expert_activation_precision(),
886 if auto_tp_attention {
887 "tensor-parallel"
888 } else {
889 "root-local"
890 },
891 if nvfp4_device_routes {
892 "nvfp4-w4a16"
893 } else {
894 "artifact-selected-host-oracle"
895 },
896 );
897 }
898 if tp_specs.is_empty() {
899 if auto_tp_attention {
900 return Err("MEMRA_PARALLEL_TP_ATTENTION=1 produced no tensor-parallel layers".into());
901 }
902 if device_arithmetic || f32_mirror || bulk_p2p {
903 return Err(
904 "MEMRA_STEP_EP_DEVICE_ARITHMETIC=1, MEMRA_STEP_TP_F32_MIRROR=1, or \
905 MEMRA_STEP_TP_BULK_P2P=1 requires MEMRA_STEP_TP; device arithmetic and bulk \
906 transport also require MEMRA_STEP_TP_NATIVE_P2P=1"
907 .into(),
908 );
909 }
910 if nvfp4_device_routes && ep_specs.is_empty() {
911 return Err(
912 "MEMRA_STEP_NVFP4_DEV_ROUTES=1 requires MEMRA_STEP_EP or MEMRA_STEP_TP".into(),
913 );
914 }
915 if nvfp4_device_routes && !native_p2p {
916 return Err("MEMRA_STEP_NVFP4_DEV_ROUTES=1 with explicit EP requires \
917 MEMRA_STEP_TP_NATIVE_P2P=1"
918 .into());
919 }
920 let expert_artifact = if ep_specs.is_empty() {
923 StepExpertArtifact::default()
924 } else if nvfp4_device_routes
925 && matches!(
926 src.expert_activation_precision(),
927 memra_gguf::source::ExpertActivationPrecision::Bf16
928 )
929 {
930 StepExpertArtifact::Nvfp4
935 } else {
936 let contract = crate::parallel::ModelParallelContract::from_model(cfg)?;
937 validate_step_expert_specs(&contract, "MEMRA_STEP_EP", &ep_specs, false)?;
938 let layer_owners = (0..trunk_layers)
939 .map(|layer| {
940 crate::pp::layer_engine(e, trunk_layers, layer)
941 .map(|engine| engine.ctx().ordinal())
942 })
943 .collect::<Result<Vec<_>, _>>()?;
944 let mut runtime_groups = Vec::<Vec<usize>>::new();
945 for spec in &ep_specs {
946 let owner = layer_owners[spec.layer];
947 if !spec.devices.contains(&owner) {
948 return Err(format!(
949 "MEMRA_STEP_EP layer {} owning device {owner} is absent from {:?}",
950 spec.layer, spec.devices
951 )
952 .into());
953 }
954 if nvfp4_device_routes && spec.devices.first().copied() != Some(owner) {
955 return Err(format!(
956 "MEMRA_STEP_NVFP4_DEV_ROUTES=1 requires the owning device first; \
957 layer {} owner={owner} devices={:?}",
958 spec.layer, spec.devices
959 )
960 .into());
961 }
962 if !runtime_groups.contains(&spec.devices) {
963 runtime_groups.push(spec.devices.clone());
964 }
965 }
966 for devices in &runtime_groups {
967 let hardware = crate::parallel::detect_uniform_hardware(devices)?;
968 if !contract.hardware_targets.contains(&hardware) {
969 return Err(format!(
970 "{} has no qualified {hardware:?} EP contract for devices {devices:?}",
971 contract.variant
972 )
973 .into());
974 }
975 }
976 let artifact = match crate::parallel::validate_fp8_expert_checkpoint(src, &contract) {
977 Ok(_) => StepExpertArtifact::E4m3,
978 Err(fp8_error) => {
979 match crate::parallel::validate_nvfp4_expert_checkpoint(src, &contract) {
980 Ok(_) => StepExpertArtifact::Nvfp4,
981 Err(nvfp4_error) => {
982 return Err(format!(
983 "Step checkpoint qualifies as neither native expert artifact \
984 class: [E4M3] {fp8_error} [NVFP4] {nvfp4_error}"
985 )
986 .into());
987 }
988 }
989 }
990 };
991 if nvfp4_device_routes && artifact != StepExpertArtifact::Nvfp4 {
992 return Err(
993 "MEMRA_STEP_NVFP4_DEV_ROUTES=1 requires a native ModelOpt NVFP4 expert \
994 artifact"
995 .into(),
996 );
997 }
998 artifact
999 };
1000 return Ok(StepParallelLoadConfig {
1001 ep_specs,
1002 native_p2p,
1003 nvfp4_device_routes,
1004 auto_parallel,
1005 expert_artifact,
1006 ..StepParallelLoadConfig::default()
1007 });
1008 }
1009 let contract = crate::parallel::ModelParallelContract::from_model(cfg)?;
1010 validate_step_expert_specs(&contract, "MEMRA_STEP_EP", &ep_specs, false)?;
1011 validate_step_expert_specs(&contract, "MEMRA_STEP_TP", &tp_specs, true)?;
1012 for spec in &tp_specs {
1013 let selection = select_step_expert_layout(spec.layer, &ep_specs, &tp_specs)?
1014 .ok_or("Step TP expert selection disappeared during preflight")?;
1015 validate_step_expert_activation_layout(cfg, "MEMRA_STEP_TP", &selection)?;
1016 }
1017
1018 let layer_owners = (0..trunk_layers)
1019 .map(|layer| {
1020 crate::pp::layer_engine(e, trunk_layers, layer).map(|engine| engine.ctx().ordinal())
1021 })
1022 .collect::<Result<Vec<_>, _>>()?;
1023 let plan = contract.preflight_step_tp_specs(
1024 tp_specs
1025 .iter()
1026 .map(|spec| (spec.layer, spec.devices.as_slice())),
1027 &layer_owners,
1028 )?;
1029
1030 for devices in &plan.runtime_groups {
1031 let hardware = crate::parallel::detect_uniform_hardware(devices)?;
1032 if !contract.hardware_targets.contains(&hardware) {
1033 return Err(format!(
1034 "{} has no qualified {hardware:?} TP contract for devices {devices:?}",
1035 contract.variant
1036 )
1037 .into());
1038 }
1039 }
1040
1041 if bulk_p2p && !native_p2p {
1042 return Err("MEMRA_STEP_TP_BULK_P2P=1 requires MEMRA_STEP_TP_NATIVE_P2P=1".into());
1043 }
1044 if device_arithmetic
1045 && (!ep_specs.is_empty()
1046 || !native_p2p
1047 || plan.expert_parallel_layers() == 0
1048 || plan.tensor_parallel_expert_layers() != 0)
1049 {
1050 return Err(
1051 "MEMRA_STEP_EP_DEVICE_ARITHMETIC=1 requires native-P2P TP4/TP8 \
1052 expert ownership for every selected routed-expert layer"
1053 .into(),
1054 );
1055 }
1056 let (qualified_experts, expert_artifact) =
1060 match crate::parallel::validate_fp8_expert_checkpoint(src, &contract) {
1061 Ok(qualified) => (qualified, StepExpertArtifact::E4m3),
1062 Err(fp8_error) => {
1063 match crate::parallel::validate_nvfp4_expert_checkpoint(src, &contract) {
1064 Ok(qualified) => (qualified, StepExpertArtifact::Nvfp4),
1065 Err(nvfp4_error) => {
1066 return Err(format!(
1067 "Step checkpoint qualifies as neither native expert artifact class: \
1068 [E4M3] {fp8_error} [NVFP4] {nvfp4_error}"
1069 )
1070 .into());
1071 }
1072 }
1073 }
1074 };
1075 if expert_artifact == StepExpertArtifact::Nvfp4 {
1076 if device_arithmetic {
1077 return Err(
1078 "MEMRA_STEP_EP_DEVICE_ARITHMETIC=1 is qualified for the E4M3 expert artifact \
1079 only; the NVFP4 expert program is host-canonical in this increment"
1080 .into(),
1081 );
1082 }
1083 if bulk_p2p {
1088 return Err(
1089 "MEMRA_STEP_TP_BULK_P2P=1 is qualified for the E4M3 expert artifact only; the \
1090 NVFP4 bank transport increment has not landed"
1091 .into(),
1092 );
1093 }
1094 }
1095
1096 if f32_mirror {
1097 eprintln!(
1098 "[step-tp-preflight] layers={} full_trunk={} runtime_groups={} \
1099 dense_attention_layers={} tensor_expert_layers={} expert_owner_layers={} \
1100 qualified_fp8_expert_projection_slices={} owner_first=true \
1101 hardware=rtx-pro-6000-blackwell \
1102 native_p2p={} bulk_p2p={} device_arithmetic={} bf16_residency=f32-mirror \
1103 weights_loaded=false performance_claim=false",
1104 plan.layers.len(),
1105 plan.full_trunk,
1106 plan.runtime_groups.len(),
1107 plan.dense_attention_layers(),
1108 plan.tensor_parallel_expert_layers(),
1109 plan.expert_parallel_layers(),
1110 qualified_experts,
1111 native_p2p,
1112 bulk_p2p,
1113 device_arithmetic,
1114 );
1115 } else {
1116 eprintln!(
1117 "[step-tp-preflight] layers={} full_trunk={} runtime_groups={} \
1118 dense_attention_layers={} tensor_expert_layers={} expert_owner_layers={} \
1119 qualified_fp8_expert_projection_slices={} owner_first=true \
1120 hardware=rtx-pro-6000-blackwell \
1121 native_p2p={} bulk_p2p={} device_arithmetic={} \
1122 weights_loaded=false performance_claim=false",
1123 plan.layers.len(),
1124 plan.full_trunk,
1125 plan.runtime_groups.len(),
1126 plan.dense_attention_layers(),
1127 plan.tensor_parallel_expert_layers(),
1128 plan.expert_parallel_layers(),
1129 qualified_experts,
1130 native_p2p,
1131 bulk_p2p,
1132 device_arithmetic,
1133 );
1134 }
1135 Ok(StepParallelLoadConfig {
1136 ep_specs,
1137 tp_specs,
1138 native_p2p,
1139 ep_device_arithmetic: device_arithmetic,
1140 f32_mirror,
1141 bulk_p2p,
1142 nvfp4_device_routes,
1143 auto_parallel,
1144 expert_artifact,
1145 })
1146}
1147
1148fn nvfp4_native_expert_bank<'a>(
1150 src: &'a dyn TensorSource,
1151 layer: usize,
1152 proj: &str,
1153) -> Result<memra_gguf::source::Nvfp4StackedNative<'a>, Box<dyn std::error::Error>> {
1154 let name = format!("blk.{layer}.ffn_{proj}_exps.weight");
1155 src.find_nvfp4_stacked_native(&name)
1156 .ok_or_else(|| format!("NVFP4 expert backend is missing native bank {name}").into())
1157}
1158
1159fn nvfp4_expert_bank_view<'a>(
1161 native: &'a memra_gguf::source::Nvfp4StackedNative<'a>,
1162) -> crate::tp::Nvfp4ExpertBank<'a> {
1163 crate::tp::Nvfp4ExpertBank {
1164 codes: native.codes,
1165 scales: native.scales,
1166 macros: &native.macros,
1167 expert_count: native.n_expert,
1168 out_features: native.out_f,
1169 in_features: native.in_f,
1170 }
1171}
1172
1173#[allow(clippy::too_many_arguments)] fn build_step_distributed_exps(
1175 e: &Engine,
1176 cfg: &ModelConfig,
1177 src: &dyn TensorSource,
1178 layer: usize,
1179 gate: &HostExps,
1180 up: &HostExps,
1181 down: &HostExps,
1182 step_runtimes: &mut StepParallelRuntimeRegistry,
1183) -> Result<(Option<StepEpExps>, Option<StepTpExps>), Box<dyn std::error::Error>> {
1184 let ep_device_arithmetic = step_runtimes.config.ep_device_arithmetic;
1185 if step_runtimes.config.ep_specs.is_empty() && step_runtimes.config.tp_specs.is_empty() {
1186 if ep_device_arithmetic {
1187 return Err(
1188 "MEMRA_STEP_EP_DEVICE_ARITHMETIC=1 requires MEMRA_STEP_TP and \
1189 MEMRA_STEP_TP_NATIVE_P2P=1"
1190 .into(),
1191 );
1192 }
1193 return Ok((None, None));
1194 }
1195 let contract = crate::parallel::ModelParallelContract::from_model(cfg)?;
1196 validate_step_expert_specs(
1197 &contract,
1198 "MEMRA_STEP_EP",
1199 &step_runtimes.config.ep_specs,
1200 false,
1201 )?;
1202 validate_step_expert_specs(
1203 &contract,
1204 "MEMRA_STEP_TP",
1205 &step_runtimes.config.tp_specs,
1206 true,
1207 )?;
1208 let Some(selection) = step_runtimes.expert_selection(layer)? else {
1209 return Ok((None, None));
1210 };
1211 validate_step_expert_activation_layout(
1212 cfg,
1213 if selection.configured_by_tp {
1214 "MEMRA_STEP_TP"
1215 } else {
1216 "MEMRA_STEP_EP"
1217 },
1218 &selection,
1219 )?;
1220 let activation_limit = match cfg.clamp_exp_at(layer as u32) {
1225 None => None,
1226 Some(SwigluClamp::Post(l)) => Some(l),
1227 Some(SwigluClamp::Pre(_)) => {
1228 return Err(format!(
1229 "MEMRA_STEP_EP/TP layer {layer}: glm5_next PRE-clamped SwiGLU has no \
1230 expert-parallel arm (the banks encode step35's post-clamp form)"
1231 )
1232 .into());
1233 }
1234 };
1235 let owner = e.ctx().ordinal();
1236 if !selection.spec.devices.contains(&owner) {
1237 let flag = if selection.configured_by_tp {
1238 "MEMRA_STEP_TP"
1239 } else {
1240 "MEMRA_STEP_EP"
1241 };
1242 return Err(format!(
1243 "{flag} layer {layer} owning PP device {owner} is absent from rank devices {:?}",
1244 selection.spec.devices
1245 )
1246 .into());
1247 }
1248 let expert_parallel = selection.layout == StepExpertLayout::ExpertParallel;
1249 if selection.configured_by_tp {
1250 contract.plan(crate::parallel::TopologyRequest {
1251 pipeline: 1,
1252 tensor: selection.spec.devices.len(),
1253 expert_parallel,
1254 available_devices: selection.spec.devices.len(),
1255 hardware: crate::parallel::HardwareTarget::RtxPro6000Blackwell,
1256 })?;
1257 }
1258 let native_p2p = selection.configured_by_tp && step_runtimes.config.native_p2p;
1259 if ep_device_arithmetic
1260 && (!selection.configured_by_tp
1261 || selection.layout != StepExpertLayout::ExpertParallel
1262 || !native_p2p)
1263 {
1264 return Err(
1265 "MEMRA_STEP_EP_DEVICE_ARITHMETIC=1 requires a MEMRA_STEP_TP TP4/TP8 \
1266 expert-owner layer and MEMRA_STEP_TP_NATIVE_P2P=1"
1267 .into(),
1268 );
1269 }
1270 let expert_artifact = step_runtimes.config.expert_artifact;
1271 match selection.layout {
1272 StepExpertLayout::ExpertParallel => {
1273 if expert_artifact == StepExpertArtifact::Nvfp4 {
1274 let w4a16_device_routes = step_runtimes.config.nvfp4_device_routes;
1278 if w4a16_device_routes
1279 && !matches!(
1280 src.expert_activation_precision(),
1281 memra_gguf::source::ExpertActivationPrecision::Bf16
1282 )
1283 {
1284 return Err(
1285 "explicit-EP MEMRA_STEP_NVFP4_DEV_ROUTES=1 requires an artifact that \
1286 declares BF16 routed-expert activations; TP keeps its separately gated \
1287 quantized-activation path"
1288 .into(),
1289 );
1290 }
1291 let runtime = step_runtimes.runtime(
1296 &selection.spec.devices,
1297 step_runtimes.config.native_p2p,
1298 false,
1299 )?;
1300 let experts = runtime.upload_expert_parallel_nvfp4_normalized(gate, up, down)?;
1301 let marker = if step_runtimes.config.auto_parallel {
1302 "parallel-ep"
1303 } else {
1304 "step-ep"
1305 };
1306 eprintln!(
1307 "[{marker}] layer={layer} devices={:?} experts={} artifact=nvfp4 \
1308 expert_layout=expert-parallel expert_transport={} \
1309 macro_fold=post-kernel-once native_p2p={} w4a16_device_routes={} \
1310 performance_claim=false",
1311 selection.spec.devices,
1312 contract.expert_count,
1313 runtime.transport_label(),
1314 runtime.native_p2p(),
1315 w4a16_device_routes,
1316 );
1317 if let Some(limit) = activation_limit {
1318 eprintln!(
1319 "[step-ep-clamp] load layer={layer} routed_clamp={limit} \
1320 formula=min-silu-times-clamped-up performance_claim=false"
1321 );
1322 }
1323 return Ok((
1324 Some(StepEpExps {
1325 runtime,
1326 experts: StepEpExpertBank::Nvfp4(experts),
1327 devices: selection.spec.devices,
1328 configured_by_tp: selection.configured_by_tp,
1329 activation_limit,
1330 nvfp4_device_routes: w4a16_device_routes,
1331 grouped_decode: None,
1332 }),
1333 None,
1334 ));
1335 }
1336 let runtime =
1337 step_runtimes.runtime(&selection.spec.devices, native_p2p, ep_device_arithmetic)?;
1338 let experts = runtime.upload_expert_parallel(
1339 host_e4m3_bank(gate)?,
1340 host_e4m3_bank(up)?,
1341 host_e4m3_bank(down)?,
1342 )?;
1343 let grouped_decode = if ep_device_arithmetic {
1344 let tokens = 1;
1345 let selected = (0..contract.experts_per_token).collect::<Vec<_>>();
1346 let input = vec![0.0f32; contract.hidden_size];
1347 let route_weights = vec![1.0f32; contract.experts_per_token];
1348 let projection = runtime.prepare_step_grouped_expert_parallel_gate_with_capacity(
1349 &experts,
1350 &input,
1351 tokens,
1352 &selected,
1353 activation_limit,
1354 tokens,
1355 )?;
1356 let combine = runtime
1357 .prepare_step_grouped_expert_parallel_combine(&projection, &route_weights)?;
1358 Some(std::sync::Mutex::new(StepEpGroupedDecode {
1359 projection,
1360 combine,
1361 }))
1362 } else {
1363 None
1364 };
1365 if selection.configured_by_tp {
1366 eprintln!(
1367 "[step-tp-ep] layer={layer} devices={:?} experts={} tp={} \
1368 attention_layout=tensor-parallel expert_layout=expert-parallel \
1369 expert_transport={} tp_transport={} native_p2p={} \
1370 activation={} accumulation={} output={} \
1371 grouped_decode_prepared={} grouped_decode_capacity=1 \
1372 performance_claim=false",
1373 selection.spec.devices,
1374 contract.expert_count,
1375 selection.spec.devices.len(),
1376 runtime.transport_label(),
1377 runtime.transport_label(),
1378 runtime.native_p2p(),
1379 runtime.expert_activation_label(),
1380 runtime.expert_accumulation_label(),
1381 runtime.expert_output_label(),
1382 grouped_decode.is_some(),
1383 );
1384 } else {
1385 eprintln!(
1386 "[step-ep] layer={layer} devices={:?} experts={} \
1387 expert_layout=expert-parallel expert_transport=host-bounce \
1388 native_p2p=false performance_claim=false",
1389 selection.spec.devices, contract.expert_count
1390 );
1391 }
1392 if let Some(limit) = activation_limit {
1393 eprintln!(
1394 "[step-ep-clamp] load layer={layer} routed_clamp={limit} \
1395 formula=min-silu-times-clamped-up performance_claim=false"
1396 );
1397 }
1398 Ok((
1399 Some(StepEpExps {
1400 runtime,
1401 experts: StepEpExpertBank::E4m3(experts),
1402 devices: selection.spec.devices,
1403 configured_by_tp: selection.configured_by_tp,
1404 activation_limit,
1405 nvfp4_device_routes: false,
1406 grouped_decode,
1407 }),
1408 None,
1409 ))
1410 }
1411 StepExpertLayout::TensorParallel => {
1412 let runtime =
1413 step_runtimes.runtime(&selection.spec.devices, native_p2p, ep_device_arithmetic)?;
1414 if activation_limit.is_some() && expert_artifact == StepExpertArtifact::E4m3 {
1415 return Err(format!(
1416 "layer {layer} uses the routed SwiGLU clamp and the E4M3 TP expert \
1417 program has no clamp arm; select EP for this layer (the NVFP4 TP \
1418 program carries the clamp)"
1419 )
1420 .into());
1421 }
1422 let experts = if expert_artifact == StepExpertArtifact::Nvfp4 {
1423 let gate_native = nvfp4_native_expert_bank(src, layer, "gate")?;
1424 let up_native = nvfp4_native_expert_bank(src, layer, "up")?;
1425 let down_native = nvfp4_native_expert_bank(src, layer, "down")?;
1426 StepTpExpertBank::Nvfp4(runtime.upload_tensor_parallel_nvfp4(
1427 nvfp4_expert_bank_view(&gate_native),
1428 nvfp4_expert_bank_view(&up_native),
1429 nvfp4_expert_bank_view(&down_native),
1430 )?)
1431 } else {
1432 StepTpExpertBank::E4m3(runtime.upload_tensor_parallel(
1433 host_e4m3_bank(gate)?,
1434 host_e4m3_bank(up)?,
1435 host_e4m3_bank(down)?,
1436 )?)
1437 };
1438 eprintln!(
1439 "[step-tp] layer={layer} devices={:?} experts={} tp={} artifact={} \
1440 expert_layout=tensor-parallel transport={} native_p2p={} \
1441 performance_claim=false",
1442 selection.spec.devices,
1443 contract.expert_count,
1444 selection.spec.devices.len(),
1445 match expert_artifact {
1446 StepExpertArtifact::E4m3 => "e4m3",
1447 StepExpertArtifact::Nvfp4 => "nvfp4",
1448 },
1449 runtime.transport_label(),
1450 runtime.native_p2p(),
1451 );
1452 if let Some(limit) = activation_limit {
1453 eprintln!(
1454 "[step-tp-clamp] load layer={layer} routed_clamp={limit} \
1455 formula=min-silu-times-clamped-up performance_claim=false"
1456 );
1457 }
1458 Ok((
1459 None,
1460 Some(StepTpExps {
1461 runtime,
1462 experts,
1463 devices: selection.spec.devices,
1464 activation_limit,
1465 }),
1466 ))
1467 }
1468 }
1469}
1470
1471fn upload_step_bf16_column(
1472 runtime: &crate::tp::TpE4m3HostBounce,
1473 src: &dyn TensorSource,
1474 name: &str,
1475 expected_in: usize,
1476 expected_out: usize,
1477 f32_mirror: bool,
1478) -> Result<crate::tp::ResidentBf16ColumnParallel, Box<dyn std::error::Error>> {
1479 let tensor = src
1480 .find(name)
1481 .ok_or_else(|| format!("Step TP projection is missing {name}"))?;
1482 if tensor.ggml_type != GgmlType::BF16 {
1483 return Err(format!(
1484 "Step TP projection {name} must preserve checkpoint BF16 bytes, got {:?}",
1485 tensor.ggml_type
1486 )
1487 .into());
1488 }
1489 if tensor.ne.len() != 2 {
1490 return Err(format!(
1491 "Step TP projection {name} must be a 2-D matrix, got shape {:?}",
1492 tensor.ne
1493 )
1494 .into());
1495 }
1496 let matrix = crate::tp::Bf16Matrix {
1497 bytes: tensor.bytes.as_ref(),
1498 in_features: tensor.ne[0] as usize,
1499 out_features: tensor.ne[1] as usize,
1500 };
1501 matrix.validate()?;
1502 if matrix.in_features != expected_in || matrix.out_features != expected_out {
1503 return Err(format!(
1504 "Step TP projection {name} shape {}x{} != registered {expected_out}x{expected_in}",
1505 matrix.out_features, matrix.in_features
1506 )
1507 .into());
1508 }
1509 Ok(if f32_mirror {
1510 runtime.upload_step_bf16_column_parallel_f32_mirror(matrix)?
1511 } else {
1512 runtime.upload_step_bf16_column_parallel(matrix)?
1513 })
1514}
1515
1516fn upload_step_bf16_row(
1517 runtime: &crate::tp::TpE4m3HostBounce,
1518 src: &dyn TensorSource,
1519 name: &str,
1520 expected_in: usize,
1521 expected_out: usize,
1522 f32_mirror: bool,
1523) -> Result<crate::tp::ResidentStepBf16RowParallel, Box<dyn std::error::Error>> {
1524 let tensor = src
1525 .find(name)
1526 .ok_or_else(|| format!("Step TP projection is missing {name}"))?;
1527 if tensor.ggml_type != GgmlType::BF16 {
1528 return Err(format!(
1529 "Step TP projection {name} must preserve checkpoint BF16 bytes, got {:?}",
1530 tensor.ggml_type
1531 )
1532 .into());
1533 }
1534 if tensor.ne.len() != 2 {
1535 return Err(format!(
1536 "Step TP projection {name} must be a 2-D matrix, got shape {:?}",
1537 tensor.ne
1538 )
1539 .into());
1540 }
1541 let matrix = crate::tp::Bf16Matrix {
1542 bytes: tensor.bytes.as_ref(),
1543 in_features: tensor.ne[0] as usize,
1544 out_features: tensor.ne[1] as usize,
1545 };
1546 matrix.validate()?;
1547 if matrix.in_features != expected_in || matrix.out_features != expected_out {
1548 return Err(format!(
1549 "Step TP projection {name} shape {}x{} != registered {expected_out}x{expected_in}",
1550 matrix.out_features, matrix.in_features
1551 )
1552 .into());
1553 }
1554 Ok(if f32_mirror {
1555 runtime.upload_step_bf16_row_parallel_f32_mirror(matrix)?
1556 } else {
1557 runtime.upload_step_bf16_row_parallel(matrix)?
1558 })
1559}
1560
1561fn upload_step_tp_f32_copies(
1562 runtime: &crate::tp::TpE4m3HostBounce,
1563 src: &dyn TensorSource,
1564 name: &str,
1565 expected: usize,
1566) -> Result<Vec<CudaSlice<f32>>, Box<dyn std::error::Error>> {
1567 let tensor = src
1568 .find(name)
1569 .ok_or_else(|| format!("Step TP attention is missing {name}"))?;
1570 let values = memra_gguf::dequant::dequantize(
1571 tensor.ggml_type,
1572 &tensor.bytes,
1573 tensor.ne.iter().product::<u64>() as usize,
1574 );
1575 if values.len() != expected || values.iter().any(|value| !value.is_finite()) {
1576 return Err(format!(
1577 "Step TP attention {name} has {} finite values, expected {expected}",
1578 values.len()
1579 )
1580 .into());
1581 }
1582 let mut copies = Vec::with_capacity(runtime.devices().len());
1583 for rank in 0..runtime.devices().len() {
1584 let engine = runtime
1585 .rank_engine(rank)
1586 .ok_or_else(|| format!("Step TP attention has no engine for rank {rank}"))?;
1587 let _main = engine.gpu.enter_main()?;
1588 copies.push(engine.htod(&values)?);
1589 }
1590 Ok(copies)
1591}
1592
1593#[allow(clippy::manual_is_multiple_of)] fn upload_step_tp_f32_row_shards(
1599 runtime: &crate::tp::TpE4m3HostBounce,
1600 src: &dyn TensorSource,
1601 name: &str,
1602 rows: usize,
1603 cols: usize,
1604) -> Result<Vec<CudaSlice<f32>>, Box<dyn std::error::Error>> {
1605 let tensor = src
1606 .find(name)
1607 .ok_or_else(|| format!("Step TP attention is missing {name}"))?;
1608 let values = memra_gguf::dequant::dequantize(
1609 tensor.ggml_type,
1610 &tensor.bytes,
1611 tensor.ne.iter().product::<u64>() as usize,
1612 );
1613 let world = runtime.devices().len();
1614 if values.len() != rows * cols || rows % world != 0 || values.iter().any(|v| !v.is_finite()) {
1615 return Err(format!(
1616 "Step TP attention {name} has {} finite values, expected {rows}x{cols} \
1617 (rows divisible by world {world})",
1618 values.len()
1619 )
1620 .into());
1621 }
1622 let local_rows = rows / world;
1623 let mut shards = Vec::with_capacity(world);
1624 for rank in 0..world {
1625 let engine = runtime
1626 .rank_engine(rank)
1627 .ok_or_else(|| format!("Step TP attention has no engine for rank {rank}"))?;
1628 let _main = engine.gpu.enter_main()?;
1629 shards
1630 .push(engine.htod(&values[rank * local_rows * cols..(rank + 1) * local_rows * cols])?);
1631 }
1632 Ok(shards)
1633}
1634
1635#[allow(clippy::manual_is_multiple_of)] fn upload_step_tp_bf16_row_shards(
1638 runtime: &crate::tp::TpE4m3HostBounce,
1639 src: &dyn TensorSource,
1640 name: &str,
1641 rows: usize,
1642 cols: usize,
1643) -> Result<Vec<CudaSlice<u8>>, Box<dyn std::error::Error>> {
1644 let tensor = src
1645 .find(name)
1646 .ok_or_else(|| format!("Step TP attention is missing {name}"))?;
1647 if tensor.ggml_type != memra_gguf::GgmlType::BF16 || tensor.bytes.len() != rows * cols * 2 {
1648 return Err(format!(
1649 "Step TP attention {name} is not a bf16 [{rows}, {cols}] tensor ({} bytes, {:?})",
1650 tensor.bytes.len(),
1651 tensor.ggml_type
1652 )
1653 .into());
1654 }
1655 let world = runtime.devices().len();
1656 if rows % world != 0 {
1657 return Err(format!("{name} rows {rows} not divisible by world {world}").into());
1658 }
1659 let local = rows / world * cols * 2;
1660 let mut shards = Vec::with_capacity(world);
1661 for rank in 0..world {
1662 let engine = runtime
1663 .rank_engine(rank)
1664 .ok_or_else(|| format!("Step TP attention has no engine for rank {rank}"))?;
1665 let _main = engine.gpu.enter_main()?;
1666 shards.push(engine.htod_bytes(&tensor.bytes[rank * local..(rank + 1) * local])?);
1667 }
1668 Ok(shards)
1669}
1670
1671#[derive(Clone, Copy, Debug, PartialEq, Eq)]
1672enum StepTpAttentionPlacement {
1673 RankLocalGlobal,
1674 RankLocalSwa,
1675 OwnerSwa,
1676 OwnerTransportFallback,
1677}
1678
1679impl StepTpAttentionPlacement {
1680 fn resolve(native_p2p: bool, window: Option<u32>) -> Self {
1681 match (native_p2p, window.is_some()) {
1682 (true, true) => Self::RankLocalSwa,
1683 (false, true) => Self::OwnerSwa,
1684 (true, false) => Self::RankLocalGlobal,
1685 (false, false) => Self::OwnerTransportFallback,
1686 }
1687 }
1688
1689 fn is_rank_local(self) -> bool {
1690 matches!(self, Self::RankLocalGlobal | Self::RankLocalSwa)
1691 }
1692
1693 fn label(self) -> &'static str {
1694 match self {
1695 Self::RankLocalGlobal => "rank-local-global",
1696 Self::RankLocalSwa => "rank-local-swa-ring",
1697 Self::OwnerSwa => "owner-swa",
1698 Self::OwnerTransportFallback => "owner-transport-fallback",
1699 }
1700 }
1701}
1702
1703fn build_step_tp_qkv(
1704 e: &Engine,
1705 src: &dyn TensorSource,
1706 cfg: &ModelConfig,
1707 layer: usize,
1708 step_runtimes: &mut StepParallelRuntimeRegistry,
1709) -> Result<Option<StepTpQkv>, Box<dyn std::error::Error>> {
1710 let Some(spec) = step_runtimes.tp_spec(layer).cloned() else {
1711 return Ok(None);
1712 };
1713 let contract = crate::parallel::ModelParallelContract::from_model(cfg)?;
1714 if layer >= contract.trunk_layers {
1715 return Err(format!(
1716 "MEMRA_STEP_TP layer {layer} is outside Step trunk layers 0..{}",
1717 contract.trunk_layers
1718 )
1719 .into());
1720 }
1721 let owner = e.ctx().ordinal();
1722 if spec.devices.first().copied() != Some(owner) {
1723 return Err(format!(
1724 "MEMRA_STEP_TP layer {layer} owning PP device {owner} must be the first QKV rank, \
1725 got {:?}",
1726 spec.devices
1727 )
1728 .into());
1729 }
1730 let plan = contract.plan(crate::parallel::TopologyRequest {
1731 pipeline: 1,
1732 tensor: spec.devices.len(),
1733 expert_parallel: spec.devices.len() > 2,
1734 available_devices: spec.devices.len(),
1735 hardware: crate::parallel::HardwareTarget::RtxPro6000Blackwell,
1736 })?;
1737 for rank in 0..spec.devices.len() {
1738 plan.query_head_range(layer, rank).ok_or_else(|| {
1739 format!("Step TP layer {layer} has no query-head range for rank {rank}")
1740 })?;
1741 plan.kv_head_range(layer, rank)
1742 .ok_or_else(|| format!("Step TP layer {layer} has no KV-head range for rank {rank}"))?;
1743 }
1744 let native_p2p = step_runtimes.config.native_p2p;
1745 let ep_device_arithmetic = step_runtimes.config.ep_device_arithmetic;
1746 let f32_mirror = step_runtimes.config.f32_mirror;
1747 if ep_device_arithmetic && (!native_p2p || !matches!(spec.devices.len(), 4 | 8)) {
1748 return Err(
1749 "MEMRA_STEP_EP_DEVICE_ARITHMETIC=1 requires a MEMRA_STEP_TP TP4/TP8 \
1750 expert-owner layer and MEMRA_STEP_TP_NATIVE_P2P=1"
1751 .into(),
1752 );
1753 }
1754 let runtime = step_runtimes.runtime(&spec.devices, native_p2p, ep_device_arithmetic)?;
1755 let p = |suffix: &str| format!("blk.{layer}.{suffix}");
1756 let q = upload_step_bf16_column(
1757 &runtime,
1758 src,
1759 &p("attn_q.weight"),
1760 contract.hidden_size,
1761 contract.query_heads[layer] * contract.head_dim,
1762 f32_mirror,
1763 )?;
1764 let k = upload_step_bf16_column(
1765 &runtime,
1766 src,
1767 &p("attn_k.weight"),
1768 contract.hidden_size,
1769 contract.kv_heads[layer] * contract.head_dim,
1770 f32_mirror,
1771 )?;
1772 let v = upload_step_bf16_column(
1773 &runtime,
1774 src,
1775 &p("attn_v.weight"),
1776 contract.hidden_size,
1777 contract.kv_heads[layer] * contract.head_dim,
1778 f32_mirror,
1779 )?;
1780 let o = upload_step_bf16_row(
1781 &runtime,
1782 src,
1783 &p("attn_output.weight"),
1784 contract.query_heads[layer] * contract.head_dim,
1785 contract.hidden_size,
1786 f32_mirror,
1787 )?;
1788 let geometry = cfg.full_attention_geometry_at(layer as u32);
1789 let attention_placement =
1790 StepTpAttentionPlacement::resolve(runtime.native_p2p(), geometry.window);
1791 let attention = if attention_placement.is_rank_local() {
1792 let decode_input = if ep_device_arithmetic || crate::tp::step_tp_decode_v2_enabled()? {
1797 Some(std::sync::Mutex::new(
1798 runtime.allocate_replicated_device_rows(1, contract.hidden_size)?,
1799 ))
1800 } else {
1801 None
1802 };
1803 let gate_fused =
1806 crate::tp::step_tp_qkv_fused_enabled()? && src.find(&p("attn_gate.weight")).is_some();
1807 let gate_shards = if gate_fused && f32_mirror {
1808 Some(upload_step_tp_f32_row_shards(
1809 &runtime,
1810 src,
1811 &p("attn_gate.weight"),
1812 contract.query_heads[layer],
1813 contract.hidden_size,
1814 )?)
1815 } else {
1816 None
1817 };
1818 let gate_shards_bf16 = if gate_fused && !f32_mirror {
1819 Some(upload_step_tp_bf16_row_shards(
1820 &runtime,
1821 src,
1822 &p("attn_gate.weight"),
1823 contract.query_heads[layer],
1824 contract.hidden_size,
1825 )?)
1826 } else {
1827 None
1828 };
1829 Some(StepTpAttention {
1830 q_norm: upload_step_tp_f32_copies(
1831 &runtime,
1832 src,
1833 &p("attn_q_norm.weight"),
1834 contract.head_dim,
1835 )?,
1836 k_norm: upload_step_tp_f32_copies(
1837 &runtime,
1838 src,
1839 &p("attn_k_norm.weight"),
1840 contract.head_dim,
1841 )?,
1842 decode_input,
1843 gate_shards,
1844 gate_shards_bf16,
1845 })
1846 } else {
1847 None
1848 };
1849 if f32_mirror {
1850 eprintln!(
1851 "[step-tp-qkv] load layer={layer} devices={:?} projections=qkv \
1852 qkv_tensor_parallel=true attention_local=true kv_local=true output_local=true \
1853 transport={} native_p2p={} bf16_residency=f32-mirror \
1854 output=root-readback performance_claim=false",
1855 spec.devices,
1856 runtime.transport_label(),
1857 runtime.native_p2p(),
1858 );
1859 } else {
1860 eprintln!(
1861 "[step-tp-qkv] load layer={layer} devices={:?} projections=qkv \
1862 qkv_tensor_parallel=true attention_local=true kv_local=true output_local=true \
1863 transport={} native_p2p={} output=root-readback performance_claim=false",
1864 spec.devices,
1865 runtime.transport_label(),
1866 runtime.native_p2p(),
1867 );
1868 }
1869 eprintln!(
1870 "[step-tp-attn-plan] load layer={layer} devices={:?} \
1871 qkv_tensor_parallel=true attention_tensor_parallel={} kv_cache_distributed={} \
1872 attention_scope={} transport={} native_p2p={} replicated_decode_input_prepared={} \
1873 performance_claim=false",
1874 spec.devices,
1875 attention_placement.is_rank_local(),
1876 attention_placement.is_rank_local(),
1877 attention_placement.label(),
1878 runtime.transport_label(),
1879 runtime.native_p2p(),
1880 attention
1881 .as_ref()
1882 .is_some_and(|attention| attention.decode_input.is_some()),
1883 );
1884 if f32_mirror {
1885 eprintln!(
1886 "[step-tp-o] load layer={layer} devices={:?} projection=o \
1887 o_tensor_parallel=true attention_local=true kv_local=true \
1888 transport={} native_p2p={} reduction=global-tp8-block-order \
1889 bf16_residency=f32-mirror output=root-readback performance_claim=false",
1890 spec.devices,
1891 runtime.transport_label(),
1892 runtime.native_p2p(),
1893 );
1894 } else {
1895 eprintln!(
1896 "[step-tp-o] load layer={layer} devices={:?} projection=o \
1897 o_tensor_parallel=true attention_local=true kv_local=true \
1898 transport={} native_p2p={} reduction=global-tp8-block-order \
1899 output=root-readback performance_claim=false",
1900 spec.devices,
1901 runtime.transport_label(),
1902 runtime.native_p2p(),
1903 );
1904 }
1905 Ok(Some(StepTpQkv {
1906 runtime,
1907 q,
1908 k,
1909 v,
1910 o,
1911 attention,
1912 devices: spec.devices,
1913 layer,
1914 }))
1915}
1916
1917fn build_dev_exps(
1930 e: &Engine,
1931 resident: &mut ResidentPlan,
1932 il: usize,
1933 gate: &HostExps,
1934 up: &HostExps,
1935 down: &HostExps,
1936) -> Result<Option<crate::hybrid::DevExps>, Box<dyn std::error::Error>> {
1937 if !gate.is_uniform_layout() || !up.is_uniform_layout() || !down.is_uniform_layout() {
1940 return Ok(None);
1941 }
1942 let fp8_host = match (&gate.fp8_blk, &up.fp8_blk, &down.fp8_blk) {
1943 (None, None, None) => None,
1944 (Some(g), Some(u), Some(d)) => Some((g, u, d)),
1945 _ => {
1946 return Err("resident expert projections disagree on block-E4M3 scale carriage".into());
1947 }
1948 };
1949 let scale_bytes = fp8_host
1950 .map(|(g, u, d)| (g.scales.len() + u.scales.len() + d.scales.len()) * size_of::<f32>())
1951 .unwrap_or(0);
1952 let per_layer = gate.bytes.as_bytes().len()
1953 + up.bytes.as_bytes().len()
1954 + down.bytes.as_bytes().len()
1955 + scale_bytes;
1956 if gate.tiers.is_some() {
1957 return Ok(None); }
1959 let fits = resident.should_reside(e, il, per_layer);
1960 if !fits {
1961 return Ok(None);
1962 }
1963 use cudarc::driver::DevicePtr;
1964 let gu_il = std::env::var("MEMRA_MOE_GU_IL").as_deref() == Ok("1")
1965 && gate.out_f == up.out_f
1966 && gate.in_f == up.in_f
1967 && fp8_host.is_none();
1968 let n_expert = gate.n_expert;
1969 let (g, u) = if gu_il {
1970 let (rbg, rbu) = (gate.row_bytes, up.row_bytes);
1972 let n_rows = gate.out_f;
1973 let gb = gate.bytes.as_bytes();
1974 let ub = up.bytes.as_bytes();
1975 let mut il = vec![0u8; n_expert * n_rows * (rbg + rbu)];
1976 for ex in 0..n_expert {
1977 for o in 0..n_rows {
1978 let dst = (ex * n_rows + o) * (rbg + rbu);
1979 let sg = ex * gate.expert_stride + o * rbg;
1980 let su = ex * up.expert_stride + o * rbu;
1981 il[dst..dst + rbg].copy_from_slice(&gb[sg..sg + rbg]);
1982 il[dst + rbg..dst + rbg + rbu].copy_from_slice(&ub[su..su + rbu]);
1983 }
1984 }
1985 let ild = e.htod_bytes_padded(&il, 8)?;
1986 (ild, e.htod_bytes(&[0u8; 16])?)
1989 } else {
1990 (
1991 e.htod_bytes_padded(gate.bytes.as_bytes(), 8)?,
1992 e.htod_bytes_padded(up.bytes.as_bytes(), 8)?,
1993 )
1994 };
1995 let d = e.htod_bytes_padded(down.bytes.as_bytes(), 144)?;
2000 let fp8_blk = match fp8_host {
2001 Some((gate, up, down)) => {
2002 if e.fp8_blk_nan_count(&g)? != 0
2003 || e.fp8_blk_nan_count(&u)? != 0
2004 || e.fp8_blk_nan_count(&d)? != 0
2005 {
2006 return Err("native stacked block-E4M3 expert bank contains NaN codes".into());
2007 }
2008 Some(DevExpertFp8BlockScales {
2009 gate: DevExpertFp8ProjectionScales::upload(e, gate, n_expert)?,
2010 up: DevExpertFp8ProjectionScales::upload(e, up, n_expert)?,
2011 down: DevExpertFp8ProjectionScales::upload(e, down, n_expert)?,
2012 })
2013 }
2014 None => None,
2015 };
2016 let mut host = vec![0u64; 3 * n_expert];
2017 let (pg, pu, pd) = {
2018 let __s_e0 = e.stream();
2019 let (pg, _e0) = g.device_ptr(&__s_e0);
2020 let __s_e1 = e.stream();
2021 let (pu, _e1) = u.device_ptr(&__s_e1);
2022 let __s_e2 = e.stream();
2023 let (pd, _e2) = d.device_ptr(&__s_e2);
2024 (pg, pu, pd)
2025 };
2026 for ex in 0..n_expert {
2027 if gu_il {
2028 let stride = gate.out_f * (gate.row_bytes + up.row_bytes);
2029 host[ex] = pg + (ex * stride) as u64;
2030 host[n_expert + ex] = pg + (ex * stride + gate.row_bytes) as u64;
2031 } else {
2032 host[ex] = pg + (ex * gate.expert_stride) as u64;
2033 host[n_expert + ex] = pu + (ex * up.expert_stride) as u64;
2034 }
2035 host[2 * n_expert + ex] = pd + (ex * down.expert_stride) as u64;
2036 }
2037 if gu_il {
2038 eprintln!("[moe] gate/up dev slab INTERLEAVED (MEMRA_MOE_GU_IL)");
2039 }
2040 let ptr_row = e.htod_u64(&host)?;
2041 Ok(Some(crate::hybrid::DevExps {
2042 gate: g,
2043 up: u,
2044 down: d,
2045 ptr_row,
2046 gu_il,
2047 dev: e.ctx().ordinal(),
2048 fp8_blk,
2049 }))
2050}
2051
2052pub struct FullAttnLayer {
2053 pub wq: GpuTensor,
2054 pub wk: GpuTensor,
2055 pub wv: GpuTensor,
2056 pub wo: GpuTensor,
2057 pub q_norm: GpuTensor,
2058 pub k_norm: GpuTensor,
2059 pub attn_gate: Option<GpuTensor>,
2070 pub step_tp_qkv: Option<StepTpQkv>,
2074}
2075
2076pub struct StepTpQkv {
2077 pub runtime: Arc<crate::tp::TpE4m3HostBounce>,
2078 pub q: crate::tp::ResidentBf16ColumnParallel,
2079 pub k: crate::tp::ResidentBf16ColumnParallel,
2080 pub v: crate::tp::ResidentBf16ColumnParallel,
2081 pub o: crate::tp::ResidentStepBf16RowParallel,
2082 pub attention: Option<StepTpAttention>,
2083 pub devices: Vec<usize>,
2084 pub layer: usize,
2085}
2086
2087pub struct StepTpAttention {
2088 pub q_norm: Vec<CudaSlice<f32>>,
2089 pub k_norm: Vec<CudaSlice<f32>>,
2090 pub decode_input: Option<std::sync::Mutex<crate::tp::ResidentReplicatedDeviceRows>>,
2091 pub gate_shards: Option<Vec<CudaSlice<f32>>>,
2094 pub gate_shards_bf16: Option<Vec<CudaSlice<u8>>>,
2096}
2097
2098#[derive(Clone, Copy, Debug, PartialEq, Eq)]
2099pub struct StepTpKvDeviceAdmission {
2100 pub device: usize,
2101 pub bytes: usize,
2102}
2103
2104#[derive(Clone, Copy, Debug)]
2108pub struct MlaGeom {
2109 pub n_head: usize, pub d_nope: usize, pub d_rope: usize, pub d_v: usize, pub kv_rank: usize, pub latent_dim: usize, pub scale: f32, }
2117
2118#[derive(Clone, Copy, Debug)]
2123pub struct MlaIndexerGeom {
2124 pub heads: usize, pub head_dim: usize, pub top_k: usize, pub pool: usize, pub always_select_tail: bool,
2129}
2130
2131impl MlaIndexerGeom {
2132 pub fn select_k(&self, n_pools: usize) -> usize {
2134 (self.top_k / self.pool).min(n_pools)
2135 }
2136
2137 pub fn index_width(&self, n_pools: usize) -> usize {
2139 self.select_k(n_pools) * self.pool
2140 + if self.always_select_tail {
2141 self.pool - 1
2142 } else {
2143 0
2144 }
2145 }
2146
2147 pub fn state_width(&self) -> usize {
2149 2 * self.head_dim
2150 }
2151}
2152
2153pub struct MlaIndexer {
2157 pub wq_b: GpuTensor, pub wk: GpuTensor, pub k_norm_w: GpuTensor, pub k_norm_b: GpuTensor, pub weights_proj: GpuTensor, pub kpool_gate: GpuTensor, pub kpool_ape: GpuTensor, pub geom: MlaIndexerGeom,
2165}
2166
2167pub struct MlaAttnLayer {
2168 pub wq_a: GpuTensor, pub q_a_norm: GpuTensor, pub wq_b: GpuTensor, pub wkv_a: GpuTensor, pub kv_a_norm: GpuTensor, pub wk_b: GpuTensor, pub wv_b: GpuTensor, pub wo: GpuTensor, pub geom: MlaGeom,
2178 pub index: Option<MlaIndexer>,
2181 pub tp: Option<Box<crate::glm5_tp::Glm5TpMla>>,
2186 pub tp_shard: bool,
2193}
2194
2195impl MlaAttnLayer {
2196 pub fn load(
2208 e: &Engine,
2209 src: &dyn TensorSource,
2210 il: u32,
2211 plan: &memra_gguf::model_plan::MlaAttentionPlan,
2212 ) -> Result<Self, Box<dyn std::error::Error>> {
2213 let memra_gguf::model_plan::MlaAttentionPlan::LatentKv {
2214 query_heads,
2215 q_lora_rank,
2216 kv_lora_rank,
2217 qk_head_dim,
2218 rope_head_dim,
2219 value_head_dim,
2220 sparse_index,
2221 ..
2222 } = plan
2223 else {
2224 return Err(format!(
2225 "native MLA loader has no compressed-KV implementation for block {il}"
2226 )
2227 .into());
2228 };
2229 let d_nope = qk_head_dim
2230 .checked_sub(*rope_head_dim)
2231 .ok_or("MLA rope head width exceeds total QK head width")?;
2232 let p = |s: &str| format!("blk.{il}.{s}");
2233 let geom = MlaGeom {
2234 n_head: *query_heads as usize,
2235 d_nope: d_nope as usize,
2236 d_rope: *rope_head_dim as usize,
2237 d_v: *value_head_dim as usize,
2238 kv_rank: *kv_lora_rank as usize,
2239 latent_dim: (*kv_lora_rank + *rope_head_dim) as usize,
2240 scale: 1.0 / (*qk_head_dim as f32).sqrt(),
2241 };
2242 let wq_a = load_t(e, src, &p("attn_q_a.weight"))?;
2243 let wq_b = load_t(e, src, &p("attn_q_b.weight"))?;
2244 let wkv_a = load_t(e, src, &p("attn_kv_a_mqa.weight"))?;
2245 let wk_b = load_t(e, src, &p("attn_k_b.weight"))?;
2246 let wv_b = load_t(e, src, &p("attn_v_b.weight"))?;
2247 let wo = load_t(e, src, &p("attn_output.weight"))?;
2248 for (w, tensor) in [(&wk_b, "attn_k_b"), (&wv_b, "attn_v_b")] {
2257 if !matches!(w, GpuTensor::Float { .. }) {
2258 return Err(format!(
2259 "blk.{il}.{tensor}.weight is not f32-resident. The MLA conversion-split \
2260 operands feed f32-only absorb/decompress kernels; the checkpoint source must \
2261 dequantize them (TensorTransform::SplitMlaKv) rather than hand the engine a \
2262 quantized plane"
2263 )
2264 .into());
2265 }
2266 }
2267 let n_head = wq_b.out_features() / (geom.d_nope + geom.d_rope);
2269 assert_eq!(
2270 wq_b.out_features(),
2271 n_head * (geom.d_nope + geom.d_rope),
2272 "wq_b out {} not a multiple of qk_head_dim {}",
2273 wq_b.out_features(),
2274 geom.d_nope + geom.d_rope
2275 );
2276 assert_eq!(
2277 wq_a.in_features(),
2278 wkv_a.in_features(),
2279 "q_a/kv_a hidden mismatch"
2280 );
2281 assert_eq!(
2282 wq_b.in_features(),
2283 *q_lora_rank as usize,
2284 "wq_b in != q_lora_rank"
2285 );
2286 assert_eq!(
2287 n_head, geom.n_head,
2288 "MLA checkpoint head count != ModelPlan"
2289 );
2290 assert_eq!(
2291 wkv_a.out_features(),
2292 geom.latent_dim,
2293 "wkv_a out != kv_lora_rank + rope"
2294 );
2295 assert_eq!(
2296 wk_b.ne(),
2297 &[geom.d_nope as u64, geom.kv_rank as u64, n_head as u64],
2298 "attn_k_b must be the TRANSPOSED (nope, kv_rank, head) conversion split"
2299 );
2300 assert_eq!(
2301 wv_b.ne(),
2302 &[geom.kv_rank as u64, geom.d_v as u64, n_head as u64],
2303 "attn_v_b must be the (kv_rank, v, head) conversion split"
2304 );
2305 assert_eq!(
2306 wo.in_features(),
2307 n_head * geom.d_v,
2308 "wo in != n_head * v_head_dim"
2309 );
2310 let index = Self::load_indexer(e, src, il, sparse_index, *q_lora_rank)?;
2311 Ok(MlaAttnLayer {
2312 wq_a,
2313 q_a_norm: load_t(e, src, &p("attn_q_a_norm.weight"))?,
2314 wq_b,
2315 wkv_a,
2316 kv_a_norm: load_t(e, src, &p("attn_kv_a_norm.weight"))?,
2317 wk_b,
2318 wv_b,
2319 wo,
2320 geom,
2321 index,
2322 tp: None,
2323 tp_shard: false,
2324 })
2325 }
2326
2327 fn load_indexer(
2339 e: &Engine,
2340 src: &dyn TensorSource,
2341 il: u32,
2342 sparse_index: &memra_gguf::model_plan::SparseIndexPlan,
2343 q_lora_rank: u32,
2344 ) -> Result<Option<MlaIndexer>, Box<dyn std::error::Error>> {
2345 let memra_gguf::model_plan::SparseIndexPlan::Own {
2346 heads,
2347 head_dim,
2348 top_k,
2349 kpool: Some(kpool),
2350 } = sparse_index
2351 else {
2352 return Ok(None);
2353 };
2354 let geom = MlaIndexerGeom {
2355 heads: *heads as usize,
2356 head_dim: *head_dim as usize,
2357 top_k: *top_k as usize,
2358 pool: kpool.pool as usize,
2359 always_select_tail: kpool.always_select_tail,
2360 };
2361 if geom.heads == 0 || geom.head_dim == 0 || geom.pool == 0 || geom.top_k < geom.pool {
2362 return Err(format!(
2363 "blk.{il}: SparseIndexPlan::Own declares an unusable k-pool indexer \
2364 (heads {}, head_dim {}, pool {}, top_k {}) — heads/head_dim/pool must be \
2365 positive and top_k must admit at least one pool",
2366 geom.heads, geom.head_dim, geom.pool, geom.top_k
2367 )
2368 .into());
2369 }
2370 let need = |suffix: &str| -> Result<GpuTensor, Box<dyn std::error::Error>> {
2375 let name = format!("blk.{il}.{suffix}");
2376 if !src.has(&name) {
2377 return Err(format!(
2378 "blk.{il}: the layer's ModelPlan declares a DSA k-pool indexer but the \
2379 checkpoint has no `{name}`. This layer MUST NOT fall back to dense \
2380 attention: dense and indexed attention are the same function only below \
2381 index_topk ({}), and glm5_next serves a 1,048,576-token context",
2382 geom.top_k
2383 )
2384 .into());
2385 }
2386 load_t(e, src, &name).map_err(|source| -> Box<dyn std::error::Error> {
2387 format!("blk.{il}: DSA k-pool indexer tensor `{name}` failed to load: {source}")
2388 .into()
2389 })
2390 };
2391 let wq_b = need("indexer.attn_q_b.weight")?;
2392 let wk = need("indexer.attn_k.weight")?;
2393 let k_norm_w = need("indexer.k_norm.weight")?;
2394 let k_norm_b = need("indexer.k_norm.bias")?;
2395 let weights_proj = need("indexer.proj.weight")?;
2396 let kpool_gate = need("indexer.kpool_gate.weight")?;
2397 let kpool_ape = need("indexer.kpool_ape.weight")?;
2398 for (w, name) in [
2401 (&k_norm_w, "indexer.k_norm.weight"),
2402 (&k_norm_b, "indexer.k_norm.bias"),
2403 (&kpool_ape, "indexer.kpool_ape.weight"),
2404 ] {
2405 if !matches!(w, GpuTensor::Float { .. }) {
2406 return Err(format!(
2407 "blk.{il}.{name} is not f32-resident. The indexer's LayerNorm affine and \
2408 k-pool positional embedding feed f32-only kernels"
2409 )
2410 .into());
2411 }
2412 }
2413 assert_eq!(
2414 wq_b.in_features(),
2415 q_lora_rank as usize,
2416 "blk.{il}.indexer.attn_q_b in != q_lora_rank"
2417 );
2418 assert_eq!(
2419 wq_b.out_features(),
2420 geom.heads * geom.head_dim,
2421 "blk.{il}.indexer.attn_q_b out != index heads * head_dim"
2422 );
2423 assert_eq!(
2424 wk.out_features(),
2425 geom.head_dim,
2426 "blk.{il}.indexer.attn_k out != index head_dim"
2427 );
2428 assert_eq!(
2429 weights_proj.out_features(),
2430 geom.heads,
2431 "blk.{il}.indexer.proj out != index heads"
2432 );
2433 assert_eq!(
2434 kpool_gate.out_features(),
2435 geom.head_dim,
2436 "blk.{il}.indexer.kpool_gate out != index head_dim"
2437 );
2438 assert_eq!(
2439 kpool_ape.float_data().len(),
2440 geom.pool * geom.head_dim,
2441 "blk.{il}.indexer.kpool_ape must hold pool * head_dim elements"
2442 );
2443 Ok(Some(MlaIndexer {
2444 wq_b,
2445 wk,
2446 k_norm_w,
2447 k_norm_b,
2448 weights_proj,
2449 kpool_gate,
2450 kpool_ape,
2451 geom,
2452 }))
2453 }
2454}
2455
2456#[track_caller]
2464pub(crate) fn mla_path_unimplemented(path: &str) -> ! {
2465 panic!(
2466 "Mixer::Mla has no {path} arm — the MLA forward is wired for the stateless forward, \
2467 the stateful prime and T=1 decode only (cu/mla_attn.cu, increment 4); this path needs \
2468 its own parity gate before it may run \
2469 (research/mla-bringup-20260801/DESIGN.md §4, increment 7)"
2470 )
2471}
2472
2473#[track_caller]
2479pub(crate) fn kda_path_unimplemented(path: &str) -> ! {
2480 panic!(
2481 "Mixer::Kda has no {path} arm — glm5_next KDA is wired for the stateless forward, the \
2482 stateful prime and T=1 decode only (crates/memra-engine/src/kda.rs); this path needs \
2483 its own parity gate before it may run"
2484 )
2485}
2486
2487pub struct LinearAttnLayer {
2488 pub geometry: memra_gguf::model_plan::GatedDeltaNetPlan,
2489 pub wqkv: GpuTensor, pub wqkv_gate: GpuTensor, pub ssm_beta: GpuTensor, pub ssm_alpha: GpuTensor, pub ssm_a: GpuTensor, pub ssm_dt: GpuTensor, pub ssm_conv1d: GpuTensor, pub ssm_norm: GpuTensor, pub ssm_out: GpuTensor, }
2499
2500#[allow(clippy::large_enum_variant)] pub enum Mixer {
2502 Full(FullAttnLayer),
2503 Linear(LinearAttnLayer),
2504 Mla(MlaAttnLayer),
2506 Kda(crate::kda::KdaAttnLayer),
2508}
2509
2510pub struct MoeWeights {
2517 pub gate_inp: GpuTensor, pub gate_inp_shexp: Option<GpuTensor>, pub exp_probs_b: Option<Vec<f32>>,
2523 pub exp_probs_b_dev: CudaSlice<f32>,
2524 pub active_experts: Option<Vec<bool>>,
2528 pub active_experts_dev: CudaSlice<u8>,
2529 pub gate_exps: HostExps, pub up_exps: HostExps, pub down_exps: HostExps, pub gate_shexp: Option<GpuTensor>,
2533 pub up_shexp: Option<GpuTensor>,
2534 pub down_shexp: Option<GpuTensor>,
2535 pub dev_exps: Option<DevExps>,
2542 pub step_ep: Option<StepEpExps>,
2546 pub step_tp: Option<StepTpExps>,
2550 pub glm5_ep: Option<crate::glm5_tp::Glm5EpExps>,
2555 pub dev_macros: cudarc::driver::CudaSlice<f32>,
2561 pub has_macros: bool,
2562 pub w4a16_bf16_activations: bool,
2565}
2566
2567#[allow(clippy::large_enum_variant)] pub enum StepEpExpertBank {
2570 E4m3(crate::tp::ResidentExpertParallel),
2571 Nvfp4(crate::tp::ResidentNvfp4ExpertParallel),
2572}
2573
2574impl StepEpExpertBank {
2575 pub fn e4m3(&self) -> Result<&crate::tp::ResidentExpertParallel, String> {
2579 match self {
2580 Self::E4m3(bank) => Ok(bank),
2581 Self::Nvfp4(_) => Err(
2582 "Step grouped expert program reached an NVFP4 bank; this path is qualified \
2583 for the E4M3 artifact only"
2584 .to_string(),
2585 ),
2586 }
2587 }
2588}
2589
2590pub struct StepEpExps {
2591 pub runtime: Arc<crate::tp::TpE4m3HostBounce>,
2592 pub experts: StepEpExpertBank,
2593 pub devices: Vec<usize>,
2594 pub configured_by_tp: bool,
2595 pub activation_limit: Option<f32>,
2596 pub nvfp4_device_routes: bool,
2598 pub grouped_decode: Option<std::sync::Mutex<StepEpGroupedDecode>>,
2601}
2602
2603pub struct StepEpGroupedDecode {
2604 pub(crate) projection: crate::tp::PreparedStepGroupedExpertParallelGate,
2605 pub(crate) combine: crate::tp::PreparedPeerWeightedRouteCombine,
2606}
2607
2608#[derive(Default)]
2609pub(crate) struct StepEpGroupedPrefill {
2610 pub(crate) state: Option<StepEpGroupedPrefillState>,
2611}
2612
2613pub(crate) struct StepEpGroupedPrefillState {
2614 pub(crate) devices: Vec<usize>,
2615 pub(crate) grouped: StepEpGroupedDecode,
2616}
2617
2618#[allow(clippy::large_enum_variant)] pub enum StepTpExpertBank {
2621 E4m3(crate::tp::ResidentTensorParallel),
2622 Nvfp4(crate::tp::ResidentNvfp4TensorParallel),
2623}
2624
2625pub struct StepTpExps {
2626 pub runtime: Arc<crate::tp::TpE4m3HostBounce>,
2627 pub experts: StepTpExpertBank,
2628 pub devices: Vec<usize>,
2629 pub activation_limit: Option<f32>,
2632}
2633
2634impl MoeWeights {
2635 #[inline]
2636 pub fn has_uniform_expert_layout(&self) -> bool {
2637 self.gate_exps.is_uniform_layout()
2638 && self.up_exps.is_uniform_layout()
2639 && self.down_exps.is_uniform_layout()
2640 }
2641
2642 #[inline]
2643 pub fn active_count(&self) -> usize {
2644 self.active_experts
2645 .as_ref()
2646 .map(|mask| mask.iter().filter(|&&active| active).count())
2647 .unwrap_or(self.gate_exps.n_expert)
2648 }
2649
2650 #[allow(clippy::too_many_arguments)]
2651 pub(crate) fn qmatvec_view(
2652 &self,
2653 e: &Engine,
2654 w: &CudaSlice<u8>,
2655 range: std::ops::Range<usize>,
2656 x: &cudarc::driver::CudaView<f32>,
2657 m: usize,
2658 in_f: usize,
2659 out_f: usize,
2660 qtype: i32,
2661 row_bytes: usize,
2662 ) -> Result<CudaSlice<f32>, Box<dyn std::error::Error>> {
2663 if self.w4a16_bf16_activations && qtype == crate::QT_NVFP4 {
2664 e.qmatvec_view_bf16_activation(w, range, x, m, in_f, out_f, qtype, row_bytes)
2665 } else {
2666 e.qmatvec_view(w, range, x, m, in_f, out_f, qtype, row_bytes)
2667 }
2668 }
2669}
2670
2671pub struct DevExps {
2674 pub gate: CudaSlice<u8>,
2675 pub up: CudaSlice<u8>,
2676 pub down: CudaSlice<u8>,
2677 pub ptr_row: CudaSlice<u64>,
2679 pub dev: usize,
2687 pub gu_il: bool,
2693 pub fp8_blk: Option<DevExpertFp8BlockScales>,
2697}
2698
2699pub struct DevExpertFp8BlockScales {
2700 pub gate: DevExpertFp8ProjectionScales,
2701 pub up: DevExpertFp8ProjectionScales,
2702 pub down: DevExpertFp8ProjectionScales,
2703}
2704
2705pub struct DevExpertFp8ProjectionScales {
2706 pub scales: CudaSlice<f32>,
2707 pub rows: usize,
2708 pub cols: usize,
2709 pub expert_stride: usize,
2710}
2711
2712impl DevExpertFp8ProjectionScales {
2713 fn validate(
2714 host: &crate::model::HostExpertFp8BlockScales,
2715 n_expert: usize,
2716 ) -> Result<(), String> {
2717 if host.expert_stride == 0 {
2718 return Err("block-E4M3 expert scale stride must be nonzero".into());
2719 }
2720 if host.rows * host.cols != host.expert_stride {
2721 return Err(format!(
2722 "block-E4M3 expert scale stride mismatch: {}x{} != {}",
2723 host.rows, host.cols, host.expert_stride
2724 ));
2725 }
2726 let want = n_expert
2727 .checked_mul(host.expert_stride)
2728 .ok_or("block-E4M3 expert scale slab length overflow")?;
2729 if host.scales.len() != want {
2730 return Err(format!(
2731 "block-E4M3 scale slab length mismatch: got {}, want {n_expert}x{}={want}",
2732 host.scales.len(),
2733 host.expert_stride
2734 ));
2735 }
2736 Ok(())
2737 }
2738
2739 fn upload(
2740 e: &Engine,
2741 host: &crate::model::HostExpertFp8BlockScales,
2742 n_expert: usize,
2743 ) -> Result<Self, Box<dyn std::error::Error>> {
2744 Self::validate(host, n_expert)?;
2745 Ok(Self {
2746 scales: e.htod(&host.scales)?,
2747 rows: host.rows,
2748 cols: host.cols,
2749 expert_stride: host.expert_stride,
2750 })
2751 }
2752}
2753
2754#[allow(clippy::large_enum_variant)] pub enum Ffn {
2757 Dense {
2758 ffn_gate: GpuTensor,
2759 ffn_up: GpuTensor,
2760 ffn_down: GpuTensor,
2761 },
2762 Moe(MoeWeights),
2763}
2764
2765pub struct HybridLayer {
2766 pub attn_norm: GpuTensor,
2767 pub post_attn_norm: GpuTensor, pub mixer: Mixer,
2769 pub ffn: Ffn,
2770 pub gemma4: Option<Gemma4LayerBits>,
2771 pub hyper: Option<crate::hyper::HyperLayer>,
2776}
2777
2778pub struct Gemma4LayerBits {
2782 pub ffn_norm: GpuTensor, pub post_ffw_norm: GpuTensor, pub moe_bits: Option<Gemma4MoeBits>,
2787 pub layer_scale: f32, pub e4b: Option<Gemma4E4bLayer>,
2790}
2791
2792pub struct Gemma4E4bLayer {
2797 pub inp_gate: GpuTensor, pub proj: GpuTensor, pub post_norm: GpuTensor, pub qkv_cat: Option<GpuTensor>,
2804 pub kv_share: Option<u32>,
2808}
2809
2810pub struct Gemma4E4bModel {
2814 pub tok_tbl_gpu: std::sync::OnceLock<CudaSlice<u8>>,
2817 pub tok_embd_bytes: Vec<u8>,
2818 pub tok_embd_qt: i32,
2819 pub tok_embd_row_bytes: usize,
2820 pub model_proj: GpuTensor, pub proj_norm: GpuTensor, pub n_epl: usize,
2823}
2824
2825pub struct Gemma4MoeBits {
2826 pub post_ffw_norm_1: GpuTensor, pub pre_ffw_norm_2: GpuTensor, pub post_ffw_norm_2: GpuTensor, pub shared_gate: GpuTensor,
2830 pub shared_up: GpuTensor,
2831 pub shared_down: GpuTensor,
2832 pub router_scale_pre: CudaSlice<f32>,
2837 pub per_expert_scale: Vec<f32>, pub per_expert_scale_d: CudaSlice<f32>, }
2840
2841fn load_mtp_head_maybe_nvfp4(
2854 e: &Engine,
2855 src: &dyn TensorSource,
2856 name: &str,
2857) -> Result<Option<GpuTensor>, Box<dyn std::error::Error>> {
2858 if !{
2859 static ENV: std::sync::OnceLock<Option<bool>> = std::sync::OnceLock::new();
2860 crate::step37_door(&ENV, "MEMRA_MTP_HEAD_NVFP4")
2861 } {
2862 return load_opt(e, src, name);
2863 }
2864 let Some(v) = src.find(name) else {
2865 return Ok(None);
2866 };
2867 if !matches!(v.ggml_type, GgmlType::BF16) || v.ne[0] % 64 != 0 {
2868 return load_opt(e, src, name);
2869 }
2870 let vals: Vec<f32> = v
2871 .bytes
2872 .chunks_exact(2)
2873 .map(|c| f32::from_bits((u16::from_le_bytes([c[0], c[1]]) as u32) << 16))
2874 .collect();
2875 let blocks = memra_gguf::nvfp4_repack::f32_to_nvfp4(&vals);
2876 eprintln!(
2877 "[mtp-head] {name}: BF16 -> NVFP4 ({} MiB, was {} MiB)",
2878 blocks.len() >> 20,
2879 v.bytes.len() >> 20
2880 );
2881 Ok(Some(GpuTensor::from_quant_bytes(
2882 e,
2883 &blocks,
2884 GgmlType::NVFP4,
2885 v.ne[0],
2886 v.ne[1],
2887 1.0,
2888 )?))
2889}
2890
2891pub(crate) fn sha256_file_hex8(
2899 path: &std::path::Path,
2900) -> Result<String, Box<dyn std::error::Error>> {
2901 use sha2::{Digest, Sha256};
2902 let mut file = std::fs::File::open(path)?;
2903 let mut hasher = Sha256::new();
2904 std::io::copy(&mut file, &mut hasher)?;
2905 let digest = hasher.finalize();
2906 Ok(digest
2907 .iter()
2908 .take(4)
2909 .map(|byte| format!("{byte:02x}"))
2910 .collect())
2911}
2912
2913pub(crate) fn frspec_trim_own_head_name(n_trunk: usize) -> String {
2914 format!("blk.{n_trunk}.nextn.shared_head_head.weight")
2915}
2916
2917pub struct DflashTrimHead {
2928 pub head: GpuTensor,
2931 pub d2t: Vec<u32>,
2933}
2934
2935fn frspec_read_d2t(path: &str) -> Result<Vec<u32>, Box<dyn std::error::Error>> {
2940 Ok(if path.ends_with(".txt") {
2941 std::fs::read_to_string(path)?
2942 .lines()
2943 .filter_map(|l| l.trim().parse::<u32>().ok())
2944 .collect()
2945 } else {
2946 let tg = GgufFile::open(path)?;
2947 let d2t_t = tg
2948 .find("d2t")
2949 .expect("MEMRA_FRSPEC_TRIM file has no d2t tensor");
2950 let d2t_bytes = tg.tensor_data(d2t_t);
2951 match d2t_t.ggml_type {
2952 GgmlType::I32 => d2t_bytes
2953 .chunks_exact(4)
2954 .map(|c| i32::from_le_bytes(c.try_into().unwrap()) as u32)
2955 .collect(),
2956 GgmlType::I64 => d2t_bytes
2957 .chunks_exact(8)
2958 .map(|c| i64::from_le_bytes(c.try_into().unwrap()) as u32)
2959 .collect(),
2960 other => panic!("d2t must be I32/I64, got {other:?}"),
2961 }
2962 })
2963}
2964
2965#[allow(clippy::type_complexity)] fn frspec_gather_trimmed_head(
2974 e: &Engine,
2975 v: &memra_gguf::source::TensorView<'_>,
2976 d2t: &[u32],
2977 want_nvfp4_env: bool,
2978 macro_scale: f32,
2979) -> Result<(GpuTensor, Option<(usize, usize)>), Box<dyn std::error::Error>> {
2980 let out_f = v.ne[1] as usize;
2981 let row_bytes = v.bytes.len() / out_f;
2982 assert!(
2983 d2t.iter().all(|&t| (t as usize) < out_f),
2984 "d2t token id >= lm_head rows {out_f}"
2985 );
2986 let mut gathered = Vec::with_capacity(d2t.len() * row_bytes);
2987 for &t in d2t {
2988 let off = t as usize * row_bytes;
2989 gathered.extend_from_slice(&v.bytes[off..off + row_bytes]);
2990 }
2991 let want_nvfp4 =
2992 want_nvfp4_env && matches!(v.ggml_type, GgmlType::BF16) && v.ne[0].is_multiple_of(64);
2993 if want_nvfp4 {
2994 let in_f = v.ne[0] as usize;
2995 let vals: Vec<f32> = gathered
2996 .chunks_exact(2)
2997 .map(|c| f32::from_bits((u16::from_le_bytes([c[0], c[1]]) as u32) << 16))
2998 .collect();
2999 debug_assert_eq!(vals.len(), d2t.len() * in_f);
3000 let blocks = memra_gguf::nvfp4_repack::f32_to_nvfp4(&vals);
3001 let sizes = (blocks.len(), gathered.len());
3002 let trimmed = GpuTensor::from_quant_bytes(
3003 e,
3004 &blocks,
3005 GgmlType::NVFP4,
3006 v.ne[0],
3007 d2t.len() as u64,
3008 1.0,
3009 )?;
3010 Ok((trimmed, Some(sizes)))
3011 } else {
3012 let trimmed = match v.ggml_type {
3013 GgmlType::BF16 => GpuTensor::FloatBf16 {
3014 data: e.htod_bytes(&gathered)?,
3015 ne: vec![v.ne[0], d2t.len() as u64],
3016 },
3017 GgmlType::F32 => GpuTensor::Float {
3018 data: e.htod(
3019 &gathered
3020 .chunks_exact(4)
3021 .map(|c| f32::from_le_bytes(c.try_into().unwrap()))
3022 .collect::<Vec<f32>>(),
3023 )?,
3024 ne: vec![v.ne[0], d2t.len() as u64],
3025 },
3026 _ => GpuTensor::from_quant_bytes(
3027 e,
3028 &gathered,
3029 v.ggml_type,
3030 v.ne[0],
3031 d2t.len() as u64,
3032 macro_scale,
3033 )?,
3034 };
3035 Ok((trimmed, None))
3036 }
3037}
3038
3039pub struct MtpHead {
3040 pub enorm: GpuTensor, pub hnorm: GpuTensor, pub eh_proj: GpuTensor, pub attn_norm: GpuTensor, pub post_attn_norm: GpuTensor, pub mixer: Mixer, pub ffn: Ffn, pub shared_head_norm: Option<GpuTensor>, pub shared_head_head: Option<GpuTensor>, pub d2t: Option<Vec<u32>>,
3054 pub d2t_from_target_head: bool,
3058 pub geom: Option<DraftGeom>,
3064 pub step35: Option<Step35MtpGeom>,
3069}
3070
3071#[derive(Debug, Clone)]
3085pub struct Step35MtpGeom {
3086 pub il: u32,
3088 pub n_head: usize, pub n_head_kv: usize, pub n_rot: usize, pub rope_base: f32, pub swa: bool, pub window: usize, pub clamp_shexp: Option<f32>,
3099}
3100
3101impl Step35MtpGeom {
3102 pub fn from_plan(layer: &memra_gguf::model_plan::LayerPlan) -> Result<Self, String> {
3104 use memra_gguf::model_plan::{ActivationPlan, AttentionPlan};
3105
3106 let (attention, window) = match &layer.attention {
3107 AttentionPlan::Full(attention) => (attention, None),
3108 AttentionPlan::SlidingWindow { attention, window } => (attention, Some(*window)),
3109 other => {
3110 return Err(format!(
3111 "MTP block {} has unsupported tuned attention {other:?}",
3112 layer.index
3113 ));
3114 }
3115 };
3116 if attention.output_gate != memra_gguf::config::AttentionGateKind::SeparateHead {
3117 return Err(format!(
3118 "MTP block {} does not declare a separate attention gate",
3119 layer.index
3120 ));
3121 }
3122 let activation = match &layer.mlp {
3123 MlpPlan::Dense(dense) => &dense.activation,
3124 MlpPlan::Moe(moe) => &moe.activation,
3125 };
3126 let clamp_shexp = match activation {
3127 ActivationPlan::SwiGluClamped { limit } if *limit > 0.0 => Some(*limit),
3128 _ => None,
3129 };
3130 Ok(Step35MtpGeom {
3131 il: layer.index,
3132 n_head: attention.query_heads as usize,
3133 n_head_kv: attention.kv_heads as usize,
3134 n_rot: attention.rope.dimensions as usize,
3135 rope_base: attention.rope.base,
3136 swa: window.is_some(),
3137 window: window.unwrap_or(0) as usize,
3138 clamp_shexp,
3139 })
3140 }
3141}
3142
3143pub struct DraftGeom {
3145 pub d_inner: usize, pub n_head: usize, pub n_head_kv: usize,
3148 pub out_up: GpuTensor, }
3150
3151pub fn draft_head_tensor(has: impl Fn(&str) -> bool, n: u32) -> String {
3160 let own = format!("blk.{n}.nextn.shared_head_head.weight");
3161 if has(&own) {
3162 return own;
3163 }
3164 let legacy = format!("blk.{n}.nextn.shared_head.weight");
3167 if has(&legacy) {
3168 return legacy;
3169 }
3170 "output.weight".to_string()
3172}
3173
3174impl MtpHead {
3175 pub fn load_draft(
3182 e: &Engine,
3183 g: &GgufFile,
3184 main_cfg: &ModelConfig,
3185 ) -> Result<Self, Box<dyn std::error::Error>> {
3186 let src = GgufSource(g);
3187 let dcfg = src.try_config().map_err(std::io::Error::other)?;
3188 let draft_plan = match memra_gguf::model_packs::for_config(&dcfg) {
3189 Some(pack) => pack.compile_plan(&dcfg)?,
3190 None => memra_gguf::model_plan::ModelPlan::compile(&dcfg)?,
3191 };
3192 let main_plan = match memra_gguf::model_packs::for_config(main_cfg) {
3193 Some(pack) => pack.compile_plan(main_cfg)?,
3194 None => memra_gguf::model_plan::ModelPlan::compile(main_cfg)?,
3195 };
3196 if dcfg.nextn_predict_layers == 0 {
3201 return Err(format!(
3202 "draft GGUF has no nextn_predict_layers (arch {:?}) — not a NextN/MTP regime \
3203 draft; gemma assistant drafters attach via MEMRA_DRAFT, not '+draft'",
3204 g.arch()
3205 )
3206 .into());
3207 }
3208 let n = dcfg.n_layer - dcfg.nextn_predict_layers;
3209 let draft_block = draft_plan
3210 .mtp_blocks
3211 .iter()
3212 .find(|block| block.layer.index == n)
3213 .ok_or_else(|| format!("draft ModelPlan has no MTP block {n}"))?;
3214 let p = |s: &str| format!("blk.{n}.{s}");
3215
3216 let student = src.has(&p("nextn.out_up.weight"));
3220 assert_eq!(dcfg.n_embd, main_cfg.n_embd, "draft n_embd != model n_embd");
3221 assert_eq!(
3222 dcfg.head_dim_k, main_cfg.head_dim_k,
3223 "draft head_dim != model head_dim"
3224 );
3225 let main_sliding_gated = crate::plan_backend::decode_batch_program(&main_plan)
3232 == crate::plan_backend::DecodeBatchProgram::SlidingGatedMoe;
3233 let draft_sliding_gated = crate::plan_backend::decode_batch_program(&draft_plan)
3234 == crate::plan_backend::DecodeBatchProgram::SlidingGatedMoe;
3235 let step35 = match (main_sliding_gated, draft_sliding_gated) {
3236 (true, true) => {
3237 let g = Step35MtpGeom::from_plan(&draft_block.layer)?;
3238 let out_f = |t: &str| -> Option<usize> {
3240 src.find(&p(t))
3241 .and_then(|v| v.ne.get(1).copied())
3242 .map(|x| x as usize)
3243 };
3244 let hd = dcfg.head_dim_k as usize;
3245 let wq_out =
3246 out_f("attn_q.weight").ok_or("step35 draft block has no attn_q.weight")?;
3247 assert_eq!(
3248 wq_out,
3249 g.n_head * hd,
3250 "step35 draft blk.{n}: attn_q out {wq_out} != n_head({}) * head_dim({hd}) — \
3251 the draft file's head_count array disagrees with its own tensors",
3252 g.n_head
3253 );
3254 let wg_out = out_f("attn_gate.weight")
3257 .ok_or("step35 draft block has no attn_gate.weight (head-wise gate)")?;
3258 assert_eq!(
3259 wg_out, g.n_head,
3260 "step35 draft blk.{n}: attn_gate out {wg_out} != n_head({})",
3261 g.n_head
3262 );
3263 assert_eq!(
3267 g.n_head_kv, main_cfg.n_head_kv as usize,
3268 "step35 draft blk.{n} KV heads {} != trunk n_head_kv {} — the MTP scratch \
3269 rows are sized from the trunk cfg, so a differing draft KV width would \
3270 write past the row",
3271 g.n_head_kv, main_cfg.n_head_kv
3272 );
3273 eprintln!(
3274 "[mtp-draft] step35 MTP geometry blk.{n}: n_head={} n_head_kv={} n_rot={} \
3275 rope_base={:.0} swa={} window={}",
3276 g.n_head, g.n_head_kv, g.n_rot, g.rope_base, g.swa, g.window
3277 );
3278 Some(g)
3279 }
3280 (true, false) => {
3281 return Err(format!(
3282 "MEMRA_MTP_DRAFT operations are incompatible with the model's \
3283 sliding-gated-MoE program (draft arch {:?})",
3284 g.arch()
3285 )
3286 .into());
3287 }
3288 (false, true) => {
3289 return Err(
3290 "MEMRA_MTP_DRAFT requires sliding-gated-MoE operations but the model does not"
3291 .into(),
3292 );
3293 }
3294 (false, false) => None,
3295 };
3296 if step35.is_none() && !student {
3297 assert_eq!(dcfg.n_head, main_cfg.n_head, "draft n_head != model n_head");
3300 assert_eq!(
3301 dcfg.n_head_kv, main_cfg.n_head_kv,
3302 "draft n_head_kv != model n_head_kv"
3303 );
3304 }
3305
3306 let head_name = draft_head_tensor(|t| src.has(t), n);
3333 let head = load_t(e, &src, &head_name)?;
3334 let head_norm = match load_opt(e, &src, &p("nextn.shared_head_norm.weight"))? {
3335 Some(t) => Some(t),
3336 None => load_opt(e, &src, "output_norm.weight")?,
3337 };
3338
3339 let d2t: Option<Vec<u32>> = g.find("d2t").map(|t| {
3341 let bytes = g.tensor_data(t);
3342 match t.ggml_type {
3343 GgmlType::I32 => bytes
3344 .chunks_exact(4)
3345 .map(|c| i32::from_le_bytes(c.try_into().unwrap()) as u32)
3346 .collect(),
3347 GgmlType::I64 => bytes
3348 .chunks_exact(8)
3349 .map(|c| i64::from_le_bytes(c.try_into().unwrap()) as u32)
3350 .collect(),
3351 other => panic!("d2t must be I32/I64, got {other:?}"),
3352 }
3353 });
3354 if let Some(map) = &d2t {
3355 assert_eq!(
3356 map.len(),
3357 head.out_features(),
3358 "d2t len {} != draft head rows {}",
3359 map.len(),
3360 head.out_features()
3361 );
3362 let n_vocab = main_cfg.n_vocab as u64;
3363 assert!(
3364 map.iter().all(|&t| (t as u64) < n_vocab),
3365 "d2t contains token id >= model n_vocab {n_vocab}"
3366 );
3367 }
3368 let eh_proj = load_t(e, &src, &p("nextn.eh_proj.weight"))?;
3369 assert_eq!(
3372 eh_proj.in_features(),
3373 2 * main_cfg.n_embd as usize,
3374 "eh_proj in dim != 2*n_embd"
3375 );
3376 let geom = if student {
3377 let out_up = load_t(e, &src, &p("nextn.out_up.weight"))?;
3378 let d_inner = eh_proj.out_features();
3379 assert_eq!(
3380 out_up.out_features(),
3381 main_cfg.n_embd as usize,
3382 "out_up out dim != n_embd"
3383 );
3384 assert_eq!(
3385 out_up.in_features(),
3386 d_inner,
3387 "out_up in dim != eh_proj out dim (d_inner)"
3388 );
3389 assert!(
3390 dcfg.n_head >= 1 && dcfg.n_head_kv >= 1 && dcfg.n_head % dcfg.n_head_kv == 0,
3391 "student head counts malformed ({}/{})",
3392 dcfg.n_head,
3393 dcfg.n_head_kv
3394 );
3395 Some(DraftGeom {
3396 d_inner,
3397 n_head: dcfg.n_head as usize,
3398 n_head_kv: dcfg.n_head_kv as usize,
3399 out_up,
3400 })
3401 } else {
3402 None
3403 };
3404 let blk_prefix = format!("blk.{n}.");
3408 let head_src = head_name.strip_prefix(&blk_prefix).unwrap_or(&head_name);
3409 eprintln!(
3410 "[mtp-draft] external draft head: blk.{n}, source={}, head_vocab={}{}{}",
3411 head_src,
3412 head.out_features(),
3413 if d2t.is_some() {
3414 " (trimmed, d2t map)"
3415 } else {
3416 " (full)"
3417 },
3418 match &geom {
3419 Some(g) => format!(
3420 " (student d_inner={} heads={}/{})",
3421 g.d_inner, g.n_head, g.n_head_kv
3422 ),
3423 None => String::new(),
3424 }
3425 );
3426
3427 let mut resident = ResidentPlan::unsharded(e, &src, &dcfg);
3428 let mut step_runtimes = StepParallelRuntimeRegistry::default();
3429 Ok(MtpHead {
3430 enorm: load_t(e, &src, &p("nextn.enorm.weight"))?,
3431 hnorm: load_t(e, &src, &p("nextn.hnorm.weight"))?,
3432 eh_proj,
3433 attn_norm: load_t(e, &src, &p("attn_norm.weight"))?,
3434 post_attn_norm: load_opt(e, &src, &p("post_attention_norm.weight"))?
3435 .or(load_opt(e, &src, &p("ffn_norm.weight"))?)
3436 .expect("draft NextN block needs post_attention_norm or ffn_norm"),
3437 mixer: load_mixer_kind(
3438 e,
3439 &src,
3440 &dcfg,
3441 n,
3442 &draft_block.layer.attention,
3443 &mut step_runtimes,
3444 )?,
3445 ffn: load_ffn(
3446 e,
3447 &src,
3448 &dcfg,
3449 &draft_block.layer.mlp,
3450 n,
3451 None,
3452 &mut resident,
3453 &mut step_runtimes,
3454 )?,
3455 shared_head_norm: head_norm,
3456 shared_head_head: Some(head),
3457 d2t,
3458 d2t_from_target_head: false,
3459 geom,
3460 step35,
3461 })
3462 }
3463}
3464
3465pub struct GemmaAux {
3467 pub rope_freqs: Option<Vec<(usize, CudaSlice<f32>)>>,
3470 pub ones: Vec<(usize, CudaSlice<f32>)>,
3473 pub suppress_d: Option<(CudaSlice<i32>, usize)>,
3476 pub e4b: Option<Gemma4E4bModel>,
3478}
3479
3480impl GemmaAux {
3481 pub fn rope_freqs(&self, e: &Engine) -> Option<&CudaSlice<f32>> {
3482 self.rope_freqs.as_ref().map(|copies| {
3483 let dev = e.ctx().ordinal();
3484 &copies
3485 .iter()
3486 .find(|(d, _)| *d == dev)
3487 .unwrap_or_else(|| panic!("gemma4 rope_freqs has no local copy for device {dev}"))
3488 .1
3489 })
3490 }
3491
3492 pub fn ones(&self, e: &Engine) -> &CudaSlice<f32> {
3493 let dev = e.ctx().ordinal();
3494 &self
3495 .ones
3496 .iter()
3497 .find(|(d, _)| *d == dev)
3498 .unwrap_or_else(|| panic!("gemma4 ones has no local copy for device {dev}"))
3499 .1
3500 }
3501}
3502
3503pub struct Step35Aux {
3506 pub rope_freqs: Option<Vec<(usize, CudaSlice<f32>)>>,
3512}
3513
3514impl Step35Aux {
3515 pub fn rope_freqs(&self, e: &Engine) -> Option<&CudaSlice<f32>> {
3516 self.rope_freqs.as_ref().map(|copies| {
3517 let dev = e.ctx().ordinal();
3518 &copies
3519 .iter()
3520 .find(|(d, _)| *d == dev)
3521 .unwrap_or_else(|| panic!("step35 rope_freqs has no local copy for device {dev}"))
3522 .1
3523 })
3524 }
3525}
3526
3527pub struct HybridModel {
3528 pub cfg: ModelConfig,
3529 pub plan: memra_gguf::model_plan::ModelPlan,
3530 pub rewrite_qualifications: Option<memra_gguf::execution_manifest::RewriteQualifications>,
3531 pub embd: EmbedHost,
3532 pub output_norm: GpuTensor,
3533 pub output: GpuTensor,
3534 pub layers: Vec<HybridLayer>,
3535 pub mtp: Option<MtpHead>, pub mtp_extra: Vec<MtpHead>,
3539 pub dflash_trim: Option<DflashTrimHead>,
3543 pub embd_gpu: std::sync::OnceLock<cudarc::driver::CudaSlice<u8>>,
3546 pub gemma4_aux: Option<GemmaAux>,
3547 pub step35_aux: Option<Step35Aux>,
3549 pub prime_slabs: std::sync::Mutex<
3557 std::collections::HashMap<
3558 usize,
3559 std::sync::Arc<std::sync::Mutex<crate::hybrid_forward::PrimeSlabs>>,
3560 >,
3561 >,
3562 pub(crate) dspark_vgraphs: std::sync::Mutex<Option<crate::spec::DsparkVerifyGraphs>>,
3575 pub(crate) step_grouped_prefill: std::sync::Mutex<StepEpGroupedPrefill>,
3581 pub(crate) step35_token_graph:
3584 std::sync::Mutex<Option<crate::hybrid_forward::Step35TokenGraphState>>,
3585 pub hyper: Option<crate::hyper::HyperTopology>,
3590 pub hyper_head: Option<crate::hyper::HyperHead>,
3593 pub glm5_dflash: Option<crate::glm_spec::Glm5DflashDrafter>,
3600 pub(crate) draft_state_bytes: std::sync::atomic::AtomicUsize,
3609}
3610
3611impl HybridModel {
3612 pub fn install_rewrite_bundle(
3613 &mut self,
3614 bundle: &std::path::Path,
3615 ) -> Result<(), Box<dyn std::error::Error>> {
3616 self.rewrite_qualifications = Some(
3617 memra_gguf::execution_manifest::RewriteQualifications::load(bundle, &self.plan)
3618 .map_err(|error| format!("rewrite qualification: {error}"))?,
3619 );
3620 Ok(())
3621 }
3622
3623 pub fn rewrite_allowed(&self, surface: memra_gguf::execution_manifest::RewriteSurface) -> bool {
3624 self.rewrite_qualifications
3625 .as_ref()
3626 .is_none_or(|qualifications| qualifications.allows(surface))
3627 }
3628
3629 pub fn record_draft_state_bytes(&self, observed: usize) -> Option<usize> {
3634 use std::sync::atomic::Ordering;
3635 let prev = self
3636 .draft_state_bytes
3637 .fetch_max(observed, Ordering::Relaxed);
3638 (observed > prev).then_some(observed)
3639 }
3640
3641 pub fn draft_session_admission_bytes(&self) -> usize {
3647 self.draft_state_bytes
3648 .load(std::sync::atomic::Ordering::Relaxed)
3649 }
3650
3651 pub fn step_tp_unmaterialized_kv_bytes(
3657 &self,
3658 cache: Option<&crate::cache::Cache>,
3659 capacity: usize,
3660 ) -> Result<Vec<StepTpKvDeviceAdmission>, String> {
3661 if let Some(cache) = cache
3662 && cache.tp_kv.len() < self.layers.len()
3663 {
3664 return Err(format!(
3665 "Step TP admission cache has {} layers, model trunk has {}",
3666 cache.tp_kv.len(),
3667 self.layers.len()
3668 ));
3669 }
3670
3671 let mut by_device: HashMap<usize, usize> = HashMap::new();
3672 for (layer, weights) in self.layers.iter().enumerate() {
3673 let Mixer::Full(attention) = &weights.mixer else {
3674 continue;
3675 };
3676 let Some(tp) = attention
3677 .step_tp_qkv
3678 .as_ref()
3679 .filter(|tp| tp.attention.is_some())
3680 else {
3681 continue;
3682 };
3683 if cache.is_some_and(|cache| cache.tp_kv[layer].is_some()) {
3684 continue;
3685 }
3686 let geometry = self.cfg.full_attention_geometry_at(layer as u32);
3687 let shape = crate::cache::tp_kv_rank_allocation_shape(
3688 geometry.n_head_kv as usize * geometry.head_dim_k as usize,
3689 geometry.n_head_kv as usize * geometry.head_dim_v as usize,
3690 tp.devices.len(),
3691 )?;
3692 let physical_rows = geometry
3693 .window
3694 .map(|window| crate::cache::swa_ring_rows(window as usize, capacity))
3695 .unwrap_or(capacity);
3696 let bytes = shape.allocation_bytes(physical_rows);
3697 for &device in &tp.devices {
3698 let total = by_device.entry(device).or_default();
3699 *total = total.saturating_add(bytes);
3700 }
3701 }
3702
3703 let mut out: Vec<_> = by_device
3704 .into_iter()
3705 .map(|(device, bytes)| StepTpKvDeviceAdmission { device, bytes })
3706 .collect();
3707 out.sort_unstable_by_key(|charge| charge.device);
3708 Ok(out)
3709 }
3710
3711 pub fn step_tp_rank_engine(&self, device: usize) -> Option<&Engine> {
3713 self.layers.iter().find_map(|weights| {
3714 let Mixer::Full(attention) = &weights.mixer else {
3715 return None;
3716 };
3717 let tp = attention.step_tp_qkv.as_ref()?;
3718 let rank = tp
3719 .runtime
3720 .devices()
3721 .iter()
3722 .position(|&rank| rank == device)?;
3723 tp.runtime.rank_engine(rank)
3724 })
3725 }
3726
3727 pub(crate) fn step_tp_runtime_for_layer(
3728 &self,
3729 layer: usize,
3730 ) -> Option<&crate::tp::TpE4m3HostBounce> {
3731 let Mixer::Full(attention) = &self.layers.get(layer)?.mixer else {
3732 return None;
3733 };
3734 let tp = attention.step_tp_qkv.as_ref()?;
3735 tp.attention.as_ref()?;
3736 Some(tp.runtime.as_ref())
3737 }
3738
3739 pub fn decode_batch_program(&self) -> crate::plan_backend::DecodeBatchProgram {
3740 crate::plan_backend::decode_batch_program(&self.plan)
3741 }
3742
3743 pub fn uses_gemma_program(&self) -> bool {
3744 self.decode_batch_program() == crate::plan_backend::DecodeBatchProgram::Gemma
3745 }
3746
3747 pub fn uses_sliding_gated_moe_program(&self) -> bool {
3748 self.decode_batch_program() == crate::plan_backend::DecodeBatchProgram::SlidingGatedMoe
3749 }
3750
3751 pub fn has_plan_operation(&self, operation: memra_gguf::model_plan::OperationKind) -> bool {
3752 self.plan.trunk_operations().contains(&operation)
3753 }
3754
3755 pub fn load(e: &Engine, g: &GgufFile) -> Result<Self, Box<dyn std::error::Error>> {
3757 Self::load_from_source(e, &GgufSource(g))
3758 }
3759
3760 pub fn load_without_mtp(e: &Engine, g: &GgufFile) -> Result<Self, Box<dyn std::error::Error>> {
3763 Self::load_from_source_impl(e, &GgufSource(g), false)
3764 }
3765
3766 pub fn load_from_source(
3770 e: &Engine,
3771 src: &dyn TensorSource,
3772 ) -> Result<Self, Box<dyn std::error::Error>> {
3773 Self::load_from_source_impl(e, src, true)
3774 }
3775
3776 pub fn load_from_source_without_mtp(
3778 e: &Engine,
3779 src: &dyn TensorSource,
3780 ) -> Result<Self, Box<dyn std::error::Error>> {
3781 Self::load_from_source_impl(e, src, false)
3782 }
3783
3784 fn load_from_source_impl(
3785 e: &Engine,
3786 src: &dyn TensorSource,
3787 load_mtp: bool,
3788 ) -> Result<Self, Box<dyn std::error::Error>> {
3789 let cfg = src.try_config().map_err(std::io::Error::other)?;
3790 let plan = match memra_gguf::model_packs::for_config(&cfg) {
3791 Some(pack) => pack.compile_plan(&cfg)?,
3792 None => memra_gguf::model_plan::ModelPlan::compile(&cfg)?,
3793 };
3794 let auto_parallel = prepare_auto_parallel(src, &cfg, &plan)?;
3795 let batch_program = crate::plan_backend::decode_batch_program(&plan);
3796 let gemma_program = batch_program == crate::plan_backend::DecodeBatchProgram::Gemma;
3797 let sliding_gated_moe_program =
3798 batch_program == crate::plan_backend::DecodeBatchProgram::SlidingGatedMoe;
3799 if matches!(
3800 src.expert_activation_precision(),
3801 memra_gguf::source::ExpertActivationPrecision::Bf16
3802 ) {
3803 eprintln!(
3804 "[w4a16] artifact contract accepted: expert_weights=nvfp4 \
3805 expert_activations=bf16-rounded q8_expert_program=disabled"
3806 );
3807 }
3808 if sliding_gated_moe_program {
3813 crate::arm_step37_serving_defaults();
3814 }
3815 cfg.validate_attention_gate_layout()?;
3820 if cfg.sigmoid_router().is_some() {
3827 let host_oracle = std::env::var("MEMRA_SIG_ROUTER").as_deref() == Ok("0");
3828 match crate::sigrouter_contract::verify_host_expf() {
3829 Ok(()) => {}
3830 Err(e) if host_oracle => return Err(e.into()),
3831 Err(e) => eprintln!(
3832 "[sigrouter] WARN: host expf probe mismatch ({e}); device routing is \
3833 unaffected, but host-oracle replay/comparison cells are invalid on this host"
3834 ),
3835 }
3836 }
3837 if std::env::var("MEMRA_DRAFT").is_ok() && std::env::var("MEMRA_MMQ_SK").is_err() {
3846 let force = if cfg.n_embd >= 3500 { 0i8 } else { -1i8 };
3847 crate::MMQ_SK_FORCE.store(force, std::sync::atomic::Ordering::Relaxed);
3848 }
3849 crate::KV_FP8_FORCE.store(0, std::sync::atomic::Ordering::Relaxed);
3853
3854 let n_trunk = (cfg.n_layer - cfg.nextn_predict_layers) as usize;
3859 let mtp_skip_requested = load_mtp
3869 && match std::env::var("MEMRA_MTP_SKIP").ok().as_deref() {
3870 None | Some("") | Some("0") => false,
3871 Some("1") => true,
3872 Some(other) => {
3873 return Err(format!(
3874 "MEMRA_MTP_SKIP={other:?}: expected 1 (skip the embedded MTP block) or \
3875 0/unset (load it); refusing to guess"
3876 )
3877 .into());
3878 }
3879 };
3880 if mtp_skip_requested && std::env::var("MEMRA_MTP_DRAFT").is_ok_and(|p| !p.is_empty()) {
3881 return Err(
3882 "MEMRA_MTP_SKIP=1 together with MEMRA_MTP_DRAFT is contradictory: the skip \
3883 removes the MTP head to reclaim VRAM while MEMRA_MTP_DRAFT attaches an \
3884 external MTP head for MTP spec decode; unset one"
3885 .into(),
3886 );
3887 }
3888 if mtp_skip_requested && cfg.nextn_predict_layers > 0 {
3889 let prefixes: Vec<String> = (0..cfg.nextn_predict_layers)
3894 .map(|off| format!("blk.{}.", n_trunk as u32 + off))
3895 .collect();
3896 let skipped_bytes: Option<u64> = src.gguf().map(|g| {
3897 g.tensors
3898 .iter()
3899 .filter(|t| prefixes.iter().any(|p| t.name.starts_with(p.as_str())))
3900 .map(|t| t.n_bytes)
3901 .sum()
3902 });
3903 eprintln!(
3904 "[mtp-skip] MEMRA_MTP_SKIP=1: skipping {} embedded MTP/NextN block(s) \
3905 blk.{}..=blk.{} ({}); MTP spec decode is unavailable for this model \
3906 (dspark/DFlash2 drafting keeps its trimmed head via the MEMRA_FRSPEC_TRIM stub)",
3907 cfg.nextn_predict_layers,
3908 n_trunk,
3909 n_trunk as u32 + cfg.nextn_predict_layers - 1,
3910 match skipped_bytes {
3911 Some(b) => format!("~{} MiB of weights not loaded", b >> 20),
3912 None => "size unknown: non-GGUF source".to_string(),
3913 },
3914 );
3915 }
3916 let mtp_skip_trim_d2t: Option<Vec<u32>> = if mtp_skip_requested
3931 && cfg.nextn_predict_layers > 0
3932 && !crate::model::full_prec_enabled()
3933 {
3934 match std::env::var("MEMRA_FRSPEC_TRIM") {
3935 Ok(path) if !path.is_empty() => {
3936 let path = memra_gguf::hf::resolve_arg(&path)
3937 .map_err(|err| format!("MEMRA_FRSPEC_TRIM={path:?}: {err}"))?;
3938 let own_head_name = frspec_trim_own_head_name(n_trunk);
3939 if src.has(&own_head_name) {
3940 return Err(format!(
3941 "MEMRA_MTP_SKIP=1 with MEMRA_FRSPEC_TRIM: this artifact ships its \
3942 own MTP-block lm_head ({own_head_name}), so the trimmed draft rows \
3943 live in the block being skipped; gathering trunk rows instead is \
3944 the wrong-head bug (acceptance 0/248 receipt, \
3945 frspec_trim_own_head_name). Unset MEMRA_MTP_SKIP or \
3946 MEMRA_FRSPEC_TRIM"
3947 )
3948 .into());
3949 }
3950 if !src.has("output.weight") && !src.has("token_embd.weight") {
3951 return Err("MEMRA_MTP_SKIP=1 with MEMRA_FRSPEC_TRIM: model has no \
3952 output.weight (or tied token_embd.weight) to gather trimmed draft \
3953 rows from"
3954 .into());
3955 }
3956 let d2t = frspec_read_d2t(&path)?;
3957 if d2t.is_empty() {
3958 return Err(format!(
3959 "MEMRA_MTP_SKIP=1 with MEMRA_FRSPEC_TRIM={path}: the rank artifact \
3960 yields an EMPTY d2t list, so no stub draft head can be built; fix \
3961 the artifact or unset MEMRA_MTP_SKIP"
3962 )
3963 .into());
3964 }
3965 Some(d2t)
3966 }
3967 _ => None,
3968 }
3969 } else {
3970 None
3971 };
3972 if let Some(fence) = crate::pp::pp_cuts(n_trunk) {
3973 let pipeline = crate::plan_backend::PIPELINE
3974 .trunk_capabilities(&plan)
3975 .pipeline;
3976 let qualified_gemma_pp2 = gemma_program && fence.len() == 3;
3980 if !pipeline.supported && !qualified_gemma_pp2 {
3981 return Err(format!(
3982 "pipeline placement is unsupported for plan operations {:?}; blockers={:?}",
3983 plan.trunk_operations(),
3984 pipeline.blockers,
3985 )
3986 .into());
3987 }
3988 let illegal = illegal_pipeline_cuts(&fence, &plan.partition_boundaries);
3989 if !illegal.is_empty() {
3990 return Err(format!(
3991 "pipeline placement cuts {illegal:?} split outside ModelPlan legal boundaries {:?}",
3992 plan.partition_boundaries,
3993 )
3994 .into());
3995 }
3996 }
3997 crate::pp::init_model_transport(e, &cfg, n_trunk)?;
3998 let step_parallel =
3999 prepare_step_parallel_load(e, src, &cfg, n_trunk, auto_parallel.as_ref())?;
4000 let glm5_tp = if crate::glm5_tp::glm5_tp_armed() {
4004 use memra_gguf::model_plan::{AttentionPlan, MlpPlan};
4005 let moe = cfg.moe.as_ref().ok_or(
4006 "MEMRA_GLM5_TP requires a MoE model (glm5_next); this plan carries no MoE \
4007 metadata",
4008 )?;
4009 let mut layer_class = Vec::with_capacity(n_trunk);
4010 let mut layer_is_moe = Vec::with_capacity(n_trunk);
4011 let (mut kda_heads, mut kda_head_dim, mut mla_heads) = (0usize, 0usize, 0usize);
4012 for (il, lp) in plan.layers.iter().take(n_trunk).enumerate() {
4013 match &lp.attention {
4014 AttentionPlan::KimiDeltaNet(k) => {
4015 layer_class.push(crate::glm5_tp::Glm5LayerClass::Kda);
4016 kda_heads = k.num_heads as usize;
4017 kda_head_dim = k.head_dim as usize;
4018 }
4019 AttentionPlan::Mla(memra_gguf::model_plan::MlaAttentionPlan::LatentKv {
4020 query_heads,
4021 ..
4022 }) => {
4023 layer_class.push(crate::glm5_tp::Glm5LayerClass::Mla);
4024 mla_heads = *query_heads as usize;
4025 }
4026 other => {
4027 return Err(format!(
4028 "MEMRA_GLM5_TP requires a glm5_next-class plan (KDA/MLA mixers): \
4029 trunk layer {il} declares {other:?}"
4030 )
4031 .into());
4032 }
4033 }
4034 layer_is_moe.push(matches!(&lp.mlp, MlpPlan::Moe(_)));
4035 }
4036 let view = crate::glm5_tp::Glm5TpModelView {
4037 trunk_layers: n_trunk,
4038 layer_class,
4039 layer_is_moe,
4040 kda_heads,
4041 kda_head_dim,
4042 mla_heads,
4043 n_routed_experts: moe.expert_count as usize,
4044 top_k: moe.expert_used_count as usize,
4045 };
4046 crate::glm5_tp::prepare_glm5_tp_load(e, &view)?
4047 } else {
4048 let glm5_class = plan.layers.iter().take(n_trunk).any(|lp| {
4055 matches!(
4056 lp.attention,
4057 memra_gguf::model_plan::AttentionPlan::KimiDeltaNet(_)
4058 )
4059 });
4060 let ep_map_armed = crate::ep_map::ep_map_env()?;
4061 if let Some((flag, _)) = ep_map_armed
4062 && glm5_class
4063 {
4064 return Err(format!(
4065 "{flag} is set but MEMRA_GLM5_TP is off: the map cannot \
4066 engage, and a placement that silently reverts to the even split is \
4067 refused by name (unset one of the two)"
4068 )
4069 .into());
4070 }
4071 if glm5_class {
4079 for (armed, flag) in [crate::ep_diet_armed(), crate::ep_grouped_prime_armed()] {
4080 if armed {
4081 return Err(format!(
4082 "{flag}=1 is set but MEMRA_GLM5_TP is off: the EP dispatch \
4083 diet only exists inside the TP-2 EP walk and cannot engage \
4084 (unset one of the two)"
4085 )
4086 .into());
4087 }
4088 }
4089 }
4090 None
4091 };
4092 let embd = EmbedHost::from_source(src, "token_embd.weight");
4093 let e_head = crate::pp::layer_engine(e, n_trunk, n_trunk - 1)?;
4097 let output_norm = load_t(e_head, src, "output_norm.weight")?;
4098 let mut output = if src.has("output.weight") {
4100 load_t(e_head, src, "output.weight")?
4101 } else {
4102 load_t(e_head, src, "token_embd.weight")?
4103 };
4104 let mut resident = ResidentPlan::pp(e, src, &cfg, n_trunk)?;
4105 resident.exclude_distributed_expert_layers(
4106 step_parallel
4107 .ep_specs
4108 .iter()
4109 .map(|spec| spec.layer)
4110 .chain(step_parallel.tp_specs.iter().map(|spec| spec.layer)),
4111 );
4112 let mut step_runtimes = StepParallelRuntimeRegistry::with_config(step_parallel);
4113
4114 let gguf: Option<&GgufFile> = src.gguf();
4121 let mut spill: Option<crate::spill::SpillCtx> = if cfg
4124 .moe
4125 .as_ref()
4126 .is_some_and(|m| m.expert_count > 0)
4127 && crate::spill::disk_tier_enabled()
4128 && gguf.is_some()
4129 {
4130 let budget = crate::spill::MemBudget::probe(e)?;
4131 #[allow(clippy::unnecessary_unwrap)]
4132 let ctx = crate::spill::SpillCtx::open(gguf.unwrap(), &budget)?;
4134 eprintln!(
4135 "[spill] disk tier ON: free_vram={} MiB free_pinnable_ram={} MiB (MemAvailable*resolved_frac)",
4136 budget.free_vram >> 20,
4137 budget.free_pinnable_ram >> 20
4138 );
4139 Some(ctx)
4140 } else {
4141 None
4142 };
4143
4144 let hyper = crate::hyper::HyperTopology::from_plan(&plan)?;
4151 let hyper_head = match hyper.as_ref() {
4152 Some(topology) => {
4153 crate::hyper::HyperHead::load(e_head, src, topology, cfg.n_embd as usize)?
4154 }
4155 None => None,
4156 };
4157 let mut layers = Vec::with_capacity(n_trunk);
4158 for il in 0..n_trunk as u32 {
4159 let p = |s: &str| format!("blk.{il}.{s}");
4160 let layer_plan = plan
4161 .layers
4162 .get(il as usize)
4163 .ok_or_else(|| format!("ModelPlan has no trunk layer {il}"))?;
4164 let e = crate::pp::layer_engine(e, n_trunk, il as usize)?;
4168 layers.push(HybridLayer {
4170 attn_norm: load_t(e, src, &p("attn_norm.weight"))?,
4171 post_attn_norm: load_opt(e, src, &p("post_attention_norm.weight"))?
4172 .or(load_opt(e, src, &p("ffn_norm.weight"))?)
4173 .expect("need post_attention_norm or ffn_norm"),
4174 mixer: {
4175 let g4_shared = cfg.gemma4.as_ref().map(|g| g.shared_kv_layers).unwrap_or(0);
4179 let kv_from = n_trunk as u32 - g4_shared;
4180 if g4_shared > 0
4181 && il >= kv_from
4182 && !src.has(&format!("blk.{il}.attn_k.weight"))
4183 {
4184 let g4 = cfg.gemma4.as_ref().unwrap();
4185 let swa = g4.swa_pattern.get(il as usize).copied().unwrap_or(true);
4186 let tgt = kv_from - if swa { 2 } else { 1 };
4187 let tp = |s: &str| format!("blk.{tgt}.{s}");
4188 Mixer::Full(FullAttnLayer {
4189 wq: load_t(e, src, &p("attn_q.weight"))?,
4190 wk: load_t(e, src, &tp("attn_k.weight"))?,
4191 wv: load_t(e, src, &tp("attn_v.weight"))?,
4192 wo: load_t(e, src, &p("attn_output.weight"))?,
4193 q_norm: load_t(e, src, &p("attn_q_norm.weight"))?,
4194 k_norm: load_t(e, src, &tp("attn_k_norm.weight"))?,
4195 attn_gate: None, step_tp_qkv: None,
4197 })
4198 } else {
4199 load_mixer_kind(
4200 e,
4201 src,
4202 &cfg,
4203 il,
4204 &layer_plan.attention,
4205 &mut step_runtimes,
4206 )?
4207 }
4208 },
4209 ffn: load_ffn(
4210 e,
4211 src,
4212 &cfg,
4213 &layer_plan.mlp,
4214 il,
4215 spill.as_mut().map(|c| (gguf.unwrap(), c)),
4216 &mut resident,
4217 &mut step_runtimes,
4218 )?,
4219 gemma4: if gemma_program {
4220 let scalar = |n: &str| -> f32 {
4221 let t = src.find(&p(n)).unwrap_or_else(|| panic!("missing {n}"));
4222 memra_gguf::dequant::dequantize(t.ggml_type, &t.bytes, 1)[0]
4223 };
4224 let vecf = |n: &str| -> Vec<f32> {
4225 let t = src.find(&p(n)).unwrap_or_else(|| panic!("missing {n}"));
4226 memra_gguf::dequant::dequantize(
4227 t.ggml_type,
4228 &t.bytes,
4229 t.ne.iter().product::<u64>() as usize,
4230 )
4231 };
4232 let moe_bits = if src.find(&p("ffn_gate_inp.scale")).is_some() {
4233 Some(crate::hybrid::Gemma4MoeBits {
4234 post_ffw_norm_1: load_t(e, src, &p("post_ffw_norm_1.weight"))?,
4235 pre_ffw_norm_2: load_t(e, src, &p("pre_ffw_norm_2.weight"))?,
4236 post_ffw_norm_2: load_t(e, src, &p("post_ffw_norm_2.weight"))?,
4237 shared_gate: load_t(e, src, &p("ffn_gate.weight"))?,
4238 shared_up: load_t(e, src, &p("ffn_up.weight"))?,
4239 shared_down: load_t(e, src, &p("ffn_down.weight"))?,
4240 router_scale_pre: {
4241 let inv = 1.0 / (cfg.n_embd as f32).sqrt();
4242 let v: Vec<f32> =
4243 vecf("ffn_gate_inp.scale").iter().map(|x| x * inv).collect();
4244 e.htod(&v)?
4245 },
4246 per_expert_scale: vecf("ffn_down_exps.scale"),
4247 per_expert_scale_d: e.htod(&vecf("ffn_down_exps.scale"))?,
4248 })
4249 } else {
4250 None
4251 };
4252 let e4b = if src.has(&p("inp_gate.weight")) {
4254 let g4 = cfg.gemma4.as_ref().unwrap();
4255 let kv_from = n_trunk as u32 - g4.shared_kv_layers;
4256 let kv_share = if g4.shared_kv_layers > 0 && il >= kv_from {
4257 let swa = g4.swa_pattern.get(il as usize).copied().unwrap_or(true);
4258 Some(kv_from - if swa { 2 } else { 1 })
4259 } else {
4260 None
4261 };
4262 Some(crate::hybrid::Gemma4E4bLayer {
4263 inp_gate: load_t(e, src, &p("inp_gate.weight"))?,
4264 proj: load_t(e, src, &p("proj.weight"))?,
4265 post_norm: load_t(e, src, &p("post_norm.weight"))?,
4266 kv_share,
4267 qkv_cat: None, })
4269 } else {
4270 None
4271 };
4272 Some(Gemma4LayerBits {
4273 ffn_norm: load_t(e, src, &p("ffn_norm.weight"))?,
4274 post_ffw_norm: load_t(e, src, &p("post_ffw_norm.weight"))?,
4275 moe_bits,
4276 layer_scale: scalar("layer_output_scale.weight"),
4277 e4b,
4278 })
4279 } else {
4280 None
4281 },
4282 hyper: match hyper.as_ref() {
4283 Some(topology) => Some(crate::hyper::HyperLayer::load(
4284 e,
4285 src,
4286 il,
4287 topology,
4288 cfg.n_embd as usize,
4289 )?),
4290 None => None,
4291 },
4292 });
4293 if let Some(tp_plan) = &glm5_tp
4296 && tp_plan.layers.contains(&(il as usize))
4297 {
4298 let mut layer = layers.pop().expect("layer just pushed");
4299 layer.mixer = match layer.mixer {
4300 Mixer::Kda(la) => {
4301 Mixer::Kda(crate::glm5_tp::shard_kda_layer(e, &tp_plan.rt, la)?)
4302 }
4303 Mixer::Mla(la) => {
4304 Mixer::Mla(crate::glm5_tp::shard_mla_layer(e, &tp_plan.rt, la)?)
4305 }
4306 _ => {
4307 return Err(format!(
4308 "MEMRA_GLM5_TP selected layer {il}, whose loaded mixer is not \
4309 KDA/MLA — preflight and loader disagree (wiring bug)"
4310 )
4311 .into());
4312 }
4313 };
4314 if let Ffn::Moe(m) = &mut layer.ffn {
4315 let placement = match &tp_plan.ep_map {
4319 Some(map) => Some(
4320 map.layers
4321 .get(&(il as usize))
4322 .ok_or_else(|| {
4323 format!(
4324 "glm5-tp EP: preflight-validated map lost layer {il} \
4325 (wiring bug)"
4326 )
4327 })?
4328 .as_slice(),
4329 ),
4330 None => None,
4331 };
4332 crate::glm5_tp::arm_moe_ep(e, &tp_plan.rt, m, placement)?;
4333 }
4334 layers.push(layer);
4335 }
4336 }
4337
4338 let external_mtp_requested =
4342 load_mtp && std::env::var("MEMRA_MTP_DRAFT").is_ok_and(|path| !path.is_empty());
4343 let trim_mtp_requested = load_mtp
4344 && !crate::model::full_prec_enabled()
4345 && std::env::var("MEMRA_FRSPEC_TRIM").is_ok_and(|path| !path.is_empty());
4346 let _ = trim_mtp_requested;
4357 let glm5_mtp_requested =
4367 !cfg.arch.is_glm5_next() || std::env::var("MEMRA_GLM5_MTP").as_deref() == Ok("1");
4368 let embedded_head_count =
4371 if external_mtp_requested || !glm5_mtp_requested || mtp_skip_requested {
4372 0
4373 } else {
4374 cfg.nextn_predict_layers
4375 };
4376 if cfg.arch.is_glm5_next()
4377 && glm5_mtp_requested
4378 && !mtp_skip_requested
4379 && cfg.nextn_predict_layers > 0
4380 {
4381 eprintln!("[mtp-glm5] MEMRA_GLM5_MTP=1: loading the glm5_next NextN block");
4382 }
4383 let embedded_head_count = match std::env::var("MEMRA_MTP_HEADS")
4388 .ok()
4389 .and_then(|v| v.parse::<u32>().ok())
4390 .filter(|&n| n > 0)
4391 {
4392 Some(cap) if cap < embedded_head_count => {
4393 eprintln!(
4394 "[mtp-chain] MEMRA_MTP_HEADS={cap}: capping the embedded chain from \
4395 {embedded_head_count} heads (measurement knob)"
4396 );
4397 cap
4398 }
4399 _ => embedded_head_count,
4400 };
4401 let mut embedded_mtp = Vec::new();
4402 if load_mtp && embedded_head_count > 0 {
4403 for offset in 0..embedded_head_count {
4404 let n = n_trunk as u32 + offset;
4405 let e = crate::pp::layer_engine(e, n_trunk, n as usize)?;
4411 let p = |s: &str| format!("blk.{n}.{s}");
4412 let mtp_plan = plan
4413 .mtp_blocks
4414 .iter()
4415 .find(|block| block.layer.index == n)
4416 .ok_or_else(|| format!("ModelPlan has no embedded MTP block {n}"))?;
4417 if !src.has(&p("nextn.eh_proj.weight")) {
4418 if offset == 0 {
4419 break;
4420 }
4421 return Err(format!(
4422 "embedded MTP chain declares {} heads but blk.{n} has no \
4423 nextn.eh_proj.weight",
4424 cfg.nextn_predict_layers
4425 )
4426 .into());
4427 }
4428 embedded_mtp.push(MtpHead {
4429 enorm: load_t(e, src, &p("nextn.enorm.weight"))?,
4430 hnorm: load_t(e, src, &p("nextn.hnorm.weight"))?,
4431 eh_proj: load_t(e, src, &p("nextn.eh_proj.weight"))?,
4432 attn_norm: load_t(e, src, &p("attn_norm.weight"))?,
4433 post_attn_norm: load_opt(e, src, &p("post_attention_norm.weight"))?
4434 .or(load_opt(e, src, &p("ffn_norm.weight"))?)
4435 .expect("MTP block needs post_attention_norm or ffn_norm"),
4436 mixer: load_mixer_kind(
4437 e,
4438 src,
4439 &cfg,
4440 n,
4441 &mtp_plan.layer.attention,
4442 &mut step_runtimes,
4443 )?,
4444 ffn: load_ffn(
4445 e,
4446 src,
4447 &cfg,
4448 &mtp_plan.layer.mlp,
4449 n,
4450 spill.as_mut().map(|c| (gguf.unwrap(), c)),
4451 &mut resident,
4452 &mut step_runtimes,
4453 )?,
4454 shared_head_norm: load_opt(e, src, &p("nextn.shared_head_norm.weight"))?,
4455 shared_head_head: load_mtp_head_maybe_nvfp4(
4464 e,
4465 src,
4466 &p("nextn.shared_head_head.weight"),
4467 )?
4468 .or(load_opt(e, src, &p("nextn.shared_head.weight"))?),
4469 d2t: None,
4470 d2t_from_target_head: false,
4471 geom: None,
4472 step35: if sliding_gated_moe_program {
4473 Some(Step35MtpGeom::from_plan(&mtp_plan.layer)?)
4474 } else {
4475 None
4476 },
4477 });
4478 }
4479 }
4480 let mut embedded_mtp = embedded_mtp.into_iter();
4481 let mut mtp = embedded_mtp.next();
4482 let mut mtp_extra: Vec<MtpHead> = embedded_mtp.collect();
4483
4484 mtp = if load_mtp {
4488 match std::env::var("MEMRA_MTP_DRAFT") {
4489 Ok(path) if !path.is_empty() => {
4490 eprintln!("[mtp-draft] loading external MTP draft: {path}");
4491 let dg = GgufFile::open(&path)?;
4492 mtp_extra.clear();
4493 Some(MtpHead::load_draft(e, &dg, &cfg)?)
4494 }
4495 _ => mtp,
4496 }
4497 } else {
4498 None
4499 };
4500
4501 let trim_env = if load_mtp {
4512 std::env::var("MEMRA_FRSPEC_TRIM")
4513 } else {
4514 Err(std::env::VarError::NotPresent)
4515 };
4516 if crate::model::full_prec_enabled()
4517 && trim_env.as_deref().map(|p| !p.is_empty()).unwrap_or(false)
4518 {
4519 eprintln!(
4520 "[frspec-trim] DISABLED under MEMRA_FULL_PREC — using the natural full MTP head"
4521 );
4522 }
4523 mtp = match (
4524 if crate::model::full_prec_enabled() {
4525 Err(std::env::VarError::NotPresent)
4526 } else {
4527 trim_env
4528 },
4529 mtp,
4530 ) {
4531 (Ok(path), Some(mut head)) if !path.is_empty() => {
4532 let e = crate::pp::layer_engine(e, n_trunk, n_trunk)?;
4536 let path = memra_gguf::hf::resolve_arg(&path)
4540 .map_err(|err| format!("MEMRA_FRSPEC_TRIM={path:?}: {err}"))?;
4541 let d2t: Vec<u32> = frspec_read_d2t(&path)?;
4545 let own_head_name = frspec_trim_own_head_name(n_trunk);
4554 let own_head = src.find(&own_head_name);
4555 let from_own_head = own_head.is_some();
4556 let v = own_head
4557 .or_else(|| src.find("output.weight"))
4558 .or_else(|| src.find("token_embd.weight"))
4559 .expect("model has no output.weight for FR-Spec trim");
4560 let (trimmed, nvfp4_sizes) = frspec_gather_trimmed_head(
4580 e,
4581 &v,
4582 &d2t,
4583 std::env::var("MEMRA_FRSPEC_TRIM_NVFP4").as_deref() == Ok("1"),
4584 match src.find("output.scale") {
4586 Some(sv) => f32::from_le_bytes(sv.bytes[..4].try_into().unwrap()),
4587 None => 1.0,
4588 },
4589 )?;
4590 match nvfp4_sizes {
4591 Some((nvfp4_bytes, gathered_bytes)) => eprintln!(
4592 "[frspec-trim] self-trimmed head: {} rows of {} re-quantized BF16 -> NVFP4 \
4593 ({} MiB, was {} MiB)",
4594 d2t.len(),
4595 if from_own_head {
4596 own_head_name.as_str()
4597 } else {
4598 "main output.weight"
4599 },
4600 nvfp4_bytes >> 20,
4601 gathered_bytes >> 20,
4602 ),
4603 None => eprintln!(
4604 "[frspec-trim] self-trimmed head: {} rows of {} ({:?})",
4605 d2t.len(),
4606 if from_own_head {
4607 own_head_name.as_str()
4608 } else {
4609 "main output.weight"
4610 },
4611 v.ggml_type
4612 ),
4613 }
4614 head.shared_head_head = Some(trimmed);
4615 head.d2t = Some(d2t);
4616 head.d2t_from_target_head = !from_own_head;
4619 Some(head)
4620 }
4621 (_, m) => m,
4622 };
4623 let dflash_trim: Option<DflashTrimHead> = match mtp_skip_trim_d2t {
4634 Some(d2t) => {
4635 let v = src
4636 .find("output.weight")
4637 .or_else(|| src.find("token_embd.weight"))
4638 .ok_or("model has no output.weight for FR-Spec trim")?;
4639 let (head, nvfp4_sizes) = frspec_gather_trimmed_head(
4640 e,
4641 &v,
4642 &d2t,
4643 std::env::var("MEMRA_FRSPEC_TRIM_NVFP4").as_deref() == Ok("1"),
4644 match src.find("output.scale") {
4645 Some(sv) => f32::from_le_bytes(sv.bytes[..4].try_into().unwrap()),
4646 None => 1.0,
4647 },
4648 )?;
4649 eprintln!(
4650 "[mtp-skip] FR-Spec stub draft head built: {} rows of main output.weight \
4651 ({}); DFlash2 trim serves without the embedded MTP block",
4652 d2t.len(),
4653 match nvfp4_sizes {
4654 Some((nvfp4_bytes, gathered_bytes)) => format!(
4655 "re-quantized BF16 -> NVFP4, {} MiB, was {} MiB",
4656 nvfp4_bytes >> 20,
4657 gathered_bytes >> 20
4658 ),
4659 None => format!("{:?}", v.ggml_type),
4660 },
4661 );
4662 Some(DflashTrimHead { head, d2t })
4663 }
4664 None => None,
4665 };
4666 if let Some(d2t) = mtp.as_ref().and_then(|head| head.d2t.clone()) {
4679 let want_nvfp4_env = std::env::var("MEMRA_FRSPEC_TRIM_NVFP4").as_deref() == Ok("1");
4680 let mut kept = 0usize;
4681 let e = crate::pp::layer_engine(e, n_trunk, n_trunk)?;
4684 for (i, head) in mtp_extra.iter_mut().enumerate() {
4685 let name = frspec_trim_own_head_name(n_trunk + 1 + i);
4686 let Some(v) = src.find(&name) else { break };
4687 let out_f = v.ne[1] as usize;
4688 let row_bytes = v.bytes.len() / out_f;
4689 if d2t.iter().any(|&t| (t as usize) >= out_f) {
4690 break;
4691 }
4692 let mut gathered = Vec::with_capacity(d2t.len() * row_bytes);
4693 for &t in &d2t {
4694 let off = t as usize * row_bytes;
4695 gathered.extend_from_slice(&v.bytes[off..off + row_bytes]);
4696 }
4697 let want_nvfp4 =
4698 want_nvfp4_env && matches!(v.ggml_type, GgmlType::BF16) && v.ne[0] % 64 == 0;
4699 let trimmed = if want_nvfp4 {
4700 let vals: Vec<f32> = gathered
4701 .chunks_exact(2)
4702 .map(|c| f32::from_bits((u16::from_le_bytes([c[0], c[1]]) as u32) << 16))
4703 .collect();
4704 let blocks = memra_gguf::nvfp4_repack::f32_to_nvfp4(&vals);
4705 GpuTensor::from_quant_bytes(
4706 e,
4707 &blocks,
4708 GgmlType::NVFP4,
4709 v.ne[0],
4710 d2t.len() as u64,
4711 1.0,
4712 )?
4713 } else {
4714 match v.ggml_type {
4715 GgmlType::BF16 => GpuTensor::FloatBf16 {
4716 data: e.htod_bytes(&gathered)?,
4717 ne: vec![v.ne[0], d2t.len() as u64],
4718 },
4719 GgmlType::F32 => GpuTensor::Float {
4720 data: e.htod(
4721 &gathered
4722 .chunks_exact(4)
4723 .map(|c| f32::from_le_bytes(c.try_into().unwrap()))
4724 .collect::<Vec<f32>>(),
4725 )?,
4726 ne: vec![v.ne[0], d2t.len() as u64],
4727 },
4728 _ => GpuTensor::from_quant_bytes(
4729 e,
4730 &gathered,
4731 v.ggml_type,
4732 v.ne[0],
4733 d2t.len() as u64,
4734 1.0,
4735 )?,
4736 }
4737 };
4738 head.shared_head_head = Some(trimmed);
4739 head.d2t = Some(d2t.clone());
4740 head.d2t_from_target_head = false;
4741 kept += 1;
4742 }
4743 let dropped = mtp_extra.len() - kept;
4744 mtp_extra.truncate(kept);
4745 eprintln!(
4746 "[frspec-trim] per-head trim: {kept} extra chain head(s) gathered from their own \
4747 blocks{}",
4748 if dropped > 0 {
4749 format!(" ({dropped} dropped: no own-head tensor)")
4750 } else {
4751 String::new()
4752 }
4753 );
4754 }
4755 if !mtp_extra.is_empty() {
4756 if plan.draft_source != memra_gguf::model_plan::DraftSourcePlan::Embedded
4757 || plan.mtp_blocks.len() != 1 + mtp_extra.len()
4758 || plan
4759 .mtp_blocks
4760 .iter()
4761 .any(|block| !matches!(block.layer.mlp, MlpPlan::Dense(_)))
4762 || mtp
4763 .iter()
4764 .chain(mtp_extra.iter())
4765 .any(|head| !matches!(head.ffn, Ffn::Dense { .. }))
4766 {
4767 return Err(
4768 "multi-head MTP requires embedded dense canonical blocks and matching loaded heads"
4769 .into(),
4770 );
4771 }
4772 eprintln!(
4773 "[mtp-draft] embedded chain: heads={} blocks={}..={} scratch=per-head",
4774 1 + mtp_extra.len(),
4775 n_trunk,
4776 n_trunk + mtp_extra.len()
4777 );
4778 }
4779
4780 let glm5_dflash = match std::env::var("MEMRA_GLM5_DFLASH") {
4796 Ok(spec) if !spec.is_empty() && cfg.arch.is_glm5_next() => {
4797 let dpath = memra_gguf::hf::resolve_arg(&spec)
4798 .map_err(|err| format!("MEMRA_GLM5_DFLASH={spec:?}: {err}"))?;
4799 let de = crate::pp::layer_engine(e, n_trunk, n_trunk)?;
4800 Some(crate::dflash::load_drafter(
4801 de,
4802 std::path::Path::new(&dpath),
4803 "MEMRA_GLM5_DFLASH",
4804 n_trunk,
4805 cfg.n_embd as usize,
4806 output.out_features(),
4807 )?)
4808 }
4809 _ => None,
4810 };
4811
4812 if cfg.arch.is_glm5_next() && crate::glm_spec::glm5_spec_on() {
4820 match (glm5_dflash.as_ref(), mtp.as_ref()) {
4821 (Some(dr), head) => {
4822 let trim_note = match head.and_then(|h| h.d2t.as_ref()) {
4823 Some(map) => {
4824 format!("draft head TRIMMED to {} rows (FR-Spec d2t)", map.len())
4825 }
4826 None => "draft head FULL target vocab".to_string(),
4827 };
4828 eprintln!(
4829 "[glm5-spec] serve route ARMED: draft source = dflash2 @ {}; {trim_note}; \
4830 native MTP head {}",
4831 dr.sha8,
4832 if head.is_some() {
4833 "ALSO loaded (idle for drafting — dflash2 wins by selection)"
4834 } else {
4835 "NOT loaded (the q38 pattern: a full MoE trunk layer of VRAM saved)"
4836 }
4837 );
4838 }
4839 (None, Some(head)) => {
4840 match head.d2t.as_ref() {
4841 Some(map) => eprintln!(
4842 "[glm5-spec] serve route ARMED: MTP head loaded; draft head TRIMMED \
4843 to {} rows (FR-Spec d2t engaged)",
4844 map.len()
4845 ),
4846 None => eprintln!(
4847 "[glm5-spec] serve route ARMED: MTP head loaded; draft head FULL \
4848 target vocab (no FR-Spec trim)"
4849 ),
4850 }
4851 eprintln!("[glm5-spec] draft source = native-mtp");
4852 }
4853 (None, None) => eprintln!(
4854 "[glm5-spec] MEMRA_GLM5_SPEC=1 but no MTP head loaded \
4855 (set MEMRA_GLM5_MTP=1 or MEMRA_GLM5_DFLASH=<drafter>) — route stays \
4856 fail-closed, plain serving"
4857 ),
4858 }
4859 }
4860
4861 if let Some(ctx) = spill.as_ref() {
4862 eprintln!(
4863 "[spill] experts placed: {} pinned (Tier 1), {} mmap'd from disk (Tier 2, {} MiB)",
4864 ctx.n_pinned,
4865 ctx.n_mmap,
4866 ctx.mmap_bytes >> 20
4867 );
4868 }
4869
4870 if cfg.n_head_kv > 0 && cfg.n_head / cfg.n_head_kv > 8 {
4884 crate::FA_V4_MAX_DEFAULT.store(0, std::sync::atomic::Ordering::Relaxed);
4885 eprintln!(
4886 "[fa] v4 decode family disabled: gqa {} > fa_v4_smem capacity 8 (v3 lane serves)",
4887 cfg.n_head / cfg.n_head_kv
4888 );
4889 }
4890
4891 if gemma_program {
4892 crate::FA_VEC_MIN_DEFAULT.store(1, std::sync::atomic::Ordering::Relaxed);
4894 let real_moe = plan
4897 .trunk_operations()
4898 .contains(&memra_gguf::model_plan::OperationKind::MoeMlp);
4899 crate::FA_SPW_DEFAULT.store(
4900 if real_moe { 32 } else { 64 },
4901 std::sync::atomic::Ordering::Relaxed,
4902 );
4903 crate::FA_SP512_DEFAULT.store(
4905 if real_moe { 16 } else { 32 },
4906 std::sync::atomic::Ordering::Relaxed,
4907 );
4908 crate::FUSED_MR1_DEFAULT.store(!real_moe, std::sync::atomic::Ordering::Relaxed);
4918 crate::RMS_BLOCK_DEFAULT.store(1024, std::sync::atomic::Ordering::Relaxed);
4920 crate::FA_SP_GEMMA.store(true, std::sync::atomic::Ordering::Relaxed);
4922 }
4926 let force_embd_gpu = gemma_program;
4929 let gemma4_aux = if gemma_program {
4930 let rope_freqs = match src.find("rope_freqs.weight") {
4931 Some(t) => {
4932 let host = memra_gguf::dequant::dequantize(
4933 t.ggml_type,
4934 &t.bytes,
4935 t.ne.iter().product::<u64>() as usize,
4936 );
4937 let mut copies = Vec::new();
4938 if let Some(fence) = crate::pp::pp_cuts(n_trunk) {
4939 #[allow(clippy::needless_range_loop)]
4940 for s in 0..fence.len() - 1 {
4942 let owner = crate::pp::layer_engine(e, n_trunk, fence[s])?;
4943 let dev = owner.ctx().ordinal();
4944 if copies.iter().all(|(d, _)| *d != dev) {
4945 copies.push((dev, owner.htod(&host)?));
4946 }
4947 }
4948 } else {
4949 copies.push((e.ctx().ordinal(), e.htod(&host)?));
4950 }
4951 Some(copies)
4952 }
4953 None => {
4961 let g4 = cfg.gemma4.as_ref().unwrap();
4962 let n = (g4.rope_dims_global / 2) as usize;
4963 let keep =
4964 ((n as f32) * g4.partial_rotary_global.clamp(0.0, 1.0)).round() as usize;
4965 let host: Vec<f32> = (0..n)
4966 .map(|i| if i < keep { 1.0 } else { 1.0e30 })
4967 .collect();
4968 eprintln!(
4969 "[gemma4] rope_freqs.weight synthesized ({n} factors, first {keep} \
4970 rotate; source ships none — native checkpoint)"
4971 );
4972 let mut copies = Vec::new();
4973 if let Some(fence) = crate::pp::pp_cuts(n_trunk) {
4974 #[allow(clippy::needless_range_loop)]
4975 for s in 0..fence.len() - 1 {
4977 let owner = crate::pp::layer_engine(e, n_trunk, fence[s])?;
4978 let dev = owner.ctx().ordinal();
4979 if copies.iter().all(|(d, _)| *d != dev) {
4980 copies.push((dev, owner.htod(&host)?));
4981 }
4982 }
4983 } else {
4984 copies.push((e.ctx().ordinal(), e.htod(&host)?));
4985 }
4986 Some(copies)
4987 }
4988 };
4989 let e4b = match src.find("per_layer_token_embd.weight") {
4991 Some(t) => {
4992 let n_epl = cfg
4993 .gemma4
4994 .as_ref()
4995 .map(|g| g.n_embd_per_layer as usize)
4996 .unwrap_or(0);
4997 let row = t.ne[0] as usize; let row_bytes = t.bytes.len() / (t.ne[1] as usize);
4999 eprintln!(
5000 "[gemma4-e4b] per-layer-embed model detected (n_epl={n_epl}, row {row}) — \
5001 first-light forward (eager decode + prime); dc/graph/spec unwired \
5002 (HANDOVER-E4B.md)"
5003 );
5004 Some(crate::hybrid::Gemma4E4bModel {
5005 tok_tbl_gpu: std::sync::OnceLock::new(),
5006 tok_embd_bytes: t.bytes.to_vec(),
5007 tok_embd_qt: match t.ggml_type {
5008 memra_gguf::GgmlType::Q6_K => crate::QT_Q6_K,
5009 memra_gguf::GgmlType::Q8_0 => crate::QT_Q8_0,
5010 other => panic!("e4b per-layer tok embd: unhandled dtype {other:?}"),
5011 },
5012 tok_embd_row_bytes: row_bytes,
5013 model_proj: load_t(e, src, "per_layer_model_proj.weight")?,
5014 proj_norm: load_t(e, src, "per_layer_proj_norm.weight")?,
5015 n_epl,
5016 })
5017 }
5018 None => None,
5019 };
5020 let suppress_d = {
5021 let sup = &cfg.gemma4.as_ref().unwrap().suppress_tokens;
5022 if sup.is_empty() {
5023 None
5024 } else {
5025 let ids: Vec<i32> = sup.iter().map(|&x| x as i32).collect();
5026 eprintln!(
5027 "[gemma4] suppress_tokens: {} ids masked at sampling",
5028 ids.len()
5029 );
5030 Some((e.htod_i32(&ids)?, ids.len()))
5031 }
5032 };
5033 let ones_host = [1.0f32; 512];
5034 let mut ones = Vec::new();
5035 if let Some(fence) = crate::pp::pp_cuts(n_trunk) {
5036 #[allow(clippy::needless_range_loop)]
5037 for s in 0..fence.len() - 1 {
5039 let owner = crate::pp::layer_engine(e, n_trunk, fence[s])?;
5040 let dev = owner.ctx().ordinal();
5041 if ones.iter().all(|(d, _)| *d != dev) {
5042 ones.push((dev, owner.htod(&ones_host)?));
5043 }
5044 }
5045 } else {
5046 ones.push((e.ctx().ordinal(), e.htod(&ones_host)?));
5047 }
5048 Some(GemmaAux {
5049 rope_freqs,
5050 ones,
5051 suppress_d,
5052 e4b,
5053 })
5054 } else {
5055 None
5056 };
5057 let step35_aux = if sliding_gated_moe_program {
5061 let rope_freqs = match src.find("rope_freqs.weight") {
5062 Some(t) => {
5063 let host = memra_gguf::dequant::dequantize(
5064 t.ggml_type,
5065 &t.bytes,
5066 t.ne.iter().product::<u64>() as usize,
5067 );
5068 let mut copies = Vec::new();
5069 if let Some(fence) = crate::pp::pp_cuts(n_trunk) {
5070 #[allow(clippy::needless_range_loop)]
5071 for s in 0..fence.len() - 1 {
5073 let owner = crate::pp::layer_engine(e, n_trunk, fence[s])?;
5074 let dev = owner.ctx().ordinal();
5075 if copies.iter().all(|(d, _)| *d != dev) {
5076 copies.push((dev, owner.htod(&host)?));
5077 }
5078 }
5079 } else {
5080 copies.push((e.ctx().ordinal(), e.htod(&host)?));
5081 }
5082 Some(copies)
5083 }
5084 None => None,
5085 };
5086 Some(Step35Aux { rope_freqs })
5087 } else {
5088 None
5089 };
5090 let mut layers = layers;
5091 {
5098 let q8rp_on = match std::env::var("MEMRA_Q8RP").as_deref() {
5099 Ok("0") => false,
5100 Ok(_) => true,
5101 Err(_) => {
5109 cfg!(memra_hopper_mma) || {
5110 let q8b = |w: &crate::model::GpuTensor| -> usize {
5111 match w {
5112 crate::model::GpuTensor::Quant {
5113 bytes,
5114 qtype,
5115 row_bytes,
5116 ne,
5117 rp4: None,
5118 ..
5119 } if *qtype == crate::QT_Q8_0
5120 && ne.len() == 2
5121 && (ne[0] as usize).is_multiple_of(32)
5122 && *row_bytes == (ne[0] as usize / 32) * 34 =>
5123 {
5124 bytes.len()
5125 }
5126 _ => 0,
5127 }
5128 };
5129 let mut need = q8b(&output);
5130 for layer in layers.iter() {
5131 match &layer.mixer {
5132 Mixer::Full(fa) => {
5133 for w in [&fa.wq, &fa.wk, &fa.wv, &fa.wo] {
5134 need += q8b(w);
5135 }
5136 }
5137 Mixer::Linear(la) => {
5138 for w in [
5139 &la.wqkv,
5140 &la.wqkv_gate,
5141 &la.ssm_beta,
5142 &la.ssm_alpha,
5143 &la.ssm_out,
5144 ] {
5145 need += q8b(w);
5146 }
5147 }
5148 Mixer::Mla(_) => {}
5149 Mixer::Kda(_) => {} }
5151 if let Ffn::Dense {
5152 ffn_gate,
5153 ffn_up,
5154 ffn_down,
5155 } = &layer.ffn
5156 {
5157 for w in [ffn_gate, ffn_up, ffn_down] {
5158 need += q8b(w);
5159 }
5160 }
5161 }
5162 need > 0
5163 && e.ctx()
5164 .mem_get_info()
5165 .map(|(free, _)| free >= need + (8usize << 30))
5166 .unwrap_or(false)
5167 }
5168 }
5169 };
5170 let kqrp_on = crate::Engine::kqrp_enabled() || {
5180 std::env::var("MEMRA_KQRP").is_err() && {
5181 let kqb = |w: &crate::model::GpuTensor| -> usize {
5182 match w {
5183 crate::model::GpuTensor::Quant {
5184 bytes,
5185 qtype,
5186 row_bytes,
5187 ne,
5188 rp4: None,
5189 ..
5190 } if ne.len() == 2 && (ne[0] as usize).is_multiple_of(256) => {
5191 let sb = if *qtype == crate::QT_Q4_K {
5192 144
5193 } else if *qtype == crate::QT_Q6_K {
5194 210
5195 } else {
5196 return 0;
5197 };
5198 if *row_bytes == (ne[0] as usize / 256) * sb {
5199 bytes.len()
5200 } else {
5201 0
5202 }
5203 }
5204 _ => 0,
5205 }
5206 };
5207 let mut need = kqb(&output);
5208 for layer in layers.iter() {
5209 if let Mixer::Full(fa) = &layer.mixer {
5210 for w in [&fa.wq, &fa.wk, &fa.wv, &fa.wo] {
5211 need += kqb(w);
5212 }
5213 }
5214 if let Ffn::Dense {
5215 ffn_gate,
5216 ffn_up,
5217 ffn_down,
5218 } = &layer.ffn
5219 {
5220 for w in [ffn_gate, ffn_up, ffn_down] {
5221 need += kqb(w);
5222 }
5223 }
5224 }
5225 need > 0
5226 && e.ctx()
5227 .mem_get_info()
5228 .map(|(free, _)| free >= need + (8usize << 30))
5229 .unwrap_or(false)
5230 }
5231 };
5232 if q8rp_on || kqrp_on {
5233 let f16_model_ok = gemma_program
5240 || plan
5241 .trunk_operations()
5242 .contains(&memra_gguf::model_plan::OperationKind::MoeMlp)
5243 || std::env::var("MEMRA_PP_F16").as_deref() == Ok("1");
5244 let mut nmir = 0usize;
5245 let mut mir = |e_ref: &crate::Engine,
5249 w: &mut crate::model::GpuTensor|
5250 -> Result<(), Box<dyn std::error::Error>> {
5251 let before = matches!(w, crate::model::GpuTensor::Quant { rp4: Some(_), .. });
5252 if q8rp_on {
5253 e_ref.build_q8_rp4(w)?;
5254 }
5255 if kqrp_on {
5256 e_ref.build_q4k_rp4(w)?;
5257 e_ref.build_q6k_rp4(w)?;
5258 }
5259 let q6k = matches!(w, crate::model::GpuTensor::Quant { qtype, .. }
5264 if *qtype == crate::QT_Q6_K);
5265 if q8rp_on && crate::f16_ffi::pp_f16_enabled() && (f16_model_ok || q6k) {
5266 e_ref.build_q8_f16(w)?;
5267 }
5268 if !before && matches!(w, crate::model::GpuTensor::Quant { rp4: Some(_), .. }) {
5269 nmir += 1;
5270 }
5271 Ok(())
5272 };
5273 for (il, layer) in layers.iter_mut().enumerate() {
5274 let el = crate::pp::layer_engine(e, n_trunk, il)?;
5275 match &mut layer.mixer {
5276 Mixer::Full(fa) => {
5277 for w in [&mut fa.wq, &mut fa.wk, &mut fa.wv, &mut fa.wo] {
5278 mir(el, w)?;
5279 }
5280 }
5281 Mixer::Linear(la) => {
5282 for w in [
5283 &mut la.wqkv,
5284 &mut la.wqkv_gate,
5285 &mut la.ssm_beta,
5286 &mut la.ssm_alpha,
5287 &mut la.ssm_out,
5288 ] {
5289 mir(el, w)?;
5290 }
5291 }
5292 Mixer::Mla(_) => {}
5295 Mixer::Kda(_) => {} }
5297 if let Ffn::Dense {
5298 ffn_gate,
5299 ffn_up,
5300 ffn_down,
5301 } = &mut layer.ffn
5302 {
5303 for w in [ffn_gate, ffn_up, ffn_down] {
5304 mir(el, w)?;
5305 }
5306 }
5307 }
5308 mir(e_head, &mut output)?;
5309 if nmir > 0 {
5310 eprintln!("[q8rp] split-plane decode mirrors built: {nmir} tensors");
5311 }
5312 if q8rp_on && crate::f16_ffi::pp_f16_enabled() {
5327 for (want, tag) in [(crate::QT_Q4_K, "q4kf16"), (crate::QT_Q5_K, "q5kf16")] {
5328 let (mut n4, mut b4) = (0usize, 0usize);
5329 let mut mirk =
5330 |e_ref: &crate::Engine,
5331 w: &mut crate::model::GpuTensor|
5332 -> Result<(), Box<dyn std::error::Error>> {
5333 if matches!(w, crate::model::GpuTensor::Quant { qtype, f16: None, .. }
5334 if *qtype == want)
5335 {
5336 e_ref.build_q8_f16(w)?;
5337 if let crate::model::GpuTensor::Quant { f16: Some(m), .. } = w {
5338 n4 += 1;
5339 b4 += m.len();
5340 }
5341 }
5342 Ok(())
5343 };
5344 for (il, layer) in layers.iter_mut().enumerate() {
5345 let el = crate::pp::layer_engine(e, n_trunk, il)?;
5346 match &mut layer.mixer {
5347 Mixer::Full(fa) => {
5348 for w in [&mut fa.wq, &mut fa.wk, &mut fa.wv, &mut fa.wo] {
5349 mirk(el, w)?;
5350 }
5351 }
5352 Mixer::Linear(la) => {
5353 for w in [
5354 &mut la.wqkv,
5355 &mut la.wqkv_gate,
5356 &mut la.ssm_beta,
5357 &mut la.ssm_alpha,
5358 &mut la.ssm_out,
5359 ] {
5360 mirk(el, w)?;
5361 }
5362 }
5363 Mixer::Mla(_) => {} Mixer::Kda(_) => {} }
5366 if let Ffn::Dense {
5367 ffn_gate,
5368 ffn_up,
5369 ffn_down,
5370 } = &mut layer.ffn
5371 {
5372 for w in [ffn_gate, ffn_up, ffn_down] {
5373 mirk(el, w)?;
5374 }
5375 }
5376 }
5377 mirk(e_head, &mut output)?;
5378 if n4 > 0 {
5379 eprintln!(
5380 "[{tag}] prefill fp16 mirrors built: {n4} tensors \
5381 ({} MB)",
5382 b4 >> 20
5383 );
5384 }
5385 }
5386 }
5387 }
5388 }
5389 if gemma_program && crate::Engine::q4rp_enabled() {
5396 let mut nmir = 0usize;
5397 for (il, layer) in layers.iter_mut().enumerate() {
5398 let e = crate::pp::layer_engine(e, n_trunk, il)?;
5400 let is_moe26 = layer.gemma4.as_ref().is_some_and(|g| g.moe_bits.is_some());
5409 let is_e4b = layer.gemma4.as_ref().is_some_and(|g| g.e4b.is_some());
5410 if !(is_moe26 || is_e4b) {
5411 continue;
5412 }
5413 if let Mixer::Full(fa) = &mut layer.mixer {
5414 for w in [&mut fa.wq, &mut fa.wk, &mut fa.wv, &mut fa.wo] {
5415 e.build_q4_rp4(w)?;
5416 nmir += 1;
5417 }
5418 }
5419 if is_e4b {
5420 let own_kv = layer
5422 .gemma4
5423 .as_ref()
5424 .unwrap()
5425 .e4b
5426 .as_ref()
5427 .is_some_and(|e4| e4.kv_share.is_none());
5428 if own_kv
5429 && let Mixer::Full(fa) = &layer.mixer
5430 && let Some(mut cat) = e.build_q4_out_concat3(&fa.wq, &fa.wk, &fa.wv)?
5431 {
5432 e.build_q4_rp4(&mut cat)?;
5433 nmir += 1;
5434 layer.gemma4.as_mut().unwrap().e4b.as_mut().unwrap().qkv_cat = Some(cat);
5435 }
5436 if let Ffn::Dense {
5437 ffn_gate,
5438 ffn_up,
5439 ffn_down,
5440 } = &mut layer.ffn
5441 {
5442 for w in [ffn_gate, ffn_up, ffn_down] {
5443 e.build_q4_rp4(w)?;
5444 nmir += 1;
5445 }
5446 }
5447 let e4 = layer.gemma4.as_mut().unwrap().e4b.as_mut().unwrap();
5448 for w in [&mut e4.inp_gate, &mut e4.proj] {
5449 e.build_q4_rp4(w)?;
5450 nmir += 1;
5451 }
5452 }
5453 if let Some(mb) = layer.gemma4.as_mut().unwrap().moe_bits.as_mut() {
5454 for w in [&mut mb.shared_gate, &mut mb.shared_up, &mut mb.shared_down] {
5455 e.build_q4_rp4(w)?;
5456 nmir += 1;
5457 }
5458 }
5459 }
5460 if nmir > 0 {
5461 eprintln!("[q4rp] split-plane decode mirrors built: {nmir} trunk tensors");
5462 }
5463 let fast_on = std::env::var("MEMRA_FAST").as_deref() != Ok("0");
5470 if fast_on {
5471 let mut nswap = 0usize;
5472 let mut nf16 = 0usize;
5473 let q4f16_model_ok = matches!(cfg.n_embd, 3840 | 5376); if let Ok(v) = std::env::var("MEMRA_Q4F16")
5492 && v != "0"
5493 && v != "1"
5494 {
5495 return Err(format!(
5496 "MEMRA_Q4F16={v} is not 0 or 1 — this env selects the prefill \
5497 ARITHMETIC (fp16 mirrors vs int8 MMQ) and must never be guessed"
5498 )
5499 .into());
5500 }
5501 let f16_need = {
5502 let f16b = |w: &crate::model::GpuTensor| -> usize {
5503 match w {
5504 crate::model::GpuTensor::Quant {
5505 qtype,
5506 ne,
5507 f16: None,
5508 ..
5509 } if ne.len() == 2
5510 && matches!(
5511 *qtype,
5512 crate::QT_Q8_0
5513 | crate::QT_Q4_0
5514 | crate::QT_Q6_K
5515 | crate::QT_Q4_K
5516 | crate::QT_Q5_K
5517 ) =>
5518 {
5519 (ne[0] as usize) * (ne[1] as usize) * 2
5520 }
5521 _ => 0,
5522 }
5523 };
5524 let mut need = 0usize;
5525 for layer in layers.iter() {
5526 if layer.gemma4.as_ref().is_none_or(|g| g.moe_bits.is_some()) {
5527 continue;
5528 }
5529 if let Mixer::Full(fa) = &layer.mixer {
5530 for w in [&fa.wq, &fa.wk, &fa.wv, &fa.wo] {
5531 need += f16b(w);
5532 }
5533 }
5534 if let Ffn::Dense {
5535 ffn_gate,
5536 ffn_up,
5537 ffn_down,
5538 } = &layer.ffn
5539 {
5540 for w in [ffn_gate, ffn_up, ffn_down] {
5541 need += f16b(w);
5542 }
5543 }
5544 }
5545 need
5546 };
5547 let f16_free = e.ctx().mem_get_info().map(|(free, _)| free).unwrap_or(0);
5548 let f16_auto = q4f16_model_ok
5549 && std::env::var("MEMRA_Q4F16").is_err()
5550 && crate::f16_ffi::pp_f16_capacity_ok(f16_free, f16_need);
5551 let (f16_on, f16_why) = match std::env::var("MEMRA_Q4F16").as_deref() {
5557 Ok("1") => (true, "env MEMRA_Q4F16=1"),
5558 Ok("0") => (false, "env MEMRA_Q4F16=0"),
5559 _ if crate::f16_ffi::pp_f16_enabled() && q4f16_model_ok => {
5560 (true, "env MEMRA_PP_F16")
5561 }
5562 _ if f16_auto => (true, "capacity-keyed auto (UNPINNED)"),
5563 _ if !q4f16_model_ok => (false, "model geometry not eligible"),
5564 _ => (false, "capacity-keyed auto REFUSED (UNPINNED)"),
5565 };
5566 eprintln!(
5573 "[q4f16] prefill program = {} (reason: {}); free {} MiB, mirror mass {} MiB, \
5574 capacity threshold {} MiB (mass + 8192 headroom) — SELECTS PREFILL ARITHMETIC",
5575 if f16_on {
5576 "FP16 MIRRORS"
5577 } else {
5578 "INT8 MMQ (no f16 mirrors)"
5579 },
5580 f16_why,
5581 f16_free >> 20,
5582 f16_need >> 20,
5583 (f16_need + (8usize << 30)) >> 20,
5584 );
5585 for (il, layer) in layers.iter_mut().enumerate() {
5586 let e = crate::pp::layer_engine(e, n_trunk, il)?;
5588 let dense_gemma = layer.gemma4.as_ref().is_some_and(|g| g.moe_bits.is_none());
5589 if !dense_gemma {
5590 continue;
5591 }
5592 if let Mixer::Full(fa) = &mut layer.mixer {
5593 for w in [&mut fa.wq, &mut fa.wk, &mut fa.wv, &mut fa.wo] {
5594 if f16_on {
5595 e.build_q8_f16(w)?;
5596 if matches!(w, crate::model::GpuTensor::Quant { f16: Some(_), .. })
5597 {
5598 nf16 += 1;
5599 }
5600 }
5601 if e.build_q4_rp_swap(w)? {
5602 nswap += 1;
5603 }
5604 }
5605 }
5606 if let Ffn::Dense {
5607 ffn_gate,
5608 ffn_up,
5609 ffn_down,
5610 } = &mut layer.ffn
5611 {
5612 for w in [ffn_gate, ffn_up, ffn_down] {
5613 if f16_on {
5614 e.build_q8_f16(w)?;
5615 if matches!(w, crate::model::GpuTensor::Quant { f16: Some(_), .. })
5616 {
5617 nf16 += 1;
5618 }
5619 }
5620 if e.build_q4_rp_swap(w)? {
5621 nswap += 1;
5622 }
5623 }
5624 }
5625 }
5626 if nswap > 0 {
5627 eprintln!("[q4rp] split-plane IN-PLACE swap: {nswap} dense trunk tensors");
5628 }
5629 if nf16 > 0 {
5630 eprintln!("[q4f16] prefill fp16 mirrors built: {nf16} dense trunk tensors");
5631 }
5632 }
5633 }
5634 let model = HybridModel {
5635 cfg,
5636 plan,
5637 rewrite_qualifications: None,
5638 embd,
5639 output_norm,
5640 output,
5641 layers,
5642 mtp,
5643 mtp_extra,
5644 dflash_trim,
5645 embd_gpu: std::sync::OnceLock::new(),
5646 gemma4_aux,
5647 step35_aux,
5648 prime_slabs: std::sync::Mutex::new(std::collections::HashMap::new()),
5649 dspark_vgraphs: std::sync::Mutex::new(None),
5650 step_grouped_prefill: std::sync::Mutex::new(StepEpGroupedPrefill::default()),
5651 step35_token_graph: std::sync::Mutex::new(None),
5652 hyper,
5653 hyper_head,
5654 glm5_dflash,
5655 draft_state_bytes: std::sync::atomic::AtomicUsize::new(0),
5656 };
5657 e.configure_moe_cache_layout(model.moe_cache_block_sizes());
5658 if force_embd_gpu {
5659 let _ = model
5660 .embd_gpu
5661 .get_or_init(|| e.upload_u8(&model.embd.raw).expect("embed table upload"));
5662 }
5663 crate::pp::sync_stages_after_load(e, n_trunk)?;
5669 Ok(model)
5670 }
5671
5672 pub fn ensure_embed_resident(&self, e: &Engine) -> Result<(), Box<dyn std::error::Error>> {
5682 if std::env::var("MEMRA_EMBED_DEV").as_deref() == Ok("0") {
5683 return Ok(());
5684 }
5685 if self.embd_gpu.get().is_none() {
5686 let buf = e.upload_u8(&self.embd.raw)?;
5687 let _ = self.embd_gpu.set(buf); }
5689 Ok(())
5690 }
5691
5692 pub fn embed(
5693 &self,
5694 e: &Engine,
5695 tokens: &[u32],
5696 ) -> Result<CudaSlice<f32>, Box<dyn std::error::Error>> {
5697 let n_embd = self.cfg.n_embd as usize;
5698 if std::env::var("MEMRA_EMBED_DEV").as_deref() != Ok("0") {
5704 let tbl = self
5705 .embd_gpu
5706 .get_or_init(|| e.upload_u8(&self.embd.raw).expect("embed table upload"));
5707 let tok_d = e.htod_u32_v(tokens)?;
5708 let (qt, rb) = self.embd.qt_and_row_bytes(n_embd);
5709 return e.embed_gather_device_td(tbl, &tok_d, tokens.len(), n_embd, qt, rb);
5710 }
5711 let x = self.embd.gather(n_embd, tokens);
5712 e.htod(&x)
5713 }
5714}
5715
5716fn illegal_pipeline_cuts(fence: &[usize], legal_boundaries: &[usize]) -> Vec<usize> {
5717 fence
5718 .get(1..fence.len().saturating_sub(1))
5719 .unwrap_or_default()
5720 .iter()
5721 .copied()
5722 .filter(|cut| !legal_boundaries.contains(cut))
5723 .collect()
5724}
5725
5726#[cfg(test)]
5727mod pipeline_cut_tests {
5728 use super::illegal_pipeline_cuts;
5729
5730 #[test]
5731 fn manual_pipeline_cuts_cannot_bypass_model_plan_boundaries() {
5732 assert!(illegal_pipeline_cuts(&[0, 8, 16, 24], &[8, 16]).is_empty());
5733 assert_eq!(illegal_pipeline_cuts(&[0, 7, 16, 24], &[8, 16]), vec![7]);
5734 assert_eq!(
5735 illegal_pipeline_cuts(&[0, 7, 15, 24], &[8, 16]),
5736 vec![7, 15]
5737 );
5738 }
5739}
5740
5741#[cfg(test)]
5742mod auto_parallel_policy_tests {
5743 use super::{parse_auto_parallel_tp_attention, parse_auto_w4a16_bf16_mmv};
5744
5745 #[test]
5746 fn automatic_w4a16_bf16_residency_defaults_on_with_explicit_rollback() {
5747 assert!(parse_auto_w4a16_bf16_mmv(None).unwrap());
5748 assert!(!parse_auto_w4a16_bf16_mmv(Some("0")).unwrap());
5749 assert!(parse_auto_w4a16_bf16_mmv(Some("1")).unwrap());
5750 assert!(parse_auto_w4a16_bf16_mmv(Some("true")).is_err());
5751 assert!(parse_auto_w4a16_bf16_mmv(Some("")).is_err());
5752 }
5753
5754 #[test]
5755 fn automatic_tp_attention_is_strict_and_defaults_off() {
5756 assert!(!parse_auto_parallel_tp_attention(None).unwrap());
5757 assert!(!parse_auto_parallel_tp_attention(Some("")).unwrap());
5758 assert!(!parse_auto_parallel_tp_attention(Some("0")).unwrap());
5759 assert!(parse_auto_parallel_tp_attention(Some("1")).unwrap());
5760 assert!(parse_auto_parallel_tp_attention(Some("true")).is_err());
5761 assert!(parse_auto_parallel_tp_attention(Some("2")).is_err());
5762 }
5763}
5764
5765#[cfg(test)]
5766mod step_expert_selection_tests {
5767 use super::{
5768 StepExpertArtifact, StepExpertLayout, StepParallelLoadConfig, StepParallelRuntimeRegistry,
5769 StepTpAttentionPlacement, select_step_expert_layout,
5770 };
5771 use crate::tp::StepEpLayerSpec;
5772
5773 fn spec(layer: usize, ranks: usize) -> StepEpLayerSpec {
5774 StepEpLayerSpec {
5775 layer,
5776 devices: (0..ranks).collect(),
5777 }
5778 }
5779
5780 #[test]
5781 fn tp2_keeps_projection_sharded_experts() {
5782 let selection = select_step_expert_layout(24, &[], &[spec(24, 2)])
5783 .unwrap()
5784 .unwrap();
5785 assert_eq!(selection.layout, StepExpertLayout::TensorParallel);
5786 assert!(selection.configured_by_tp);
5787 }
5788
5789 #[test]
5790 fn tp4_and_tp8_use_expert_ownership_without_a_second_flag() {
5791 for ranks in [4, 8] {
5792 let selection = select_step_expert_layout(24, &[], &[spec(24, ranks)])
5793 .unwrap()
5794 .unwrap();
5795 assert_eq!(selection.layout, StepExpertLayout::ExpertParallel);
5796 assert!(selection.configured_by_tp);
5797 assert_eq!(selection.spec.devices.len(), ranks);
5798 }
5799 }
5800
5801 #[test]
5802 fn explicit_ep_remains_expert_parallel() {
5803 let selection = select_step_expert_layout(24, &[spec(24, 2)], &[])
5804 .unwrap()
5805 .unwrap();
5806 assert_eq!(selection.layout, StepExpertLayout::ExpertParallel);
5807 assert!(!selection.configured_by_tp);
5808 }
5809
5810 #[test]
5811 fn conflicting_ep_and_tp_assignments_fail_closed() {
5812 let error = select_step_expert_layout(24, &[spec(24, 4)], &[spec(24, 4)]).unwrap_err();
5813 assert!(error.contains("cannot enable MEMRA_STEP_EP and MEMRA_STEP_TP together"));
5814 }
5815
5816 #[test]
5817 fn runtime_registry_owns_one_immutable_load_snapshot() {
5818 let mut source_specs = vec![spec(24, 8)];
5819 let registry = StepParallelRuntimeRegistry::with_config(StepParallelLoadConfig {
5820 ep_specs: Vec::new(),
5821 tp_specs: source_specs.clone(),
5822 native_p2p: true,
5823 ep_device_arithmetic: true,
5824 f32_mirror: true,
5825 bulk_p2p: true,
5826 nvfp4_device_routes: true,
5827 auto_parallel: true,
5828 expert_artifact: StepExpertArtifact::default(),
5829 });
5830 source_specs[0].devices.clear();
5831
5832 let stored = registry.tp_spec(24).unwrap();
5833 assert_eq!(stored.devices, (0..8).collect::<Vec<_>>());
5834 assert!(registry.config.native_p2p);
5835 assert!(registry.config.ep_device_arithmetic);
5836 assert!(registry.config.f32_mirror);
5837 assert!(registry.config.bulk_p2p);
5838 assert!(registry.config.nvfp4_device_routes);
5839 assert!(registry.config.auto_parallel);
5840 assert_eq!(
5841 registry.expert_selection(24).unwrap().unwrap().layout,
5842 StepExpertLayout::ExpertParallel
5843 );
5844
5845 let standalone = StepParallelRuntimeRegistry::default();
5846 assert!(standalone.tp_spec(24).is_none());
5847 assert!(!standalone.config.native_p2p);
5848 assert!(!standalone.config.ep_device_arithmetic);
5849 assert!(!standalone.config.f32_mirror);
5850 assert!(!standalone.config.bulk_p2p);
5851 }
5852
5853 #[test]
5854 fn rank_local_attention_uses_bounded_swa_rings_only_with_native_p2p() {
5855 assert_eq!(
5856 StepTpAttentionPlacement::resolve(true, None),
5857 StepTpAttentionPlacement::RankLocalGlobal
5858 );
5859 assert_eq!(
5860 StepTpAttentionPlacement::resolve(true, Some(512)),
5861 StepTpAttentionPlacement::RankLocalSwa
5862 );
5863 assert_eq!(
5864 StepTpAttentionPlacement::resolve(false, None),
5865 StepTpAttentionPlacement::OwnerTransportFallback
5866 );
5867 assert_eq!(
5868 StepTpAttentionPlacement::resolve(false, Some(512)),
5869 StepTpAttentionPlacement::OwnerSwa
5870 );
5871 }
5872}
5873
5874#[cfg(test)]
5875mod residency_tests {
5876 use super::{DevExpertFp8ProjectionScales, ResidentPlan, residency_bytes_by_device};
5877 use crate::model::HostExpertFp8BlockScales;
5878 use std::collections::HashMap;
5879
5880 #[test]
5881 fn pp_residency_counts_only_each_devices_expert_slice() {
5882 let tensors = [
5883 ("blk.0.ffn_gate_exps.weight", 10usize),
5884 ("blk.0.ffn_up_exps.weight", 20),
5885 ("blk.1.ffn_down_exps.weight", 30),
5886 ("blk.2.ffn_gate_exps.weight", 40),
5887 ("blk.3.ffn_up_exps.weight", 50),
5888 ("blk.0.attn_q.weight", 7),
5889 ("output.weight", 11),
5890 ];
5891 let bytes = residency_bytes_by_device(tensors, &[0, 0, 1, 1], 0);
5892 assert_eq!(bytes.experts.get(&0), Some(&60));
5893 assert_eq!(bytes.experts.get(&1), Some(&90));
5894 assert_eq!(bytes.rest, 18);
5895 assert!(bytes.saw_experts);
5896 }
5897
5898 #[test]
5899 fn pp_residency_combines_stages_that_share_one_device() {
5900 let tensors = [
5901 ("blk.0.ffn_gate_exps.weight", 10usize),
5902 ("blk.1.ffn_gate_exps.weight", 20),
5903 ("blk.2.ffn_gate_exps.weight", 30),
5904 ("blk.3.ffn_gate_exps.weight", 40),
5905 ];
5906 let bytes = residency_bytes_by_device(tensors, &[0, 0, 0, 0], 0);
5907 assert_eq!(bytes.experts.get(&0), Some(&100));
5908 assert_eq!(bytes.experts.len(), 1);
5909 }
5910
5911 #[test]
5912 fn distributed_trunk_layers_do_not_poison_local_mtp_residency_estimates() {
5913 let mut plan = ResidentPlan {
5914 primary_device: 0,
5915 layer_devices: vec![0; 81],
5916 layer_counts: HashMap::from([(0, 81)]),
5917 exact_expert_bytes: None,
5918 trunk_bytes: 0,
5919 decisions: HashMap::new(),
5920 pp: false,
5921 };
5922 plan.exclude_distributed_expert_layers(1..80);
5923 assert_eq!(plan.layer_counts.get(&0), Some(&2));
5924 }
5925
5926 #[test]
5927 fn resident_fp8_scale_slab_must_match_every_expert() {
5928 let valid = HostExpertFp8BlockScales {
5929 scales: vec![1.0; 12],
5930 rows: 2,
5931 cols: 3,
5932 expert_stride: 6,
5933 };
5934 DevExpertFp8ProjectionScales::validate(&valid, 2).unwrap();
5935
5936 let short = HostExpertFp8BlockScales {
5937 scales: vec![1.0; 11],
5938 ..valid
5939 };
5940 assert_eq!(
5941 DevExpertFp8ProjectionScales::validate(&short, 2).unwrap_err(),
5942 "block-E4M3 scale slab length mismatch: got 11, want 2x6=12"
5943 );
5944 }
5945
5946 #[test]
5947 fn resident_fp8_scale_stride_must_match_its_grid() {
5948 let invalid = HostExpertFp8BlockScales {
5949 scales: vec![1.0; 8],
5950 rows: 2,
5951 cols: 2,
5952 expert_stride: 0,
5953 };
5954 assert_eq!(
5955 DevExpertFp8ProjectionScales::validate(&invalid, 2).unwrap_err(),
5956 "block-E4M3 expert scale stride must be nonzero"
5957 );
5958 }
5959}
5960
5961#[cfg(test)]
5962mod draft_head_tests {
5963 use super::{draft_head_tensor, frspec_trim_own_head_name};
5964
5965 const STEP37_DRAFTER: &[&str] = &[
5972 "output.weight",
5973 "output_norm.weight",
5974 "token_embd.weight",
5975 "blk.45.nextn.shared_head_norm.weight",
5976 "blk.45.nextn.shared_head_head.weight",
5977 "blk.46.nextn.shared_head_head.weight",
5978 "blk.47.nextn.shared_head_head.weight",
5979 ];
5980
5981 fn present(names: &'static [&'static str]) -> impl Fn(&str) -> bool {
5982 move |t: &str| names.contains(&t)
5983 }
5984
5985 #[test]
5993 fn step37_drafter_prefers_the_blocks_own_nextn_head_over_file_level_output() {
5994 assert_eq!(
5995 draft_head_tensor(present(STEP37_DRAFTER), 45),
5996 "blk.45.nextn.shared_head_head.weight"
5997 );
5998 }
5999
6000 #[test]
6004 fn each_nextn_block_selects_its_own_head() {
6005 for n in 45..=47u32 {
6006 assert_eq!(
6007 draft_head_tensor(present(STEP37_DRAFTER), n),
6008 format!("blk.{n}.nextn.shared_head_head.weight")
6009 );
6010 }
6011 }
6012
6013 #[test]
6017 fn draft_without_a_nextn_head_falls_back_to_file_level_output() {
6018 let fr_spec: &[&str] = &["output.weight", "output_norm.weight", "d2t.weight"];
6019 assert_eq!(draft_head_tensor(present(fr_spec), 45), "output.weight");
6020 }
6021
6022 #[test]
6027 fn legacy_shared_head_is_probed_but_loses_to_shared_head_head() {
6028 let legacy_only: &[&str] = &["output.weight", "blk.45.nextn.shared_head.weight"];
6029 assert_eq!(
6030 draft_head_tensor(present(legacy_only), 45),
6031 "blk.45.nextn.shared_head.weight"
6032 );
6033
6034 let both: &[&str] = &[
6035 "output.weight",
6036 "blk.45.nextn.shared_head.weight",
6037 "blk.45.nextn.shared_head_head.weight",
6038 ];
6039 assert_eq!(
6040 draft_head_tensor(present(both), 45),
6041 "blk.45.nextn.shared_head_head.weight"
6042 );
6043 }
6044
6045 #[test]
6049 fn a_different_blocks_nextn_head_is_never_borrowed() {
6050 let wrong_block: &[&str] = &[
6051 "output.weight",
6052 "blk.46.nextn.shared_head_head.weight",
6053 "blk.47.nextn.shared_head_head.weight",
6054 ];
6055 assert_eq!(draft_head_tensor(present(wrong_block), 45), "output.weight");
6056 }
6057
6058 #[test]
6063 fn frspec_trim_prefers_the_nextn_blocks_own_head_name() {
6064 assert_eq!(
6065 frspec_trim_own_head_name(45),
6066 "blk.45.nextn.shared_head_head.weight"
6067 );
6068 assert_eq!(
6070 frspec_trim_own_head_name(45),
6071 format!("blk.{}.nextn.shared_head_head.weight", 45)
6072 );
6073 assert_eq!(
6074 frspec_trim_own_head_name(40),
6075 "blk.40.nextn.shared_head_head.weight"
6076 );
6077 }
6078}