1use crate::Engine;
6use crate::model::{EmbedHost, GpuTensor, HostExps};
7use cudarc::driver::CudaSlice;
8use memra_gguf::config::ModelConfig;
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 expert_artifact: StepExpertArtifact,
103}
104
105#[derive(Default)]
106pub(crate) struct StepParallelRuntimeRegistry {
107 config: StepParallelLoadConfig,
108 runtimes: HashMap<(Vec<usize>, bool, bool, bool), Arc<crate::tp::TpE4m3HostBounce>>,
109}
110
111#[derive(Clone, Copy, Debug, PartialEq, Eq)]
112enum StepExpertLayout {
113 TensorParallel,
114 ExpertParallel,
115}
116
117#[derive(Clone, Debug, PartialEq, Eq)]
118struct StepExpertSelection {
119 spec: crate::tp::StepEpLayerSpec,
120 layout: StepExpertLayout,
121 configured_by_tp: bool,
122}
123
124fn select_step_expert_layout(
125 layer: usize,
126 ep_specs: &[crate::tp::StepEpLayerSpec],
127 tp_specs: &[crate::tp::StepTpLayerSpec],
128) -> Result<Option<StepExpertSelection>, String> {
129 let ep = ep_specs.iter().find(|spec| spec.layer == layer);
130 let tp = tp_specs.iter().find(|spec| spec.layer == layer);
131 if ep.is_some() && tp.is_some() {
132 return Err(format!(
133 "Step layer {layer} cannot enable MEMRA_STEP_EP and MEMRA_STEP_TP together"
134 ));
135 }
136 Ok(match (ep, tp) {
137 (Some(spec), None) => Some(StepExpertSelection {
138 spec: spec.clone(),
139 layout: StepExpertLayout::ExpertParallel,
140 configured_by_tp: false,
141 }),
142 (None, Some(spec)) => Some(StepExpertSelection {
143 spec: spec.clone(),
144 layout: if spec.devices.len() > 2 {
145 StepExpertLayout::ExpertParallel
146 } else {
147 StepExpertLayout::TensorParallel
148 },
149 configured_by_tp: true,
150 }),
151 (None, None) => None,
152 (Some(_), Some(_)) => unreachable!(),
153 })
154}
155
156impl StepParallelRuntimeRegistry {
157 fn with_config(config: StepParallelLoadConfig) -> Self {
158 Self {
159 config,
160 runtimes: HashMap::new(),
161 }
162 }
163
164 fn tp_spec(&self, layer: usize) -> Option<&crate::tp::StepTpLayerSpec> {
165 self.config.tp_specs.iter().find(|spec| spec.layer == layer)
166 }
167
168 fn expert_selection(&self, layer: usize) -> Result<Option<StepExpertSelection>, String> {
169 select_step_expert_layout(layer, &self.config.ep_specs, &self.config.tp_specs)
170 }
171
172 fn runtime(
173 &mut self,
174 devices: &[usize],
175 native_p2p: bool,
176 ep_device_arithmetic: bool,
177 ) -> Result<Arc<crate::tp::TpE4m3HostBounce>, Box<dyn std::error::Error>> {
178 let bulk_p2p = self.config.bulk_p2p && native_p2p;
179 let key = (devices.to_vec(), native_p2p, ep_device_arithmetic, bulk_p2p);
180 if let Some(runtime) = self.runtimes.get(&key) {
181 return Ok(Arc::clone(runtime));
182 }
183 let runtime = Arc::new(crate::tp::TpE4m3HostBounce::new_configured(
184 devices,
185 native_p2p,
186 ep_device_arithmetic,
187 bulk_p2p,
188 )?);
189 let names = runtime.device_names()?;
190 if names
191 .iter()
192 .any(|name| !name.contains("RTX PRO 6000") || !name.contains("Blackwell"))
193 {
194 return Err(format!(
195 "Step distributed execution is qualified only on RTX PRO 6000 Blackwell, \
196 got {names:?}"
197 )
198 .into());
199 }
200 self.runtimes.insert(key, Arc::clone(&runtime));
201 Ok(runtime)
202 }
203}
204
205impl ResidentPlan {
206 fn from_layout(
207 src: &dyn TensorSource,
208 primary_device: usize,
209 layer_devices: Vec<usize>,
210 pp: bool,
211 ) -> Self {
212 let mut layer_counts = HashMap::new();
213 for &device in &layer_devices {
214 *layer_counts.entry(device).or_default() += 1;
215 }
216 let (exact_expert_bytes, trunk_bytes) = match src.gguf() {
217 Some(g) => {
218 let bytes = residency_bytes_by_device(
219 g.tensors
220 .iter()
221 .map(|t| (t.name.as_str(), t.n_bytes as usize)),
222 &layer_devices,
223 primary_device,
224 );
225 if bytes.saw_experts {
226 (Some(bytes.experts), bytes.rest)
227 } else {
228 (None, 0)
229 }
230 }
231 None => (None, 0),
232 };
233 Self {
234 primary_device,
235 layer_devices,
236 layer_counts,
237 exact_expert_bytes,
238 trunk_bytes,
239 decisions: HashMap::new(),
240 pp,
241 }
242 }
243
244 pub(crate) fn unsharded(e: &Engine, src: &dyn TensorSource, cfg: &ModelConfig) -> Self {
245 let device = e.ctx().ordinal();
246 Self::from_layout(src, device, vec![device; cfg.n_layer as usize], false)
247 }
248
249 pub(crate) fn pp(
250 e: &Engine,
251 src: &dyn TensorSource,
252 cfg: &ModelConfig,
253 n_trunk: usize,
254 ) -> Result<Self, Box<dyn std::error::Error>> {
255 let primary = e.ctx().ordinal();
256 let Some(_fence) = crate::pp::pp_cuts(n_trunk) else {
257 return Ok(Self::unsharded(e, src, cfg));
258 };
259 let mut layer_devices = vec![primary; cfg.n_layer as usize];
260 for (il, device) in layer_devices.iter_mut().take(n_trunk).enumerate() {
261 *device = crate::pp::layer_engine(e, n_trunk, il)?.ctx().ordinal();
262 }
263 Ok(Self::from_layout(src, primary, layer_devices, true))
264 }
265
266 fn should_reside(&mut self, e: &Engine, il: usize, per_layer: usize) -> bool {
267 let device = self
268 .layer_devices
269 .get(il)
270 .copied()
271 .unwrap_or(self.primary_device);
272 debug_assert_eq!(e.ctx().ordinal(), device);
273 if let Some(&decision) = self.decisions.get(&device) {
274 return decision;
275 }
276 if std::env::var("MEMRA_MOE_RESIDENT").as_deref() == Ok("0") {
277 self.decisions.insert(device, false);
278 return false;
279 }
280 let (free, _total) = match e.ctx().mem_get_info() {
281 Ok(v) => v,
282 Err(_) => {
283 self.decisions.insert(device, false);
284 return false;
285 }
286 };
287 let projected = self
288 .exact_expert_bytes
289 .as_ref()
290 .map(|bytes| bytes.get(&device).copied().unwrap_or(0))
291 .unwrap_or(per_layer * self.layer_counts.get(&device).copied().unwrap_or(1));
292 let budget = std::env::var("MEMRA_MOE_RESIDENT_GB")
293 .ok()
294 .and_then(|v| v.parse::<f64>().ok())
295 .map(|gb| (gb * 1e9) as usize)
296 .unwrap_or_else(|| {
297 let reserve = std::env::var("MEMRA_MOE_RESIDENT_HEADROOM_GB")
298 .ok()
299 .and_then(|v| v.parse::<f64>().ok())
300 .map(|gb| (gb * 1e9) as usize)
301 .unwrap_or(2_000_000_000);
302 (free as usize).saturating_sub(self.trunk_bytes + reserve)
303 });
304 let ok = projected <= budget;
305 eprintln!(
306 "[moe] resident-experts decision ({}dev{}): experts {:.2}GB + trunk {:.2}GB vs free {:.2}GB (expert budget {:.2}GB) -> {}",
307 if self.pp { "PP " } else { "" },
308 device,
309 projected as f64 / 1e9,
310 self.trunk_bytes as f64 / 1e9,
311 free as f64 / 1e9,
312 budget as f64 / 1e9,
313 if ok { "RESIDENT" } else { "SLRU cache" }
314 );
315 self.decisions.insert(device, ok);
316 ok
317 }
318}
319
320fn load_mixer_kind(
322 e: &Engine,
323 src: &dyn TensorSource,
324 cfg: &ModelConfig,
325 il: u32,
326 attention: &AttentionPlan,
327 step_runtimes: &mut StepParallelRuntimeRegistry,
328) -> Result<Mixer, Box<dyn std::error::Error>> {
329 let p = |s: &str| format!("blk.{il}.{s}");
330 Ok(match attention {
331 AttentionPlan::Mla(mla) => Mixer::Mla(MlaAttnLayer::load(e, src, il, mla)?),
332 AttentionPlan::Full(full)
333 | AttentionPlan::SlidingWindow {
334 attention: full, ..
335 } => {
336 Mixer::Full(FullAttnLayer {
337 wq: load_t(e, src, &p("attn_q.weight"))?,
338 wk: load_t(e, src, &p("attn_k.weight"))?,
339 wv: match load_opt(e, src, &p("attn_v.weight"))? {
344 Some(v) => v,
345 None => load_t(e, src, &p("attn_k.weight"))?,
346 },
347 wo: load_t(e, src, &p("attn_output.weight"))?,
348 q_norm: load_t(e, src, &p("attn_q_norm.weight"))?,
349 k_norm: load_t(e, src, &p("attn_k_norm.weight"))?,
350 attn_gate: if full.output_gate
354 == memra_gguf::config::AttentionGateKind::SeparateHead
355 {
356 Some(load_t(e, src, &p("attn_gate.weight"))?)
357 } else {
358 None
359 },
360 step_tp_qkv: build_step_tp_qkv(e, src, cfg, il as usize, step_runtimes)?,
361 })
362 }
363 AttentionPlan::GatedDeltaNet(geometry) => Mixer::Linear(LinearAttnLayer {
364 geometry: *geometry,
365 wqkv: load_t(e, src, &p("attn_qkv.weight"))?,
366 wqkv_gate: load_t(e, src, &p("attn_gate.weight"))?,
367 ssm_beta: load_t(e, src, &p("ssm_beta.weight"))?,
368 ssm_alpha: load_t(e, src, &p("ssm_alpha.weight"))?,
369 ssm_a: load_t(e, src, &p("ssm_a"))?,
370 ssm_dt: load_t(e, src, &p("ssm_dt.bias"))?,
371 ssm_conv1d: load_t(e, src, &p("ssm_conv1d.weight"))?,
372 ssm_norm: load_t(e, src, &p("ssm_norm.weight"))?,
373 ssm_out: load_t(e, src, &p("ssm_out.weight"))?,
374 }),
375 })
376}
377
378pub(crate) fn load_ffn(
385 e: &Engine,
386 src: &dyn TensorSource,
387 cfg: &ModelConfig,
388 mlp: &MlpPlan,
389 il: u32,
390 spill: Option<(&GgufFile, &mut crate::spill::SpillCtx)>,
391 resident: &mut ResidentPlan,
392 step_runtimes: &mut StepParallelRuntimeRegistry,
393) -> Result<Ffn, Box<dyn std::error::Error>> {
394 let p = |s: &str| format!("blk.{il}.{s}");
395 let artifact_dense = matches!(mlp, MlpPlan::Moe(_))
402 && !src.has(&p("ffn_gate_exps.weight"))
403 && !src.has(&p("ffn_gate_up_exps.weight"))
404 && src.has(&p("ffn_gate.weight"));
405 Ok(if artifact_dense {
406 Ffn::Dense {
407 ffn_gate: GpuTensor::load_from_source(e, src, &p("ffn_gate.weight"))?,
408 ffn_up: GpuTensor::load_from_source(e, src, &p("ffn_up.weight"))?,
409 ffn_down: GpuTensor::load_from_source(e, src, &p("ffn_down.weight"))?,
410 }
411 } else if let MlpPlan::Moe(moe) = mlp {
412 let n_expert = moe.expert_count as usize;
413 let (gate_exps, up_exps, down_exps) = match spill {
419 Some((g, ctx)) => (
420 HostExps::load_tiered(e, g, &p("ffn_gate_exps.weight"), ctx)?,
421 HostExps::load_tiered(e, g, &p("ffn_up_exps.weight"), ctx)?,
422 HostExps::load_tiered(e, g, &p("ffn_down_exps.weight"), ctx)?,
423 ),
424 None => {
425 let exps = |e: &Engine, n: &str| -> Result<HostExps, Box<dyn std::error::Error>> {
426 if src.has(n) {
427 HostExps::load_stacked_from_source(e, src, n)
428 } else {
429 HostExps::load_from_source(e, src, n, n_expert)
430 }
431 };
432 let fused = p("ffn_gate_up_exps.weight");
434 if !src.has(&p("ffn_gate_exps.weight")) && src.has(&fused) {
435 let ff = moe.expert_intermediate_size as usize;
436 (
437 HostExps::load_stacked_split_from_source(e, src, &fused, 0, ff)?,
438 HostExps::load_stacked_split_from_source(e, src, &fused, ff, 2 * ff)?,
439 exps(e, &p("ffn_down_exps.weight"))?,
440 )
441 } else {
442 (
443 exps(e, &p("ffn_gate_exps.weight"))?,
444 exps(e, &p("ffn_up_exps.weight"))?,
445 exps(e, &p("ffn_down_exps.weight"))?,
446 )
447 }
448 }
449 };
450 let (step_ep, step_tp) = build_step_distributed_exps(
451 e,
452 cfg,
453 src,
454 il as usize,
455 &gate_exps,
456 &up_exps,
457 &down_exps,
458 step_runtimes,
459 )?;
460 let dev_exps = if step_ep.is_some() || step_tp.is_some() {
466 None
467 } else {
468 build_dev_exps(e, resident, il as usize, &gate_exps, &up_exps, &down_exps)?
469 };
470 let mut macro_row = vec![1.0f32; 3 * n_expert];
472 for (slot, exps) in [(0usize, &gate_exps), (1, &up_exps), (2, &down_exps)] {
473 if let Some(ms) = exps.macros.as_ref() {
474 macro_row[slot * n_expert..(slot + 1) * n_expert].copy_from_slice(ms);
475 }
476 }
477 let has_macros = macro_row.iter().any(|&m| m != 1.0);
478 let dev_macros = e.htod(¯o_row)?;
479 let exp_probs_b = src
482 .find(&p("exp_probs_b.bias"))
483 .map(|v| memra_gguf::dequant::dequantize(v.ggml_type, &v.bytes, n_expert));
484 let active_experts = src.active_experts(il).map(<[bool]>::to_vec);
485 let route_bias = exp_probs_b.clone().unwrap_or_else(|| vec![0.0; n_expert]);
486 let active_row: Vec<u8> = active_experts
487 .as_ref()
488 .map(|mask| mask.iter().map(|&is_active| u8::from(is_active)).collect())
489 .unwrap_or_else(|| vec![1; n_expert]);
490 let exp_probs_b_dev = e.htod(&route_bias)?;
491 let active_experts_dev = e.htod_bytes(&active_row)?;
492 Ffn::Moe(MoeWeights {
493 gate_inp: load_t(e, src, &p("ffn_gate_inp.weight"))?,
494 gate_inp_shexp: load_opt(e, src, &p("ffn_gate_inp_shexp.weight"))?,
495 exp_probs_b,
496 exp_probs_b_dev,
497 active_experts,
498 active_experts_dev,
499 gate_exps,
500 up_exps,
501 down_exps,
502 gate_shexp: load_opt(e, src, &p("ffn_gate_shexp.weight"))?,
503 up_shexp: load_opt(e, src, &p("ffn_up_shexp.weight"))?,
504 down_shexp: load_opt(e, src, &p("ffn_down_shexp.weight"))?,
505 dev_exps,
506 step_ep,
507 step_tp,
508 dev_macros,
509 has_macros,
510 })
511 } else {
512 Ffn::Dense {
513 ffn_gate: load_t(e, src, &p("ffn_gate.weight"))?,
514 ffn_up: load_t(e, src, &p("ffn_up.weight"))?,
515 ffn_down: load_t(e, src, &p("ffn_down.weight"))?,
516 }
517 })
518}
519
520fn host_e4m3_bank(
521 exps: &HostExps,
522) -> Result<crate::tp::E4m3ExpertBank<'_>, Box<dyn std::error::Error>> {
523 if exps.qtype != crate::QT_F8_E4M3_BLK {
524 return Err(format!(
525 "Step EP requires native block-E4M3 expert banks, got qtype {}",
526 exps.qtype
527 )
528 .into());
529 }
530 let scales = exps
531 .fp8_blk
532 .as_ref()
533 .ok_or("Step EP native expert bank has no block-E4M3 scale plane")?;
534 Ok(crate::tp::E4m3ExpertBank {
535 codes: exps.bytes.as_bytes(),
536 scales: &scales.scales,
537 expert_count: exps.n_expert,
538 out_features: exps.out_f,
539 in_features: exps.in_f,
540 })
541}
542
543fn validate_step_expert_specs(
544 contract: &crate::parallel::ModelParallelContract,
545 flag: &str,
546 specs: &[crate::tp::StepEpLayerSpec],
547 allow_dense_attention_only: bool,
548) -> Result<(), Box<dyn std::error::Error>> {
549 for candidate in specs {
550 if candidate.layer >= contract.trunk_layers {
551 return Err(format!(
552 "{flag} layer {} is outside Step trunk layers 0..{}",
553 candidate.layer, contract.trunk_layers
554 )
555 .into());
556 }
557 if candidate.layer < contract.dense_prefix_layers {
558 if allow_dense_attention_only {
559 continue;
560 }
561 return Err(format!(
562 "{flag} layer {} is outside Step routed-expert layers {}..{}",
563 candidate.layer, contract.dense_prefix_layers, contract.trunk_layers
564 )
565 .into());
566 }
567 }
568 Ok(())
569}
570
571fn validate_step_expert_activation_layout(
572 cfg: &ModelConfig,
573 flag: &str,
574 selection: &StepExpertSelection,
575) -> Result<(), Box<dyn std::error::Error>> {
576 let _ = (cfg, flag, selection);
582 Ok(())
583}
584
585fn prepare_step_parallel_load(
586 e: &Engine,
587 src: &dyn TensorSource,
588 cfg: &ModelConfig,
589 trunk_layers: usize,
590) -> Result<StepParallelLoadConfig, Box<dyn std::error::Error>> {
591 let tp_specs = crate::tp::step_tp_layer_specs()?;
592 let ep_specs = crate::tp::step_ep_layer_specs()?;
593 let device_arithmetic = crate::tp::step_ep_device_arithmetic_enabled()?;
594 let f32_mirror = crate::tp::step_tp_f32_mirror_enabled()?;
595 let bulk_p2p = crate::tp::step_tp_bulk_p2p_enabled()?;
596 if tp_specs.is_empty() {
597 if device_arithmetic || f32_mirror || bulk_p2p {
598 return Err(
599 "MEMRA_STEP_EP_DEVICE_ARITHMETIC=1, MEMRA_STEP_TP_F32_MIRROR=1, or \
600 MEMRA_STEP_TP_BULK_P2P=1 requires MEMRA_STEP_TP; device arithmetic and bulk \
601 transport also require MEMRA_STEP_TP_NATIVE_P2P=1"
602 .into(),
603 );
604 }
605 let expert_artifact = if ep_specs.is_empty() {
608 StepExpertArtifact::default()
609 } else {
610 let contract = crate::parallel::ModelParallelContract::from_model(cfg)?;
611 match crate::parallel::validate_step_fp8_checkpoint(src, &contract) {
612 Ok(_) => StepExpertArtifact::E4m3,
613 Err(fp8_error) => {
614 match crate::parallel::validate_step_nvfp4_checkpoint(src, &contract) {
615 Ok(_) => StepExpertArtifact::Nvfp4,
616 Err(nvfp4_error) => {
617 return Err(format!(
618 "Step checkpoint qualifies as neither native expert artifact \
619 class: [E4M3] {fp8_error} [NVFP4] {nvfp4_error}"
620 )
621 .into());
622 }
623 }
624 }
625 }
626 };
627 return Ok(StepParallelLoadConfig {
628 ep_specs,
629 expert_artifact,
630 ..StepParallelLoadConfig::default()
631 });
632 }
633 let contract = crate::parallel::ModelParallelContract::from_model(cfg)?;
634 validate_step_expert_specs(&contract, "MEMRA_STEP_EP", &ep_specs, false)?;
635 validate_step_expert_specs(&contract, "MEMRA_STEP_TP", &tp_specs, true)?;
636 for spec in &tp_specs {
637 let selection = select_step_expert_layout(spec.layer, &ep_specs, &tp_specs)?
638 .ok_or("Step TP expert selection disappeared during preflight")?;
639 validate_step_expert_activation_layout(cfg, "MEMRA_STEP_TP", &selection)?;
640 }
641
642 let layer_owners = (0..trunk_layers)
643 .map(|layer| {
644 crate::pp::layer_engine(e, trunk_layers, layer).map(|engine| engine.ctx().ordinal())
645 })
646 .collect::<Result<Vec<_>, _>>()?;
647 let plan = contract.preflight_step_tp_specs(
648 tp_specs
649 .iter()
650 .map(|spec| (spec.layer, spec.devices.as_slice())),
651 &layer_owners,
652 )?;
653
654 for devices in &plan.runtime_groups {
655 let hardware = crate::parallel::detect_uniform_hardware(devices)?;
656 if !contract.hardware_targets.contains(&hardware) {
657 return Err(format!(
658 "{} has no qualified {hardware:?} TP contract for devices {devices:?}",
659 contract.variant
660 )
661 .into());
662 }
663 }
664
665 let native_p2p = crate::tp::step_tp_native_p2p_enabled()?;
666 if bulk_p2p && !native_p2p {
667 return Err("MEMRA_STEP_TP_BULK_P2P=1 requires MEMRA_STEP_TP_NATIVE_P2P=1".into());
668 }
669 if device_arithmetic
670 && (!ep_specs.is_empty()
671 || !native_p2p
672 || plan.expert_parallel_layers() == 0
673 || plan.tensor_parallel_expert_layers() != 0)
674 {
675 return Err(
676 "MEMRA_STEP_EP_DEVICE_ARITHMETIC=1 requires native-P2P TP4/TP8 \
677 expert ownership for every selected routed-expert layer"
678 .into(),
679 );
680 }
681 let (qualified_experts, expert_artifact) =
685 match crate::parallel::validate_step_fp8_checkpoint(src, &contract) {
686 Ok(qualified) => (qualified, StepExpertArtifact::E4m3),
687 Err(fp8_error) => match crate::parallel::validate_step_nvfp4_checkpoint(src, &contract)
688 {
689 Ok(qualified) => (qualified, StepExpertArtifact::Nvfp4),
690 Err(nvfp4_error) => {
691 return Err(format!(
692 "Step checkpoint qualifies as neither native expert artifact class: \
693 [E4M3] {fp8_error} [NVFP4] {nvfp4_error}"
694 )
695 .into());
696 }
697 },
698 };
699 if expert_artifact == StepExpertArtifact::Nvfp4 {
700 if device_arithmetic {
701 return Err(
702 "MEMRA_STEP_EP_DEVICE_ARITHMETIC=1 is qualified for the E4M3 expert artifact \
703 only; the NVFP4 expert program is host-canonical in this increment"
704 .into(),
705 );
706 }
707 if bulk_p2p {
712 return Err(
713 "MEMRA_STEP_TP_BULK_P2P=1 is qualified for the E4M3 expert artifact only; the \
714 NVFP4 bank transport increment has not landed"
715 .into(),
716 );
717 }
718 }
719
720 if f32_mirror {
721 eprintln!(
722 "[step-tp-preflight] layers={} full_trunk={} runtime_groups={} \
723 dense_attention_layers={} tensor_expert_layers={} expert_owner_layers={} \
724 qualified_fp8_expert_projection_slices={} owner_first=true \
725 hardware=rtx-pro-6000-blackwell \
726 native_p2p={} bulk_p2p={} device_arithmetic={} bf16_residency=f32-mirror \
727 weights_loaded=false performance_claim=false",
728 plan.layers.len(),
729 plan.full_trunk,
730 plan.runtime_groups.len(),
731 plan.dense_attention_layers(),
732 plan.tensor_parallel_expert_layers(),
733 plan.expert_parallel_layers(),
734 qualified_experts,
735 native_p2p,
736 bulk_p2p,
737 device_arithmetic,
738 );
739 } else {
740 eprintln!(
741 "[step-tp-preflight] layers={} full_trunk={} runtime_groups={} \
742 dense_attention_layers={} tensor_expert_layers={} expert_owner_layers={} \
743 qualified_fp8_expert_projection_slices={} owner_first=true \
744 hardware=rtx-pro-6000-blackwell \
745 native_p2p={} bulk_p2p={} device_arithmetic={} \
746 weights_loaded=false performance_claim=false",
747 plan.layers.len(),
748 plan.full_trunk,
749 plan.runtime_groups.len(),
750 plan.dense_attention_layers(),
751 plan.tensor_parallel_expert_layers(),
752 plan.expert_parallel_layers(),
753 qualified_experts,
754 native_p2p,
755 bulk_p2p,
756 device_arithmetic,
757 );
758 }
759 Ok(StepParallelLoadConfig {
760 ep_specs,
761 tp_specs,
762 native_p2p,
763 ep_device_arithmetic: device_arithmetic,
764 f32_mirror,
765 bulk_p2p,
766 expert_artifact,
767 })
768}
769
770fn step_nvfp4_native<'a>(
772 src: &'a dyn TensorSource,
773 layer: usize,
774 proj: &str,
775) -> Result<memra_gguf::source::Nvfp4StackedNative<'a>, Box<dyn std::error::Error>> {
776 let name = format!("blk.{layer}.ffn_{proj}_exps.weight");
777 src.find_nvfp4_stacked_native(&name)
778 .ok_or_else(|| format!("Step NVFP4 expert program is missing native bank {name}").into())
779}
780
781fn step_nvfp4_bank<'a>(
783 native: &'a memra_gguf::source::Nvfp4StackedNative<'a>,
784) -> crate::tp::Nvfp4ExpertBank<'a> {
785 crate::tp::Nvfp4ExpertBank {
786 codes: native.codes,
787 scales: native.scales,
788 macros: &native.macros,
789 expert_count: native.n_expert,
790 out_features: native.out_f,
791 in_features: native.in_f,
792 }
793}
794
795fn build_step_distributed_exps(
796 e: &Engine,
797 cfg: &ModelConfig,
798 src: &dyn TensorSource,
799 layer: usize,
800 gate: &HostExps,
801 up: &HostExps,
802 down: &HostExps,
803 step_runtimes: &mut StepParallelRuntimeRegistry,
804) -> Result<(Option<StepEpExps>, Option<StepTpExps>), Box<dyn std::error::Error>> {
805 let ep_device_arithmetic = step_runtimes.config.ep_device_arithmetic;
806 if step_runtimes.config.ep_specs.is_empty() && step_runtimes.config.tp_specs.is_empty() {
807 if ep_device_arithmetic {
808 return Err(
809 "MEMRA_STEP_EP_DEVICE_ARITHMETIC=1 requires MEMRA_STEP_TP and \
810 MEMRA_STEP_TP_NATIVE_P2P=1"
811 .into(),
812 );
813 }
814 return Ok((None, None));
815 }
816 let contract = crate::parallel::ModelParallelContract::from_model(cfg)?;
817 validate_step_expert_specs(
818 &contract,
819 "MEMRA_STEP_EP",
820 &step_runtimes.config.ep_specs,
821 false,
822 )?;
823 validate_step_expert_specs(
824 &contract,
825 "MEMRA_STEP_TP",
826 &step_runtimes.config.tp_specs,
827 true,
828 )?;
829 let Some(selection) = step_runtimes.expert_selection(layer)? else {
830 return Ok((None, None));
831 };
832 validate_step_expert_activation_layout(
833 cfg,
834 if selection.configured_by_tp {
835 "MEMRA_STEP_TP"
836 } else {
837 "MEMRA_STEP_EP"
838 },
839 &selection,
840 )?;
841 let activation_limit = cfg.clamp_exp_at(layer as u32);
842 let owner = e.ctx().ordinal();
843 if !selection.spec.devices.contains(&owner) {
844 let flag = if selection.configured_by_tp {
845 "MEMRA_STEP_TP"
846 } else {
847 "MEMRA_STEP_EP"
848 };
849 return Err(format!(
850 "{flag} layer {layer} owning PP device {owner} is absent from rank devices {:?}",
851 selection.spec.devices
852 )
853 .into());
854 }
855 let expert_parallel = selection.layout == StepExpertLayout::ExpertParallel;
856 if selection.configured_by_tp {
857 contract.plan(crate::parallel::TopologyRequest {
858 pipeline: 1,
859 tensor: selection.spec.devices.len(),
860 expert_parallel,
861 available_devices: selection.spec.devices.len(),
862 hardware: crate::parallel::HardwareTarget::RtxPro6000Blackwell,
863 })?;
864 }
865 let native_p2p = selection.configured_by_tp && step_runtimes.config.native_p2p;
866 if ep_device_arithmetic
867 && (!selection.configured_by_tp
868 || selection.layout != StepExpertLayout::ExpertParallel
869 || !native_p2p)
870 {
871 return Err(
872 "MEMRA_STEP_EP_DEVICE_ARITHMETIC=1 requires a MEMRA_STEP_TP TP4/TP8 \
873 expert-owner layer and MEMRA_STEP_TP_NATIVE_P2P=1"
874 .into(),
875 );
876 }
877 let runtime =
878 step_runtimes.runtime(&selection.spec.devices, native_p2p, ep_device_arithmetic)?;
879 let expert_artifact = step_runtimes.config.expert_artifact;
880 match selection.layout {
881 StepExpertLayout::ExpertParallel => {
882 if expert_artifact == StepExpertArtifact::Nvfp4 {
883 let runtime = step_runtimes.runtime(
890 &selection.spec.devices,
891 step_runtimes.config.native_p2p,
892 false,
893 )?;
894 let gate_native = step_nvfp4_native(src, layer, "gate")?;
895 let up_native = step_nvfp4_native(src, layer, "up")?;
896 let down_native = step_nvfp4_native(src, layer, "down")?;
897 let experts = runtime.upload_expert_parallel_nvfp4(
898 step_nvfp4_bank(&gate_native),
899 step_nvfp4_bank(&up_native),
900 step_nvfp4_bank(&down_native),
901 )?;
902 eprintln!(
903 "[step-ep] layer={layer} devices={:?} experts={} artifact=nvfp4 \
904 expert_layout=expert-parallel expert_transport=host-bounce \
905 macro_fold=post-kernel-once native_p2p=false performance_claim=false",
906 selection.spec.devices, contract.expert_count
907 );
908 if let Some(limit) = activation_limit {
909 eprintln!(
910 "[step-ep-clamp] load layer={layer} routed_clamp={limit} \
911 formula=min-silu-times-clamped-up performance_claim=false"
912 );
913 }
914 return Ok((
915 Some(StepEpExps {
916 runtime,
917 experts: StepEpExpertBank::Nvfp4(experts),
918 devices: selection.spec.devices,
919 configured_by_tp: selection.configured_by_tp,
920 activation_limit,
921 grouped_decode: None,
922 }),
923 None,
924 ));
925 }
926 let experts = runtime.upload_expert_parallel(
927 host_e4m3_bank(gate)?,
928 host_e4m3_bank(up)?,
929 host_e4m3_bank(down)?,
930 )?;
931 let grouped_decode = if ep_device_arithmetic {
932 let tokens = 1;
933 let selected = (0..contract.experts_per_token).collect::<Vec<_>>();
934 let input = vec![0.0f32; contract.hidden_size];
935 let route_weights = vec![1.0f32; contract.experts_per_token];
936 let projection = runtime.prepare_step_grouped_expert_parallel_gate_with_capacity(
937 &experts,
938 &input,
939 tokens,
940 &selected,
941 activation_limit,
942 tokens,
943 )?;
944 let combine = runtime
945 .prepare_step_grouped_expert_parallel_combine(&projection, &route_weights)?;
946 Some(std::sync::Mutex::new(StepEpGroupedDecode {
947 projection,
948 combine,
949 }))
950 } else {
951 None
952 };
953 if selection.configured_by_tp {
954 eprintln!(
955 "[step-tp-ep] layer={layer} devices={:?} experts={} tp={} \
956 attention_layout=tensor-parallel expert_layout=expert-parallel \
957 expert_transport={} tp_transport={} native_p2p={} \
958 activation={} accumulation={} output={} \
959 grouped_decode_prepared={} grouped_decode_capacity=1 \
960 performance_claim=false",
961 selection.spec.devices,
962 contract.expert_count,
963 selection.spec.devices.len(),
964 runtime.transport_label(),
965 runtime.transport_label(),
966 runtime.native_p2p(),
967 runtime.expert_activation_label(),
968 runtime.expert_accumulation_label(),
969 runtime.expert_output_label(),
970 grouped_decode.is_some(),
971 );
972 } else {
973 eprintln!(
974 "[step-ep] layer={layer} devices={:?} experts={} \
975 expert_layout=expert-parallel expert_transport=host-bounce \
976 native_p2p=false performance_claim=false",
977 selection.spec.devices, contract.expert_count
978 );
979 }
980 if let Some(limit) = activation_limit {
981 eprintln!(
982 "[step-ep-clamp] load layer={layer} routed_clamp={limit} \
983 formula=min-silu-times-clamped-up performance_claim=false"
984 );
985 }
986 Ok((
987 Some(StepEpExps {
988 runtime,
989 experts: StepEpExpertBank::E4m3(experts),
990 devices: selection.spec.devices,
991 configured_by_tp: selection.configured_by_tp,
992 activation_limit,
993 grouped_decode,
994 }),
995 None,
996 ))
997 }
998 StepExpertLayout::TensorParallel => {
999 if activation_limit.is_some() && expert_artifact == StepExpertArtifact::E4m3 {
1000 return Err(format!(
1001 "layer {layer} uses the routed SwiGLU clamp and the E4M3 TP expert \
1002 program has no clamp arm; select EP for this layer (the NVFP4 TP \
1003 program carries the clamp)"
1004 )
1005 .into());
1006 }
1007 let experts = if expert_artifact == StepExpertArtifact::Nvfp4 {
1008 let gate_native = step_nvfp4_native(src, layer, "gate")?;
1009 let up_native = step_nvfp4_native(src, layer, "up")?;
1010 let down_native = step_nvfp4_native(src, layer, "down")?;
1011 StepTpExpertBank::Nvfp4(runtime.upload_tensor_parallel_nvfp4(
1012 step_nvfp4_bank(&gate_native),
1013 step_nvfp4_bank(&up_native),
1014 step_nvfp4_bank(&down_native),
1015 )?)
1016 } else {
1017 StepTpExpertBank::E4m3(runtime.upload_tensor_parallel(
1018 host_e4m3_bank(gate)?,
1019 host_e4m3_bank(up)?,
1020 host_e4m3_bank(down)?,
1021 )?)
1022 };
1023 eprintln!(
1024 "[step-tp] layer={layer} devices={:?} experts={} tp={} artifact={} \
1025 expert_layout=tensor-parallel transport={} native_p2p={} \
1026 performance_claim=false",
1027 selection.spec.devices,
1028 contract.expert_count,
1029 selection.spec.devices.len(),
1030 match expert_artifact {
1031 StepExpertArtifact::E4m3 => "e4m3",
1032 StepExpertArtifact::Nvfp4 => "nvfp4",
1033 },
1034 runtime.transport_label(),
1035 runtime.native_p2p(),
1036 );
1037 if let Some(limit) = activation_limit {
1038 eprintln!(
1039 "[step-tp-clamp] load layer={layer} routed_clamp={limit} \
1040 formula=min-silu-times-clamped-up performance_claim=false"
1041 );
1042 }
1043 Ok((
1044 None,
1045 Some(StepTpExps {
1046 runtime,
1047 experts,
1048 devices: selection.spec.devices,
1049 activation_limit,
1050 }),
1051 ))
1052 }
1053 }
1054}
1055
1056fn upload_step_bf16_column(
1057 runtime: &crate::tp::TpE4m3HostBounce,
1058 src: &dyn TensorSource,
1059 name: &str,
1060 expected_in: usize,
1061 expected_out: usize,
1062 f32_mirror: bool,
1063) -> Result<crate::tp::ResidentBf16ColumnParallel, Box<dyn std::error::Error>> {
1064 let tensor = src
1065 .find(name)
1066 .ok_or_else(|| format!("Step TP projection is missing {name}"))?;
1067 if tensor.ggml_type != GgmlType::BF16 {
1068 return Err(format!(
1069 "Step TP projection {name} must preserve checkpoint BF16 bytes, got {:?}",
1070 tensor.ggml_type
1071 )
1072 .into());
1073 }
1074 if tensor.ne.len() != 2 {
1075 return Err(format!(
1076 "Step TP projection {name} must be a 2-D matrix, got shape {:?}",
1077 tensor.ne
1078 )
1079 .into());
1080 }
1081 let matrix = crate::tp::Bf16Matrix {
1082 bytes: tensor.bytes.as_ref(),
1083 in_features: tensor.ne[0] as usize,
1084 out_features: tensor.ne[1] as usize,
1085 };
1086 matrix.validate()?;
1087 if matrix.in_features != expected_in || matrix.out_features != expected_out {
1088 return Err(format!(
1089 "Step TP projection {name} shape {}x{} != registered {expected_out}x{expected_in}",
1090 matrix.out_features, matrix.in_features
1091 )
1092 .into());
1093 }
1094 Ok(if f32_mirror {
1095 runtime.upload_step_bf16_column_parallel_f32_mirror(matrix)?
1096 } else {
1097 runtime.upload_step_bf16_column_parallel(matrix)?
1098 })
1099}
1100
1101fn upload_step_bf16_row(
1102 runtime: &crate::tp::TpE4m3HostBounce,
1103 src: &dyn TensorSource,
1104 name: &str,
1105 expected_in: usize,
1106 expected_out: usize,
1107 f32_mirror: bool,
1108) -> Result<crate::tp::ResidentStepBf16RowParallel, Box<dyn std::error::Error>> {
1109 let tensor = src
1110 .find(name)
1111 .ok_or_else(|| format!("Step TP projection is missing {name}"))?;
1112 if tensor.ggml_type != GgmlType::BF16 {
1113 return Err(format!(
1114 "Step TP projection {name} must preserve checkpoint BF16 bytes, got {:?}",
1115 tensor.ggml_type
1116 )
1117 .into());
1118 }
1119 if tensor.ne.len() != 2 {
1120 return Err(format!(
1121 "Step TP projection {name} must be a 2-D matrix, got shape {:?}",
1122 tensor.ne
1123 )
1124 .into());
1125 }
1126 let matrix = crate::tp::Bf16Matrix {
1127 bytes: tensor.bytes.as_ref(),
1128 in_features: tensor.ne[0] as usize,
1129 out_features: tensor.ne[1] as usize,
1130 };
1131 matrix.validate()?;
1132 if matrix.in_features != expected_in || matrix.out_features != expected_out {
1133 return Err(format!(
1134 "Step TP projection {name} shape {}x{} != registered {expected_out}x{expected_in}",
1135 matrix.out_features, matrix.in_features
1136 )
1137 .into());
1138 }
1139 Ok(if f32_mirror {
1140 runtime.upload_step_bf16_row_parallel_f32_mirror(matrix)?
1141 } else {
1142 runtime.upload_step_bf16_row_parallel(matrix)?
1143 })
1144}
1145
1146fn upload_step_tp_f32_copies(
1147 runtime: &crate::tp::TpE4m3HostBounce,
1148 src: &dyn TensorSource,
1149 name: &str,
1150 expected: usize,
1151) -> Result<Vec<CudaSlice<f32>>, Box<dyn std::error::Error>> {
1152 let tensor = src
1153 .find(name)
1154 .ok_or_else(|| format!("Step TP attention is missing {name}"))?;
1155 let values = memra_gguf::dequant::dequantize(
1156 tensor.ggml_type,
1157 &tensor.bytes,
1158 tensor.ne.iter().product::<u64>() as usize,
1159 );
1160 if values.len() != expected || values.iter().any(|value| !value.is_finite()) {
1161 return Err(format!(
1162 "Step TP attention {name} has {} finite values, expected {expected}",
1163 values.len()
1164 )
1165 .into());
1166 }
1167 let mut copies = Vec::with_capacity(runtime.devices().len());
1168 for rank in 0..runtime.devices().len() {
1169 let engine = runtime
1170 .rank_engine(rank)
1171 .ok_or_else(|| format!("Step TP attention has no engine for rank {rank}"))?;
1172 let _main = engine.gpu.enter_main()?;
1173 copies.push(engine.htod(&values)?);
1174 }
1175 Ok(copies)
1176}
1177
1178fn upload_step_tp_f32_row_shards(
1183 runtime: &crate::tp::TpE4m3HostBounce,
1184 src: &dyn TensorSource,
1185 name: &str,
1186 rows: usize,
1187 cols: usize,
1188) -> Result<Vec<CudaSlice<f32>>, Box<dyn std::error::Error>> {
1189 let tensor = src
1190 .find(name)
1191 .ok_or_else(|| format!("Step TP attention is missing {name}"))?;
1192 let values = memra_gguf::dequant::dequantize(
1193 tensor.ggml_type,
1194 &tensor.bytes,
1195 tensor.ne.iter().product::<u64>() as usize,
1196 );
1197 let world = runtime.devices().len();
1198 if values.len() != rows * cols || rows % world != 0 || values.iter().any(|v| !v.is_finite()) {
1199 return Err(format!(
1200 "Step TP attention {name} has {} finite values, expected {rows}x{cols} \
1201 (rows divisible by world {world})",
1202 values.len()
1203 )
1204 .into());
1205 }
1206 let local_rows = rows / world;
1207 let mut shards = Vec::with_capacity(world);
1208 for rank in 0..world {
1209 let engine = runtime
1210 .rank_engine(rank)
1211 .ok_or_else(|| format!("Step TP attention has no engine for rank {rank}"))?;
1212 let _main = engine.gpu.enter_main()?;
1213 shards
1214 .push(engine.htod(&values[rank * local_rows * cols..(rank + 1) * local_rows * cols])?);
1215 }
1216 Ok(shards)
1217}
1218
1219fn upload_step_tp_bf16_row_shards(
1221 runtime: &crate::tp::TpE4m3HostBounce,
1222 src: &dyn TensorSource,
1223 name: &str,
1224 rows: usize,
1225 cols: usize,
1226) -> Result<Vec<CudaSlice<u8>>, Box<dyn std::error::Error>> {
1227 let tensor = src
1228 .find(name)
1229 .ok_or_else(|| format!("Step TP attention is missing {name}"))?;
1230 if tensor.ggml_type != memra_gguf::GgmlType::BF16 || tensor.bytes.len() != rows * cols * 2 {
1231 return Err(format!(
1232 "Step TP attention {name} is not a bf16 [{rows}, {cols}] tensor ({} bytes, {:?})",
1233 tensor.bytes.len(),
1234 tensor.ggml_type
1235 )
1236 .into());
1237 }
1238 let world = runtime.devices().len();
1239 if rows % world != 0 {
1240 return Err(format!("{name} rows {rows} not divisible by world {world}").into());
1241 }
1242 let local = rows / world * cols * 2;
1243 let mut shards = Vec::with_capacity(world);
1244 for rank in 0..world {
1245 let engine = runtime
1246 .rank_engine(rank)
1247 .ok_or_else(|| format!("Step TP attention has no engine for rank {rank}"))?;
1248 let _main = engine.gpu.enter_main()?;
1249 shards.push(engine.htod_bytes(&tensor.bytes[rank * local..(rank + 1) * local])?);
1250 }
1251 Ok(shards)
1252}
1253
1254#[derive(Clone, Copy, Debug, PartialEq, Eq)]
1255enum StepTpAttentionPlacement {
1256 RankLocalGlobal,
1257 RankLocalSwa,
1258 OwnerSwa,
1259 OwnerTransportFallback,
1260}
1261
1262impl StepTpAttentionPlacement {
1263 fn resolve(native_p2p: bool, window: Option<u32>) -> Self {
1264 match (native_p2p, window.is_some()) {
1265 (true, true) => Self::RankLocalSwa,
1266 (false, true) => Self::OwnerSwa,
1267 (true, false) => Self::RankLocalGlobal,
1268 (false, false) => Self::OwnerTransportFallback,
1269 }
1270 }
1271
1272 fn is_rank_local(self) -> bool {
1273 matches!(self, Self::RankLocalGlobal | Self::RankLocalSwa)
1274 }
1275
1276 fn label(self) -> &'static str {
1277 match self {
1278 Self::RankLocalGlobal => "rank-local-global",
1279 Self::RankLocalSwa => "rank-local-swa-ring",
1280 Self::OwnerSwa => "owner-swa",
1281 Self::OwnerTransportFallback => "owner-transport-fallback",
1282 }
1283 }
1284}
1285
1286fn build_step_tp_qkv(
1287 e: &Engine,
1288 src: &dyn TensorSource,
1289 cfg: &ModelConfig,
1290 layer: usize,
1291 step_runtimes: &mut StepParallelRuntimeRegistry,
1292) -> Result<Option<StepTpQkv>, Box<dyn std::error::Error>> {
1293 let Some(spec) = step_runtimes.tp_spec(layer).cloned() else {
1294 return Ok(None);
1295 };
1296 let contract = crate::parallel::ModelParallelContract::from_model(cfg)?;
1297 if layer >= contract.trunk_layers {
1298 return Err(format!(
1299 "MEMRA_STEP_TP layer {layer} is outside Step trunk layers 0..{}",
1300 contract.trunk_layers
1301 )
1302 .into());
1303 }
1304 let owner = e.ctx().ordinal();
1305 if spec.devices.first().copied() != Some(owner) {
1306 return Err(format!(
1307 "MEMRA_STEP_TP layer {layer} owning PP device {owner} must be the first QKV rank, \
1308 got {:?}",
1309 spec.devices
1310 )
1311 .into());
1312 }
1313 let plan = contract.plan(crate::parallel::TopologyRequest {
1314 pipeline: 1,
1315 tensor: spec.devices.len(),
1316 expert_parallel: spec.devices.len() > 2,
1317 available_devices: spec.devices.len(),
1318 hardware: crate::parallel::HardwareTarget::RtxPro6000Blackwell,
1319 })?;
1320 for rank in 0..spec.devices.len() {
1321 plan.query_head_range(layer, rank).ok_or_else(|| {
1322 format!("Step TP layer {layer} has no query-head range for rank {rank}")
1323 })?;
1324 plan.kv_head_range(layer, rank)
1325 .ok_or_else(|| format!("Step TP layer {layer} has no KV-head range for rank {rank}"))?;
1326 }
1327 let native_p2p = step_runtimes.config.native_p2p;
1328 let ep_device_arithmetic = step_runtimes.config.ep_device_arithmetic;
1329 let f32_mirror = step_runtimes.config.f32_mirror;
1330 if ep_device_arithmetic && (!native_p2p || !matches!(spec.devices.len(), 4 | 8)) {
1331 return Err(
1332 "MEMRA_STEP_EP_DEVICE_ARITHMETIC=1 requires a MEMRA_STEP_TP TP4/TP8 \
1333 expert-owner layer and MEMRA_STEP_TP_NATIVE_P2P=1"
1334 .into(),
1335 );
1336 }
1337 let runtime = step_runtimes.runtime(&spec.devices, native_p2p, ep_device_arithmetic)?;
1338 let p = |suffix: &str| format!("blk.{layer}.{suffix}");
1339 let q = upload_step_bf16_column(
1340 &runtime,
1341 src,
1342 &p("attn_q.weight"),
1343 contract.hidden_size,
1344 contract.query_heads[layer] * contract.head_dim,
1345 f32_mirror,
1346 )?;
1347 let k = upload_step_bf16_column(
1348 &runtime,
1349 src,
1350 &p("attn_k.weight"),
1351 contract.hidden_size,
1352 contract.kv_heads[layer] * contract.head_dim,
1353 f32_mirror,
1354 )?;
1355 let v = upload_step_bf16_column(
1356 &runtime,
1357 src,
1358 &p("attn_v.weight"),
1359 contract.hidden_size,
1360 contract.kv_heads[layer] * contract.head_dim,
1361 f32_mirror,
1362 )?;
1363 let o = upload_step_bf16_row(
1364 &runtime,
1365 src,
1366 &p("attn_output.weight"),
1367 contract.query_heads[layer] * contract.head_dim,
1368 contract.hidden_size,
1369 f32_mirror,
1370 )?;
1371 let geometry = cfg.full_attention_geometry_at(layer as u32);
1372 let attention_placement =
1373 StepTpAttentionPlacement::resolve(runtime.native_p2p(), geometry.window);
1374 let attention = if attention_placement.is_rank_local() {
1375 let decode_input = if ep_device_arithmetic || crate::tp::step_tp_decode_v2_enabled()? {
1380 Some(std::sync::Mutex::new(
1381 runtime.allocate_replicated_device_rows(1, contract.hidden_size)?,
1382 ))
1383 } else {
1384 None
1385 };
1386 let gate_fused =
1389 crate::tp::step_tp_qkv_fused_enabled()? && src.find(&p("attn_gate.weight")).is_some();
1390 let gate_shards = if gate_fused && f32_mirror {
1391 Some(upload_step_tp_f32_row_shards(
1392 &runtime,
1393 src,
1394 &p("attn_gate.weight"),
1395 contract.query_heads[layer],
1396 contract.hidden_size,
1397 )?)
1398 } else {
1399 None
1400 };
1401 let gate_shards_bf16 = if gate_fused && !f32_mirror {
1402 Some(upload_step_tp_bf16_row_shards(
1403 &runtime,
1404 src,
1405 &p("attn_gate.weight"),
1406 contract.query_heads[layer],
1407 contract.hidden_size,
1408 )?)
1409 } else {
1410 None
1411 };
1412 Some(StepTpAttention {
1413 q_norm: upload_step_tp_f32_copies(
1414 &runtime,
1415 src,
1416 &p("attn_q_norm.weight"),
1417 contract.head_dim,
1418 )?,
1419 k_norm: upload_step_tp_f32_copies(
1420 &runtime,
1421 src,
1422 &p("attn_k_norm.weight"),
1423 contract.head_dim,
1424 )?,
1425 decode_input,
1426 gate_shards,
1427 gate_shards_bf16,
1428 })
1429 } else {
1430 None
1431 };
1432 if f32_mirror {
1433 eprintln!(
1434 "[step-tp-qkv] load layer={layer} devices={:?} projections=qkv \
1435 qkv_tensor_parallel=true attention_local=true kv_local=true output_local=true \
1436 transport={} native_p2p={} bf16_residency=f32-mirror \
1437 output=root-readback performance_claim=false",
1438 spec.devices,
1439 runtime.transport_label(),
1440 runtime.native_p2p(),
1441 );
1442 } else {
1443 eprintln!(
1444 "[step-tp-qkv] load layer={layer} devices={:?} projections=qkv \
1445 qkv_tensor_parallel=true attention_local=true kv_local=true output_local=true \
1446 transport={} native_p2p={} output=root-readback performance_claim=false",
1447 spec.devices,
1448 runtime.transport_label(),
1449 runtime.native_p2p(),
1450 );
1451 }
1452 eprintln!(
1453 "[step-tp-attn-plan] load layer={layer} devices={:?} \
1454 qkv_tensor_parallel=true attention_tensor_parallel={} kv_cache_distributed={} \
1455 attention_scope={} transport={} native_p2p={} replicated_decode_input_prepared={} \
1456 performance_claim=false",
1457 spec.devices,
1458 attention_placement.is_rank_local(),
1459 attention_placement.is_rank_local(),
1460 attention_placement.label(),
1461 runtime.transport_label(),
1462 runtime.native_p2p(),
1463 attention
1464 .as_ref()
1465 .is_some_and(|attention| attention.decode_input.is_some()),
1466 );
1467 if f32_mirror {
1468 eprintln!(
1469 "[step-tp-o] load layer={layer} devices={:?} projection=o \
1470 o_tensor_parallel=true attention_local=true kv_local=true \
1471 transport={} native_p2p={} reduction=global-tp8-block-order \
1472 bf16_residency=f32-mirror output=root-readback performance_claim=false",
1473 spec.devices,
1474 runtime.transport_label(),
1475 runtime.native_p2p(),
1476 );
1477 } else {
1478 eprintln!(
1479 "[step-tp-o] load layer={layer} devices={:?} projection=o \
1480 o_tensor_parallel=true attention_local=true kv_local=true \
1481 transport={} native_p2p={} reduction=global-tp8-block-order \
1482 output=root-readback performance_claim=false",
1483 spec.devices,
1484 runtime.transport_label(),
1485 runtime.native_p2p(),
1486 );
1487 }
1488 Ok(Some(StepTpQkv {
1489 runtime,
1490 q,
1491 k,
1492 v,
1493 o,
1494 attention,
1495 devices: spec.devices,
1496 layer,
1497 }))
1498}
1499
1500fn build_dev_exps(
1513 e: &Engine,
1514 resident: &mut ResidentPlan,
1515 il: usize,
1516 gate: &HostExps,
1517 up: &HostExps,
1518 down: &HostExps,
1519) -> Result<Option<crate::hybrid::DevExps>, Box<dyn std::error::Error>> {
1520 if !gate.is_uniform_layout() || !up.is_uniform_layout() || !down.is_uniform_layout() {
1523 return Ok(None);
1524 }
1525 let fp8_host = match (&gate.fp8_blk, &up.fp8_blk, &down.fp8_blk) {
1526 (None, None, None) => None,
1527 (Some(g), Some(u), Some(d)) => Some((g, u, d)),
1528 _ => {
1529 return Err("resident expert projections disagree on block-E4M3 scale carriage".into());
1530 }
1531 };
1532 let scale_bytes = fp8_host
1533 .map(|(g, u, d)| (g.scales.len() + u.scales.len() + d.scales.len()) * size_of::<f32>())
1534 .unwrap_or(0);
1535 let per_layer = gate.bytes.as_bytes().len()
1536 + up.bytes.as_bytes().len()
1537 + down.bytes.as_bytes().len()
1538 + scale_bytes;
1539 if gate.tiers.is_some() {
1540 return Ok(None); }
1542 let fits = resident.should_reside(e, il, per_layer);
1543 if !fits {
1544 return Ok(None);
1545 }
1546 use cudarc::driver::DevicePtr;
1547 let gu_il = std::env::var("MEMRA_MOE_GU_IL").as_deref() == Ok("1")
1548 && gate.out_f == up.out_f
1549 && gate.in_f == up.in_f
1550 && fp8_host.is_none();
1551 let n_expert = gate.n_expert;
1552 let (g, u) = if gu_il {
1553 let (rbg, rbu) = (gate.row_bytes, up.row_bytes);
1555 let n_rows = gate.out_f;
1556 let gb = gate.bytes.as_bytes();
1557 let ub = up.bytes.as_bytes();
1558 let mut il = vec![0u8; n_expert * n_rows * (rbg + rbu)];
1559 for ex in 0..n_expert {
1560 for o in 0..n_rows {
1561 let dst = (ex * n_rows + o) * (rbg + rbu);
1562 let sg = ex * gate.expert_stride + o * rbg;
1563 let su = ex * up.expert_stride + o * rbu;
1564 il[dst..dst + rbg].copy_from_slice(&gb[sg..sg + rbg]);
1565 il[dst + rbg..dst + rbg + rbu].copy_from_slice(&ub[su..su + rbu]);
1566 }
1567 }
1568 let ild = e.htod_bytes_padded(&il, 8)?;
1569 (ild, e.htod_bytes(&[0u8; 16])?)
1572 } else {
1573 (
1574 e.htod_bytes_padded(gate.bytes.as_bytes(), 8)?,
1575 e.htod_bytes_padded(up.bytes.as_bytes(), 8)?,
1576 )
1577 };
1578 let d = e.htod_bytes_padded(down.bytes.as_bytes(), 144)?;
1583 let fp8_blk = match fp8_host {
1584 Some((gate, up, down)) => {
1585 if e.fp8_blk_nan_count(&g)? != 0
1586 || e.fp8_blk_nan_count(&u)? != 0
1587 || e.fp8_blk_nan_count(&d)? != 0
1588 {
1589 return Err("native stacked block-E4M3 expert bank contains NaN codes".into());
1590 }
1591 Some(DevExpertFp8BlockScales {
1592 gate: DevExpertFp8ProjectionScales::upload(e, gate, n_expert)?,
1593 up: DevExpertFp8ProjectionScales::upload(e, up, n_expert)?,
1594 down: DevExpertFp8ProjectionScales::upload(e, down, n_expert)?,
1595 })
1596 }
1597 None => None,
1598 };
1599 let mut host = vec![0u64; 3 * n_expert];
1600 let (pg, pu, pd) = {
1601 let __s_e0 = e.stream();
1602 let (pg, _e0) = g.device_ptr(&__s_e0);
1603 let __s_e1 = e.stream();
1604 let (pu, _e1) = u.device_ptr(&__s_e1);
1605 let __s_e2 = e.stream();
1606 let (pd, _e2) = d.device_ptr(&__s_e2);
1607 (pg as u64, pu as u64, pd as u64)
1608 };
1609 for ex in 0..n_expert {
1610 if gu_il {
1611 let stride = gate.out_f * (gate.row_bytes + up.row_bytes);
1612 host[ex] = pg + (ex * stride) as u64;
1613 host[n_expert + ex] = pg + (ex * stride + gate.row_bytes) as u64;
1614 } else {
1615 host[ex] = pg + (ex * gate.expert_stride) as u64;
1616 host[n_expert + ex] = pu + (ex * up.expert_stride) as u64;
1617 }
1618 host[2 * n_expert + ex] = pd + (ex * down.expert_stride) as u64;
1619 }
1620 if gu_il {
1621 eprintln!("[moe] gate/up dev slab INTERLEAVED (MEMRA_MOE_GU_IL)");
1622 }
1623 let ptr_row = e.htod_u64(&host)?;
1624 Ok(Some(crate::hybrid::DevExps {
1625 gate: g,
1626 up: u,
1627 down: d,
1628 ptr_row,
1629 gu_il,
1630 dev: e.ctx().ordinal(),
1631 fp8_blk,
1632 }))
1633}
1634
1635pub struct FullAttnLayer {
1636 pub wq: GpuTensor,
1637 pub wk: GpuTensor,
1638 pub wv: GpuTensor,
1639 pub wo: GpuTensor,
1640 pub q_norm: GpuTensor,
1641 pub k_norm: GpuTensor,
1642 pub attn_gate: Option<GpuTensor>,
1653 pub step_tp_qkv: Option<StepTpQkv>,
1657}
1658
1659pub struct StepTpQkv {
1660 pub runtime: Arc<crate::tp::TpE4m3HostBounce>,
1661 pub q: crate::tp::ResidentBf16ColumnParallel,
1662 pub k: crate::tp::ResidentBf16ColumnParallel,
1663 pub v: crate::tp::ResidentBf16ColumnParallel,
1664 pub o: crate::tp::ResidentStepBf16RowParallel,
1665 pub attention: Option<StepTpAttention>,
1666 pub devices: Vec<usize>,
1667 pub layer: usize,
1668}
1669
1670pub struct StepTpAttention {
1671 pub q_norm: Vec<CudaSlice<f32>>,
1672 pub k_norm: Vec<CudaSlice<f32>>,
1673 pub decode_input: Option<std::sync::Mutex<crate::tp::ResidentReplicatedDeviceRows>>,
1674 pub gate_shards: Option<Vec<CudaSlice<f32>>>,
1677 pub gate_shards_bf16: Option<Vec<CudaSlice<u8>>>,
1679}
1680
1681#[derive(Clone, Copy, Debug, PartialEq, Eq)]
1682pub struct StepTpKvDeviceAdmission {
1683 pub device: usize,
1684 pub bytes: usize,
1685}
1686
1687#[derive(Clone, Copy, Debug)]
1691pub struct MlaGeom {
1692 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, }
1700
1701pub struct MlaAttnLayer {
1705 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,
1715}
1716
1717impl MlaAttnLayer {
1718 pub fn load(
1727 e: &Engine,
1728 src: &dyn TensorSource,
1729 il: u32,
1730 plan: &memra_gguf::model_plan::MlaAttentionPlan,
1731 ) -> Result<Self, Box<dyn std::error::Error>> {
1732 let memra_gguf::model_plan::MlaAttentionPlan::LatentKv {
1733 query_heads,
1734 q_lora_rank,
1735 kv_lora_rank,
1736 qk_head_dim,
1737 rope_head_dim,
1738 value_head_dim,
1739 ..
1740 } = plan
1741 else {
1742 return Err(format!(
1743 "native MLA loader has no compressed-KV implementation for block {il}"
1744 )
1745 .into());
1746 };
1747 let d_nope = qk_head_dim
1748 .checked_sub(*rope_head_dim)
1749 .ok_or("MLA rope head width exceeds total QK head width")?;
1750 let p = |s: &str| format!("blk.{il}.{s}");
1751 let geom = MlaGeom {
1752 n_head: *query_heads as usize,
1753 d_nope: d_nope as usize,
1754 d_rope: *rope_head_dim as usize,
1755 d_v: *value_head_dim as usize,
1756 kv_rank: *kv_lora_rank as usize,
1757 latent_dim: (*kv_lora_rank + *rope_head_dim) as usize,
1758 scale: 1.0 / (*qk_head_dim as f32).sqrt(),
1759 };
1760 let wq_a = load_t(e, src, &p("attn_q_a.weight"))?;
1761 let wq_b = load_t(e, src, &p("attn_q_b.weight"))?;
1762 let wkv_a = load_t(e, src, &p("attn_kv_a_mqa.weight"))?;
1763 let wk_b = load_t(e, src, &p("attn_k_b.weight"))?;
1764 let wv_b = load_t(e, src, &p("attn_v_b.weight"))?;
1765 let wo = load_t(e, src, &p("attn_output.weight"))?;
1766 let n_head = wq_b.out_features() / (geom.d_nope + geom.d_rope);
1768 assert_eq!(
1769 wq_b.out_features(),
1770 n_head * (geom.d_nope + geom.d_rope),
1771 "wq_b out {} not a multiple of qk_head_dim {}",
1772 wq_b.out_features(),
1773 geom.d_nope + geom.d_rope
1774 );
1775 assert_eq!(
1776 wq_a.in_features(),
1777 wkv_a.in_features(),
1778 "q_a/kv_a hidden mismatch"
1779 );
1780 assert_eq!(
1781 wq_b.in_features(),
1782 *q_lora_rank as usize,
1783 "wq_b in != q_lora_rank"
1784 );
1785 assert_eq!(
1786 n_head, geom.n_head,
1787 "MLA checkpoint head count != ModelPlan"
1788 );
1789 assert_eq!(
1790 wkv_a.out_features(),
1791 geom.latent_dim,
1792 "wkv_a out != kv_lora_rank + rope"
1793 );
1794 assert_eq!(
1795 wk_b.ne(),
1796 &[geom.d_nope as u64, geom.kv_rank as u64, n_head as u64],
1797 "attn_k_b must be the TRANSPOSED (nope, kv_rank, head) conversion split"
1798 );
1799 assert_eq!(
1800 wv_b.ne(),
1801 &[geom.kv_rank as u64, geom.d_v as u64, n_head as u64],
1802 "attn_v_b must be the (kv_rank, v, head) conversion split"
1803 );
1804 assert_eq!(
1805 wo.in_features(),
1806 n_head * geom.d_v,
1807 "wo in != n_head * v_head_dim"
1808 );
1809 Ok(MlaAttnLayer {
1810 wq_a,
1811 q_a_norm: load_t(e, src, &p("attn_q_a_norm.weight"))?,
1812 wq_b,
1813 wkv_a,
1814 kv_a_norm: load_t(e, src, &p("attn_kv_a_norm.weight"))?,
1815 wk_b,
1816 wv_b,
1817 wo,
1818 geom,
1819 })
1820 }
1821}
1822
1823#[track_caller]
1827pub(crate) fn mla_forward_unimplemented() -> ! {
1828 panic!(
1829 "Mixer::Mla has no forward arm yet — glm-dsa is loader-only in increment 2; \
1830 the CUDA forward lands in increment 4 (research/mla-bringup-20260801/DESIGN.md §4)"
1831 )
1832}
1833
1834pub struct LinearAttnLayer {
1835 pub geometry: memra_gguf::model_plan::GatedDeltaNetPlan,
1836 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, }
1846
1847pub enum Mixer {
1848 Full(FullAttnLayer),
1849 Linear(LinearAttnLayer),
1850 Mla(MlaAttnLayer),
1852}
1853
1854pub struct MoeWeights {
1861 pub gate_inp: GpuTensor, pub gate_inp_shexp: Option<GpuTensor>, pub exp_probs_b: Option<Vec<f32>>,
1867 pub exp_probs_b_dev: CudaSlice<f32>,
1868 pub active_experts: Option<Vec<bool>>,
1872 pub active_experts_dev: CudaSlice<u8>,
1873 pub gate_exps: HostExps, pub up_exps: HostExps, pub down_exps: HostExps, pub gate_shexp: Option<GpuTensor>,
1877 pub up_shexp: Option<GpuTensor>,
1878 pub down_shexp: Option<GpuTensor>,
1879 pub dev_exps: Option<DevExps>,
1886 pub step_ep: Option<StepEpExps>,
1890 pub step_tp: Option<StepTpExps>,
1894 pub dev_macros: cudarc::driver::CudaSlice<f32>,
1900 pub has_macros: bool,
1901}
1902
1903pub enum StepEpExpertBank {
1905 E4m3(crate::tp::ResidentExpertParallel),
1906 Nvfp4(crate::tp::ResidentNvfp4ExpertParallel),
1907}
1908
1909impl StepEpExpertBank {
1910 pub fn e4m3(&self) -> Result<&crate::tp::ResidentExpertParallel, String> {
1914 match self {
1915 Self::E4m3(bank) => Ok(bank),
1916 Self::Nvfp4(_) => Err(
1917 "Step grouped expert program reached an NVFP4 bank; this path is qualified \
1918 for the E4M3 artifact only"
1919 .to_string(),
1920 ),
1921 }
1922 }
1923}
1924
1925pub struct StepEpExps {
1926 pub runtime: Arc<crate::tp::TpE4m3HostBounce>,
1927 pub experts: StepEpExpertBank,
1928 pub devices: Vec<usize>,
1929 pub configured_by_tp: bool,
1930 pub activation_limit: Option<f32>,
1931 pub grouped_decode: Option<std::sync::Mutex<StepEpGroupedDecode>>,
1934}
1935
1936pub struct StepEpGroupedDecode {
1937 pub(crate) projection: crate::tp::PreparedStepGroupedExpertParallelGate,
1938 pub(crate) combine: crate::tp::PreparedPeerWeightedRouteCombine,
1939}
1940
1941#[derive(Default)]
1942pub(crate) struct StepEpGroupedPrefill {
1943 pub(crate) state: Option<StepEpGroupedPrefillState>,
1944}
1945
1946pub(crate) struct StepEpGroupedPrefillState {
1947 pub(crate) devices: Vec<usize>,
1948 pub(crate) grouped: StepEpGroupedDecode,
1949}
1950
1951pub enum StepTpExpertBank {
1953 E4m3(crate::tp::ResidentTensorParallel),
1954 Nvfp4(crate::tp::ResidentNvfp4TensorParallel),
1955}
1956
1957pub struct StepTpExps {
1958 pub runtime: Arc<crate::tp::TpE4m3HostBounce>,
1959 pub experts: StepTpExpertBank,
1960 pub devices: Vec<usize>,
1961 pub activation_limit: Option<f32>,
1964}
1965
1966impl MoeWeights {
1967 #[inline]
1968 pub fn has_uniform_expert_layout(&self) -> bool {
1969 self.gate_exps.is_uniform_layout()
1970 && self.up_exps.is_uniform_layout()
1971 && self.down_exps.is_uniform_layout()
1972 }
1973
1974 #[inline]
1975 pub fn active_count(&self) -> usize {
1976 self.active_experts
1977 .as_ref()
1978 .map(|mask| mask.iter().filter(|&&active| active).count())
1979 .unwrap_or(self.gate_exps.n_expert)
1980 }
1981}
1982
1983pub struct DevExps {
1986 pub gate: CudaSlice<u8>,
1987 pub up: CudaSlice<u8>,
1988 pub down: CudaSlice<u8>,
1989 pub ptr_row: CudaSlice<u64>,
1991 pub dev: usize,
1999 pub gu_il: bool,
2005 pub fp8_blk: Option<DevExpertFp8BlockScales>,
2009}
2010
2011pub struct DevExpertFp8BlockScales {
2012 pub gate: DevExpertFp8ProjectionScales,
2013 pub up: DevExpertFp8ProjectionScales,
2014 pub down: DevExpertFp8ProjectionScales,
2015}
2016
2017pub struct DevExpertFp8ProjectionScales {
2018 pub scales: CudaSlice<f32>,
2019 pub rows: usize,
2020 pub cols: usize,
2021 pub expert_stride: usize,
2022}
2023
2024impl DevExpertFp8ProjectionScales {
2025 fn validate(
2026 host: &crate::model::HostExpertFp8BlockScales,
2027 n_expert: usize,
2028 ) -> Result<(), String> {
2029 if host.expert_stride == 0 {
2030 return Err("block-E4M3 expert scale stride must be nonzero".into());
2031 }
2032 if host.rows * host.cols != host.expert_stride {
2033 return Err(format!(
2034 "block-E4M3 expert scale stride mismatch: {}x{} != {}",
2035 host.rows, host.cols, host.expert_stride
2036 ));
2037 }
2038 let want = n_expert
2039 .checked_mul(host.expert_stride)
2040 .ok_or("block-E4M3 expert scale slab length overflow")?;
2041 if host.scales.len() != want {
2042 return Err(format!(
2043 "block-E4M3 scale slab length mismatch: got {}, want {n_expert}x{}={want}",
2044 host.scales.len(),
2045 host.expert_stride
2046 ));
2047 }
2048 Ok(())
2049 }
2050
2051 fn upload(
2052 e: &Engine,
2053 host: &crate::model::HostExpertFp8BlockScales,
2054 n_expert: usize,
2055 ) -> Result<Self, Box<dyn std::error::Error>> {
2056 Self::validate(host, n_expert)?;
2057 Ok(Self {
2058 scales: e.htod(&host.scales)?,
2059 rows: host.rows,
2060 cols: host.cols,
2061 expert_stride: host.expert_stride,
2062 })
2063 }
2064}
2065
2066pub enum Ffn {
2068 Dense {
2069 ffn_gate: GpuTensor,
2070 ffn_up: GpuTensor,
2071 ffn_down: GpuTensor,
2072 },
2073 Moe(MoeWeights),
2074}
2075
2076pub struct HybridLayer {
2077 pub attn_norm: GpuTensor,
2078 pub post_attn_norm: GpuTensor, pub mixer: Mixer,
2080 pub ffn: Ffn,
2081 pub gemma4: Option<Gemma4LayerBits>,
2082}
2083
2084pub struct Gemma4LayerBits {
2088 pub ffn_norm: GpuTensor, pub post_ffw_norm: GpuTensor, pub moe_bits: Option<Gemma4MoeBits>,
2093 pub layer_scale: f32, pub e4b: Option<Gemma4E4bLayer>,
2096}
2097
2098pub struct Gemma4E4bLayer {
2103 pub inp_gate: GpuTensor, pub proj: GpuTensor, pub post_norm: GpuTensor, pub qkv_cat: Option<GpuTensor>,
2110 pub kv_share: Option<u32>,
2114}
2115
2116pub struct Gemma4E4bModel {
2120 pub tok_tbl_gpu: std::sync::OnceLock<CudaSlice<u8>>,
2123 pub tok_embd_bytes: Vec<u8>,
2124 pub tok_embd_qt: i32,
2125 pub tok_embd_row_bytes: usize,
2126 pub model_proj: GpuTensor, pub proj_norm: GpuTensor, pub n_epl: usize,
2129}
2130
2131pub struct Gemma4MoeBits {
2132 pub post_ffw_norm_1: GpuTensor, pub pre_ffw_norm_2: GpuTensor, pub post_ffw_norm_2: GpuTensor, pub shared_gate: GpuTensor,
2136 pub shared_up: GpuTensor,
2137 pub shared_down: GpuTensor,
2138 pub router_scale_pre: CudaSlice<f32>,
2143 pub per_expert_scale: Vec<f32>, pub per_expert_scale_d: CudaSlice<f32>, }
2146
2147fn load_mtp_head_maybe_nvfp4(
2160 e: &Engine,
2161 src: &dyn TensorSource,
2162 name: &str,
2163) -> Result<Option<GpuTensor>, Box<dyn std::error::Error>> {
2164 if !{
2165 static ENV: std::sync::OnceLock<Option<bool>> = std::sync::OnceLock::new();
2166 crate::step37_door(&ENV, "MEMRA_MTP_HEAD_NVFP4")
2167 } {
2168 return load_opt(e, src, name);
2169 }
2170 let Some(v) = src.find(name) else {
2171 return Ok(None);
2172 };
2173 if !matches!(v.ggml_type, GgmlType::BF16) || v.ne[0] % 64 != 0 {
2174 return load_opt(e, src, name);
2175 }
2176 let vals: Vec<f32> = v
2177 .bytes
2178 .chunks_exact(2)
2179 .map(|c| f32::from_bits((u16::from_le_bytes([c[0], c[1]]) as u32) << 16))
2180 .collect();
2181 let blocks = memra_gguf::nvfp4_repack::f32_to_nvfp4(&vals);
2182 eprintln!(
2183 "[mtp-head] {name}: BF16 -> NVFP4 ({} MiB, was {} MiB)",
2184 blocks.len() >> 20,
2185 v.bytes.len() >> 20
2186 );
2187 Ok(Some(GpuTensor::from_quant_bytes(
2188 e,
2189 &blocks,
2190 GgmlType::NVFP4,
2191 v.ne[0],
2192 v.ne[1],
2193 1.0,
2194 )?))
2195}
2196
2197pub(crate) fn frspec_trim_own_head_name(n_trunk: usize) -> String {
2202 format!("blk.{n_trunk}.nextn.shared_head_head.weight")
2203}
2204
2205pub struct DflashTrimHead {
2216 pub head: GpuTensor,
2219 pub d2t: Vec<u32>,
2221}
2222
2223fn frspec_read_d2t(path: &str) -> Result<Vec<u32>, Box<dyn std::error::Error>> {
2228 Ok(if path.ends_with(".txt") {
2229 std::fs::read_to_string(path)?
2230 .lines()
2231 .filter_map(|l| l.trim().parse::<u32>().ok())
2232 .collect()
2233 } else {
2234 let tg = GgufFile::open(path)?;
2235 let d2t_t = tg
2236 .find("d2t")
2237 .expect("MEMRA_FRSPEC_TRIM file has no d2t tensor");
2238 let d2t_bytes = tg.tensor_data(d2t_t);
2239 match d2t_t.ggml_type {
2240 GgmlType::I32 => d2t_bytes
2241 .chunks_exact(4)
2242 .map(|c| i32::from_le_bytes(c.try_into().unwrap()) as u32)
2243 .collect(),
2244 GgmlType::I64 => d2t_bytes
2245 .chunks_exact(8)
2246 .map(|c| i64::from_le_bytes(c.try_into().unwrap()) as u32)
2247 .collect(),
2248 other => panic!("d2t must be I32/I64, got {other:?}"),
2249 }
2250 })
2251}
2252
2253fn frspec_gather_trimmed_head(
2261 e: &Engine,
2262 v: &memra_gguf::source::TensorView<'_>,
2263 d2t: &[u32],
2264 want_nvfp4_env: bool,
2265 macro_scale: f32,
2266) -> Result<(GpuTensor, Option<(usize, usize)>), Box<dyn std::error::Error>> {
2267 let out_f = v.ne[1] as usize;
2268 let row_bytes = v.bytes.len() / out_f;
2269 assert!(
2270 d2t.iter().all(|&t| (t as usize) < out_f),
2271 "d2t token id >= lm_head rows {out_f}"
2272 );
2273 let mut gathered = Vec::with_capacity(d2t.len() * row_bytes);
2274 for &t in d2t {
2275 let off = t as usize * row_bytes;
2276 gathered.extend_from_slice(&v.bytes[off..off + row_bytes]);
2277 }
2278 let want_nvfp4 = want_nvfp4_env && matches!(v.ggml_type, GgmlType::BF16) && v.ne[0] % 64 == 0;
2279 if want_nvfp4 {
2280 let in_f = v.ne[0] as usize;
2281 let vals: Vec<f32> = gathered
2282 .chunks_exact(2)
2283 .map(|c| f32::from_bits((u16::from_le_bytes([c[0], c[1]]) as u32) << 16))
2284 .collect();
2285 debug_assert_eq!(vals.len(), d2t.len() * in_f);
2286 let blocks = memra_gguf::nvfp4_repack::f32_to_nvfp4(&vals);
2287 let sizes = (blocks.len(), gathered.len());
2288 let trimmed = GpuTensor::from_quant_bytes(
2289 e,
2290 &blocks,
2291 GgmlType::NVFP4,
2292 v.ne[0],
2293 d2t.len() as u64,
2294 1.0,
2295 )?;
2296 Ok((trimmed, Some(sizes)))
2297 } else {
2298 let trimmed = match v.ggml_type {
2299 GgmlType::BF16 => GpuTensor::FloatBf16 {
2300 data: e.htod_bytes(&gathered)?,
2301 ne: vec![v.ne[0], d2t.len() as u64],
2302 },
2303 GgmlType::F32 => GpuTensor::Float {
2304 data: e.htod(
2305 &gathered
2306 .chunks_exact(4)
2307 .map(|c| f32::from_le_bytes(c.try_into().unwrap()))
2308 .collect::<Vec<f32>>(),
2309 )?,
2310 ne: vec![v.ne[0], d2t.len() as u64],
2311 },
2312 _ => GpuTensor::from_quant_bytes(
2313 e,
2314 &gathered,
2315 v.ggml_type,
2316 v.ne[0],
2317 d2t.len() as u64,
2318 macro_scale,
2319 )?,
2320 };
2321 Ok((trimmed, None))
2322 }
2323}
2324
2325pub struct MtpHead {
2326 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>>,
2340 pub d2t_from_target_head: bool,
2344 pub geom: Option<DraftGeom>,
2350 pub step35: Option<Step35MtpGeom>,
2355}
2356
2357#[derive(Debug, Clone)]
2371pub struct Step35MtpGeom {
2372 pub il: u32,
2374 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>,
2385}
2386
2387impl Step35MtpGeom {
2388 pub fn from_plan(layer: &memra_gguf::model_plan::LayerPlan) -> Result<Self, String> {
2390 use memra_gguf::model_plan::{ActivationPlan, AttentionPlan};
2391
2392 let (attention, window) = match &layer.attention {
2393 AttentionPlan::Full(attention) => (attention, None),
2394 AttentionPlan::SlidingWindow { attention, window } => (attention, Some(*window)),
2395 other => {
2396 return Err(format!(
2397 "MTP block {} has unsupported tuned attention {other:?}",
2398 layer.index
2399 ));
2400 }
2401 };
2402 if attention.output_gate != memra_gguf::config::AttentionGateKind::SeparateHead {
2403 return Err(format!(
2404 "MTP block {} does not declare a separate attention gate",
2405 layer.index
2406 ));
2407 }
2408 let activation = match &layer.mlp {
2409 MlpPlan::Dense(dense) => &dense.activation,
2410 MlpPlan::Moe(moe) => &moe.activation,
2411 };
2412 let clamp_shexp = match activation {
2413 ActivationPlan::SwiGluClamped { limit } if *limit > 0.0 => Some(*limit),
2414 _ => None,
2415 };
2416 Ok(Step35MtpGeom {
2417 il: layer.index,
2418 n_head: attention.query_heads as usize,
2419 n_head_kv: attention.kv_heads as usize,
2420 n_rot: attention.rope.dimensions as usize,
2421 rope_base: attention.rope.base,
2422 swa: window.is_some(),
2423 window: window.unwrap_or(0) as usize,
2424 clamp_shexp,
2425 })
2426 }
2427}
2428
2429pub struct DraftGeom {
2431 pub d_inner: usize, pub n_head: usize, pub n_head_kv: usize,
2434 pub out_up: GpuTensor, }
2436
2437pub fn draft_head_tensor(has: impl Fn(&str) -> bool, n: u32) -> String {
2446 let own = format!("blk.{n}.nextn.shared_head_head.weight");
2447 if has(&own) {
2448 return own;
2449 }
2450 let legacy = format!("blk.{n}.nextn.shared_head.weight");
2453 if has(&legacy) {
2454 return legacy;
2455 }
2456 "output.weight".to_string()
2458}
2459
2460impl MtpHead {
2461 pub fn load_draft(
2468 e: &Engine,
2469 g: &GgufFile,
2470 main_cfg: &ModelConfig,
2471 ) -> Result<Self, Box<dyn std::error::Error>> {
2472 let src = GgufSource(g);
2473 let dcfg = src.try_config().map_err(std::io::Error::other)?;
2474 let draft_plan = match memra_gguf::model_packs::for_config(&dcfg) {
2475 Some(pack) => pack.compile_plan(&dcfg)?,
2476 None => memra_gguf::model_plan::ModelPlan::compile(&dcfg)?,
2477 };
2478 let main_plan = match memra_gguf::model_packs::for_config(main_cfg) {
2479 Some(pack) => pack.compile_plan(main_cfg)?,
2480 None => memra_gguf::model_plan::ModelPlan::compile(main_cfg)?,
2481 };
2482 if dcfg.nextn_predict_layers == 0 {
2487 return Err(format!(
2488 "draft GGUF has no nextn_predict_layers (arch {:?}) — not a NextN/MTP regime \
2489 draft; gemma assistant drafters attach via MEMRA_DRAFT, not '+draft'",
2490 g.arch()
2491 )
2492 .into());
2493 }
2494 let n = dcfg.n_layer - dcfg.nextn_predict_layers;
2495 let draft_block = draft_plan
2496 .mtp_blocks
2497 .iter()
2498 .find(|block| block.layer.index == n)
2499 .ok_or_else(|| format!("draft ModelPlan has no MTP block {n}"))?;
2500 let p = |s: &str| format!("blk.{n}.{s}");
2501
2502 let student = src.has(&p("nextn.out_up.weight"));
2506 assert_eq!(dcfg.n_embd, main_cfg.n_embd, "draft n_embd != model n_embd");
2507 assert_eq!(
2508 dcfg.head_dim_k, main_cfg.head_dim_k,
2509 "draft head_dim != model head_dim"
2510 );
2511 let main_sliding_gated = crate::plan_backend::decode_batch_program(&main_plan)
2518 == crate::plan_backend::DecodeBatchProgram::SlidingGatedMoe;
2519 let draft_sliding_gated = crate::plan_backend::decode_batch_program(&draft_plan)
2520 == crate::plan_backend::DecodeBatchProgram::SlidingGatedMoe;
2521 let step35 = match (main_sliding_gated, draft_sliding_gated) {
2522 (true, true) => {
2523 let g = Step35MtpGeom::from_plan(&draft_block.layer)?;
2524 let out_f = |t: &str| -> Option<usize> {
2526 src.find(&p(t))
2527 .and_then(|v| v.ne.get(1).copied())
2528 .map(|x| x as usize)
2529 };
2530 let hd = dcfg.head_dim_k as usize;
2531 let wq_out =
2532 out_f("attn_q.weight").ok_or("step35 draft block has no attn_q.weight")?;
2533 assert_eq!(
2534 wq_out,
2535 g.n_head * hd,
2536 "step35 draft blk.{n}: attn_q out {wq_out} != n_head({}) * head_dim({hd}) — \
2537 the draft file's head_count array disagrees with its own tensors",
2538 g.n_head
2539 );
2540 let wg_out = out_f("attn_gate.weight")
2543 .ok_or("step35 draft block has no attn_gate.weight (head-wise gate)")?;
2544 assert_eq!(
2545 wg_out, g.n_head,
2546 "step35 draft blk.{n}: attn_gate out {wg_out} != n_head({})",
2547 g.n_head
2548 );
2549 assert_eq!(
2553 g.n_head_kv, main_cfg.n_head_kv as usize,
2554 "step35 draft blk.{n} KV heads {} != trunk n_head_kv {} — the MTP scratch \
2555 rows are sized from the trunk cfg, so a differing draft KV width would \
2556 write past the row",
2557 g.n_head_kv, main_cfg.n_head_kv
2558 );
2559 eprintln!(
2560 "[mtp-draft] step35 MTP geometry blk.{n}: n_head={} n_head_kv={} n_rot={} \
2561 rope_base={:.0} swa={} window={}",
2562 g.n_head, g.n_head_kv, g.n_rot, g.rope_base, g.swa, g.window
2563 );
2564 Some(g)
2565 }
2566 (true, false) => {
2567 return Err(format!(
2568 "MEMRA_MTP_DRAFT operations are incompatible with the model's \
2569 sliding-gated-MoE program (draft arch {:?})",
2570 g.arch()
2571 )
2572 .into());
2573 }
2574 (false, true) => {
2575 return Err(
2576 "MEMRA_MTP_DRAFT requires sliding-gated-MoE operations but the model does not"
2577 .into(),
2578 );
2579 }
2580 (false, false) => None,
2581 };
2582 if step35.is_none() && !student {
2583 assert_eq!(dcfg.n_head, main_cfg.n_head, "draft n_head != model n_head");
2586 assert_eq!(
2587 dcfg.n_head_kv, main_cfg.n_head_kv,
2588 "draft n_head_kv != model n_head_kv"
2589 );
2590 }
2591
2592 let head_name = draft_head_tensor(|t| src.has(t), n);
2619 let head = load_t(e, &src, &head_name)?;
2620 let head_norm = match load_opt(e, &src, &p("nextn.shared_head_norm.weight"))? {
2621 Some(t) => Some(t),
2622 None => load_opt(e, &src, "output_norm.weight")?,
2623 };
2624
2625 let d2t: Option<Vec<u32>> = g.find("d2t").map(|t| {
2627 let bytes = g.tensor_data(t);
2628 match t.ggml_type {
2629 GgmlType::I32 => bytes
2630 .chunks_exact(4)
2631 .map(|c| i32::from_le_bytes(c.try_into().unwrap()) as u32)
2632 .collect(),
2633 GgmlType::I64 => bytes
2634 .chunks_exact(8)
2635 .map(|c| i64::from_le_bytes(c.try_into().unwrap()) as u32)
2636 .collect(),
2637 other => panic!("d2t must be I32/I64, got {other:?}"),
2638 }
2639 });
2640 if let Some(map) = &d2t {
2641 assert_eq!(
2642 map.len(),
2643 head.out_features(),
2644 "d2t len {} != draft head rows {}",
2645 map.len(),
2646 head.out_features()
2647 );
2648 let n_vocab = main_cfg.n_vocab as u64;
2649 assert!(
2650 map.iter().all(|&t| (t as u64) < n_vocab),
2651 "d2t contains token id >= model n_vocab {n_vocab}"
2652 );
2653 }
2654 let eh_proj = load_t(e, &src, &p("nextn.eh_proj.weight"))?;
2655 assert_eq!(
2658 eh_proj.in_features(),
2659 2 * main_cfg.n_embd as usize,
2660 "eh_proj in dim != 2*n_embd"
2661 );
2662 let geom = if student {
2663 let out_up = load_t(e, &src, &p("nextn.out_up.weight"))?;
2664 let d_inner = eh_proj.out_features();
2665 assert_eq!(
2666 out_up.out_features(),
2667 main_cfg.n_embd as usize,
2668 "out_up out dim != n_embd"
2669 );
2670 assert_eq!(
2671 out_up.in_features(),
2672 d_inner,
2673 "out_up in dim != eh_proj out dim (d_inner)"
2674 );
2675 assert!(
2676 dcfg.n_head >= 1 && dcfg.n_head_kv >= 1 && dcfg.n_head % dcfg.n_head_kv == 0,
2677 "student head counts malformed ({}/{})",
2678 dcfg.n_head,
2679 dcfg.n_head_kv
2680 );
2681 Some(DraftGeom {
2682 d_inner,
2683 n_head: dcfg.n_head as usize,
2684 n_head_kv: dcfg.n_head_kv as usize,
2685 out_up,
2686 })
2687 } else {
2688 None
2689 };
2690 let blk_prefix = format!("blk.{n}.");
2694 let head_src = head_name.strip_prefix(&blk_prefix).unwrap_or(&head_name);
2695 eprintln!(
2696 "[mtp-draft] external draft head: blk.{n}, source={}, head_vocab={}{}{}",
2697 head_src,
2698 head.out_features(),
2699 if d2t.is_some() {
2700 " (trimmed, d2t map)"
2701 } else {
2702 " (full)"
2703 },
2704 match &geom {
2705 Some(g) => format!(
2706 " (student d_inner={} heads={}/{})",
2707 g.d_inner, g.n_head, g.n_head_kv
2708 ),
2709 None => String::new(),
2710 }
2711 );
2712
2713 let mut resident = ResidentPlan::unsharded(e, &src, &dcfg);
2714 let mut step_runtimes = StepParallelRuntimeRegistry::default();
2715 Ok(MtpHead {
2716 enorm: load_t(e, &src, &p("nextn.enorm.weight"))?,
2717 hnorm: load_t(e, &src, &p("nextn.hnorm.weight"))?,
2718 eh_proj,
2719 attn_norm: load_t(e, &src, &p("attn_norm.weight"))?,
2720 post_attn_norm: load_opt(e, &src, &p("post_attention_norm.weight"))?
2721 .or(load_opt(e, &src, &p("ffn_norm.weight"))?)
2722 .expect("draft NextN block needs post_attention_norm or ffn_norm"),
2723 mixer: load_mixer_kind(
2724 e,
2725 &src,
2726 &dcfg,
2727 n,
2728 &draft_block.layer.attention,
2729 &mut step_runtimes,
2730 )?,
2731 ffn: load_ffn(
2732 e,
2733 &src,
2734 &dcfg,
2735 &draft_block.layer.mlp,
2736 n,
2737 None,
2738 &mut resident,
2739 &mut step_runtimes,
2740 )?,
2741 shared_head_norm: head_norm,
2742 shared_head_head: Some(head),
2743 d2t,
2744 d2t_from_target_head: false,
2745 geom,
2746 step35,
2747 })
2748 }
2749}
2750
2751pub struct GemmaAux {
2753 pub rope_freqs: Option<Vec<(usize, CudaSlice<f32>)>>,
2756 pub ones: Vec<(usize, CudaSlice<f32>)>,
2759 pub suppress_d: Option<(CudaSlice<i32>, usize)>,
2762 pub e4b: Option<Gemma4E4bModel>,
2764}
2765
2766impl GemmaAux {
2767 pub fn rope_freqs(&self, e: &Engine) -> Option<&CudaSlice<f32>> {
2768 self.rope_freqs.as_ref().map(|copies| {
2769 let dev = e.ctx().ordinal();
2770 &copies
2771 .iter()
2772 .find(|(d, _)| *d == dev)
2773 .unwrap_or_else(|| panic!("gemma4 rope_freqs has no local copy for device {dev}"))
2774 .1
2775 })
2776 }
2777
2778 pub fn ones(&self, e: &Engine) -> &CudaSlice<f32> {
2779 let dev = e.ctx().ordinal();
2780 &self
2781 .ones
2782 .iter()
2783 .find(|(d, _)| *d == dev)
2784 .unwrap_or_else(|| panic!("gemma4 ones has no local copy for device {dev}"))
2785 .1
2786 }
2787}
2788
2789pub struct Step35Aux {
2792 pub rope_freqs: Option<Vec<(usize, CudaSlice<f32>)>>,
2798}
2799
2800impl Step35Aux {
2801 pub fn rope_freqs(&self, e: &Engine) -> Option<&CudaSlice<f32>> {
2802 self.rope_freqs.as_ref().map(|copies| {
2803 let dev = e.ctx().ordinal();
2804 &copies
2805 .iter()
2806 .find(|(d, _)| *d == dev)
2807 .unwrap_or_else(|| panic!("step35 rope_freqs has no local copy for device {dev}"))
2808 .1
2809 })
2810 }
2811}
2812
2813pub struct HybridModel {
2814 pub cfg: ModelConfig,
2815 pub plan: memra_gguf::model_plan::ModelPlan,
2816 pub rewrite_qualifications: Option<memra_gguf::execution_manifest::RewriteQualifications>,
2817 pub embd: EmbedHost,
2818 pub output_norm: GpuTensor,
2819 pub output: GpuTensor,
2820 pub layers: Vec<HybridLayer>,
2821 pub mtp: Option<MtpHead>, pub mtp_extra: Vec<MtpHead>,
2825 pub dflash_trim: Option<DflashTrimHead>,
2829 pub embd_gpu: std::sync::OnceLock<cudarc::driver::CudaSlice<u8>>,
2832 pub gemma4_aux: Option<GemmaAux>,
2833 pub step35_aux: Option<Step35Aux>,
2835 pub prime_slabs: std::sync::Mutex<
2843 std::collections::HashMap<
2844 usize,
2845 std::sync::Arc<std::sync::Mutex<crate::hybrid_forward::PrimeSlabs>>,
2846 >,
2847 >,
2848 pub(crate) dspark_vgraphs: std::sync::Mutex<Option<crate::spec::DsparkVerifyGraphs>>,
2861 pub(crate) step_grouped_prefill: std::sync::Mutex<StepEpGroupedPrefill>,
2867 pub(crate) step35_token_graph:
2870 std::sync::Mutex<Option<crate::hybrid_forward::Step35TokenGraphState>>,
2871 pub(crate) draft_state_bytes: std::sync::atomic::AtomicUsize,
2880}
2881
2882impl HybridModel {
2883 pub fn install_rewrite_bundle(
2884 &mut self,
2885 bundle: &std::path::Path,
2886 ) -> Result<(), Box<dyn std::error::Error>> {
2887 self.rewrite_qualifications = Some(
2888 memra_gguf::execution_manifest::RewriteQualifications::load(bundle, &self.plan)
2889 .map_err(|error| format!("rewrite qualification: {error}"))?,
2890 );
2891 Ok(())
2892 }
2893
2894 pub fn rewrite_allowed(&self, surface: memra_gguf::execution_manifest::RewriteSurface) -> bool {
2895 self.rewrite_qualifications
2896 .as_ref()
2897 .is_none_or(|qualifications| qualifications.allows(surface))
2898 }
2899
2900 pub fn record_draft_state_bytes(&self, observed: usize) -> Option<usize> {
2905 use std::sync::atomic::Ordering;
2906 let prev = self
2907 .draft_state_bytes
2908 .fetch_max(observed, Ordering::Relaxed);
2909 (observed > prev).then_some(observed)
2910 }
2911
2912 pub fn draft_session_admission_bytes(&self) -> usize {
2918 self.draft_state_bytes
2919 .load(std::sync::atomic::Ordering::Relaxed)
2920 }
2921
2922 pub fn step_tp_unmaterialized_kv_bytes(
2928 &self,
2929 cache: Option<&crate::cache::Cache>,
2930 capacity: usize,
2931 ) -> Result<Vec<StepTpKvDeviceAdmission>, String> {
2932 if let Some(cache) = cache
2933 && cache.tp_kv.len() < self.layers.len()
2934 {
2935 return Err(format!(
2936 "Step TP admission cache has {} layers, model trunk has {}",
2937 cache.tp_kv.len(),
2938 self.layers.len()
2939 ));
2940 }
2941
2942 let mut by_device: HashMap<usize, usize> = HashMap::new();
2943 for (layer, weights) in self.layers.iter().enumerate() {
2944 let Mixer::Full(attention) = &weights.mixer else {
2945 continue;
2946 };
2947 let Some(tp) = attention
2948 .step_tp_qkv
2949 .as_ref()
2950 .filter(|tp| tp.attention.is_some())
2951 else {
2952 continue;
2953 };
2954 if cache.is_some_and(|cache| cache.tp_kv[layer].is_some()) {
2955 continue;
2956 }
2957 let geometry = self.cfg.full_attention_geometry_at(layer as u32);
2958 let shape = crate::cache::tp_kv_rank_allocation_shape(
2959 geometry.n_head_kv as usize * geometry.head_dim_k as usize,
2960 geometry.n_head_kv as usize * geometry.head_dim_v as usize,
2961 tp.devices.len(),
2962 )?;
2963 let physical_rows = geometry
2964 .window
2965 .map(|window| crate::cache::swa_ring_rows(window as usize, capacity))
2966 .unwrap_or(capacity);
2967 let bytes = shape.allocation_bytes(physical_rows);
2968 for &device in &tp.devices {
2969 let total = by_device.entry(device).or_default();
2970 *total = total.saturating_add(bytes);
2971 }
2972 }
2973
2974 let mut out: Vec<_> = by_device
2975 .into_iter()
2976 .map(|(device, bytes)| StepTpKvDeviceAdmission { device, bytes })
2977 .collect();
2978 out.sort_unstable_by_key(|charge| charge.device);
2979 Ok(out)
2980 }
2981
2982 pub fn step_tp_rank_engine(&self, device: usize) -> Option<&Engine> {
2984 self.layers.iter().find_map(|weights| {
2985 let Mixer::Full(attention) = &weights.mixer else {
2986 return None;
2987 };
2988 let tp = attention.step_tp_qkv.as_ref()?;
2989 let rank = tp
2990 .runtime
2991 .devices()
2992 .iter()
2993 .position(|&rank| rank == device)?;
2994 tp.runtime.rank_engine(rank)
2995 })
2996 }
2997
2998 pub(crate) fn step_tp_runtime_for_layer(
2999 &self,
3000 layer: usize,
3001 ) -> Option<&crate::tp::TpE4m3HostBounce> {
3002 let Mixer::Full(attention) = &self.layers.get(layer)?.mixer else {
3003 return None;
3004 };
3005 let tp = attention.step_tp_qkv.as_ref()?;
3006 tp.attention.as_ref()?;
3007 Some(tp.runtime.as_ref())
3008 }
3009
3010 pub fn decode_batch_program(&self) -> crate::plan_backend::DecodeBatchProgram {
3011 crate::plan_backend::decode_batch_program(&self.plan)
3012 }
3013
3014 pub fn uses_gemma_program(&self) -> bool {
3015 self.decode_batch_program() == crate::plan_backend::DecodeBatchProgram::Gemma
3016 }
3017
3018 pub fn uses_sliding_gated_moe_program(&self) -> bool {
3019 self.decode_batch_program() == crate::plan_backend::DecodeBatchProgram::SlidingGatedMoe
3020 }
3021
3022 pub fn has_plan_operation(&self, operation: memra_gguf::model_plan::OperationKind) -> bool {
3023 self.plan.trunk_operations().contains(&operation)
3024 }
3025
3026 pub fn load(e: &Engine, g: &GgufFile) -> Result<Self, Box<dyn std::error::Error>> {
3028 Self::load_from_source(e, &GgufSource(g))
3029 }
3030
3031 pub fn load_without_mtp(e: &Engine, g: &GgufFile) -> Result<Self, Box<dyn std::error::Error>> {
3034 Self::load_from_source_impl(e, &GgufSource(g), false)
3035 }
3036
3037 pub fn load_from_source(
3041 e: &Engine,
3042 src: &dyn TensorSource,
3043 ) -> Result<Self, Box<dyn std::error::Error>> {
3044 Self::load_from_source_impl(e, src, true)
3045 }
3046
3047 pub fn load_from_source_without_mtp(
3049 e: &Engine,
3050 src: &dyn TensorSource,
3051 ) -> Result<Self, Box<dyn std::error::Error>> {
3052 Self::load_from_source_impl(e, src, false)
3053 }
3054
3055 fn load_from_source_impl(
3056 e: &Engine,
3057 src: &dyn TensorSource,
3058 load_mtp: bool,
3059 ) -> Result<Self, Box<dyn std::error::Error>> {
3060 let cfg = src.try_config().map_err(std::io::Error::other)?;
3061 let plan = match memra_gguf::model_packs::for_config(&cfg) {
3062 Some(pack) => pack.compile_plan(&cfg)?,
3063 None => memra_gguf::model_plan::ModelPlan::compile(&cfg)?,
3064 };
3065 let batch_program = crate::plan_backend::decode_batch_program(&plan);
3066 let gemma_program = batch_program == crate::plan_backend::DecodeBatchProgram::Gemma;
3067 let sliding_gated_moe_program =
3068 batch_program == crate::plan_backend::DecodeBatchProgram::SlidingGatedMoe;
3069 if sliding_gated_moe_program {
3074 crate::arm_step37_serving_defaults();
3075 }
3076 cfg.validate_attention_gate_layout()?;
3081 if cfg.sigmoid_router().is_some() {
3088 let host_oracle = std::env::var("MEMRA_SIG_ROUTER").as_deref() == Ok("0");
3089 match crate::sigrouter_contract::verify_host_expf() {
3090 Ok(()) => {}
3091 Err(e) if host_oracle => return Err(e.into()),
3092 Err(e) => eprintln!(
3093 "[sigrouter] WARN: host expf probe mismatch ({e}); device routing is \
3094 unaffected, but host-oracle replay/comparison cells are invalid on this host"
3095 ),
3096 }
3097 }
3098 if std::env::var("MEMRA_DRAFT").is_ok() && std::env::var("MEMRA_MMQ_SK").is_err() {
3107 let force = if cfg.n_embd >= 3500 { 0i8 } else { -1i8 };
3108 crate::MMQ_SK_FORCE.store(force, std::sync::atomic::Ordering::Relaxed);
3109 }
3110 crate::KV_FP8_FORCE.store(0, std::sync::atomic::Ordering::Relaxed);
3114
3115 let n_trunk = (cfg.n_layer - cfg.nextn_predict_layers) as usize;
3120 let mtp_skip_requested = load_mtp
3130 && match std::env::var("MEMRA_MTP_SKIP").ok().as_deref() {
3131 None | Some("") | Some("0") => false,
3132 Some("1") => true,
3133 Some(other) => {
3134 return Err(format!(
3135 "MEMRA_MTP_SKIP={other:?}: expected 1 (skip the embedded MTP block) or \
3136 0/unset (load it); refusing to guess"
3137 )
3138 .into());
3139 }
3140 };
3141 if mtp_skip_requested && std::env::var("MEMRA_MTP_DRAFT").is_ok_and(|p| !p.is_empty()) {
3142 return Err(
3143 "MEMRA_MTP_SKIP=1 together with MEMRA_MTP_DRAFT is contradictory: the skip \
3144 removes the MTP head to reclaim VRAM while MEMRA_MTP_DRAFT attaches an \
3145 external MTP head for MTP spec decode; unset one"
3146 .into(),
3147 );
3148 }
3149 if mtp_skip_requested && cfg.nextn_predict_layers > 0 {
3150 let prefixes: Vec<String> = (0..cfg.nextn_predict_layers)
3155 .map(|off| format!("blk.{}.", n_trunk as u32 + off))
3156 .collect();
3157 let skipped_bytes: Option<u64> = src.gguf().map(|g| {
3158 g.tensors
3159 .iter()
3160 .filter(|t| prefixes.iter().any(|p| t.name.starts_with(p.as_str())))
3161 .map(|t| t.n_bytes)
3162 .sum()
3163 });
3164 eprintln!(
3165 "[mtp-skip] MEMRA_MTP_SKIP=1: skipping {} embedded MTP/NextN block(s) \
3166 blk.{}..=blk.{} ({}); MTP spec decode is unavailable for this model \
3167 (dspark/DFlash2 drafting keeps its trimmed head via the MEMRA_FRSPEC_TRIM stub)",
3168 cfg.nextn_predict_layers,
3169 n_trunk,
3170 n_trunk as u32 + cfg.nextn_predict_layers - 1,
3171 match skipped_bytes {
3172 Some(b) => format!("~{} MiB of weights not loaded", b >> 20),
3173 None => "size unknown: non-GGUF source".to_string(),
3174 },
3175 );
3176 }
3177 let mtp_skip_trim_d2t: Option<Vec<u32>> = if mtp_skip_requested
3192 && cfg.nextn_predict_layers > 0
3193 && !crate::model::full_prec_enabled()
3194 {
3195 match std::env::var("MEMRA_FRSPEC_TRIM") {
3196 Ok(path) if !path.is_empty() => {
3197 let path = memra_gguf::hf::resolve_arg(&path)
3198 .map_err(|err| format!("MEMRA_FRSPEC_TRIM={path:?}: {err}"))?;
3199 let own_head_name = frspec_trim_own_head_name(n_trunk);
3200 if src.has(&own_head_name) {
3201 return Err(format!(
3202 "MEMRA_MTP_SKIP=1 with MEMRA_FRSPEC_TRIM: this artifact ships its \
3203 own MTP-block lm_head ({own_head_name}), so the trimmed draft rows \
3204 live in the block being skipped; gathering trunk rows instead is \
3205 the wrong-head bug (acceptance 0/248 receipt, \
3206 frspec_trim_own_head_name). Unset MEMRA_MTP_SKIP or \
3207 MEMRA_FRSPEC_TRIM"
3208 )
3209 .into());
3210 }
3211 if !src.has("output.weight") && !src.has("token_embd.weight") {
3212 return Err("MEMRA_MTP_SKIP=1 with MEMRA_FRSPEC_TRIM: model has no \
3213 output.weight (or tied token_embd.weight) to gather trimmed draft \
3214 rows from"
3215 .into());
3216 }
3217 let d2t = frspec_read_d2t(&path)?;
3218 if d2t.is_empty() {
3219 return Err(format!(
3220 "MEMRA_MTP_SKIP=1 with MEMRA_FRSPEC_TRIM={path}: the rank artifact \
3221 yields an EMPTY d2t list, so no stub draft head can be built; fix \
3222 the artifact or unset MEMRA_MTP_SKIP"
3223 )
3224 .into());
3225 }
3226 Some(d2t)
3227 }
3228 _ => None,
3229 }
3230 } else {
3231 None
3232 };
3233 crate::pp::init_model_transport(e, &cfg, n_trunk)?;
3234 let step_parallel = prepare_step_parallel_load(e, src, &cfg, n_trunk)?;
3235 let embd = EmbedHost::from_source(src, "token_embd.weight");
3236 let e_head = crate::pp::layer_engine(e, n_trunk, n_trunk - 1)?;
3240 let output_norm = load_t(e_head, src, "output_norm.weight")?;
3241 let mut output = if src.has("output.weight") {
3243 load_t(e_head, src, "output.weight")?
3244 } else {
3245 load_t(e_head, src, "token_embd.weight")?
3246 };
3247 let mut resident = ResidentPlan::pp(e, src, &cfg, n_trunk)?;
3248 let mut step_runtimes = StepParallelRuntimeRegistry::with_config(step_parallel);
3249
3250 let gguf: Option<&GgufFile> = src.gguf();
3257 let mut spill: Option<crate::spill::SpillCtx> = if cfg
3260 .moe
3261 .as_ref()
3262 .is_some_and(|m| m.expert_count > 0)
3263 && crate::spill::disk_tier_enabled()
3264 && gguf.is_some()
3265 {
3266 let budget = crate::spill::MemBudget::probe(e)?;
3267 let ctx = crate::spill::SpillCtx::open(gguf.unwrap(), &budget)?;
3268 eprintln!(
3269 "[spill] disk tier ON: free_vram={} MiB free_pinnable_ram={} MiB (MemAvailable*resolved_frac)",
3270 budget.free_vram >> 20,
3271 budget.free_pinnable_ram >> 20
3272 );
3273 Some(ctx)
3274 } else {
3275 None
3276 };
3277
3278 let mut layers = Vec::with_capacity(n_trunk);
3281 for il in 0..n_trunk as u32 {
3282 let p = |s: &str| format!("blk.{il}.{s}");
3283 let layer_plan = plan
3284 .layers
3285 .get(il as usize)
3286 .ok_or_else(|| format!("ModelPlan has no trunk layer {il}"))?;
3287 let e = crate::pp::layer_engine(e, n_trunk, il as usize)?;
3291 layers.push(HybridLayer {
3293 attn_norm: load_t(e, src, &p("attn_norm.weight"))?,
3294 post_attn_norm: load_opt(e, src, &p("post_attention_norm.weight"))?
3295 .or(load_opt(e, src, &p("ffn_norm.weight"))?)
3296 .expect("need post_attention_norm or ffn_norm"),
3297 mixer: {
3298 let g4_shared = cfg.gemma4.as_ref().map(|g| g.shared_kv_layers).unwrap_or(0);
3302 let kv_from = n_trunk as u32 - g4_shared;
3303 if g4_shared > 0
3304 && il >= kv_from
3305 && !src.has(&format!("blk.{il}.attn_k.weight"))
3306 {
3307 let g4 = cfg.gemma4.as_ref().unwrap();
3308 let swa = g4.swa_pattern.get(il as usize).copied().unwrap_or(true);
3309 let tgt = kv_from - if swa { 2 } else { 1 };
3310 let tp = |s: &str| format!("blk.{tgt}.{s}");
3311 Mixer::Full(FullAttnLayer {
3312 wq: load_t(e, src, &p("attn_q.weight"))?,
3313 wk: load_t(e, src, &tp("attn_k.weight"))?,
3314 wv: load_t(e, src, &tp("attn_v.weight"))?,
3315 wo: load_t(e, src, &p("attn_output.weight"))?,
3316 q_norm: load_t(e, src, &p("attn_q_norm.weight"))?,
3317 k_norm: load_t(e, src, &tp("attn_k_norm.weight"))?,
3318 attn_gate: None, step_tp_qkv: None,
3320 })
3321 } else {
3322 load_mixer_kind(
3323 e,
3324 src,
3325 &cfg,
3326 il,
3327 &layer_plan.attention,
3328 &mut step_runtimes,
3329 )?
3330 }
3331 },
3332 ffn: load_ffn(
3333 e,
3334 src,
3335 &cfg,
3336 &layer_plan.mlp,
3337 il,
3338 spill.as_mut().map(|c| (gguf.unwrap(), c)),
3339 &mut resident,
3340 &mut step_runtimes,
3341 )?,
3342 gemma4: if gemma_program {
3343 let scalar = |n: &str| -> f32 {
3344 let t = src.find(&p(n)).unwrap_or_else(|| panic!("missing {n}"));
3345 memra_gguf::dequant::dequantize(t.ggml_type, &t.bytes, 1)[0]
3346 };
3347 let vecf = |n: &str| -> Vec<f32> {
3348 let t = src.find(&p(n)).unwrap_or_else(|| panic!("missing {n}"));
3349 memra_gguf::dequant::dequantize(
3350 t.ggml_type,
3351 &t.bytes,
3352 t.ne.iter().product::<u64>() as usize,
3353 )
3354 };
3355 let moe_bits = if src.find(&p("ffn_gate_inp.scale")).is_some() {
3356 Some(crate::hybrid::Gemma4MoeBits {
3357 post_ffw_norm_1: load_t(e, src, &p("post_ffw_norm_1.weight"))?,
3358 pre_ffw_norm_2: load_t(e, src, &p("pre_ffw_norm_2.weight"))?,
3359 post_ffw_norm_2: load_t(e, src, &p("post_ffw_norm_2.weight"))?,
3360 shared_gate: load_t(e, src, &p("ffn_gate.weight"))?,
3361 shared_up: load_t(e, src, &p("ffn_up.weight"))?,
3362 shared_down: load_t(e, src, &p("ffn_down.weight"))?,
3363 router_scale_pre: {
3364 let inv = 1.0 / (cfg.n_embd as f32).sqrt();
3365 let v: Vec<f32> =
3366 vecf("ffn_gate_inp.scale").iter().map(|x| x * inv).collect();
3367 e.htod(&v)?
3368 },
3369 per_expert_scale: vecf("ffn_down_exps.scale"),
3370 per_expert_scale_d: e.htod(&vecf("ffn_down_exps.scale"))?,
3371 })
3372 } else {
3373 None
3374 };
3375 let e4b = if src.has(&p("inp_gate.weight")) {
3377 let g4 = cfg.gemma4.as_ref().unwrap();
3378 let kv_from = n_trunk as u32 - g4.shared_kv_layers;
3379 let kv_share = if g4.shared_kv_layers > 0 && il >= kv_from {
3380 let swa = g4.swa_pattern.get(il as usize).copied().unwrap_or(true);
3381 Some(kv_from - if swa { 2 } else { 1 })
3382 } else {
3383 None
3384 };
3385 Some(crate::hybrid::Gemma4E4bLayer {
3386 inp_gate: load_t(e, src, &p("inp_gate.weight"))?,
3387 proj: load_t(e, src, &p("proj.weight"))?,
3388 post_norm: load_t(e, src, &p("post_norm.weight"))?,
3389 kv_share,
3390 qkv_cat: None, })
3392 } else {
3393 None
3394 };
3395 Some(Gemma4LayerBits {
3396 ffn_norm: load_t(e, src, &p("ffn_norm.weight"))?,
3397 post_ffw_norm: load_t(e, src, &p("post_ffw_norm.weight"))?,
3398 moe_bits,
3399 layer_scale: scalar("layer_output_scale.weight"),
3400 e4b,
3401 })
3402 } else {
3403 None
3404 },
3405 });
3406 }
3407
3408 let external_mtp_requested =
3412 load_mtp && std::env::var("MEMRA_MTP_DRAFT").is_ok_and(|path| !path.is_empty());
3413 let trim_mtp_requested = load_mtp
3414 && !crate::model::full_prec_enabled()
3415 && std::env::var("MEMRA_FRSPEC_TRIM").is_ok_and(|path| !path.is_empty());
3416 let _ = trim_mtp_requested;
3427 let embedded_head_count = if external_mtp_requested || mtp_skip_requested {
3430 0
3431 } else {
3432 cfg.nextn_predict_layers
3433 };
3434 let embedded_head_count = match std::env::var("MEMRA_MTP_HEADS")
3439 .ok()
3440 .and_then(|v| v.parse::<u32>().ok())
3441 .filter(|&n| n > 0)
3442 {
3443 Some(cap) if cap < embedded_head_count => {
3444 eprintln!(
3445 "[mtp-chain] MEMRA_MTP_HEADS={cap}: capping the embedded chain from \
3446 {embedded_head_count} heads (measurement knob)"
3447 );
3448 cap
3449 }
3450 _ => embedded_head_count,
3451 };
3452 let mut embedded_mtp = Vec::new();
3453 if load_mtp && embedded_head_count > 0 {
3454 for offset in 0..embedded_head_count {
3455 let n = n_trunk as u32 + offset;
3456 let p = |s: &str| format!("blk.{n}.{s}");
3457 let mtp_plan = plan
3458 .mtp_blocks
3459 .iter()
3460 .find(|block| block.layer.index == n)
3461 .ok_or_else(|| format!("ModelPlan has no embedded MTP block {n}"))?;
3462 if !src.has(&p("nextn.eh_proj.weight")) {
3463 if offset == 0 {
3464 break;
3465 }
3466 return Err(format!(
3467 "embedded MTP chain declares {} heads but blk.{n} has no \
3468 nextn.eh_proj.weight",
3469 cfg.nextn_predict_layers
3470 )
3471 .into());
3472 }
3473 embedded_mtp.push(MtpHead {
3474 enorm: load_t(e, src, &p("nextn.enorm.weight"))?,
3475 hnorm: load_t(e, src, &p("nextn.hnorm.weight"))?,
3476 eh_proj: load_t(e, src, &p("nextn.eh_proj.weight"))?,
3477 attn_norm: load_t(e, src, &p("attn_norm.weight"))?,
3478 post_attn_norm: load_opt(e, src, &p("post_attention_norm.weight"))?
3479 .or(load_opt(e, src, &p("ffn_norm.weight"))?)
3480 .expect("MTP block needs post_attention_norm or ffn_norm"),
3481 mixer: load_mixer_kind(
3482 e,
3483 src,
3484 &cfg,
3485 n,
3486 &mtp_plan.layer.attention,
3487 &mut step_runtimes,
3488 )?,
3489 ffn: load_ffn(
3490 e,
3491 src,
3492 &cfg,
3493 &mtp_plan.layer.mlp,
3494 n,
3495 spill.as_mut().map(|c| (gguf.unwrap(), c)),
3496 &mut resident,
3497 &mut step_runtimes,
3498 )?,
3499 shared_head_norm: load_opt(e, src, &p("nextn.shared_head_norm.weight"))?,
3500 shared_head_head: load_mtp_head_maybe_nvfp4(
3509 e,
3510 src,
3511 &p("nextn.shared_head_head.weight"),
3512 )?
3513 .or(load_opt(e, src, &p("nextn.shared_head.weight"))?),
3514 d2t: None,
3515 d2t_from_target_head: false,
3516 geom: None,
3517 step35: if sliding_gated_moe_program {
3518 Some(Step35MtpGeom::from_plan(&mtp_plan.layer)?)
3519 } else {
3520 None
3521 },
3522 });
3523 }
3524 }
3525 let mut embedded_mtp = embedded_mtp.into_iter();
3526 let mut mtp = embedded_mtp.next();
3527 let mut mtp_extra: Vec<MtpHead> = embedded_mtp.collect();
3528
3529 mtp = if load_mtp {
3533 match std::env::var("MEMRA_MTP_DRAFT") {
3534 Ok(path) if !path.is_empty() => {
3535 eprintln!("[mtp-draft] loading external MTP draft: {path}");
3536 let dg = GgufFile::open(&path)?;
3537 mtp_extra.clear();
3538 Some(MtpHead::load_draft(e, &dg, &cfg)?)
3539 }
3540 _ => mtp,
3541 }
3542 } else {
3543 None
3544 };
3545
3546 let trim_env = if load_mtp {
3557 std::env::var("MEMRA_FRSPEC_TRIM")
3558 } else {
3559 Err(std::env::VarError::NotPresent)
3560 };
3561 if crate::model::full_prec_enabled()
3562 && trim_env.as_deref().map(|p| !p.is_empty()).unwrap_or(false)
3563 {
3564 eprintln!(
3565 "[frspec-trim] DISABLED under MEMRA_FULL_PREC — using the natural full MTP head"
3566 );
3567 }
3568 mtp = match (
3569 if crate::model::full_prec_enabled() {
3570 Err(std::env::VarError::NotPresent)
3571 } else {
3572 trim_env
3573 },
3574 mtp,
3575 ) {
3576 (Ok(path), Some(mut head)) if !path.is_empty() => {
3577 let path = memra_gguf::hf::resolve_arg(&path)
3581 .map_err(|err| format!("MEMRA_FRSPEC_TRIM={path:?}: {err}"))?;
3582 let d2t: Vec<u32> = frspec_read_d2t(&path)?;
3586 let own_head_name = frspec_trim_own_head_name(n_trunk);
3595 let own_head = src.find(&own_head_name);
3596 let from_own_head = own_head.is_some();
3597 let v = own_head
3598 .or_else(|| src.find("output.weight"))
3599 .or_else(|| src.find("token_embd.weight"))
3600 .expect("model has no output.weight for FR-Spec trim");
3601 let (trimmed, nvfp4_sizes) = frspec_gather_trimmed_head(
3621 e,
3622 &v,
3623 &d2t,
3624 std::env::var("MEMRA_FRSPEC_TRIM_NVFP4").as_deref() == Ok("1"),
3625 match src.find("output.scale") {
3627 Some(sv) => f32::from_le_bytes(sv.bytes[..4].try_into().unwrap()),
3628 None => 1.0,
3629 },
3630 )?;
3631 match nvfp4_sizes {
3632 Some((nvfp4_bytes, gathered_bytes)) => eprintln!(
3633 "[frspec-trim] self-trimmed head: {} rows of {} re-quantized BF16 -> NVFP4 \
3634 ({} MiB, was {} MiB)",
3635 d2t.len(),
3636 if from_own_head {
3637 own_head_name.as_str()
3638 } else {
3639 "main output.weight"
3640 },
3641 nvfp4_bytes >> 20,
3642 gathered_bytes >> 20,
3643 ),
3644 None => eprintln!(
3645 "[frspec-trim] self-trimmed head: {} rows of {} ({:?})",
3646 d2t.len(),
3647 if from_own_head {
3648 own_head_name.as_str()
3649 } else {
3650 "main output.weight"
3651 },
3652 v.ggml_type
3653 ),
3654 }
3655 head.shared_head_head = Some(trimmed);
3656 head.d2t = Some(d2t);
3657 head.d2t_from_target_head = !from_own_head;
3660 Some(head)
3661 }
3662 (_, m) => m,
3663 };
3664 let dflash_trim: Option<DflashTrimHead> = match mtp_skip_trim_d2t {
3675 Some(d2t) => {
3676 let v = src
3677 .find("output.weight")
3678 .or_else(|| src.find("token_embd.weight"))
3679 .ok_or("model has no output.weight for FR-Spec trim")?;
3680 let (head, nvfp4_sizes) = frspec_gather_trimmed_head(
3681 e,
3682 &v,
3683 &d2t,
3684 std::env::var("MEMRA_FRSPEC_TRIM_NVFP4").as_deref() == Ok("1"),
3685 match src.find("output.scale") {
3686 Some(sv) => f32::from_le_bytes(sv.bytes[..4].try_into().unwrap()),
3687 None => 1.0,
3688 },
3689 )?;
3690 eprintln!(
3691 "[mtp-skip] FR-Spec stub draft head built: {} rows of main output.weight \
3692 ({}); DFlash2 trim serves without the embedded MTP block",
3693 d2t.len(),
3694 match nvfp4_sizes {
3695 Some((nvfp4_bytes, gathered_bytes)) => format!(
3696 "re-quantized BF16 -> NVFP4, {} MiB, was {} MiB",
3697 nvfp4_bytes >> 20,
3698 gathered_bytes >> 20
3699 ),
3700 None => format!("{:?}", v.ggml_type),
3701 },
3702 );
3703 Some(DflashTrimHead { head, d2t })
3704 }
3705 None => None,
3706 };
3707 if let Some(d2t) = mtp.as_ref().and_then(|head| head.d2t.clone()) {
3720 let want_nvfp4_env = std::env::var("MEMRA_FRSPEC_TRIM_NVFP4").as_deref() == Ok("1");
3721 let mut kept = 0usize;
3722 for (i, head) in mtp_extra.iter_mut().enumerate() {
3723 let name = frspec_trim_own_head_name(n_trunk + 1 + i);
3724 let Some(v) = src.find(&name) else { break };
3725 let out_f = v.ne[1] as usize;
3726 let row_bytes = v.bytes.len() / out_f;
3727 if d2t.iter().any(|&t| (t as usize) >= out_f) {
3728 break;
3729 }
3730 let mut gathered = Vec::with_capacity(d2t.len() * row_bytes);
3731 for &t in &d2t {
3732 let off = t as usize * row_bytes;
3733 gathered.extend_from_slice(&v.bytes[off..off + row_bytes]);
3734 }
3735 let want_nvfp4 =
3736 want_nvfp4_env && matches!(v.ggml_type, GgmlType::BF16) && v.ne[0] % 64 == 0;
3737 let trimmed = if want_nvfp4 {
3738 let vals: Vec<f32> = gathered
3739 .chunks_exact(2)
3740 .map(|c| f32::from_bits((u16::from_le_bytes([c[0], c[1]]) as u32) << 16))
3741 .collect();
3742 let blocks = memra_gguf::nvfp4_repack::f32_to_nvfp4(&vals);
3743 GpuTensor::from_quant_bytes(
3744 e,
3745 &blocks,
3746 GgmlType::NVFP4,
3747 v.ne[0],
3748 d2t.len() as u64,
3749 1.0,
3750 )?
3751 } else {
3752 match v.ggml_type {
3753 GgmlType::BF16 => GpuTensor::FloatBf16 {
3754 data: e.htod_bytes(&gathered)?,
3755 ne: vec![v.ne[0], d2t.len() as u64],
3756 },
3757 GgmlType::F32 => GpuTensor::Float {
3758 data: e.htod(
3759 &gathered
3760 .chunks_exact(4)
3761 .map(|c| f32::from_le_bytes(c.try_into().unwrap()))
3762 .collect::<Vec<f32>>(),
3763 )?,
3764 ne: vec![v.ne[0], d2t.len() as u64],
3765 },
3766 _ => GpuTensor::from_quant_bytes(
3767 e,
3768 &gathered,
3769 v.ggml_type,
3770 v.ne[0],
3771 d2t.len() as u64,
3772 1.0,
3773 )?,
3774 }
3775 };
3776 head.shared_head_head = Some(trimmed);
3777 head.d2t = Some(d2t.clone());
3778 head.d2t_from_target_head = false;
3779 kept += 1;
3780 }
3781 let dropped = mtp_extra.len() - kept;
3782 mtp_extra.truncate(kept);
3783 eprintln!(
3784 "[frspec-trim] per-head trim: {kept} extra chain head(s) gathered from their own \
3785 blocks{}",
3786 if dropped > 0 {
3787 format!(" ({dropped} dropped: no own-head tensor)")
3788 } else {
3789 String::new()
3790 }
3791 );
3792 }
3793 if !mtp_extra.is_empty() {
3794 if plan.draft_source != memra_gguf::model_plan::DraftSourcePlan::Embedded
3795 || plan.mtp_blocks.len() != 1 + mtp_extra.len()
3796 || plan
3797 .mtp_blocks
3798 .iter()
3799 .any(|block| !matches!(block.layer.mlp, MlpPlan::Dense(_)))
3800 || mtp
3801 .iter()
3802 .chain(mtp_extra.iter())
3803 .any(|head| !matches!(head.ffn, Ffn::Dense { .. }))
3804 {
3805 return Err(
3806 "multi-head MTP requires embedded dense canonical blocks and matching loaded heads"
3807 .into(),
3808 );
3809 }
3810 eprintln!(
3811 "[mtp-draft] embedded chain: heads={} blocks={}..={} scratch=per-head",
3812 1 + mtp_extra.len(),
3813 n_trunk,
3814 n_trunk + mtp_extra.len()
3815 );
3816 }
3817
3818 if let Some(ctx) = spill.as_ref() {
3819 eprintln!(
3820 "[spill] experts placed: {} pinned (Tier 1), {} mmap'd from disk (Tier 2, {} MiB)",
3821 ctx.n_pinned,
3822 ctx.n_mmap,
3823 ctx.mmap_bytes >> 20
3824 );
3825 }
3826
3827 if cfg.n_head_kv > 0 && cfg.n_head / cfg.n_head_kv > 8 {
3841 crate::FA_V4_MAX_DEFAULT.store(0, std::sync::atomic::Ordering::Relaxed);
3842 eprintln!(
3843 "[fa] v4 decode family disabled: gqa {} > fa_v4_smem capacity 8 (v3 lane serves)",
3844 cfg.n_head / cfg.n_head_kv
3845 );
3846 }
3847
3848 if gemma_program {
3849 crate::FA_VEC_MIN_DEFAULT.store(1, std::sync::atomic::Ordering::Relaxed);
3851 let real_moe = plan
3854 .trunk_operations()
3855 .contains(&memra_gguf::model_plan::OperationKind::MoeMlp);
3856 crate::FA_SPW_DEFAULT.store(
3857 if real_moe { 32 } else { 64 },
3858 std::sync::atomic::Ordering::Relaxed,
3859 );
3860 crate::FA_SP512_DEFAULT.store(
3862 if real_moe { 16 } else { 32 },
3863 std::sync::atomic::Ordering::Relaxed,
3864 );
3865 crate::FUSED_MR1_DEFAULT.store(!real_moe, std::sync::atomic::Ordering::Relaxed);
3875 crate::RMS_BLOCK_DEFAULT.store(1024, std::sync::atomic::Ordering::Relaxed);
3877 crate::FA_SP_GEMMA.store(true, std::sync::atomic::Ordering::Relaxed);
3879 }
3883 let force_embd_gpu = gemma_program;
3886 let gemma4_aux = if gemma_program {
3887 let rope_freqs = match src.find("rope_freqs.weight") {
3888 Some(t) => {
3889 let host = memra_gguf::dequant::dequantize(
3890 t.ggml_type,
3891 &t.bytes,
3892 t.ne.iter().product::<u64>() as usize,
3893 );
3894 let mut copies = Vec::new();
3895 if let Some(fence) = crate::pp::pp_cuts(n_trunk) {
3896 for s in 0..fence.len() - 1 {
3897 let owner = crate::pp::layer_engine(e, n_trunk, fence[s])?;
3898 let dev = owner.ctx().ordinal();
3899 if copies.iter().all(|(d, _)| *d != dev) {
3900 copies.push((dev, owner.htod(&host)?));
3901 }
3902 }
3903 } else {
3904 copies.push((e.ctx().ordinal(), e.htod(&host)?));
3905 }
3906 Some(copies)
3907 }
3908 None => {
3916 let g4 = cfg.gemma4.as_ref().unwrap();
3917 let n = (g4.rope_dims_global / 2) as usize;
3918 let keep =
3919 ((n as f32) * g4.partial_rotary_global.clamp(0.0, 1.0)).round() as usize;
3920 let host: Vec<f32> = (0..n)
3921 .map(|i| if i < keep { 1.0 } else { 1.0e30 })
3922 .collect();
3923 eprintln!(
3924 "[gemma4] rope_freqs.weight synthesized ({n} factors, first {keep} \
3925 rotate; source ships none — native checkpoint)"
3926 );
3927 let mut copies = Vec::new();
3928 if let Some(fence) = crate::pp::pp_cuts(n_trunk) {
3929 for s in 0..fence.len() - 1 {
3930 let owner = crate::pp::layer_engine(e, n_trunk, fence[s])?;
3931 let dev = owner.ctx().ordinal();
3932 if copies.iter().all(|(d, _)| *d != dev) {
3933 copies.push((dev, owner.htod(&host)?));
3934 }
3935 }
3936 } else {
3937 copies.push((e.ctx().ordinal(), e.htod(&host)?));
3938 }
3939 Some(copies)
3940 }
3941 };
3942 let e4b = match src.find("per_layer_token_embd.weight") {
3944 Some(t) => {
3945 let n_epl = cfg
3946 .gemma4
3947 .as_ref()
3948 .map(|g| g.n_embd_per_layer as usize)
3949 .unwrap_or(0);
3950 let row = t.ne[0] as usize; let row_bytes = t.bytes.len() / (t.ne[1] as usize);
3952 eprintln!(
3953 "[gemma4-e4b] per-layer-embed model detected (n_epl={n_epl}, row {row}) — \
3954 first-light forward (eager decode + prime); dc/graph/spec unwired \
3955 (HANDOVER-E4B.md)"
3956 );
3957 Some(crate::hybrid::Gemma4E4bModel {
3958 tok_tbl_gpu: std::sync::OnceLock::new(),
3959 tok_embd_bytes: t.bytes.to_vec(),
3960 tok_embd_qt: match t.ggml_type {
3961 memra_gguf::GgmlType::Q6_K => crate::QT_Q6_K,
3962 memra_gguf::GgmlType::Q8_0 => crate::QT_Q8_0,
3963 other => panic!("e4b per-layer tok embd: unhandled dtype {other:?}"),
3964 },
3965 tok_embd_row_bytes: row_bytes,
3966 model_proj: load_t(e, src, "per_layer_model_proj.weight")?,
3967 proj_norm: load_t(e, src, "per_layer_proj_norm.weight")?,
3968 n_epl,
3969 })
3970 }
3971 None => None,
3972 };
3973 let suppress_d = {
3974 let sup = &cfg.gemma4.as_ref().unwrap().suppress_tokens;
3975 if sup.is_empty() {
3976 None
3977 } else {
3978 let ids: Vec<i32> = sup.iter().map(|&x| x as i32).collect();
3979 eprintln!(
3980 "[gemma4] suppress_tokens: {} ids masked at sampling",
3981 ids.len()
3982 );
3983 Some((e.htod_i32(&ids)?, ids.len()))
3984 }
3985 };
3986 let ones_host = [1.0f32; 512];
3987 let mut ones = Vec::new();
3988 if let Some(fence) = crate::pp::pp_cuts(n_trunk) {
3989 for s in 0..fence.len() - 1 {
3990 let owner = crate::pp::layer_engine(e, n_trunk, fence[s])?;
3991 let dev = owner.ctx().ordinal();
3992 if ones.iter().all(|(d, _)| *d != dev) {
3993 ones.push((dev, owner.htod(&ones_host)?));
3994 }
3995 }
3996 } else {
3997 ones.push((e.ctx().ordinal(), e.htod(&ones_host)?));
3998 }
3999 Some(GemmaAux {
4000 rope_freqs,
4001 ones,
4002 suppress_d,
4003 e4b,
4004 })
4005 } else {
4006 None
4007 };
4008 let step35_aux = if sliding_gated_moe_program {
4012 let rope_freqs = match src.find("rope_freqs.weight") {
4013 Some(t) => {
4014 let host = memra_gguf::dequant::dequantize(
4015 t.ggml_type,
4016 &t.bytes,
4017 t.ne.iter().product::<u64>() as usize,
4018 );
4019 let mut copies = Vec::new();
4020 if let Some(fence) = crate::pp::pp_cuts(n_trunk) {
4021 for s in 0..fence.len() - 1 {
4022 let owner = crate::pp::layer_engine(e, n_trunk, fence[s])?;
4023 let dev = owner.ctx().ordinal();
4024 if copies.iter().all(|(d, _)| *d != dev) {
4025 copies.push((dev, owner.htod(&host)?));
4026 }
4027 }
4028 } else {
4029 copies.push((e.ctx().ordinal(), e.htod(&host)?));
4030 }
4031 Some(copies)
4032 }
4033 None => None,
4034 };
4035 Some(Step35Aux { rope_freqs })
4036 } else {
4037 None
4038 };
4039 let mut layers = layers;
4040 {
4047 let q8rp_on = match std::env::var("MEMRA_Q8RP").as_deref() {
4048 Ok("0") => false,
4049 Ok(_) => true,
4050 Err(_) => {
4058 cfg!(memra_hopper_mma) || {
4059 let q8b = |w: &crate::model::GpuTensor| -> usize {
4060 match w {
4061 crate::model::GpuTensor::Quant {
4062 bytes,
4063 qtype,
4064 row_bytes,
4065 ne,
4066 rp4: None,
4067 ..
4068 } if *qtype == crate::QT_Q8_0
4069 && ne.len() == 2
4070 && (ne[0] as usize) % 32 == 0
4071 && *row_bytes == (ne[0] as usize / 32) * 34 =>
4072 {
4073 bytes.len()
4074 }
4075 _ => 0,
4076 }
4077 };
4078 let mut need = q8b(&output);
4079 for layer in layers.iter() {
4080 match &layer.mixer {
4081 Mixer::Full(fa) => {
4082 for w in [&fa.wq, &fa.wk, &fa.wv, &fa.wo] {
4083 need += q8b(w);
4084 }
4085 }
4086 Mixer::Linear(la) => {
4087 for w in [
4088 &la.wqkv,
4089 &la.wqkv_gate,
4090 &la.ssm_beta,
4091 &la.ssm_alpha,
4092 &la.ssm_out,
4093 ] {
4094 need += q8b(w);
4095 }
4096 }
4097 Mixer::Mla(_) => {}
4098 }
4099 if let Ffn::Dense {
4100 ffn_gate,
4101 ffn_up,
4102 ffn_down,
4103 } = &layer.ffn
4104 {
4105 for w in [ffn_gate, ffn_up, ffn_down] {
4106 need += q8b(w);
4107 }
4108 }
4109 }
4110 need > 0
4111 && e.ctx()
4112 .mem_get_info()
4113 .map(|(free, _)| free >= need + (8usize << 30))
4114 .unwrap_or(false)
4115 }
4116 }
4117 };
4118 let kqrp_on = crate::Engine::kqrp_enabled() || {
4128 std::env::var("MEMRA_KQRP").is_err() && {
4129 let kqb = |w: &crate::model::GpuTensor| -> usize {
4130 match w {
4131 crate::model::GpuTensor::Quant {
4132 bytes,
4133 qtype,
4134 row_bytes,
4135 ne,
4136 rp4: None,
4137 ..
4138 } if ne.len() == 2 && (ne[0] as usize) % 256 == 0 => {
4139 let sb = if *qtype == crate::QT_Q4_K {
4140 144
4141 } else if *qtype == crate::QT_Q6_K {
4142 210
4143 } else {
4144 return 0;
4145 };
4146 if *row_bytes == (ne[0] as usize / 256) * sb {
4147 bytes.len()
4148 } else {
4149 0
4150 }
4151 }
4152 _ => 0,
4153 }
4154 };
4155 let mut need = kqb(&output);
4156 for layer in layers.iter() {
4157 if let Mixer::Full(fa) = &layer.mixer {
4158 for w in [&fa.wq, &fa.wk, &fa.wv, &fa.wo] {
4159 need += kqb(w);
4160 }
4161 }
4162 if let Ffn::Dense {
4163 ffn_gate,
4164 ffn_up,
4165 ffn_down,
4166 } = &layer.ffn
4167 {
4168 for w in [ffn_gate, ffn_up, ffn_down] {
4169 need += kqb(w);
4170 }
4171 }
4172 }
4173 need > 0
4174 && e.ctx()
4175 .mem_get_info()
4176 .map(|(free, _)| free >= need + (8usize << 30))
4177 .unwrap_or(false)
4178 }
4179 };
4180 if q8rp_on || kqrp_on {
4181 let f16_model_ok = gemma_program
4188 || plan
4189 .trunk_operations()
4190 .contains(&memra_gguf::model_plan::OperationKind::MoeMlp)
4191 || std::env::var("MEMRA_PP_F16").as_deref() == Ok("1");
4192 let mut nmir = 0usize;
4193 let mut mir = |e_ref: &crate::Engine,
4197 w: &mut crate::model::GpuTensor|
4198 -> Result<(), Box<dyn std::error::Error>> {
4199 let before = matches!(w, crate::model::GpuTensor::Quant { rp4: Some(_), .. });
4200 if q8rp_on {
4201 e_ref.build_q8_rp4(w)?;
4202 }
4203 if kqrp_on {
4204 e_ref.build_q4k_rp4(w)?;
4205 e_ref.build_q6k_rp4(w)?;
4206 }
4207 let q6k = matches!(w, crate::model::GpuTensor::Quant { qtype, .. }
4212 if *qtype == crate::QT_Q6_K);
4213 if q8rp_on && crate::f16_ffi::pp_f16_enabled() && (f16_model_ok || q6k) {
4214 e_ref.build_q8_f16(w)?;
4215 }
4216 if !before && matches!(w, crate::model::GpuTensor::Quant { rp4: Some(_), .. }) {
4217 nmir += 1;
4218 }
4219 Ok(())
4220 };
4221 for (il, layer) in layers.iter_mut().enumerate() {
4222 let el = crate::pp::layer_engine(e, n_trunk, il)?;
4223 match &mut layer.mixer {
4224 Mixer::Full(fa) => {
4225 for w in [&mut fa.wq, &mut fa.wk, &mut fa.wv, &mut fa.wo] {
4226 mir(el, w)?;
4227 }
4228 }
4229 Mixer::Linear(la) => {
4230 for w in [
4231 &mut la.wqkv,
4232 &mut la.wqkv_gate,
4233 &mut la.ssm_beta,
4234 &mut la.ssm_alpha,
4235 &mut la.ssm_out,
4236 ] {
4237 mir(el, w)?;
4238 }
4239 }
4240 Mixer::Mla(_) => {}
4243 }
4244 if let Ffn::Dense {
4245 ffn_gate,
4246 ffn_up,
4247 ffn_down,
4248 } = &mut layer.ffn
4249 {
4250 for w in [ffn_gate, ffn_up, ffn_down] {
4251 mir(el, w)?;
4252 }
4253 }
4254 }
4255 mir(e_head, &mut output)?;
4256 if nmir > 0 {
4257 eprintln!("[q8rp] split-plane decode mirrors built: {nmir} tensors");
4258 }
4259 if q8rp_on && crate::f16_ffi::pp_f16_enabled() {
4274 for (want, tag) in [(crate::QT_Q4_K, "q4kf16"), (crate::QT_Q5_K, "q5kf16")] {
4275 let (mut n4, mut b4) = (0usize, 0usize);
4276 let mut mirk =
4277 |e_ref: &crate::Engine,
4278 w: &mut crate::model::GpuTensor|
4279 -> Result<(), Box<dyn std::error::Error>> {
4280 if matches!(w, crate::model::GpuTensor::Quant { qtype, f16: None, .. }
4281 if *qtype == want)
4282 {
4283 e_ref.build_q8_f16(w)?;
4284 if let crate::model::GpuTensor::Quant { f16: Some(m), .. } = w {
4285 n4 += 1;
4286 b4 += m.len();
4287 }
4288 }
4289 Ok(())
4290 };
4291 for (il, layer) in layers.iter_mut().enumerate() {
4292 let el = crate::pp::layer_engine(e, n_trunk, il)?;
4293 match &mut layer.mixer {
4294 Mixer::Full(fa) => {
4295 for w in [&mut fa.wq, &mut fa.wk, &mut fa.wv, &mut fa.wo] {
4296 mirk(el, w)?;
4297 }
4298 }
4299 Mixer::Linear(la) => {
4300 for w in [
4301 &mut la.wqkv,
4302 &mut la.wqkv_gate,
4303 &mut la.ssm_beta,
4304 &mut la.ssm_alpha,
4305 &mut la.ssm_out,
4306 ] {
4307 mirk(el, w)?;
4308 }
4309 }
4310 Mixer::Mla(_) => {} }
4312 if let Ffn::Dense {
4313 ffn_gate,
4314 ffn_up,
4315 ffn_down,
4316 } = &mut layer.ffn
4317 {
4318 for w in [ffn_gate, ffn_up, ffn_down] {
4319 mirk(el, w)?;
4320 }
4321 }
4322 }
4323 mirk(e_head, &mut output)?;
4324 if n4 > 0 {
4325 eprintln!(
4326 "[{tag}] prefill fp16 mirrors built: {n4} tensors \
4327 ({} MB)",
4328 b4 >> 20
4329 );
4330 }
4331 }
4332 }
4333 }
4334 }
4335 if gemma_program && crate::Engine::q4rp_enabled() {
4342 let mut nmir = 0usize;
4343 for (il, layer) in layers.iter_mut().enumerate() {
4344 let e = crate::pp::layer_engine(e, n_trunk, il)?;
4346 let is_moe26 = layer.gemma4.as_ref().is_some_and(|g| g.moe_bits.is_some());
4355 let is_e4b = layer.gemma4.as_ref().is_some_and(|g| g.e4b.is_some());
4356 if !(is_moe26 || is_e4b) {
4357 continue;
4358 }
4359 if let Mixer::Full(fa) = &mut layer.mixer {
4360 for w in [&mut fa.wq, &mut fa.wk, &mut fa.wv, &mut fa.wo] {
4361 e.build_q4_rp4(w)?;
4362 nmir += 1;
4363 }
4364 }
4365 if is_e4b {
4366 let own_kv = layer
4368 .gemma4
4369 .as_ref()
4370 .unwrap()
4371 .e4b
4372 .as_ref()
4373 .is_some_and(|e4| e4.kv_share.is_none());
4374 if own_kv {
4375 if let Mixer::Full(fa) = &layer.mixer {
4376 if let Some(mut cat) = e.build_q4_out_concat3(&fa.wq, &fa.wk, &fa.wv)? {
4377 e.build_q4_rp4(&mut cat)?;
4378 nmir += 1;
4379 layer.gemma4.as_mut().unwrap().e4b.as_mut().unwrap().qkv_cat =
4380 Some(cat);
4381 }
4382 }
4383 }
4384 if let Ffn::Dense {
4385 ffn_gate,
4386 ffn_up,
4387 ffn_down,
4388 } = &mut layer.ffn
4389 {
4390 for w in [ffn_gate, ffn_up, ffn_down] {
4391 e.build_q4_rp4(w)?;
4392 nmir += 1;
4393 }
4394 }
4395 let e4 = layer.gemma4.as_mut().unwrap().e4b.as_mut().unwrap();
4396 for w in [&mut e4.inp_gate, &mut e4.proj] {
4397 e.build_q4_rp4(w)?;
4398 nmir += 1;
4399 }
4400 }
4401 if let Some(mb) = layer.gemma4.as_mut().unwrap().moe_bits.as_mut() {
4402 for w in [&mut mb.shared_gate, &mut mb.shared_up, &mut mb.shared_down] {
4403 e.build_q4_rp4(w)?;
4404 nmir += 1;
4405 }
4406 }
4407 }
4408 if nmir > 0 {
4409 eprintln!("[q4rp] split-plane decode mirrors built: {nmir} trunk tensors");
4410 }
4411 let fast_on = std::env::var("MEMRA_FAST").as_deref() != Ok("0");
4418 if fast_on {
4419 let mut nswap = 0usize;
4420 let mut nf16 = 0usize;
4421 let q4f16_model_ok = matches!(cfg.n_embd, 3840 | 5376); if let Ok(v) = std::env::var("MEMRA_Q4F16") {
4440 if v != "0" && v != "1" {
4441 return Err(format!(
4442 "MEMRA_Q4F16={v} is not 0 or 1 — this env selects the prefill \
4443 ARITHMETIC (fp16 mirrors vs int8 MMQ) and must never be guessed"
4444 )
4445 .into());
4446 }
4447 }
4448 let f16_need = {
4449 let f16b = |w: &crate::model::GpuTensor| -> usize {
4450 match w {
4451 crate::model::GpuTensor::Quant {
4452 qtype,
4453 ne,
4454 f16: None,
4455 ..
4456 } if ne.len() == 2
4457 && matches!(
4458 *qtype,
4459 crate::QT_Q8_0
4460 | crate::QT_Q4_0
4461 | crate::QT_Q6_K
4462 | crate::QT_Q4_K
4463 | crate::QT_Q5_K
4464 ) =>
4465 {
4466 (ne[0] as usize) * (ne[1] as usize) * 2
4467 }
4468 _ => 0,
4469 }
4470 };
4471 let mut need = 0usize;
4472 for layer in layers.iter() {
4473 if layer.gemma4.as_ref().is_none_or(|g| g.moe_bits.is_some()) {
4474 continue;
4475 }
4476 if let Mixer::Full(fa) = &layer.mixer {
4477 for w in [&fa.wq, &fa.wk, &fa.wv, &fa.wo] {
4478 need += f16b(w);
4479 }
4480 }
4481 if let Ffn::Dense {
4482 ffn_gate,
4483 ffn_up,
4484 ffn_down,
4485 } = &layer.ffn
4486 {
4487 for w in [ffn_gate, ffn_up, ffn_down] {
4488 need += f16b(w);
4489 }
4490 }
4491 }
4492 need
4493 };
4494 let f16_free = e.ctx().mem_get_info().map(|(free, _)| free).unwrap_or(0);
4495 let f16_auto = q4f16_model_ok
4496 && std::env::var("MEMRA_Q4F16").is_err()
4497 && crate::f16_ffi::pp_f16_capacity_ok(f16_free, f16_need);
4498 let (f16_on, f16_why) = match std::env::var("MEMRA_Q4F16").as_deref() {
4504 Ok("1") => (true, "env MEMRA_Q4F16=1"),
4505 Ok("0") => (false, "env MEMRA_Q4F16=0"),
4506 _ if crate::f16_ffi::pp_f16_enabled() && q4f16_model_ok => {
4507 (true, "env MEMRA_PP_F16")
4508 }
4509 _ if f16_auto => (true, "capacity-keyed auto (UNPINNED)"),
4510 _ if !q4f16_model_ok => (false, "model geometry not eligible"),
4511 _ => (false, "capacity-keyed auto REFUSED (UNPINNED)"),
4512 };
4513 eprintln!(
4520 "[q4f16] prefill program = {} (reason: {}); free {} MiB, mirror mass {} MiB, \
4521 capacity threshold {} MiB (mass + 8192 headroom) — SELECTS PREFILL ARITHMETIC",
4522 if f16_on {
4523 "FP16 MIRRORS"
4524 } else {
4525 "INT8 MMQ (no f16 mirrors)"
4526 },
4527 f16_why,
4528 f16_free >> 20,
4529 f16_need >> 20,
4530 (f16_need + (8usize << 30)) >> 20,
4531 );
4532 for (il, layer) in layers.iter_mut().enumerate() {
4533 let e = crate::pp::layer_engine(e, n_trunk, il)?;
4535 let dense_gemma = layer.gemma4.as_ref().is_some_and(|g| g.moe_bits.is_none());
4536 if !dense_gemma {
4537 continue;
4538 }
4539 if let Mixer::Full(fa) = &mut layer.mixer {
4540 for w in [&mut fa.wq, &mut fa.wk, &mut fa.wv, &mut fa.wo] {
4541 if f16_on {
4542 e.build_q8_f16(w)?;
4543 if matches!(w, crate::model::GpuTensor::Quant { f16: Some(_), .. })
4544 {
4545 nf16 += 1;
4546 }
4547 }
4548 if e.build_q4_rp_swap(w)? {
4549 nswap += 1;
4550 }
4551 }
4552 }
4553 if let Ffn::Dense {
4554 ffn_gate,
4555 ffn_up,
4556 ffn_down,
4557 } = &mut layer.ffn
4558 {
4559 for w in [ffn_gate, ffn_up, ffn_down] {
4560 if f16_on {
4561 e.build_q8_f16(w)?;
4562 if matches!(w, crate::model::GpuTensor::Quant { f16: Some(_), .. })
4563 {
4564 nf16 += 1;
4565 }
4566 }
4567 if e.build_q4_rp_swap(w)? {
4568 nswap += 1;
4569 }
4570 }
4571 }
4572 }
4573 if nswap > 0 {
4574 eprintln!("[q4rp] split-plane IN-PLACE swap: {nswap} dense trunk tensors");
4575 }
4576 if nf16 > 0 {
4577 eprintln!("[q4f16] prefill fp16 mirrors built: {nf16} dense trunk tensors");
4578 }
4579 }
4580 }
4581 let model = HybridModel {
4582 cfg,
4583 plan,
4584 rewrite_qualifications: None,
4585 embd,
4586 output_norm,
4587 output,
4588 layers,
4589 mtp,
4590 mtp_extra,
4591 dflash_trim,
4592 embd_gpu: std::sync::OnceLock::new(),
4593 gemma4_aux,
4594 step35_aux,
4595 prime_slabs: std::sync::Mutex::new(std::collections::HashMap::new()),
4596 dspark_vgraphs: std::sync::Mutex::new(None),
4597 step_grouped_prefill: std::sync::Mutex::new(StepEpGroupedPrefill::default()),
4598 step35_token_graph: std::sync::Mutex::new(None),
4599 draft_state_bytes: std::sync::atomic::AtomicUsize::new(0),
4600 };
4601 e.configure_moe_cache_layout(model.moe_cache_block_sizes());
4602 if force_embd_gpu {
4603 let _ = model
4604 .embd_gpu
4605 .get_or_init(|| e.upload_u8(&model.embd.raw).expect("embed table upload"));
4606 }
4607 crate::pp::sync_stages_after_load(e, n_trunk)?;
4613 Ok(model)
4614 }
4615
4616 pub fn ensure_embed_resident(&self, e: &Engine) -> Result<(), Box<dyn std::error::Error>> {
4626 if std::env::var("MEMRA_EMBED_DEV").as_deref() == Ok("0") {
4627 return Ok(());
4628 }
4629 if self.embd_gpu.get().is_none() {
4630 let buf = e.upload_u8(&self.embd.raw)?;
4631 let _ = self.embd_gpu.set(buf); }
4633 Ok(())
4634 }
4635
4636 pub fn embed(
4637 &self,
4638 e: &Engine,
4639 tokens: &[u32],
4640 ) -> Result<CudaSlice<f32>, Box<dyn std::error::Error>> {
4641 let n_embd = self.cfg.n_embd as usize;
4642 if std::env::var("MEMRA_EMBED_DEV").as_deref() != Ok("0") {
4648 let tbl = self
4649 .embd_gpu
4650 .get_or_init(|| e.upload_u8(&self.embd.raw).expect("embed table upload"));
4651 let tok_d = e.htod_u32_v(tokens)?;
4652 let (qt, rb) = self.embd.qt_and_row_bytes(n_embd);
4653 return e.embed_gather_device_td(tbl, &tok_d, tokens.len(), n_embd, qt, rb);
4654 }
4655 let x = self.embd.gather(n_embd, tokens);
4656 Ok(e.htod(&x)?)
4657 }
4658}
4659
4660#[cfg(test)]
4661mod step_expert_selection_tests {
4662 use super::{
4663 StepExpertArtifact, StepExpertLayout, StepParallelLoadConfig, StepParallelRuntimeRegistry,
4664 StepTpAttentionPlacement, select_step_expert_layout,
4665 };
4666 use crate::tp::StepEpLayerSpec;
4667
4668 fn spec(layer: usize, ranks: usize) -> StepEpLayerSpec {
4669 StepEpLayerSpec {
4670 layer,
4671 devices: (0..ranks).collect(),
4672 }
4673 }
4674
4675 #[test]
4676 fn tp2_keeps_projection_sharded_experts() {
4677 let selection = select_step_expert_layout(24, &[], &[spec(24, 2)])
4678 .unwrap()
4679 .unwrap();
4680 assert_eq!(selection.layout, StepExpertLayout::TensorParallel);
4681 assert!(selection.configured_by_tp);
4682 }
4683
4684 #[test]
4685 fn tp4_and_tp8_use_expert_ownership_without_a_second_flag() {
4686 for ranks in [4, 8] {
4687 let selection = select_step_expert_layout(24, &[], &[spec(24, ranks)])
4688 .unwrap()
4689 .unwrap();
4690 assert_eq!(selection.layout, StepExpertLayout::ExpertParallel);
4691 assert!(selection.configured_by_tp);
4692 assert_eq!(selection.spec.devices.len(), ranks);
4693 }
4694 }
4695
4696 #[test]
4697 fn explicit_ep_remains_expert_parallel() {
4698 let selection = select_step_expert_layout(24, &[spec(24, 2)], &[])
4699 .unwrap()
4700 .unwrap();
4701 assert_eq!(selection.layout, StepExpertLayout::ExpertParallel);
4702 assert!(!selection.configured_by_tp);
4703 }
4704
4705 #[test]
4706 fn conflicting_ep_and_tp_assignments_fail_closed() {
4707 let error = select_step_expert_layout(24, &[spec(24, 4)], &[spec(24, 4)]).unwrap_err();
4708 assert!(error.contains("cannot enable MEMRA_STEP_EP and MEMRA_STEP_TP together"));
4709 }
4710
4711 #[test]
4712 fn runtime_registry_owns_one_immutable_load_snapshot() {
4713 let mut source_specs = vec![spec(24, 8)];
4714 let registry = StepParallelRuntimeRegistry::with_config(StepParallelLoadConfig {
4715 ep_specs: Vec::new(),
4716 tp_specs: source_specs.clone(),
4717 native_p2p: true,
4718 ep_device_arithmetic: true,
4719 f32_mirror: true,
4720 bulk_p2p: true,
4721 expert_artifact: StepExpertArtifact::default(),
4722 });
4723 source_specs[0].devices.clear();
4724
4725 let stored = registry.tp_spec(24).unwrap();
4726 assert_eq!(stored.devices, (0..8).collect::<Vec<_>>());
4727 assert!(registry.config.native_p2p);
4728 assert!(registry.config.ep_device_arithmetic);
4729 assert!(registry.config.f32_mirror);
4730 assert!(registry.config.bulk_p2p);
4731 assert_eq!(
4732 registry.expert_selection(24).unwrap().unwrap().layout,
4733 StepExpertLayout::ExpertParallel
4734 );
4735
4736 let standalone = StepParallelRuntimeRegistry::default();
4737 assert!(standalone.tp_spec(24).is_none());
4738 assert!(!standalone.config.native_p2p);
4739 assert!(!standalone.config.ep_device_arithmetic);
4740 assert!(!standalone.config.f32_mirror);
4741 assert!(!standalone.config.bulk_p2p);
4742 }
4743
4744 #[test]
4745 fn rank_local_attention_uses_bounded_swa_rings_only_with_native_p2p() {
4746 assert_eq!(
4747 StepTpAttentionPlacement::resolve(true, None),
4748 StepTpAttentionPlacement::RankLocalGlobal
4749 );
4750 assert_eq!(
4751 StepTpAttentionPlacement::resolve(true, Some(512)),
4752 StepTpAttentionPlacement::RankLocalSwa
4753 );
4754 assert_eq!(
4755 StepTpAttentionPlacement::resolve(false, None),
4756 StepTpAttentionPlacement::OwnerTransportFallback
4757 );
4758 assert_eq!(
4759 StepTpAttentionPlacement::resolve(false, Some(512)),
4760 StepTpAttentionPlacement::OwnerSwa
4761 );
4762 }
4763}
4764
4765#[cfg(test)]
4766mod residency_tests {
4767 use super::{DevExpertFp8ProjectionScales, residency_bytes_by_device};
4768 use crate::model::HostExpertFp8BlockScales;
4769
4770 #[test]
4771 fn pp_residency_counts_only_each_devices_expert_slice() {
4772 let tensors = [
4773 ("blk.0.ffn_gate_exps.weight", 10usize),
4774 ("blk.0.ffn_up_exps.weight", 20),
4775 ("blk.1.ffn_down_exps.weight", 30),
4776 ("blk.2.ffn_gate_exps.weight", 40),
4777 ("blk.3.ffn_up_exps.weight", 50),
4778 ("blk.0.attn_q.weight", 7),
4779 ("output.weight", 11),
4780 ];
4781 let bytes = residency_bytes_by_device(tensors, &[0, 0, 1, 1], 0);
4782 assert_eq!(bytes.experts.get(&0), Some(&60));
4783 assert_eq!(bytes.experts.get(&1), Some(&90));
4784 assert_eq!(bytes.rest, 18);
4785 assert!(bytes.saw_experts);
4786 }
4787
4788 #[test]
4789 fn pp_residency_combines_stages_that_share_one_device() {
4790 let tensors = [
4791 ("blk.0.ffn_gate_exps.weight", 10usize),
4792 ("blk.1.ffn_gate_exps.weight", 20),
4793 ("blk.2.ffn_gate_exps.weight", 30),
4794 ("blk.3.ffn_gate_exps.weight", 40),
4795 ];
4796 let bytes = residency_bytes_by_device(tensors, &[0, 0, 0, 0], 0);
4797 assert_eq!(bytes.experts.get(&0), Some(&100));
4798 assert_eq!(bytes.experts.len(), 1);
4799 }
4800
4801 #[test]
4802 fn resident_fp8_scale_slab_must_match_every_expert() {
4803 let valid = HostExpertFp8BlockScales {
4804 scales: vec![1.0; 12],
4805 rows: 2,
4806 cols: 3,
4807 expert_stride: 6,
4808 };
4809 DevExpertFp8ProjectionScales::validate(&valid, 2).unwrap();
4810
4811 let short = HostExpertFp8BlockScales {
4812 scales: vec![1.0; 11],
4813 ..valid
4814 };
4815 assert_eq!(
4816 DevExpertFp8ProjectionScales::validate(&short, 2).unwrap_err(),
4817 "block-E4M3 scale slab length mismatch: got 11, want 2x6=12"
4818 );
4819 }
4820
4821 #[test]
4822 fn resident_fp8_scale_stride_must_match_its_grid() {
4823 let invalid = HostExpertFp8BlockScales {
4824 scales: vec![1.0; 8],
4825 rows: 2,
4826 cols: 2,
4827 expert_stride: 0,
4828 };
4829 assert_eq!(
4830 DevExpertFp8ProjectionScales::validate(&invalid, 2).unwrap_err(),
4831 "block-E4M3 expert scale stride must be nonzero"
4832 );
4833 }
4834}
4835
4836#[cfg(test)]
4837mod draft_head_tests {
4838 use super::{draft_head_tensor, frspec_trim_own_head_name};
4839
4840 const STEP37_DRAFTER: &[&str] = &[
4847 "output.weight",
4848 "output_norm.weight",
4849 "token_embd.weight",
4850 "blk.45.nextn.shared_head_norm.weight",
4851 "blk.45.nextn.shared_head_head.weight",
4852 "blk.46.nextn.shared_head_head.weight",
4853 "blk.47.nextn.shared_head_head.weight",
4854 ];
4855
4856 fn present(names: &'static [&'static str]) -> impl Fn(&str) -> bool {
4857 move |t: &str| names.contains(&t)
4858 }
4859
4860 #[test]
4868 fn step37_drafter_prefers_the_blocks_own_nextn_head_over_file_level_output() {
4869 assert_eq!(
4870 draft_head_tensor(present(STEP37_DRAFTER), 45),
4871 "blk.45.nextn.shared_head_head.weight"
4872 );
4873 }
4874
4875 #[test]
4879 fn each_nextn_block_selects_its_own_head() {
4880 for n in 45..=47u32 {
4881 assert_eq!(
4882 draft_head_tensor(present(STEP37_DRAFTER), n),
4883 format!("blk.{n}.nextn.shared_head_head.weight")
4884 );
4885 }
4886 }
4887
4888 #[test]
4892 fn draft_without_a_nextn_head_falls_back_to_file_level_output() {
4893 let fr_spec: &[&str] = &["output.weight", "output_norm.weight", "d2t.weight"];
4894 assert_eq!(draft_head_tensor(present(fr_spec), 45), "output.weight");
4895 }
4896
4897 #[test]
4902 fn legacy_shared_head_is_probed_but_loses_to_shared_head_head() {
4903 let legacy_only: &[&str] = &["output.weight", "blk.45.nextn.shared_head.weight"];
4904 assert_eq!(
4905 draft_head_tensor(present(legacy_only), 45),
4906 "blk.45.nextn.shared_head.weight"
4907 );
4908
4909 let both: &[&str] = &[
4910 "output.weight",
4911 "blk.45.nextn.shared_head.weight",
4912 "blk.45.nextn.shared_head_head.weight",
4913 ];
4914 assert_eq!(
4915 draft_head_tensor(present(both), 45),
4916 "blk.45.nextn.shared_head_head.weight"
4917 );
4918 }
4919
4920 #[test]
4924 fn a_different_blocks_nextn_head_is_never_borrowed() {
4925 let wrong_block: &[&str] = &[
4926 "output.weight",
4927 "blk.46.nextn.shared_head_head.weight",
4928 "blk.47.nextn.shared_head_head.weight",
4929 ];
4930 assert_eq!(draft_head_tensor(present(wrong_block), 45), "output.weight");
4931 }
4932
4933 #[test]
4938 fn frspec_trim_prefers_the_nextn_blocks_own_head_name() {
4939 assert_eq!(
4940 frspec_trim_own_head_name(45),
4941 "blk.45.nextn.shared_head_head.weight"
4942 );
4943 assert_eq!(
4945 frspec_trim_own_head_name(45),
4946 format!("blk.{}.nextn.shared_head_head.weight", 45)
4947 );
4948 assert_eq!(
4949 frspec_trim_own_head_name(40),
4950 "blk.40.nextn.shared_head_head.weight"
4951 );
4952 }
4953}