Skip to main content

eredu_runtime/
expert.rs

1//! Runtime ownership boundary for routed expert acquisition and residency.
2
3use eredu_nn::{
4    DistributedNeuralBackend, GroupSelection, GroupedGatedProductOperator, GroupedNeuralBackend,
5    GroupedRelu2Operator, Tensor, TensorParallelGroupedOutput,
6};
7
8use crate::ExpertPass;
9use crate::{
10    observe_and_intervene, ActivationObserver, ParameterBankAccess, ParameterBankKey,
11    ReplicatedTextMaterializationTask, ReplicatedTextParameterOwner, RoutingObservation,
12    WeightLoweringKind,
13};
14
15/// One exact selected parameter in an independently addressable bank member.
16#[derive(Debug, Clone, Eq, PartialEq)]
17pub struct AddressableBankParameter {
18    binding_name: String,
19    task: ReplicatedTextMaterializationTask,
20    recipe: eredu_checkpoint::recipe::DerivedWeightRecipe,
21    source_output: eredu_checkpoint::recipe::RecipeMetadata,
22    selected_bytes: u64,
23    quantization_companions: Option<crate::QuantizationCompanionBindings>,
24}
25
26impl AddressableBankParameter {
27    /// Retains and validates one selected task and its member-local recipe.
28    pub fn new(
29        binding_name: impl Into<String>,
30        task: ReplicatedTextMaterializationTask,
31        recipe: eredu_checkpoint::recipe::DerivedWeightRecipe,
32        source_output: eredu_checkpoint::recipe::RecipeMetadata,
33        selected_bytes: u64,
34        quantization_companions: Option<crate::QuantizationCompanionBindings>,
35    ) -> Result<Self, AddressableBankMemberError> {
36        let binding_name = binding_name.into();
37        if binding_name.trim().is_empty() {
38            return Err(AddressableBankMemberError::InvalidParameter {
39                parameter: task.name().to_owned(),
40                detail: "addressable binding name is empty".into(),
41            });
42        }
43        task.source_recipe()
44            .map_err(|error| AddressableBankMemberError::InvalidParameter {
45                parameter: task.name().to_owned(),
46                detail: error.to_string(),
47            })?;
48        let descriptor = task.lowering_descriptor();
49        if descriptor.source() != task.source_encoding()
50            || descriptor.executable() != task.executable()
51            || descriptor.physical_shape() != task.physical_shape()
52            || descriptor.logical_shape() != task.logical_shape()
53        {
54            return Err(AddressableBankMemberError::InvalidParameter {
55                parameter: task.name().to_owned(),
56                detail: "selected source, executable, or lowering descriptor drifted".into(),
57            });
58        }
59        let declared_sources = task
60            .sources()
61            .iter()
62            .map(String::as_str)
63            .collect::<std::collections::BTreeSet<_>>();
64        let recipe_sources = recipe
65            .source_keys()
66            .into_iter()
67            .collect::<std::collections::BTreeSet<_>>();
68        if recipe_sources.is_empty() || !recipe_sources.is_subset(&declared_sources) {
69            return Err(AddressableBankMemberError::InvalidParameter {
70                parameter: task.name().to_owned(),
71                detail: "member recipe consumes sources outside the selected task".into(),
72            });
73        }
74        if source_output.byte_len() == 0 {
75            return Err(AddressableBankMemberError::ZeroSourceBytes {
76                parameter: task.name().to_owned(),
77            });
78        }
79        if selected_bytes == 0 {
80            return Err(AddressableBankMemberError::ZeroSelectedBytes {
81                parameter: task.name().to_owned(),
82            });
83        }
84        let transforms = matches!(
85            task.lowering(),
86            WeightLoweringKind::Transform | WeightLoweringKind::DerivedTransform
87        );
88        if !transforms && quantization_companions.is_some() {
89            return Err(AddressableBankMemberError::InvalidParameter {
90                parameter: task.name().to_owned(),
91                detail: "non-transform lowering declared local transform companions".into(),
92            });
93        }
94        if transforms
95            && quantization_companions.is_none()
96            && source_output.dtype() != &eredu_checkpoint::recipe::RecipeDtype::F4
97        {
98            return Err(AddressableBankMemberError::InvalidParameter {
99                parameter: task.name().to_owned(),
100                detail: "floating transform omitted its local output companions".into(),
101            });
102        }
103        if transforms
104            && source_output.dtype() != &eredu_checkpoint::recipe::RecipeDtype::F4
105            && task.output_companions().is_empty()
106        {
107            return Err(AddressableBankMemberError::InvalidParameter {
108                parameter: task.name().to_owned(),
109                detail: "floating transform omitted its exact selected output companions".into(),
110            });
111        }
112        if let Some(companions) = quantization_companions.as_ref() {
113            let declared_roles = task
114                .output_companions()
115                .iter()
116                .map(|companion| companion.role())
117                .collect::<std::collections::BTreeSet<_>>();
118            let mut bound_roles =
119                std::collections::BTreeSet::from([eredu_nn::LinearCompanionRole::Scale]);
120            if companions.affine_bias().is_some() {
121                bound_roles.insert(eredu_nn::LinearCompanionRole::AffineBias);
122            }
123            if declared_roles != bound_roles {
124                return Err(AddressableBankMemberError::InvalidParameter {
125                    parameter: task.name().to_owned(),
126                    detail: "selected quantization companion roles differ from exact outputs"
127                        .into(),
128                });
129            }
130            for companion in task.output_companions() {
131                let local = match companion.role() {
132                    eredu_nn::LinearCompanionRole::Scale => companions.scale(),
133                    eredu_nn::LinearCompanionRole::AffineBias => companions
134                        .affine_bias()
135                        .expect("validated affine-bias role has one binding"),
136                };
137                if companion.name() != local && !companion.name().ends_with(&format!(".{local}")) {
138                    return Err(AddressableBankMemberError::InvalidParameter {
139                        parameter: task.name().to_owned(),
140                        detail: format!(
141                            "local companion {local:?} differs from selected output {:?}",
142                            companion.name()
143                        ),
144                    });
145                }
146                let owner_matches = match (task.owner(), companion.owner()) {
147                    (
148                        ReplicatedTextParameterOwner::ExecutionUnit { group, unit },
149                        crate::ParameterGroupOwner::ExecutionUnit {
150                            group: companion_group,
151                            global_unit,
152                        },
153                    ) => group == companion_group.as_str() && unit == global_unit,
154                    (
155                        ReplicatedTextParameterOwner::StaticRole(role),
156                        crate::ParameterGroupOwner::StaticRole(companion_role),
157                    ) => role == companion_role,
158                    _ => false,
159                };
160                if !owner_matches {
161                    return Err(AddressableBankMemberError::InvalidParameter {
162                        parameter: task.name().to_owned(),
163                        detail: format!(
164                            "selected companion {:?} has a different owner",
165                            companion.name()
166                        ),
167                    });
168                }
169            }
170        }
171        let expected = selected_addressable_parameter_bytes(&task, &source_output)?;
172        if selected_bytes != expected {
173            return Err(AddressableBankMemberError::SelectedByteMismatch {
174                parameter: task.name().to_owned(),
175                expected,
176                actual: selected_bytes,
177            });
178        }
179        Ok(Self {
180            binding_name,
181            task,
182            recipe,
183            source_output,
184            selected_bytes,
185            quantization_companions,
186        })
187    }
188
189    /// Returns the local grouped-operator binding name.
190    pub fn binding_name(&self) -> &str {
191        &self.binding_name
192    }
193
194    /// Returns the complete authoritative selected materialization task.
195    pub const fn task(&self) -> &ReplicatedTextMaterializationTask {
196        &self.task
197    }
198
199    /// Returns the exact member-local source recipe.
200    pub const fn recipe(&self) -> &eredu_checkpoint::recipe::DerivedWeightRecipe {
201        &self.recipe
202    }
203
204    /// Returns admitted metadata for the member-local source recipe.
205    pub const fn source_output(&self) -> &eredu_checkpoint::recipe::RecipeMetadata {
206        &self.source_output
207    }
208
209    /// Returns source bytes before an optional lowering.
210    pub const fn source_bytes(&self) -> u64 {
211        self.source_output.byte_len()
212    }
213
214    /// Returns executable bytes after the selected lowering.
215    pub const fn selected_bytes(&self) -> u64 {
216        self.selected_bytes
217    }
218
219    /// Returns exact local scale and affine-bias binding names for a transform.
220    pub const fn quantization_companions(&self) -> Option<&crate::QuantizationCompanionBindings> {
221        self.quantization_companions.as_ref()
222    }
223}
224
225/// Exact generic storage member projected from an architecture-owned bank catalog.
226#[derive(Debug, Clone, Eq, PartialEq)]
227pub struct AddressableBankMember {
228    key: ParameterBankKey,
229    placement: AddressableBankMemberPlacement,
230    parameters: Vec<AddressableBankParameter>,
231    source_bytes: u64,
232    selected_bytes: u64,
233}
234
235/// Neutral placement class for one independently addressable member.
236#[derive(Debug, Clone, Copy, Eq, PartialEq)]
237#[non_exhaustive]
238pub enum AddressableBankDistribution {
239    /// The member is present on every execution rank.
240    Replicated,
241    /// The member follows an architecture-selected expert partition.
242    ExpertParallel,
243}
244
245/// Architecture-selected ownership retained with an addressable member.
246#[derive(Debug, Clone, Eq, PartialEq)]
247pub struct AddressableBankMemberPlacement {
248    owner_group: crate::ExecutionGroupId,
249    owner_unit: usize,
250    unit_path: String,
251    distribution: AddressableBankDistribution,
252    owner_rank: Option<usize>,
253}
254
255impl AddressableBankMemberPlacement {
256    /// Creates exact architecture-global member placement.
257    pub fn new(
258        owner_group: crate::ExecutionGroupId,
259        owner_unit: usize,
260        unit_path: impl Into<String>,
261        distribution: AddressableBankDistribution,
262    ) -> Result<Self, AddressableBankMemberError> {
263        let unit_path = unit_path.into();
264        if unit_path.trim().is_empty() {
265            return Err(AddressableBankMemberError::InvalidPlacement(
266                "addressable member unit path is empty".into(),
267            ));
268        }
269        Ok(Self {
270            owner_group,
271            owner_unit,
272            unit_path,
273            distribution,
274            owner_rank: None,
275        })
276    }
277
278    /// Binds this selected member projection to one global partition rank.
279    pub fn with_owner_rank(mut self, owner_rank: usize) -> Self {
280        self.owner_rank = Some(owner_rank);
281        self
282    }
283
284    /// Returns the architecture execution group that owns this member.
285    pub const fn owner_group(&self) -> &crate::ExecutionGroupId {
286        &self.owner_group
287    }
288    /// Returns the architecture-global execution-unit index.
289    pub const fn owner_unit(&self) -> usize {
290        self.owner_unit
291    }
292    /// Returns the stable architecture path of the owning unit.
293    pub fn unit_path(&self) -> &str {
294        &self.unit_path
295    }
296    /// Returns the architecture-selected distribution class.
297    pub const fn distribution(&self) -> AddressableBankDistribution {
298        self.distribution
299    }
300    /// Returns the global partition rank after rank-local projection.
301    pub const fn owner_rank(&self) -> Option<usize> {
302        self.owner_rank
303    }
304}
305
306impl AddressableBankMember {
307    /// Validates one atomic member and all of its selected parameter tasks.
308    pub fn new(
309        key: ParameterBankKey,
310        placement: AddressableBankMemberPlacement,
311        parameters: impl IntoIterator<Item = AddressableBankParameter>,
312    ) -> Result<Self, AddressableBankMemberError> {
313        let parameters = parameters.into_iter().collect::<Vec<_>>();
314        if parameters.is_empty() {
315            return Err(AddressableBankMemberError::EmptyMember { key });
316        }
317        if placement.owner_unit() != key.unit() {
318            return Err(AddressableBankMemberError::InvalidPlacement(format!(
319                "addressable member unit {} differs from placement unit {}",
320                key.unit(),
321                placement.owner_unit()
322            )));
323        }
324        let mut bindings = std::collections::BTreeSet::new();
325        let mut targets = std::collections::BTreeSet::new();
326        let mut source_bytes = 0u64;
327        let mut selected_bytes = 0u64;
328        for parameter in &parameters {
329            if !bindings.insert(parameter.binding_name())
330                || !targets.insert(parameter.task().name())
331            {
332                return Err(AddressableBankMemberError::DuplicateParameter { key });
333            }
334            if !matches!(
335                parameter.task().owner(),
336                ReplicatedTextParameterOwner::ExecutionUnit { group, unit }
337                    if *unit == placement.owner_unit()
338                        && group == placement.owner_group().as_str()
339            ) {
340                return Err(AddressableBankMemberError::InvalidParameter {
341                    parameter: parameter.task().name().to_owned(),
342                    detail: "selected task has a non-bank owner".into(),
343                });
344            }
345            source_bytes = source_bytes
346                .checked_add(parameter.source_bytes())
347                .ok_or(AddressableBankMemberError::SourceByteOverflow { key })?;
348            selected_bytes = selected_bytes
349                .checked_add(parameter.selected_bytes())
350                .ok_or(AddressableBankMemberError::SelectedByteOverflow { key })?;
351        }
352        Ok(Self {
353            key,
354            placement,
355            parameters,
356            source_bytes,
357            selected_bytes,
358        })
359    }
360
361    /// Returns the generic bank key selected by neutral composition.
362    pub const fn key(&self) -> ParameterBankKey {
363        self.key
364    }
365
366    /// Exact group/unit/distribution/rank ownership selected for this member.
367    pub const fn placement(&self) -> &AddressableBankMemberPlacement {
368        &self.placement
369    }
370
371    /// Retains the global rank selected by a partition projection.
372    pub fn with_owner_rank(mut self, owner_rank: usize) -> Self {
373        self.placement = self.placement.with_owner_rank(owner_rank);
374        self
375    }
376
377    /// Returns every exact selected parameter in deterministic binding order.
378    pub fn parameters(&self) -> &[AddressableBankParameter] {
379        &self.parameters
380    }
381
382    /// Returns admitted source bytes before optional lowerings.
383    pub const fn source_bytes(&self) -> u64 {
384        self.source_bytes
385    }
386
387    /// Returns executable bytes after selected lowerings.
388    pub const fn selected_bytes(&self) -> u64 {
389        self.selected_bytes
390    }
391}
392
393/// Backend-neutral transform retained for one addressable binding.
394#[derive(Debug, Clone, Eq, PartialEq)]
395pub struct AddressableBindingTransform {
396    quantization: eredu_checkpoint::WeightQuantization,
397    companion_dtype: eredu_checkpoint::recipe::RecipeDtype,
398}
399
400impl AddressableBindingTransform {
401    /// Selected packed executable format.
402    pub const fn quantization(&self) -> eredu_checkpoint::WeightQuantization {
403        self.quantization
404    }
405    /// Selected scale and affine-bias scalar representation.
406    pub const fn companion_dtype(&self) -> &eredu_checkpoint::recipe::RecipeDtype {
407        &self.companion_dtype
408    }
409}
410
411/// Canonical binding and transformation plan for one addressable member.
412#[derive(Debug, Clone)]
413pub struct AddressableBankBindingPlan {
414    key: ParameterBankKey,
415    bindings: Vec<crate::WeightBinding>,
416    transformations: std::collections::BTreeMap<String, AddressableBindingTransform>,
417    selected_bytes: u64,
418    placement: AddressableBankMemberPlacement,
419}
420
421impl AddressableBankBindingPlan {
422    /// Generic member identity.
423    pub const fn key(&self) -> ParameterBankKey {
424        self.key
425    }
426    /// Source-side canonical bindings.
427    pub fn bindings(&self) -> &[crate::WeightBinding] {
428        &self.bindings
429    }
430    /// Per-binding transforms selected by architecture admission.
431    pub const fn transformations(
432        &self,
433    ) -> &std::collections::BTreeMap<String, AddressableBindingTransform> {
434        &self.transformations
435    }
436    /// Exact executable bytes after all transforms.
437    pub const fn selected_bytes(&self) -> u64 {
438        self.selected_bytes
439    }
440    /// Exact rank-local architecture placement.
441    pub const fn placement(&self) -> &AddressableBankMemberPlacement {
442        &self.placement
443    }
444    /// Consumes the complete canonical member plan.
445    #[allow(clippy::type_complexity)]
446    pub fn into_parts(
447        self,
448    ) -> (
449        ParameterBankKey,
450        Vec<crate::WeightBinding>,
451        std::collections::BTreeMap<String, AddressableBindingTransform>,
452        u64,
453        AddressableBankMemberPlacement,
454    ) {
455        (
456            self.key,
457            self.bindings,
458            self.transformations,
459            self.selected_bytes,
460            self.placement,
461        )
462    }
463}
464
465/// Validates exact addressable tasks and derives their singular canonical binding plans.
466pub fn plan_addressable_bank_bindings<L, E>(
467    members: &[AddressableBankMember],
468    source: &dyn eredu_checkpoint::store::CheckpointSource,
469    mut lower_mxfp4: L,
470) -> Result<Vec<AddressableBankBindingPlan>, AddressableBankMemberError>
471where
472    L: FnMut(
473        &ReplicatedTextMaterializationTask,
474        eredu_checkpoint::recipe::DerivedWeightRecipe,
475        &dyn eredu_checkpoint::store::CheckpointSource,
476    ) -> Result<eredu_checkpoint::recipe::DerivedWeightRecipe, E>,
477    E: std::fmt::Display,
478{
479    let mut plans = Vec::with_capacity(members.len());
480    for member in members {
481        let mut bindings = Vec::with_capacity(member.parameters().len());
482        let mut transformations = std::collections::BTreeMap::new();
483        for parameter in member.parameters() {
484            let task = parameter.task();
485            let declared = task
486                .sources()
487                .iter()
488                .map(String::as_str)
489                .collect::<std::collections::BTreeSet<_>>();
490            let physical = task
491                .physical_sources()
492                .iter()
493                .map(|item| item.catalog_key())
494                .collect::<std::collections::BTreeSet<_>>();
495            if declared != physical || physical.len() != task.physical_sources().len() {
496                return Err(AddressableBankMemberError::InvalidParameter {
497                    parameter: task.name().to_owned(),
498                    detail: "selected physical provenance does not exactly cover task sources"
499                        .into(),
500                });
501            }
502            for admitted in task.physical_sources() {
503                let actual = source
504                    .source_provenance(admitted.catalog_key())
505                    .map_err(|error| AddressableBankMemberError::InvalidParameter {
506                        parameter: task.name().to_owned(),
507                        detail: error.to_string(),
508                    })?;
509                let metadata = source
510                    .source_metadata(admitted.catalog_key())
511                    .map_err(|error| AddressableBankMemberError::InvalidParameter {
512                        parameter: task.name().to_owned(),
513                        detail: error.to_string(),
514                    })?;
515                if actual.catalog_key != admitted.catalog_key()
516                    || actual.physical_tensor != admitted.tensor()
517                    || actual.output != admitted.output()
518                    || actual.backing_shard.as_deref() != Some(admitted.shard())
519                    || actual.source_encoding != *admitted.source_encoding()
520                    || metadata.encoded_byte_len != admitted.encoded_byte_len()
521                {
522                    return Err(AddressableBankMemberError::InvalidParameter {
523                        parameter: task.name().to_owned(),
524                        detail: format!(
525                            "source {:?} differs from admitted provenance",
526                            admitted.catalog_key()
527                        ),
528                    });
529                }
530            }
531            let mut recipe = parameter.recipe().clone();
532            let inferred = recipe.infer(source).map_err(|error| {
533                AddressableBankMemberError::InvalidParameter {
534                    parameter: task.name().to_owned(),
535                    detail: error.to_string(),
536                }
537            })?;
538            if &inferred != parameter.source_output() {
539                return Err(AddressableBankMemberError::InvalidParameter {
540                    parameter: task.name().to_owned(),
541                    detail: "member-local recipe output drifted".into(),
542                });
543            }
544            if task.executable() == eredu_checkpoint::LinearFormat::MxFp4
545                && inferred.dtype() == &eredu_checkpoint::recipe::RecipeDtype::F4
546                && parameter.quantization_companions().is_none()
547            {
548                recipe = lower_mxfp4(task, recipe, source).map_err(|error| {
549                    AddressableBankMemberError::InvalidParameter {
550                        parameter: task.name().to_owned(),
551                        detail: error.to_string(),
552                    }
553                })?;
554            }
555            let metadata = recipe.infer(source).map_err(|error| {
556                AddressableBankMemberError::InvalidParameter {
557                    parameter: task.name().to_owned(),
558                    detail: error.to_string(),
559                }
560            })?;
561            let mut binding = crate::WeightBinding::from_recipe(
562                parameter.binding_name(),
563                recipe,
564                metadata.byte_len(),
565            )
566            .and_then(|binding| binding.with_logical_target(task.name()))
567            .map_err(|error| AddressableBankMemberError::InvalidParameter {
568                parameter: task.name().to_owned(),
569                detail: error.to_string(),
570            })?;
571            if let Some(companions) = parameter.quantization_companions() {
572                let quantization = task.executable().weight_quantization().ok_or_else(|| {
573                    AddressableBankMemberError::InvalidParameter {
574                        parameter: task.name().to_owned(),
575                        detail: "transformed task has no packed format".into(),
576                    }
577                })?;
578                transformations.insert(
579                    parameter.binding_name().to_owned(),
580                    AddressableBindingTransform {
581                        quantization,
582                        companion_dtype: parameter.source_output().dtype().clone(),
583                    },
584                );
585                binding = binding
586                    .with_quantization_companions(
587                        companions.scale(),
588                        companions.affine_bias().map(str::to_owned),
589                    )
590                    .map_err(|error| AddressableBankMemberError::InvalidParameter {
591                        parameter: task.name().to_owned(),
592                        detail: error.to_string(),
593                    })?;
594            }
595            bindings.push(binding);
596        }
597        crate::WeightBindingPlan::new(&bindings).map_err(|error| {
598            AddressableBankMemberError::InvalidParameter {
599                parameter: format!("{:?}", member.key()),
600                detail: error.to_string(),
601            }
602        })?;
603        plans.push(AddressableBankBindingPlan {
604            key: member.key(),
605            bindings,
606            transformations,
607            selected_bytes: member.selected_bytes(),
608            placement: member.placement().clone(),
609        });
610    }
611    Ok(plans)
612}
613
614/// Computes executable storage bytes for one admitted member-local task output.
615///
616/// A task's own derived output describes whole-parameter derivation before a
617/// member projection. `metadata` instead describes the exact member-local
618/// recipe retained by [`AddressableBankParameter`], including rank-local
619/// sharding. This function is the single neutral authority for their selected
620/// executable byte geometry.
621pub fn selected_addressable_parameter_bytes(
622    task: &ReplicatedTextMaterializationTask,
623    metadata: &eredu_checkpoint::recipe::RecipeMetadata,
624) -> Result<u64, AddressableBankMemberError> {
625    if !matches!(
626        task.lowering(),
627        WeightLoweringKind::Transform | WeightLoweringKind::DerivedTransform
628    ) {
629        return Ok(metadata.byte_len());
630    }
631    let quantization = task.executable().weight_quantization().ok_or_else(|| {
632        AddressableBankMemberError::InvalidParameter {
633            parameter: task.name().to_owned(),
634            detail: "transform lowering has no packed executable format".into(),
635        }
636    })?;
637    if matches!(
638        quantization,
639        eredu_checkpoint::WeightQuantization::GgufIQuant { .. }
640    ) {
641        return Err(AddressableBankMemberError::InvalidParameter {
642            parameter: task.name().to_owned(),
643            detail: "load-time transform selected checkpoint-native GGUF encoding".into(),
644        });
645    }
646    if task.lowering_descriptor().packed_axis() != metadata.shape().len().checked_sub(1) {
647        return Err(AddressableBankMemberError::InvalidParameter {
648            parameter: task.name().to_owned(),
649            detail: "transform packed axis is not the final logical matrix axis".into(),
650        });
651    }
652    let shape = metadata.shape();
653    let (&columns, row_shape) =
654        shape
655            .split_last()
656            .ok_or_else(|| AddressableBankMemberError::InvalidParameter {
657                parameter: task.name().to_owned(),
658                detail: "transform target is not a matrix".into(),
659            })?;
660    let rows = row_shape
661        .iter()
662        .try_fold(1u64, |total, dimension| {
663            total.checked_mul(*dimension as u64)
664        })
665        .ok_or_else(|| AddressableBankMemberError::InvalidParameter {
666            parameter: task.name().to_owned(),
667            detail: "transform row geometry overflowed".into(),
668        })?;
669    let group = usize::try_from(quantization.group_size()).map_err(|_| {
670        AddressableBankMemberError::InvalidParameter {
671            parameter: task.name().to_owned(),
672            detail: "transform group size is invalid".into(),
673        }
674    })?;
675    if group == 0 || !columns.is_multiple_of(group) || !columns.is_multiple_of(32) {
676        return Err(AddressableBankMemberError::InvalidParameter {
677            parameter: task.name().to_owned(),
678            detail: "transform geometry is incompatible with its packed format".into(),
679        });
680    }
681    let groups = (columns / group) as u64;
682    let packed = (columns as u64)
683        .checked_mul(quantization.bits() as u64)
684        .and_then(|bits| bits.checked_div(8))
685        .ok_or_else(|| AddressableBankMemberError::InvalidParameter {
686            parameter: task.name().to_owned(),
687            detail: "packed row byte geometry overflowed".into(),
688        })?;
689    let scalar_bytes = metadata.dtype().bit_width().map_err(|error| {
690        AddressableBankMemberError::InvalidParameter {
691            parameter: task.name().to_owned(),
692            detail: error.to_string(),
693        }
694    })? / 8;
695    let companion = if matches!(quantization, eredu_checkpoint::WeightQuantization::MxFp4) {
696        groups
697    } else {
698        groups.checked_mul(scalar_bytes).ok_or_else(|| {
699            AddressableBankMemberError::InvalidParameter {
700                parameter: task.name().to_owned(),
701                detail: "scale byte geometry overflowed".into(),
702            }
703        })?
704    };
705    let bias = if quantization.has_biases() {
706        companion
707    } else {
708        0
709    };
710    rows.checked_mul(
711        packed
712            .checked_add(companion)
713            .and_then(|bytes| bytes.checked_add(bias))
714            .ok_or_else(|| AddressableBankMemberError::InvalidParameter {
715                parameter: task.name().to_owned(),
716                detail: "selected row byte geometry overflowed".into(),
717            })?,
718    )
719    .ok_or_else(|| AddressableBankMemberError::InvalidParameter {
720        parameter: task.name().to_owned(),
721        detail: "selected byte geometry overflowed".into(),
722    })
723}
724
725/// Invalid generic addressable-bank member projection.
726#[derive(Debug, Clone, Eq, PartialEq, thiserror::Error)]
727pub enum AddressableBankMemberError {
728    /// Member placement was empty or disagreed with task ownership.
729    #[error("invalid addressable bank member placement: {0}")]
730    InvalidPlacement(String),
731    /// A member contained no parameter task.
732    #[error("addressable bank member {key:?} is empty")]
733    EmptyMember {
734        /// Invalid member identity.
735        key: ParameterBankKey,
736    },
737    /// A member repeated a local binding or logical task.
738    #[error("addressable bank member {key:?} repeats a parameter")]
739    DuplicateParameter {
740        /// Invalid member identity.
741        key: ParameterBankKey,
742    },
743    /// One parameter's exact task closure was inconsistent.
744    #[error("invalid addressable bank parameter {parameter:?}: {detail}")]
745    InvalidParameter {
746        /// Invalid logical parameter.
747        parameter: String,
748        /// Exact validation failure.
749        detail: String,
750    },
751    /// A source recipe selected no bytes.
752    #[error("addressable bank parameter {parameter:?} source byte geometry is zero")]
753    ZeroSourceBytes {
754        /// Invalid logical parameter.
755        parameter: String,
756    },
757    /// Source binding byte accounting overflowed.
758    #[error("addressable bank member {key:?} source byte geometry overflowed")]
759    SourceByteOverflow {
760        /// Invalid member.
761        key: ParameterBankKey,
762    },
763    /// Selected executable byte accounting overflowed.
764    #[error("addressable bank member {key:?} selected byte geometry overflowed")]
765    SelectedByteOverflow {
766        /// Invalid member identity.
767        key: ParameterBankKey,
768    },
769    /// Selected executable storage was empty.
770    #[error("addressable bank parameter {parameter:?} selected byte geometry is zero")]
771    ZeroSelectedBytes {
772        /// Invalid parameter.
773        parameter: String,
774    },
775    /// Selected executable byte geometry differed from the task.
776    #[error("addressable bank parameter {parameter:?} selected bytes differ: expected {expected}, got {actual}")]
777    SelectedByteMismatch {
778        /// Invalid logical parameter.
779        parameter: String,
780        /// Authoritative computed byte total.
781        expected: u64,
782        /// Supplied selected byte total.
783        actual: u64,
784    },
785}
786
787/// Generic indexed tensor movement required by bounded grouped execution.
788///
789/// Implementations expose integer-index discovery and tensor movement without
790/// receiving architecture plans, bank meaning, or text lifecycle policy.
791pub trait IndexedMovement<B>
792where
793    B: GroupedNeuralBackend,
794{
795    /// Indexed movement failure.
796    type Error;
797
798    /// Returns deterministic demand counts for integer indices below `upper_bound`.
799    fn index_demands(
800        &mut self,
801        indices: &B::Tensor,
802        upper_bound: usize,
803        context: &<B::Tensor as Tensor>::Context,
804    ) -> Result<Vec<(usize, u64)>, Self::Error>;
805
806    /// Rewrites source indices through one exact source-to-compact mapping.
807    fn remap_indices(
808        &mut self,
809        indices: &B::Tensor,
810        mapping: &[(usize, usize)],
811        context: &<B::Tensor as Tensor>::Context,
812    ) -> Result<B::Tensor, Self::Error>;
813
814    /// Selects a contiguous range along the leading row axis.
815    fn select_rows(
816        &mut self,
817        value: &B::Tensor,
818        start: usize,
819        end: usize,
820        context: &<B::Tensor as Tensor>::Context,
821    ) -> Result<B::Tensor, Self::Error>;
822
823    /// Concatenates row partitions in their original order.
824    fn concatenate_rows(
825        &mut self,
826        values: &[B::Tensor],
827        context: &<B::Tensor as Tensor>::Context,
828    ) -> Result<B::Tensor, Self::Error>;
829}
830
831/// Backend-neutral tensor movement needed by an expert-exchange protocol.
832///
833/// Architecture code supplies already validated row and flattened-route
834/// indices. Implementations retain tensor storage and completion ownership;
835/// they do not receive expert identities, topology, or model-family policy.
836pub trait ExpertRouteTensorMovement<T> {
837    /// Tensor movement failure.
838    type Error;
839
840    /// Returns the logical tensor shape without materializing its values.
841    fn shape(&self, value: &T) -> Vec<usize>;
842
843    /// Duplicates and reorders leading-axis rows in the supplied order.
844    fn gather_rows(&mut self, value: &T, rows: &[usize]) -> Result<T, Self::Error>;
845
846    /// Selects flattened route scalars and returns them as `[routes, 1]`.
847    fn gather_route_values(
848        &mut self,
849        value: &T,
850        flattened_routes: &[usize],
851    ) -> Result<T, Self::Error>;
852
853    /// Additively combines route rows into their architecture source rows.
854    ///
855    /// Every input row must be consumed exactly once. Repeated destination
856    /// rows are intentional and implement weighted routed-expert summation.
857    fn scatter_add_rows(
858        &mut self,
859        value: T,
860        destination_rows: &[usize],
861        output_rows: usize,
862    ) -> Result<T, Self::Error>;
863}
864
865/// Opaque variable-count transport used by architecture-owned expert routing.
866///
867/// Implementations must preserve peer-block and within-block order, validate
868/// every tensor against the selected communication requirement, and retain all
869/// native resources until the exact completion has finished.
870pub trait ExpertRouteExchange<T> {
871    /// Communication or metadata transport failure.
872    type Error;
873
874    /// Exchanges one tensor whose leading rows match the supplied peer counts.
875    fn exchange_tensor(
876        &mut self,
877        counts: &crate::CommunicationPeerCounts,
878        value: T,
879    ) -> Result<T, Self::Error>;
880
881    /// Exchanges one unsigned metadata value per leading tensor row.
882    fn exchange_indices(
883        &mut self,
884        counts: &crate::CommunicationPeerCounts,
885        values: Vec<usize>,
886    ) -> Result<Vec<usize>, Self::Error>;
887}
888
889/// Architecture-selected combination for one expert-exchange batch.
890#[derive(Debug, Clone, Copy, Eq, PartialEq)]
891#[non_exhaustive]
892pub enum ExpertRouteCombination {
893    /// Apply each route coefficient once, then add routes targeting one token.
894    CoefficientWeightedSum,
895}
896
897/// One owner-local grouped batch submitted after expert exchange.
898pub struct AddressableExpertRouteRequest<'a, T> {
899    /// Global execution unit containing the addressable expert bank.
900    pub unit: usize,
901    /// Rows received from every source peer.
902    pub input: &'a T,
903    /// Checkpoint-global expert identity for every received row.
904    ///
905    /// Addressable storage keys must be derived from this identity. It is
906    /// deliberately kept separate from `owner_local_experts`, whose values
907    /// are valid only as indices into the rank-local grouped operator.
908    pub global_experts: &'a [usize],
909    /// Dense owner-local expert identity for every received row.
910    pub owner_local_experts: &'a [usize],
911    /// Selected router scores aligned one-for-one with received rows.
912    pub selected_scores: &'a T,
913    /// Final route coefficients aligned one-for-one with received rows.
914    pub coefficients: &'a T,
915    /// Prefill or decode execution classification.
916    pub pass: ExpertPass,
917    /// Storage access classification derived from `pass`.
918    pub access: ParameterBankAccess,
919    /// Architecture-declared route combination.
920    pub combination: ExpertRouteCombination,
921}
922
923impl<T> AddressableExpertRouteRequest<'_, T> {
924    /// Returns the only valid addressable-bank key for one routed row.
925    ///
926    /// The owner-local ID is intentionally not accepted here: it addresses the
927    /// compact grouped operator, not checkpoint-global storage.
928    pub fn addressable_bank_key(&self, row: usize) -> Option<ParameterBankKey> {
929        self.global_experts
930            .get(row)
931            .copied()
932            .map(|global| ParameterBankKey::new(self.unit, global))
933    }
934
935    /// Returns the rank-local grouped-operator ID for one routed row.
936    pub fn owner_local_execution_id(&self, row: usize) -> Option<usize> {
937        self.owner_local_experts.get(row).copied()
938    }
939}
940
941/// Local addressable grouped execution used by expert exchange.
942///
943/// The provider must consume every submitted row exactly once, select its
944/// corresponding owner-local expert, and apply its route coefficient exactly
945/// once. Acquired bank resources remain provider-owned until the returned
946/// tensor is natively complete.
947pub trait AddressableExpertRouteProvider<T> {
948    /// Acquisition or grouped execution failure.
949    type Error;
950
951    /// Executes one owner-local grouped batch.
952    fn execute_addressable_routes(
953        &mut self,
954        request: AddressableExpertRouteRequest<'_, T>,
955    ) -> Result<T, Self::Error>;
956
957    /// Executes one owner-local grouped batch while retaining tensor-parallel
958    /// reduction structure.
959    ///
960    /// Providers without rank-local TP work inherit complete-output behavior.
961    /// A TP provider overrides this method and returns its reducible activation
962    /// contribution plus the optional selection-weighted post-reduction bias.
963    /// The exchange protocol returns both values to their source-token order;
964    /// it must not add the bias before the caller's tensor all-sum.
965    fn execute_addressable_routes_tensor_parallel(
966        &mut self,
967        request: AddressableExpertRouteRequest<'_, T>,
968    ) -> Result<RoutedExpertTensorParallelOutput<T>, Self::Error> {
969        self.execute_addressable_routes(request)
970            .map(RoutedExpertTensorParallelOutput::Complete)
971    }
972}
973
974/// Exact generic request for an independently addressable bank acquisition.
975#[derive(Debug, Clone, Copy)]
976pub struct ParameterBankAcquisition<'a> {
977    entries: &'a [(ParameterBankKey, u64)],
978    access: ParameterBankAccess,
979}
980
981impl<'a> ParameterBankAcquisition<'a> {
982    /// Creates one deterministic acquisition request in compact-bank order.
983    pub const fn new(entries: &'a [(ParameterBankKey, u64)], access: ParameterBankAccess) -> Self {
984        Self { entries, access }
985    }
986
987    /// Returns generic bank keys and duplicate-preserving demand counts.
988    pub const fn entries(&self) -> &'a [(ParameterBankKey, u64)] {
989        self.entries
990    }
991
992    /// Returns the selected generic storage access class.
993    pub const fn access(&self) -> ParameterBankAccess {
994        self.access
995    }
996}
997
998/// Generic addressable storage and grouped-operator construction mechanisms.
999///
1000/// The mechanism receives already translated bank keys, compact specifications,
1001/// and access classes. Architecture identity, routing policy, global identity
1002/// mapping, chunking, and text-session behavior remain outside this contract.
1003pub trait AddressableGroupedBank<B>
1004where
1005    B: GroupedNeuralBackend,
1006{
1007    /// Live native storage retained across grouped execution.
1008    type Acquisition;
1009    /// Generic bank telemetry snapshot.
1010    type Report;
1011    /// Storage, transfer, lowering, or construction failure.
1012    type Error;
1013
1014    /// Returns the selected byte geometry for one admitted bank member.
1015    fn member_bytes(&self, key: ParameterBankKey) -> Option<u64>;
1016
1017    /// Acquires exact generic keys in caller-supplied compact order.
1018    fn acquire(
1019        &mut self,
1020        request: ParameterBankAcquisition<'_>,
1021        context: &<B::Tensor as Tensor>::Context,
1022    ) -> Result<Self::Acquisition, Self::Error>;
1023
1024    /// Constructs one compact gated-product operator from acquired bindings.
1025    fn gated_product_groups(
1026        &mut self,
1027        acquisition: &Self::Acquisition,
1028        spec: &eredu_nn::GroupedGatedProductSpec,
1029        context: &<B::Tensor as Tensor>::Context,
1030    ) -> Result<B::GatedProductGroups, Self::Error>;
1031
1032    /// Constructs one compact ReLU-squared operator from acquired bindings.
1033    fn relu2_groups(
1034        &mut self,
1035        acquisition: &Self::Acquisition,
1036        spec: &eredu_nn::GroupedRelu2Spec,
1037        context: &<B::Tensor as Tensor>::Context,
1038    ) -> Result<B::Relu2Groups, Self::Error>;
1039
1040    /// Retains acquired storage until the grouped output is natively complete.
1041    fn complete(
1042        &mut self,
1043        acquisition: Self::Acquisition,
1044        output: &B::Tensor,
1045        context: &<B::Tensor as Tensor>::Context,
1046    ) -> Result<(), Self::Error>;
1047
1048    /// Returns generic key, byte, tier, acquisition, and eviction telemetry.
1049    fn report(&self) -> Result<Self::Report, Self::Error>;
1050}
1051
1052/// Mechanism-only lookup of one grouped operator in an addressable parameter bank.
1053pub trait AddressableGatedProductBank<B>
1054where
1055    B: GroupedNeuralBackend,
1056{
1057    /// Bank lookup or construction failure.
1058    type Error;
1059
1060    /// Resolves one generic bank key and exact grouped construction specification.
1061    fn acquire(
1062        &mut self,
1063        key: ParameterBankKey,
1064        spec: &eredu_nn::GroupedGatedProductSpec,
1065        context: &<B::Tensor as Tensor>::Context,
1066    ) -> Result<&mut B::GatedProductGroups, Self::Error>;
1067}
1068
1069/// One architecture route batch submitted to a runtime expert provider.
1070pub struct RoutedExpertRequest<'a, T> {
1071    /// Global decoder layer requesting experts.
1072    pub layer: usize,
1073    /// Flattened token rows submitted to the selected experts.
1074    pub input: &'a T,
1075    /// Backend-native selected expert IDs, scores, and weights.
1076    pub routes: &'a GroupSelection<T>,
1077    /// Whether this route batch belongs to prefill or decode.
1078    pub pass: ExpertPass,
1079}
1080
1081impl<T> RoutedExpertRequest<'_, T> {
1082    /// Projects architecture execution semantics into the storage workload
1083    /// class exposed to backend parameter-bank mechanisms.
1084    pub const fn parameter_bank_access(&self) -> ParameterBankAccess {
1085        self.pass.parameter_bank_access()
1086    }
1087}
1088
1089/// Provider result that distinguishes complete outputs from rank-local TP work.
1090pub enum RoutedExpertTensorParallelOutput<T> {
1091    /// Provider already completed every required collective and bias addition.
1092    Complete(T),
1093    /// Caller must all-sum `reducible`, then add `post_reduce` exactly once.
1094    Partial(TensorParallelGroupedOutput<T>),
1095}
1096
1097/// Completes one rank-local expert output with one all-sum and one post-bias add.
1098pub fn reduce_tensor_parallel_expert_output<B>(
1099    output: TensorParallelGroupedOutput<B::Tensor>,
1100    parallel: &B::ParallelContext,
1101    context: &<B::Tensor as Tensor>::Context,
1102) -> Result<B::Tensor, eredu_nn::Error>
1103where
1104    B: GroupedNeuralBackend + DistributedNeuralBackend,
1105{
1106    let reduced = B::sum_parallel(output.reducible().clone(), parallel, context)?;
1107    match output.post_reduce().cloned() {
1108        Some(bias) => reduced.add(&bias, context),
1109        None => Ok(reduced),
1110    }
1111}
1112
1113/// Combines two rank-local expert partials without introducing another collective.
1114pub fn combine_tensor_parallel_expert_outputs<B>(
1115    left: TensorParallelGroupedOutput<B::Tensor>,
1116    right: TensorParallelGroupedOutput<B::Tensor>,
1117    context: &<B::Tensor as Tensor>::Context,
1118) -> Result<TensorParallelGroupedOutput<B::Tensor>, eredu_nn::Error>
1119where
1120    B: GroupedNeuralBackend,
1121{
1122    let post_reduce = match (left.post_reduce().cloned(), right.post_reduce().cloned()) {
1123        (Some(left), Some(right)) => Some(left.add(&right, context)?),
1124        (Some(bias), None) | (None, Some(bias)) => Some(bias),
1125        (None, None) => None,
1126    };
1127    Ok(TensorParallelGroupedOutput::new(
1128        left.reducible().add(right.reducible(), context)?,
1129        post_reduce,
1130    ))
1131}
1132
1133/// Combines routed/shared provider outputs while requiring one coherent TP mode.
1134pub fn combine_routed_expert_tensor_parallel<B>(
1135    left: RoutedExpertTensorParallelOutput<B::Tensor>,
1136    right: RoutedExpertTensorParallelOutput<B::Tensor>,
1137    context: &<B::Tensor as Tensor>::Context,
1138) -> Result<RoutedExpertTensorParallelOutput<B::Tensor>, eredu_nn::Error>
1139where
1140    B: GroupedNeuralBackend,
1141{
1142    match (left, right) {
1143        (
1144            RoutedExpertTensorParallelOutput::Complete(left),
1145            RoutedExpertTensorParallelOutput::Complete(right),
1146        ) => Ok(RoutedExpertTensorParallelOutput::Complete(
1147            left.add(&right, context)?,
1148        )),
1149        (
1150            RoutedExpertTensorParallelOutput::Partial(left),
1151            RoutedExpertTensorParallelOutput::Partial(right),
1152        ) => combine_tensor_parallel_expert_outputs::<B>(left, right, context)
1153            .map(RoutedExpertTensorParallelOutput::Partial),
1154        _ => Err(eredu_nn::Error::backend(
1155            "provider mixed complete and rank-local expert outputs in one block",
1156        )),
1157    }
1158}
1159
1160/// Completes a provider TP result while preserving provider-owned collectives.
1161pub fn reduce_routed_expert_tensor_parallel<B>(
1162    output: RoutedExpertTensorParallelOutput<B::Tensor>,
1163    parallel: &B::ParallelContext,
1164    context: &<B::Tensor as Tensor>::Context,
1165) -> Result<B::Tensor, eredu_nn::Error>
1166where
1167    B: GroupedNeuralBackend + DistributedNeuralBackend,
1168{
1169    match output {
1170        RoutedExpertTensorParallelOutput::Complete(output) => Ok(output),
1171        RoutedExpertTensorParallelOutput::Partial(output) => {
1172            reduce_tensor_parallel_expert_output::<B>(output, parallel, context)
1173        }
1174    }
1175}
1176
1177/// Runtime boundary for resident or independently cached routed experts.
1178///
1179/// Implementations own identity ordering, acquisition, leases, chunking,
1180/// budgets, and residency reports. They keep every lease alive until the
1181/// backend-native routed result is safe to return. The backend retains tensor
1182/// storage, transfers, compact-bank construction, and execution kernels.
1183pub trait RoutedExpertProvider<B>
1184where
1185    B: GroupedNeuralBackend,
1186{
1187    /// Provider-specific acquisition or execution failure.
1188    type Error;
1189
1190    /// Executes one typed route batch while retaining its acquired resources.
1191    fn forward_grouped(
1192        &mut self,
1193        resident_bank: &mut B::GatedProductGroups,
1194        request: RoutedExpertRequest<'_, B::Tensor>,
1195        context: &<B::Tensor as Tensor>::Context,
1196    ) -> Result<B::Tensor, Self::Error>;
1197
1198    /// Executes destination-local rows that were already expanded to one
1199    /// owner-local expert per row by the neutral expert exchange.
1200    ///
1201    /// The compact request deliberately has route cardinality one; providers
1202    /// must not compare it with the architecture's original top-k cardinality.
1203    fn forward_compact_grouped(
1204        &mut self,
1205        resident_bank: &mut B::GatedProductGroups,
1206        request: RoutedExpertRequest<'_, B::Tensor>,
1207        context: &<B::Tensor as Tensor>::Context,
1208    ) -> Result<B::Tensor, Self::Error> {
1209        self.forward_grouped(resident_bank, request, context)
1210    }
1211
1212    /// Executes one ReLU-squared route batch through the same residency boundary.
1213    fn forward_relu2_routed(
1214        &mut self,
1215        resident_bank: &mut B::Relu2Groups,
1216        request: RoutedExpertRequest<'_, B::Tensor>,
1217        context: &<B::Tensor as Tensor>::Context,
1218    ) -> Result<B::Tensor, Self::Error>;
1219}
1220
1221/// Additive provider mechanism for tensor-parallel grouped partials.
1222pub trait TensorParallelRoutedExpertProvider<B>: RoutedExpertProvider<B>
1223where
1224    B: GroupedNeuralBackend,
1225{
1226    /// Executes a rank-local gated-product contribution.
1227    fn forward_grouped_tensor_parallel(
1228        &mut self,
1229        resident_bank: &mut B::GatedProductGroups,
1230        request: RoutedExpertRequest<'_, B::Tensor>,
1231        partitions: usize,
1232        context: &<B::Tensor as Tensor>::Context,
1233    ) -> Result<RoutedExpertTensorParallelOutput<B::Tensor>, Self::Error>;
1234
1235    /// Executes destination-local, one-expert-per-row contributions while
1236    /// preserving the backend's TP reduction and post-bias structure.
1237    fn forward_compact_grouped_tensor_parallel(
1238        &mut self,
1239        resident_bank: &mut B::GatedProductGroups,
1240        request: RoutedExpertRequest<'_, B::Tensor>,
1241        partitions: usize,
1242        context: &<B::Tensor as Tensor>::Context,
1243    ) -> Result<RoutedExpertTensorParallelOutput<B::Tensor>, Self::Error> {
1244        self.forward_grouped_tensor_parallel(resident_bank, request, partitions, context)
1245    }
1246
1247    /// Executes a rank-local ReLU-squared contribution.
1248    fn forward_relu2_routed_tensor_parallel(
1249        &mut self,
1250        resident_bank: &mut B::Relu2Groups,
1251        request: RoutedExpertRequest<'_, B::Tensor>,
1252        partitions: usize,
1253        context: &<B::Tensor as Tensor>::Context,
1254    ) -> Result<RoutedExpertTensorParallelOutput<B::Tensor>, Self::Error>;
1255}
1256
1257/// Stable routing metadata supplied by an architecture composition at one
1258/// canonical unit boundary.
1259#[derive(Debug, Clone, Eq, PartialEq)]
1260pub struct RoutedObservationPoint {
1261    path: String,
1262    expert_count: i32,
1263}
1264
1265impl RoutedObservationPoint {
1266    /// Creates one routed observation point.
1267    pub fn new(path: impl Into<String>, expert_count: i32) -> Self {
1268        Self {
1269            path: path.into(),
1270            expert_count,
1271        }
1272    }
1273
1274    /// Returns the stable routed-module path.
1275    pub fn path(&self) -> &str {
1276        &self.path
1277    }
1278
1279    /// Returns the total number of routed experts.
1280    pub const fn expert_count(&self) -> i32 {
1281        self.expert_count
1282    }
1283}
1284
1285/// Failure from either canonical expert execution or its observation hook.
1286#[derive(Debug)]
1287pub enum ObservedExpertProviderError<P, O> {
1288    /// The wrapped provider rejected or failed the expert request.
1289    Provider(P),
1290    /// The observer rejected the normalized routing event.
1291    Observer(O),
1292}
1293
1294impl<P, O> std::fmt::Display for ObservedExpertProviderError<P, O>
1295where
1296    P: std::fmt::Display,
1297    O: std::fmt::Display,
1298{
1299    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
1300        match self {
1301            Self::Provider(error) => write!(formatter, "routed expert provider failed: {error}"),
1302            Self::Observer(error) => write!(formatter, "routed expert observer failed: {error}"),
1303        }
1304    }
1305}
1306
1307impl<P, O> std::error::Error for ObservedExpertProviderError<P, O>
1308where
1309    P: std::error::Error + 'static,
1310    O: std::error::Error + 'static,
1311{
1312}
1313
1314/// Decorates a routed provider with normalized routing observation.
1315///
1316/// The decorator sees the exact request and output of canonical provider
1317/// execution. It therefore adds observation without reimplementing a model
1318/// family's block, routing, shape, or residency lifecycle. Tensor-parallel
1319/// requests are delegated without an event because their provider result may
1320/// still require an architecture-owned reduction before it is observable.
1321pub struct ObservedExpertProvider<'a, P, O: ?Sized, E> {
1322    provider: &'a mut P,
1323    observer: &'a mut O,
1324    point: RoutedObservationPoint,
1325    error: std::marker::PhantomData<fn() -> E>,
1326}
1327
1328impl<'a, P, O: ?Sized, E> ObservedExpertProvider<'a, P, O, E> {
1329    /// Wraps `provider` for one canonical routed module invocation.
1330    pub fn new(provider: &'a mut P, observer: &'a mut O, point: RoutedObservationPoint) -> Self {
1331        Self {
1332            provider,
1333            observer,
1334            point,
1335            error: std::marker::PhantomData,
1336        }
1337    }
1338
1339    fn observe<T, ObservationError>(
1340        &mut self,
1341        routes: &eredu_nn::GroupSelection<T>,
1342        output: &T,
1343    ) -> Result<T, ObservationError>
1344    where
1345        T: Clone,
1346        O: ActivationObserver<T, ObservationError>,
1347    {
1348        self.observer.observe_routing(RoutingObservation {
1349            path: self.point.path(),
1350            selected_experts: routes.group_indices(),
1351            selected_scores: routes.selected_scores(),
1352            coefficients: routes.coefficients(),
1353            routed_output: output,
1354            local_routed_output: None,
1355            reduced_routed_output: None,
1356            shared_output: None,
1357            combined_output: None,
1358            expert_count: self.point.expert_count(),
1359        })?;
1360        observe_and_intervene(
1361            self.observer,
1362            &format!("{}.output", self.point.path()),
1363            output,
1364        )
1365    }
1366}
1367
1368impl<B, P, O, E> RoutedExpertProvider<B> for ObservedExpertProvider<'_, P, O, E>
1369where
1370    B: GroupedNeuralBackend,
1371    P: RoutedExpertProvider<B>,
1372    O: ActivationObserver<B::Tensor, E> + ?Sized,
1373{
1374    type Error = ObservedExpertProviderError<P::Error, E>;
1375
1376    fn forward_grouped(
1377        &mut self,
1378        resident_bank: &mut B::GatedProductGroups,
1379        request: RoutedExpertRequest<'_, B::Tensor>,
1380        context: &<B::Tensor as Tensor>::Context,
1381    ) -> Result<B::Tensor, Self::Error> {
1382        let routes = request.routes;
1383        let output = self
1384            .provider
1385            .forward_grouped(resident_bank, request, context)
1386            .map_err(ObservedExpertProviderError::Provider)?;
1387        self.observe(routes, &output)
1388            .map_err(ObservedExpertProviderError::Observer)
1389    }
1390
1391    fn forward_relu2_routed(
1392        &mut self,
1393        resident_bank: &mut B::Relu2Groups,
1394        request: RoutedExpertRequest<'_, B::Tensor>,
1395        context: &<B::Tensor as Tensor>::Context,
1396    ) -> Result<B::Tensor, Self::Error> {
1397        let routes = request.routes;
1398        let output = self
1399            .provider
1400            .forward_relu2_routed(resident_bank, request, context)
1401            .map_err(ObservedExpertProviderError::Provider)?;
1402        self.observe(routes, &output)
1403            .map_err(ObservedExpertProviderError::Observer)
1404    }
1405}
1406
1407impl<B, P, O, E> TensorParallelRoutedExpertProvider<B> for ObservedExpertProvider<'_, P, O, E>
1408where
1409    B: GroupedNeuralBackend,
1410    P: TensorParallelRoutedExpertProvider<B>,
1411    O: ActivationObserver<B::Tensor, E> + ?Sized,
1412{
1413    fn forward_grouped_tensor_parallel(
1414        &mut self,
1415        resident_bank: &mut B::GatedProductGroups,
1416        request: RoutedExpertRequest<'_, B::Tensor>,
1417        partitions: usize,
1418        context: &<B::Tensor as Tensor>::Context,
1419    ) -> Result<RoutedExpertTensorParallelOutput<B::Tensor>, Self::Error> {
1420        self.provider
1421            .forward_grouped_tensor_parallel(resident_bank, request, partitions, context)
1422            .map_err(ObservedExpertProviderError::Provider)
1423    }
1424
1425    fn forward_relu2_routed_tensor_parallel(
1426        &mut self,
1427        resident_bank: &mut B::Relu2Groups,
1428        request: RoutedExpertRequest<'_, B::Tensor>,
1429        partitions: usize,
1430        context: &<B::Tensor as Tensor>::Context,
1431    ) -> Result<RoutedExpertTensorParallelOutput<B::Tensor>, Self::Error> {
1432        self.provider
1433            .forward_relu2_routed_tensor_parallel(resident_bank, request, partitions, context)
1434            .map_err(ObservedExpertProviderError::Provider)
1435    }
1436}
1437
1438/// Provider for a fully resident expert bank.
1439#[derive(Debug, Default, Clone, Copy)]
1440pub struct ResidentExpertProvider;
1441
1442impl<B> RoutedExpertProvider<B> for ResidentExpertProvider
1443where
1444    B: GroupedNeuralBackend,
1445{
1446    type Error = eredu_nn::Error;
1447
1448    fn forward_grouped(
1449        &mut self,
1450        resident_bank: &mut B::GatedProductGroups,
1451        request: RoutedExpertRequest<'_, B::Tensor>,
1452        context: &<B::Tensor as Tensor>::Context,
1453    ) -> Result<B::Tensor, Self::Error> {
1454        resident_bank.forward_grouped(request.input, request.routes, context)
1455    }
1456
1457    fn forward_relu2_routed(
1458        &mut self,
1459        resident_bank: &mut B::Relu2Groups,
1460        request: RoutedExpertRequest<'_, B::Tensor>,
1461        context: &<B::Tensor as Tensor>::Context,
1462    ) -> Result<B::Tensor, Self::Error> {
1463        resident_bank.forward_grouped(request.input, request.routes, context)
1464    }
1465}
1466
1467impl<B> TensorParallelRoutedExpertProvider<B> for ResidentExpertProvider
1468where
1469    B: eredu_nn::TensorParallelGroupedNeuralBackend,
1470{
1471    fn forward_grouped_tensor_parallel(
1472        &mut self,
1473        resident_bank: &mut B::GatedProductGroups,
1474        request: RoutedExpertRequest<'_, B::Tensor>,
1475        partitions: usize,
1476        context: &<B::Tensor as Tensor>::Context,
1477    ) -> Result<RoutedExpertTensorParallelOutput<B::Tensor>, Self::Error> {
1478        B::gated_product_groups_tensor_parallel(
1479            resident_bank,
1480            request.input,
1481            request.routes,
1482            partitions,
1483            context,
1484        )
1485        .map(RoutedExpertTensorParallelOutput::Partial)
1486    }
1487
1488    fn forward_relu2_routed_tensor_parallel(
1489        &mut self,
1490        resident_bank: &mut B::Relu2Groups,
1491        request: RoutedExpertRequest<'_, B::Tensor>,
1492        partitions: usize,
1493        context: &<B::Tensor as Tensor>::Context,
1494    ) -> Result<RoutedExpertTensorParallelOutput<B::Tensor>, Self::Error> {
1495        B::relu2_groups_tensor_parallel(
1496            resident_bank,
1497            request.input,
1498            request.routes,
1499            partitions,
1500            context,
1501        )
1502        .map(RoutedExpertTensorParallelOutput::Partial)
1503    }
1504}