1use std::{collections::BTreeSet, num::NonZeroUsize};
4
5use eredu_core::{ParallelTopology, RealtimeSpeechConfig, SessionCapabilities};
6use eredu_nn::NeuralOperatorCapabilities;
7
8use crate::{
9 ArchitectureParameterDescription, CacheResidencyPolicy, ExecutionGraph, ExecutionResidency,
10 ExecutionUnitLayout, LayerWeightResidency, StateComponentMechanism, StateComponentPlacement,
11 StateLayout, StateMechanismCapabilities, WeightLoweringCapability, WeightLoweringDescriptor,
12 WeightLoweringKind,
13};
14
15#[derive(Debug, Clone, Eq, Hash, Ord, PartialEq, PartialOrd)]
17pub struct RealtimeIdentity(String);
18
19impl RealtimeIdentity {
20 pub fn new(value: impl Into<String>) -> Result<Self, RealtimeContractError> {
22 let value = value.into();
23 if value.trim().is_empty() {
24 return Err(RealtimeContractError::EmptyIdentity);
25 }
26 Ok(Self(value))
27 }
28
29 pub fn as_str(&self) -> &str {
31 &self.0
32 }
33}
34
35impl std::fmt::Display for RealtimeIdentity {
36 fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
37 formatter.write_str(&self.0)
38 }
39}
40
41#[derive(Debug, Clone, Copy, Eq, Hash, Ord, PartialEq, PartialOrd)]
43#[non_exhaustive]
44pub enum RealtimeMechanism {
45 TensorOperations,
47 NeuralOperations,
49 ParameterMaterialization,
51 ParameterStorage,
53 StateStorage,
55 CoordinateStorage,
57 Sampling,
59 Randomness,
61 HostConversion,
63 ExactCompletion,
65 ResourceRetention,
67 Transfer,
69 Observation,
71 Collectives,
73 Timing,
75}
76
77#[derive(Debug, Clone, Copy, Eq, Hash, Ord, PartialEq, PartialOrd)]
79pub enum RealtimeWeightComponentRole {
80 Primary,
82 Scale,
84 AffineBias,
86}
87
88#[derive(Debug, Clone, Eq, PartialEq)]
90pub struct RealtimeWeightComponentRequirement {
91 target: RealtimeIdentity,
92 recipe_owner: Option<RealtimeIdentity>,
93 recipe_identity: Option<RealtimeIdentity>,
94 recipe: Option<eredu_checkpoint::recipe::DerivedWeightRecipe>,
95 recipe_output: Option<eredu_checkpoint::recipe::RecipeMetadata>,
96 source_occurrences: Vec<RealtimeIdentity>,
97 physical_shape: Vec<usize>,
98 role: RealtimeWeightComponentRole,
99}
100
101impl RealtimeWeightComponentRequirement {
102 #[allow(clippy::too_many_arguments)]
104 pub fn new(
105 target: RealtimeIdentity,
106 recipe_owner: Option<RealtimeIdentity>,
107 recipe_identity: Option<RealtimeIdentity>,
108 recipe: Option<eredu_checkpoint::recipe::DerivedWeightRecipe>,
109 recipe_output: Option<eredu_checkpoint::recipe::RecipeMetadata>,
110 source_occurrences: impl IntoIterator<Item = RealtimeIdentity>,
111 physical_shape: impl Into<Vec<usize>>,
112 role: RealtimeWeightComponentRole,
113 ) -> Result<Self, RealtimeContractError> {
114 let source_occurrences = source_occurrences.into_iter().collect::<Vec<_>>();
115 let physical_shape = physical_shape.into();
116 if physical_shape.is_empty() || physical_shape.contains(&0) {
117 return Err(RealtimeContractError::InvalidWeightComponentShape);
118 }
119 match (
120 &recipe_owner,
121 &recipe_identity,
122 &recipe,
123 &recipe_output,
124 source_occurrences.is_empty(),
125 ) {
126 (Some(_), Some(_), Some(recipe), Some(output), false) => {
127 let declared = source_occurrences
128 .iter()
129 .map(RealtimeIdentity::as_str)
130 .collect::<Vec<_>>();
131 if recipe.source_occurrences() != declared || output.shape != physical_shape {
132 return Err(RealtimeContractError::WeightComponentRecipeMismatch);
133 }
134 }
135 (Some(_), Some(_), Some(_), Some(_), true) => {
136 return Err(RealtimeContractError::EmptyRecipeComponentSources);
137 }
138 (None, None, None, None, true) => {}
139 (None, None, None, None, false) => {
140 return Err(RealtimeContractError::GeneratedComponentHasSources);
141 }
142 _ => return Err(RealtimeContractError::IncompleteWeightComponentRecipe),
143 }
144 Ok(Self {
145 target,
146 recipe_owner,
147 recipe_identity,
148 recipe,
149 recipe_output,
150 source_occurrences,
151 physical_shape,
152 role,
153 })
154 }
155
156 pub const fn target(&self) -> &RealtimeIdentity {
158 &self.target
159 }
160
161 pub const fn recipe_owner(&self) -> Option<&RealtimeIdentity> {
163 self.recipe_owner.as_ref()
164 }
165
166 pub const fn recipe_identity(&self) -> Option<&RealtimeIdentity> {
168 self.recipe_identity.as_ref()
169 }
170
171 pub const fn recipe(&self) -> Option<&eredu_checkpoint::recipe::DerivedWeightRecipe> {
173 self.recipe.as_ref()
174 }
175
176 pub const fn recipe_output(&self) -> Option<&eredu_checkpoint::recipe::RecipeMetadata> {
178 self.recipe_output.as_ref()
179 }
180
181 pub fn source_occurrences(&self) -> &[RealtimeIdentity] {
183 &self.source_occurrences
184 }
185
186 pub fn physical_shape(&self) -> &[usize] {
188 &self.physical_shape
189 }
190
191 pub const fn role(&self) -> RealtimeWeightComponentRole {
193 self.role
194 }
195
196 pub const fn is_recipe_backed(&self) -> bool {
198 self.recipe_identity.is_some()
199 }
200}
201
202#[derive(Debug, Clone, Eq, PartialEq)]
204pub struct RealtimeWeightLoweringRequirement {
205 target: RealtimeIdentity,
206 components: Vec<RealtimeWeightComponentRequirement>,
207 descriptor: WeightLoweringDescriptor,
208 kind: WeightLoweringKind,
209}
210
211impl RealtimeWeightLoweringRequirement {
212 pub fn new(
214 target: RealtimeIdentity,
215 components: impl IntoIterator<Item = RealtimeWeightComponentRequirement>,
216 descriptor: WeightLoweringDescriptor,
217 kind: WeightLoweringKind,
218 ) -> Result<Self, RealtimeContractError> {
219 let components = components.into_iter().collect::<Vec<_>>();
220 let primary = components
221 .iter()
222 .filter(|component| component.role == RealtimeWeightComponentRole::Primary)
223 .collect::<Vec<_>>();
224 match primary.as_slice() {
225 [] => return Err(RealtimeContractError::MissingPrimaryWeightComponent),
226 [primary] if primary.target != target => {
227 return Err(RealtimeContractError::PrimaryWeightComponentTargetMismatch);
228 }
229 [_] => {}
230 _ => return Err(RealtimeContractError::DuplicatePrimaryWeightComponent),
231 }
232 let targets = components
233 .iter()
234 .map(RealtimeWeightComponentRequirement::target)
235 .collect::<BTreeSet<_>>();
236 if targets.len() != components.len() {
237 return Err(RealtimeContractError::DuplicateWeightComponentTarget);
238 }
239 for role in [
240 RealtimeWeightComponentRole::Scale,
241 RealtimeWeightComponentRole::AffineBias,
242 ] {
243 if components
244 .iter()
245 .filter(|component| component.role == role)
246 .count()
247 > 1
248 {
249 return Err(RealtimeContractError::DuplicateWeightComponentRole { role });
250 }
251 }
252 Ok(Self {
253 target,
254 components,
255 descriptor,
256 kind,
257 })
258 }
259
260 pub const fn target(&self) -> &RealtimeIdentity {
262 &self.target
263 }
264
265 pub fn components(&self) -> &[RealtimeWeightComponentRequirement] {
267 &self.components
268 }
269
270 pub fn primary(&self) -> &RealtimeWeightComponentRequirement {
272 self.component(RealtimeWeightComponentRole::Primary)
273 .expect("validated realtime lowering contains one primary component")
274 }
275
276 pub fn scale(&self) -> Option<&RealtimeWeightComponentRequirement> {
278 self.component(RealtimeWeightComponentRole::Scale)
279 }
280
281 pub fn affine_bias(&self) -> Option<&RealtimeWeightComponentRequirement> {
283 self.component(RealtimeWeightComponentRole::AffineBias)
284 }
285
286 fn component(
287 &self,
288 role: RealtimeWeightComponentRole,
289 ) -> Option<&RealtimeWeightComponentRequirement> {
290 self.components
291 .iter()
292 .find(|component| component.role == role)
293 }
294
295 pub const fn descriptor(&self) -> &WeightLoweringDescriptor {
297 &self.descriptor
298 }
299
300 pub const fn kind(&self) -> WeightLoweringKind {
302 self.kind
303 }
304}
305
306#[derive(Debug, Clone, Eq, PartialEq)]
308pub struct RealtimeExecutionRequirements {
309 identity: RealtimeIdentity,
310 parameters: ArchitectureParameterDescription,
311 weight_lowerings: Vec<RealtimeWeightLoweringRequirement>,
312}
313
314impl RealtimeExecutionRequirements {
315 pub fn new(
317 identity: RealtimeIdentity,
318 parameters: ArchitectureParameterDescription,
319 weight_lowerings: Vec<RealtimeWeightLoweringRequirement>,
320 ) -> Result<Self, RealtimeContractError> {
321 if weight_lowerings.is_empty() {
322 return Err(RealtimeContractError::EmptyWeightLowerings);
323 }
324 let targets = weight_lowerings
325 .iter()
326 .map(RealtimeWeightLoweringRequirement::target)
327 .collect::<BTreeSet<_>>();
328 if targets.len() != weight_lowerings.len() {
329 return Err(RealtimeContractError::DuplicateWeightLoweringTarget);
330 }
331 let members = parameters
332 .groups()
333 .iter()
334 .flat_map(|group| group.group().members())
335 .collect::<Vec<_>>();
336 let expected_targets = members
337 .iter()
338 .filter(|member| member.linear_companion().is_none())
339 .map(|member| member.target())
340 .collect::<BTreeSet<_>>();
341 let actual_targets = weight_lowerings
342 .iter()
343 .map(|lowering| lowering.target().as_str())
344 .collect::<BTreeSet<_>>();
345 if expected_targets != actual_targets {
346 return Err(RealtimeContractError::WeightLoweringParameterMismatch);
347 }
348 for lowering in &weight_lowerings {
349 let expected_components = members
350 .iter()
351 .filter(|member| {
352 member.target() == lowering.target().as_str()
353 || member.linear_companion_of() == Some(lowering.target().as_str())
354 })
355 .map(|member| member.target())
356 .collect::<BTreeSet<_>>();
357 let actual_components = lowering
358 .components()
359 .iter()
360 .map(|component| component.target().as_str())
361 .collect::<BTreeSet<_>>();
362 if expected_components != actual_components {
363 return Err(RealtimeContractError::WeightComponentParameterMismatch {
364 target: lowering.target().clone(),
365 });
366 }
367 }
368 Ok(Self {
369 identity,
370 parameters,
371 weight_lowerings,
372 })
373 }
374
375 pub const fn identity(&self) -> &RealtimeIdentity {
377 &self.identity
378 }
379
380 pub const fn parameters(&self) -> &ArchitectureParameterDescription {
382 &self.parameters
383 }
384
385 pub fn weight_lowerings(&self) -> &[RealtimeWeightLoweringRequirement] {
387 &self.weight_lowerings
388 }
389}
390
391#[derive(Debug, Clone, Eq, PartialEq)]
393pub struct RealtimeMechanismRequirements {
394 mechanisms: BTreeSet<RealtimeMechanism>,
395}
396
397impl RealtimeMechanismRequirements {
398 pub fn new(
400 mechanisms: impl IntoIterator<Item = RealtimeMechanism>,
401 ) -> Result<Self, RealtimeContractError> {
402 let mechanisms = mechanisms.into_iter().collect::<BTreeSet<_>>();
403 if mechanisms.is_empty() {
404 return Err(RealtimeContractError::EmptyMechanismRequirements);
405 }
406 Ok(Self { mechanisms })
407 }
408
409 pub const fn mechanisms(&self) -> &BTreeSet<RealtimeMechanism> {
411 &self.mechanisms
412 }
413
414 pub fn requires(&self, mechanism: RealtimeMechanism) -> bool {
416 self.mechanisms.contains(&mechanism)
417 }
418}
419
420#[derive(Debug, Clone, Eq, PartialEq)]
422pub struct RealtimeTopologyPolicy {
423 identity: RealtimeIdentity,
424 replicated: bool,
425 pure_tensor_parallel_sizes: BTreeSet<usize>,
426}
427
428impl RealtimeTopologyPolicy {
429 pub fn new(
431 identity: RealtimeIdentity,
432 replicated: bool,
433 pure_tensor_parallel_sizes: impl IntoIterator<Item = usize>,
434 ) -> Result<Self, RealtimeContractError> {
435 let pure_tensor_parallel_sizes = pure_tensor_parallel_sizes
436 .into_iter()
437 .collect::<BTreeSet<_>>();
438 if let Some(size) = pure_tensor_parallel_sizes
439 .iter()
440 .copied()
441 .find(|size| *size < 2)
442 {
443 return Err(RealtimeContractError::InvalidPureTensorParallelSize { size });
444 }
445 if !replicated && pure_tensor_parallel_sizes.is_empty() {
446 return Err(RealtimeContractError::EmptyTopologyPolicy);
447 }
448 Ok(Self {
449 identity,
450 replicated,
451 pure_tensor_parallel_sizes,
452 })
453 }
454
455 pub const fn identity(&self) -> &RealtimeIdentity {
457 &self.identity
458 }
459
460 pub const fn admits_replicated(&self) -> bool {
462 self.replicated
463 }
464
465 pub const fn pure_tensor_parallel_sizes(&self) -> &BTreeSet<usize> {
467 &self.pure_tensor_parallel_sizes
468 }
469
470 pub fn admits(&self, topology: ParallelTopology) -> bool {
472 if topology.is_replicated() {
473 return self.replicated;
474 }
475 topology.pipeline() == 1
476 && topology.expert() == 1
477 && topology.data() == 1
478 && self.pure_tensor_parallel_sizes.contains(&topology.tensor())
479 }
480}
481
482#[derive(Debug, Clone, Eq, PartialEq)]
484pub struct RealtimeArchitectureRequirements {
485 architecture: RealtimeIdentity,
486 source: RealtimeIdentity,
487 execution_graph: ExecutionGraph,
488 execution_units: ExecutionUnitLayout,
489 source_parameters: ArchitectureParameterDescription,
490 executions: Vec<RealtimeExecutionRequirements>,
491 operators: NeuralOperatorCapabilities,
492 speech_schedule_identity: RealtimeIdentity,
493 speech_schedule: RealtimeSpeechConfig,
494 state_layout_identity: RealtimeIdentity,
495 state_layout: StateLayout,
496 mechanisms: RealtimeMechanismRequirements,
497 topology: RealtimeTopologyPolicy,
498 residencies: Vec<ExecutionResidency>,
499}
500
501impl RealtimeArchitectureRequirements {
502 #[allow(clippy::too_many_arguments)]
504 pub fn new(
505 architecture: RealtimeIdentity,
506 source: RealtimeIdentity,
507 execution_graph: ExecutionGraph,
508 execution_units: ExecutionUnitLayout,
509 source_parameters: ArchitectureParameterDescription,
510 executions: impl IntoIterator<Item = RealtimeExecutionRequirements>,
511 operators: NeuralOperatorCapabilities,
512 speech_schedule_identity: RealtimeIdentity,
513 speech_schedule: RealtimeSpeechConfig,
514 state_layout_identity: RealtimeIdentity,
515 state_layout: StateLayout,
516 mechanisms: RealtimeMechanismRequirements,
517 topology: RealtimeTopologyPolicy,
518 residencies: impl IntoIterator<Item = ExecutionResidency>,
519 ) -> Result<Self, RealtimeContractError> {
520 if execution_graph.groups().len() != execution_units.group_count()
521 || execution_graph
522 .groups()
523 .iter()
524 .enumerate()
525 .any(|(index, group)| {
526 execution_units
527 .group_id(index)
528 .is_none_or(|identity| identity.as_str() != group.id())
529 })
530 {
531 return Err(RealtimeContractError::ExecutionGraphLayoutMismatch);
532 }
533 if source_parameters.graph() != &execution_graph
534 || source_parameters.unit_layout() != &execution_units
535 {
536 return Err(RealtimeContractError::SourceParameterDescriptionMismatch);
537 }
538 let executions = executions.into_iter().collect::<Vec<_>>();
539 if executions.is_empty() {
540 return Err(RealtimeContractError::EmptyExecutionIdentities);
541 }
542 let execution_identities = executions
543 .iter()
544 .map(RealtimeExecutionRequirements::identity)
545 .collect::<BTreeSet<_>>();
546 if execution_identities.len() != executions.len() {
547 return Err(RealtimeContractError::DuplicateExecutionIdentity);
548 }
549 if let Some(execution) = executions.iter().find(|execution| {
550 execution.parameters.graph() != &execution_graph
551 || execution.parameters.unit_layout() != &execution_units
552 }) {
553 return Err(
554 RealtimeContractError::ExecutionParameterDescriptionMismatch {
555 execution: execution.identity.clone(),
556 },
557 );
558 }
559 let residencies = residencies.into_iter().collect::<Vec<_>>();
560 if residencies.is_empty() {
561 return Err(RealtimeContractError::EmptyResidencies);
562 }
563 if residencies
564 .iter()
565 .enumerate()
566 .any(|(index, residency)| residencies[..index].contains(residency))
567 {
568 return Err(RealtimeContractError::DuplicateResidency);
569 }
570 Ok(Self {
571 architecture,
572 source,
573 execution_graph,
574 execution_units,
575 source_parameters,
576 executions,
577 operators,
578 speech_schedule_identity,
579 speech_schedule,
580 state_layout_identity,
581 state_layout,
582 mechanisms,
583 topology,
584 residencies,
585 })
586 }
587
588 pub const fn architecture(&self) -> &RealtimeIdentity {
590 &self.architecture
591 }
592
593 pub const fn source(&self) -> &RealtimeIdentity {
595 &self.source
596 }
597
598 pub const fn execution_graph(&self) -> &ExecutionGraph {
600 &self.execution_graph
601 }
602
603 pub const fn execution_units(&self) -> &ExecutionUnitLayout {
605 &self.execution_units
606 }
607
608 pub const fn source_parameters(&self) -> &ArchitectureParameterDescription {
610 &self.source_parameters
611 }
612
613 pub fn executions(&self) -> &[RealtimeExecutionRequirements] {
615 &self.executions
616 }
617
618 pub const fn operators(&self) -> NeuralOperatorCapabilities {
620 self.operators
621 }
622
623 pub const fn speech_schedule_identity(&self) -> &RealtimeIdentity {
625 &self.speech_schedule_identity
626 }
627
628 pub const fn speech_schedule(&self) -> &RealtimeSpeechConfig {
630 &self.speech_schedule
631 }
632
633 pub const fn state_layout_identity(&self) -> &RealtimeIdentity {
635 &self.state_layout_identity
636 }
637
638 pub const fn state_layout(&self) -> &StateLayout {
640 &self.state_layout
641 }
642
643 pub const fn mechanisms(&self) -> &RealtimeMechanismRequirements {
645 &self.mechanisms
646 }
647
648 pub const fn topology(&self) -> &RealtimeTopologyPolicy {
650 &self.topology
651 }
652
653 pub fn residencies(&self) -> &[ExecutionResidency] {
655 &self.residencies
656 }
657}
658
659#[derive(Debug, Clone, Eq, PartialEq)]
661pub struct RealtimeMechanismCapabilities {
662 operators: NeuralOperatorCapabilities,
663 mechanisms: BTreeSet<RealtimeMechanism>,
664 residencies: Vec<ExecutionResidency>,
665 weight_lowerings: Vec<WeightLoweringCapability>,
666 observation_identities: BTreeSet<RealtimeIdentity>,
667 state: StateMechanismCapabilities,
668 maximum_tensor_parallel_size: NonZeroUsize,
669 completion: crate::CommunicationCompletionCapabilities,
670 session: SessionCapabilities,
671}
672
673impl RealtimeMechanismCapabilities {
674 #[allow(clippy::too_many_arguments)]
676 pub fn new(
677 operators: NeuralOperatorCapabilities,
678 mechanisms: impl IntoIterator<Item = RealtimeMechanism>,
679 residencies: impl IntoIterator<Item = ExecutionResidency>,
680 weight_lowerings: Vec<WeightLoweringCapability>,
681 state: StateMechanismCapabilities,
682 maximum_tensor_parallel_size: NonZeroUsize,
683 completion: crate::CommunicationCompletionCapabilities,
684 session: SessionCapabilities,
685 ) -> Self {
686 Self {
687 operators,
688 mechanisms: mechanisms.into_iter().collect(),
689 residencies: residencies.into_iter().collect(),
690 weight_lowerings,
691 observation_identities: BTreeSet::new(),
692 state,
693 maximum_tensor_parallel_size,
694 completion,
695 session,
696 }
697 }
698
699 pub const fn operators(&self) -> NeuralOperatorCapabilities {
701 self.operators
702 }
703
704 pub fn supports(&self, mechanism: RealtimeMechanism) -> bool {
706 self.mechanisms.contains(&mechanism)
707 }
708
709 pub fn residencies(&self) -> &[ExecutionResidency] {
711 &self.residencies
712 }
713
714 pub fn weight_lowerings(&self) -> &[WeightLoweringCapability] {
716 &self.weight_lowerings
717 }
718
719 pub fn with_observation_identities(
721 mut self,
722 observations: impl IntoIterator<Item = RealtimeIdentity>,
723 ) -> Self {
724 self.observation_identities = observations.into_iter().collect();
725 self
726 }
727
728 pub const fn observation_identities(&self) -> &BTreeSet<RealtimeIdentity> {
730 &self.observation_identities
731 }
732
733 pub fn supports_observation(&self, observation: &RealtimeIdentity) -> bool {
735 self.observation_identities.contains(observation)
736 }
737
738 pub const fn state(&self) -> &StateMechanismCapabilities {
740 &self.state
741 }
742
743 pub const fn maximum_tensor_parallel_size(&self) -> NonZeroUsize {
745 self.maximum_tensor_parallel_size
746 }
747
748 pub const fn completion(&self) -> &crate::CommunicationCompletionCapabilities {
750 &self.completion
751 }
752
753 pub const fn session(&self) -> SessionCapabilities {
755 self.session
756 }
757}
758
759#[derive(Debug, Clone, Eq, PartialEq)]
761pub struct RealtimeArchitectureProof {
762 architecture: RealtimeIdentity,
763 speech_schedule: RealtimeIdentity,
764 state_layout: RealtimeIdentity,
765 topology_policy: RealtimeIdentity,
766 topology: ParallelTopology,
767 rank: usize,
768}
769
770impl RealtimeArchitectureProof {
771 pub fn new(
773 architecture: RealtimeIdentity,
774 speech_schedule: RealtimeIdentity,
775 state_layout: RealtimeIdentity,
776 topology_policy: RealtimeIdentity,
777 topology: ParallelTopology,
778 rank: usize,
779 ) -> Self {
780 Self {
781 architecture,
782 speech_schedule,
783 state_layout,
784 topology_policy,
785 topology,
786 rank,
787 }
788 }
789
790 pub const fn topology(&self) -> ParallelTopology {
792 self.topology
793 }
794
795 pub const fn rank(&self) -> usize {
797 self.rank
798 }
799}
800
801#[derive(Debug, Clone, Default, Eq, PartialEq)]
803pub struct RealtimeObservationRequirements {
804 output: bool,
805 activations: BTreeSet<RealtimeIdentity>,
806}
807
808impl RealtimeObservationRequirements {
809 pub fn new(output: bool, activations: impl IntoIterator<Item = RealtimeIdentity>) -> Self {
811 Self {
812 output,
813 activations: activations.into_iter().collect(),
814 }
815 }
816
817 pub const fn output(&self) -> bool {
819 self.output
820 }
821
822 pub const fn activations(&self) -> &BTreeSet<RealtimeIdentity> {
824 &self.activations
825 }
826
827 pub fn requires_activations(&self) -> bool {
829 !self.activations.is_empty()
830 }
831}
832
833#[derive(Debug, Clone, Eq, PartialEq)]
835pub struct RealtimeSelectionRequest {
836 max_cached_shards: usize,
837 required_session_capabilities: SessionCapabilities,
838 source: RealtimeIdentity,
839 execution: RealtimeIdentity,
840 residency: LayerWeightResidency,
841 state: CacheResidencyPolicy,
842 architecture_proof: Option<RealtimeArchitectureProof>,
843 completion: crate::CommunicationCompletionPolicy,
844 observations: RealtimeObservationRequirements,
845}
846
847impl RealtimeSelectionRequest {
848 pub const fn with_required_session_capabilities(
850 mut self,
851 required: SessionCapabilities,
852 ) -> Self {
853 self.required_session_capabilities = required;
854 self
855 }
856 pub const fn with_max_cached_shards(mut self, maximum: std::num::NonZeroUsize) -> Self {
858 self.max_cached_shards = maximum.get();
859 self
860 }
861 pub fn new(
863 source: RealtimeIdentity,
864 execution: RealtimeIdentity,
865 residency: LayerWeightResidency,
866 state: CacheResidencyPolicy,
867 architecture_proof: Option<RealtimeArchitectureProof>,
868 completion: crate::CommunicationCompletionPolicy,
869 observations: RealtimeObservationRequirements,
870 ) -> Self {
871 Self {
872 source,
873 execution,
874 max_cached_shards: residency.max_cached_shards(),
875 required_session_capabilities: SessionCapabilities::default(),
876 residency,
877 state,
878 architecture_proof,
879 completion,
880 observations,
881 }
882 }
883}
884
885#[derive(Debug, Clone, Eq, PartialEq)]
887pub struct SelectedRealtimeStateComponentRealization {
888 mechanism: StateComponentMechanism,
889 placement: StateComponentPlacement,
890}
891
892impl SelectedRealtimeStateComponentRealization {
893 pub const fn layer(&self) -> usize {
895 self.mechanism.layer()
896 }
897
898 pub const fn component(&self) -> &eredu_core::cache::StateComponentPolicy {
900 self.mechanism.component()
901 }
902
903 pub const fn placement(&self) -> StateComponentPlacement {
905 self.placement
906 }
907}
908
909#[derive(Debug, Clone, Eq, PartialEq)]
911pub struct SelectedRealtimeStateRealization {
912 layout: StateLayout,
913 policy: CacheResidencyPolicy,
914 components: Vec<SelectedRealtimeStateComponentRealization>,
915 checkpoint: bool,
916 rollback: bool,
917 reset: bool,
918 observation_retention: bool,
919}
920
921impl SelectedRealtimeStateRealization {
922 pub const fn layout(&self) -> &StateLayout {
924 &self.layout
925 }
926
927 pub const fn policy(&self) -> &CacheResidencyPolicy {
929 &self.policy
930 }
931
932 pub fn components(&self) -> &[SelectedRealtimeStateComponentRealization] {
934 &self.components
935 }
936
937 pub const fn checkpoint(&self) -> bool {
939 self.checkpoint
940 }
941
942 pub const fn rollback(&self) -> bool {
944 self.rollback
945 }
946
947 pub const fn reset(&self) -> bool {
949 self.reset
950 }
951
952 pub const fn observation_retention(&self) -> bool {
954 self.observation_retention
955 }
956}
957
958#[derive(Debug, Clone, Eq, PartialEq)]
960pub struct SelectedRealtimeRealization {
961 max_cached_shards: usize,
962 requirements: RealtimeArchitectureRequirements,
963 source: RealtimeIdentity,
964 execution: RealtimeExecutionRequirements,
965 residency: LayerWeightResidency,
966 state: SelectedRealtimeStateRealization,
967 topology: ParallelTopology,
968 rank: usize,
969 completion: crate::CommunicationCompletionPolicy,
970 observations: RealtimeObservationRequirements,
971}
972
973impl SelectedRealtimeRealization {
974 pub const fn max_cached_shards(&self) -> usize {
976 self.max_cached_shards
977 }
978 pub const fn requirements(&self) -> &RealtimeArchitectureRequirements {
980 &self.requirements
981 }
982
983 pub const fn source(&self) -> &RealtimeIdentity {
985 &self.source
986 }
987
988 pub const fn execution_graph(&self) -> &ExecutionGraph {
990 self.requirements.execution_graph()
991 }
992
993 pub const fn execution_units(&self) -> &ExecutionUnitLayout {
995 self.requirements.execution_units()
996 }
997
998 pub const fn source_parameters(&self) -> &ArchitectureParameterDescription {
1000 self.requirements.source_parameters()
1001 }
1002
1003 pub const fn operators(&self) -> NeuralOperatorCapabilities {
1005 self.requirements.operators()
1006 }
1007
1008 pub const fn execution(&self) -> &RealtimeIdentity {
1010 self.execution.identity()
1011 }
1012
1013 pub const fn execution_requirements(&self) -> &RealtimeExecutionRequirements {
1015 &self.execution
1016 }
1017
1018 pub const fn execution_parameters(&self) -> &ArchitectureParameterDescription {
1020 self.execution.parameters()
1021 }
1022
1023 pub fn weight_lowerings(&self) -> &[RealtimeWeightLoweringRequirement] {
1025 self.execution.weight_lowerings()
1026 }
1027
1028 pub const fn residency(&self) -> LayerWeightResidency {
1030 self.residency
1031 }
1032
1033 pub const fn state(&self) -> &SelectedRealtimeStateRealization {
1035 &self.state
1036 }
1037
1038 pub const fn topology(&self) -> ParallelTopology {
1040 self.topology
1041 }
1042
1043 pub const fn rank(&self) -> usize {
1045 self.rank
1046 }
1047
1048 pub const fn completion(&self) -> crate::CommunicationCompletionPolicy {
1050 self.completion
1051 }
1052
1053 pub const fn observations(&self) -> &RealtimeObservationRequirements {
1055 &self.observations
1056 }
1057}
1058
1059#[derive(Debug, Clone, Eq, PartialEq, thiserror::Error)]
1061#[non_exhaustive]
1062pub enum RealtimeSelectionIssue {
1063 #[error("architecture topology and rank proof is missing")]
1065 MissingArchitectureProof,
1066 #[error("architecture proof identity mismatch")]
1068 ArchitectureIdentityMismatch,
1069 #[error("speech schedule proof identity mismatch")]
1071 SpeechScheduleIdentityMismatch,
1072 #[error("state layout proof identity mismatch")]
1074 StateLayoutIdentityMismatch,
1075 #[error("topology policy proof identity mismatch")]
1077 TopologyPolicyIdentityMismatch,
1078 #[error("source artifact identity mismatch")]
1080 SourceIdentityMismatch,
1081 #[error("execution identity is not architecture-admitted")]
1083 ExecutionIdentityNotAdmitted,
1084 #[error("required neural operator {operator} is unavailable")]
1086 MissingNeuralOperator {
1087 operator: String,
1089 },
1090 #[error("execution weight lowering {index} for target {target} is unavailable")]
1092 WeightLoweringUnavailable {
1093 index: usize,
1095 target: RealtimeIdentity,
1097 },
1098 #[error("residency is not architecture-admitted")]
1100 ResidencyNotAdmitted,
1101 #[error("residency mechanism is unavailable")]
1103 ResidencyUnavailable,
1104 #[error("topology is not architecture-admitted as replicated or pure tensor parallel")]
1106 TopologyNotAdmitted,
1107 #[error("rank lies outside the proven topology")]
1109 RankOutOfRange,
1110 #[error("tensor-parallel width exceeds backend capability")]
1112 TensorParallelWidthUnavailable,
1113 #[error("required realtime mechanism {0:?} is unavailable")]
1115 MissingMechanism(RealtimeMechanism),
1116 #[error("requested completion timeout disposition is unavailable")]
1118 CompletionPolicyUnavailable,
1119 #[error("state component {component} at layer {layer} is unavailable")]
1121 StateComponentUnavailable {
1122 layer: usize,
1124 component: String,
1126 },
1127 #[error("state checkpoint mechanism is unavailable")]
1129 MissingStateCheckpoint,
1130 #[error("state rollback mechanism is unavailable")]
1132 MissingStateRollback,
1133 #[error("state reset mechanism is unavailable")]
1135 MissingStateReset,
1136 #[error("state observation retention is unavailable")]
1138 MissingStateObservationRetention,
1139 #[error("persistent session state is unavailable")]
1141 MissingPersistentState,
1142 #[error("completed output observation is unavailable")]
1144 MissingOutputObservation,
1145 #[error("activation inspection is unavailable")]
1147 MissingActivationInspection,
1148 #[error("requested activation observation {observation} is unavailable")]
1150 ObservationUnavailable {
1151 observation: RealtimeIdentity,
1153 },
1154}
1155
1156#[derive(Debug, Clone, Eq, PartialEq, thiserror::Error)]
1158#[error("realtime realization is unsupported: {message}", message = selection_message(.issues))]
1159pub struct RealtimeSelectionError {
1160 issues: Vec<RealtimeSelectionIssue>,
1161}
1162
1163impl RealtimeSelectionError {
1164 pub fn issues(&self) -> &[RealtimeSelectionIssue] {
1166 &self.issues
1167 }
1168}
1169
1170fn selection_message(issues: &[RealtimeSelectionIssue]) -> String {
1171 issues
1172 .iter()
1173 .map(ToString::to_string)
1174 .collect::<Vec<_>>()
1175 .join("; ")
1176}
1177
1178pub fn select_realtime_realization(
1180 requirements: &RealtimeArchitectureRequirements,
1181 request: &RealtimeSelectionRequest,
1182 capabilities: &RealtimeMechanismCapabilities,
1183) -> Result<SelectedRealtimeRealization, RealtimeSelectionError> {
1184 let mut issues = Vec::new();
1185 let proof = request.architecture_proof.as_ref();
1186 match proof {
1187 Some(proof) => {
1188 if proof.architecture != requirements.architecture {
1189 issues.push(RealtimeSelectionIssue::ArchitectureIdentityMismatch);
1190 }
1191 if proof.speech_schedule != requirements.speech_schedule_identity {
1192 issues.push(RealtimeSelectionIssue::SpeechScheduleIdentityMismatch);
1193 }
1194 if proof.state_layout != requirements.state_layout_identity {
1195 issues.push(RealtimeSelectionIssue::StateLayoutIdentityMismatch);
1196 }
1197 if proof.topology_policy != *requirements.topology.identity() {
1198 issues.push(RealtimeSelectionIssue::TopologyPolicyIdentityMismatch);
1199 }
1200 }
1201 None => issues.push(RealtimeSelectionIssue::MissingArchitectureProof),
1202 }
1203 if request.source != requirements.source {
1204 issues.push(RealtimeSelectionIssue::SourceIdentityMismatch);
1205 }
1206 let execution = requirements
1207 .executions
1208 .iter()
1209 .find(|execution| execution.identity == request.execution);
1210 if execution.is_none() {
1211 issues.push(RealtimeSelectionIssue::ExecutionIdentityNotAdmitted);
1212 }
1213 for operator in capabilities
1214 .operators
1215 .missing_capability_names(requirements.operators)
1216 {
1217 issues.push(RealtimeSelectionIssue::MissingNeuralOperator {
1218 operator: operator.to_owned(),
1219 });
1220 }
1221 if let Some(execution) = execution {
1222 for (index, required) in execution.weight_lowerings().iter().enumerate() {
1223 if !capabilities.weight_lowerings.iter().any(|available| {
1224 available.descriptor() == required.descriptor()
1225 && available.kind() == required.kind()
1226 }) {
1227 issues.push(RealtimeSelectionIssue::WeightLoweringUnavailable {
1228 index,
1229 target: required.target().clone(),
1230 });
1231 }
1232 }
1233 }
1234 let execution_residency = request.residency.execution_residency();
1235 if !requirements.residencies.contains(&execution_residency) {
1236 issues.push(RealtimeSelectionIssue::ResidencyNotAdmitted);
1237 }
1238 if !capabilities.residencies.contains(&execution_residency) {
1239 issues.push(RealtimeSelectionIssue::ResidencyUnavailable);
1240 }
1241 if let Some(proof) = proof {
1242 if !requirements.topology.admits(proof.topology) {
1243 issues.push(RealtimeSelectionIssue::TopologyNotAdmitted);
1244 }
1245 if proof.rank >= proof.topology.world_size() {
1246 issues.push(RealtimeSelectionIssue::RankOutOfRange);
1247 }
1248 if proof.topology.tensor() > capabilities.maximum_tensor_parallel_size.get() {
1249 issues.push(RealtimeSelectionIssue::TensorParallelWidthUnavailable);
1250 }
1251 }
1252 for mechanism in requirements.mechanisms.mechanisms() {
1253 if !capabilities.supports(*mechanism) {
1254 issues.push(RealtimeSelectionIssue::MissingMechanism(*mechanism));
1255 }
1256 }
1257 if !capabilities.completion.supports(request.completion) {
1258 issues.push(RealtimeSelectionIssue::CompletionPolicyUnavailable);
1259 }
1260 let mut state_components = Vec::new();
1261 for layer in 0..requirements.state_layout.len() {
1262 for component in requirements
1263 .state_layout
1264 .components(layer)
1265 .expect("state layout exposes every validated layer")
1266 {
1267 let matches = capabilities
1268 .state
1269 .components()
1270 .iter()
1271 .filter(|mechanism| {
1272 mechanism.layer() == layer && mechanism.component() == component
1273 })
1274 .cloned()
1275 .collect::<Vec<_>>();
1276 match matches.as_slice() {
1277 [mechanism] => match mechanism.placement(&request.state) {
1278 Some(placement)
1279 if crate::replicated_text::placement_is_compatible(
1280 component,
1281 &request.state,
1282 placement,
1283 ) =>
1284 {
1285 state_components.push(SelectedRealtimeStateComponentRealization {
1286 mechanism: mechanism.clone(),
1287 placement,
1288 });
1289 }
1290 _ => issues.push(RealtimeSelectionIssue::StateComponentUnavailable {
1291 layer,
1292 component: component.role().stable_name().to_owned(),
1293 }),
1294 },
1295 _ => issues.push(RealtimeSelectionIssue::StateComponentUnavailable {
1296 layer,
1297 component: component.role().stable_name().to_owned(),
1298 }),
1299 }
1300 }
1301 }
1302 if !capabilities.state.checkpoint() {
1303 issues.push(RealtimeSelectionIssue::MissingStateCheckpoint);
1304 }
1305 if !capabilities.state.rollback() {
1306 issues.push(RealtimeSelectionIssue::MissingStateRollback);
1307 }
1308 if !capabilities.state.reset() {
1309 issues.push(RealtimeSelectionIssue::MissingStateReset);
1310 }
1311 if (request.observations.output()
1312 || request.observations.requires_activations()
1313 || request.required_session_capabilities.output_observation()
1314 || request
1315 .required_session_capabilities
1316 .activation_inspection())
1317 && !capabilities.state.observation_retention()
1318 {
1319 issues.push(RealtimeSelectionIssue::MissingStateObservationRetention);
1320 }
1321 if let Some(proof) = proof {
1322 if !proof.topology.is_replicated()
1323 && !capabilities.supports(RealtimeMechanism::Collectives)
1324 && !requirements
1325 .mechanisms
1326 .requires(RealtimeMechanism::Collectives)
1327 {
1328 issues.push(RealtimeSelectionIssue::MissingMechanism(
1329 RealtimeMechanism::Collectives,
1330 ));
1331 }
1332 }
1333 if !capabilities.session.persistent_cache() {
1334 issues.push(RealtimeSelectionIssue::MissingPersistentState);
1335 }
1336 if (request.observations.output() || request.required_session_capabilities.output_observation())
1337 && !capabilities.session.output_observation()
1338 {
1339 issues.push(RealtimeSelectionIssue::MissingOutputObservation);
1340 }
1341 if (request.observations.requires_activations()
1342 || request
1343 .required_session_capabilities
1344 .activation_inspection())
1345 && !capabilities.session.activation_inspection()
1346 {
1347 issues.push(RealtimeSelectionIssue::MissingActivationInspection);
1348 }
1349 for observation in request.observations.activations() {
1350 if !capabilities.supports_observation(observation) {
1351 issues.push(RealtimeSelectionIssue::ObservationUnavailable {
1352 observation: observation.clone(),
1353 });
1354 }
1355 }
1356 if !issues.is_empty() {
1357 return Err(RealtimeSelectionError { issues });
1358 }
1359 let proof = proof.expect("successful realtime selection has architecture proof");
1360 let state = SelectedRealtimeStateRealization {
1361 layout: requirements.state_layout.clone(),
1362 policy: request.state.clone(),
1363 components: state_components,
1364 checkpoint: capabilities.state.checkpoint(),
1365 rollback: capabilities.state.rollback(),
1366 reset: capabilities.state.reset(),
1367 observation_retention: capabilities.state.observation_retention(),
1368 };
1369 Ok(SelectedRealtimeRealization {
1370 max_cached_shards: request.max_cached_shards,
1371 requirements: requirements.clone(),
1372 source: request.source.clone(),
1373 execution: execution
1374 .expect("successful realtime selection has admitted execution")
1375 .clone(),
1376 residency: request.residency,
1377 state,
1378 topology: proof.topology,
1379 rank: proof.rank,
1380 completion: request.completion,
1381 observations: request.observations.clone(),
1382 })
1383}
1384
1385#[derive(Debug)]
1387pub struct ConstructedRealtimeResources<P, M, S, Q, G> {
1388 payload: P,
1389 modules: M,
1390 state: S,
1391 queue: Q,
1392 group: Option<G>,
1393}
1394
1395impl<P, M, S, Q, G> ConstructedRealtimeResources<P, M, S, Q, G> {
1396 pub fn into_parts(self) -> (P, M, S, Q, Option<G>) {
1398 (
1399 self.payload,
1400 self.modules,
1401 self.state,
1402 self.queue,
1403 self.group,
1404 )
1405 }
1406}
1407
1408#[derive(Debug)]
1410pub struct PreparedRealtimeRealization<P, M, S, Q, G> {
1411 selected: SelectedRealtimeRealization,
1412 resources: ConstructedRealtimeResources<P, M, S, Q, G>,
1413}
1414
1415impl<P, M, S, Q, G> PreparedRealtimeRealization<P, M, S, Q, G> {
1416 pub const fn selected(&self) -> &SelectedRealtimeRealization {
1418 &self.selected
1419 }
1420
1421 pub fn into_parts(
1423 self,
1424 ) -> (
1425 SelectedRealtimeRealization,
1426 ConstructedRealtimeResources<P, M, S, Q, G>,
1427 ) {
1428 (self.selected, self.resources)
1429 }
1430}
1431
1432#[derive(Debug, thiserror::Error)]
1434pub enum RealtimePreparationError<E> {
1435 #[error(transparent)]
1437 Selection(#[from] RealtimeSelectionError),
1438 #[error("realtime payload construction failed")]
1440 Payload(#[source] E),
1441 #[error("realtime module construction failed")]
1443 Modules(#[source] E),
1444 #[error("realtime state construction failed")]
1446 State(#[source] E),
1447 #[error("realtime queue construction failed")]
1449 Queue(#[source] E),
1450 #[error("realtime communication group construction failed")]
1452 Group(#[source] E),
1453}
1454
1455#[allow(clippy::too_many_arguments)]
1457pub fn select_and_prepare_realtime_realization<P, M, S, Q, G, E>(
1458 requirements: &RealtimeArchitectureRequirements,
1459 request: &RealtimeSelectionRequest,
1460 capabilities: &RealtimeMechanismCapabilities,
1461 payload: impl FnOnce(&SelectedRealtimeRealization) -> Result<P, E>,
1462 modules: impl FnOnce(&SelectedRealtimeRealization, &P) -> Result<M, E>,
1463 state: impl FnOnce(&SelectedRealtimeRealization, &M) -> Result<S, E>,
1464 queue: impl FnOnce(&SelectedRealtimeRealization) -> Result<Q, E>,
1465 group: impl FnOnce(&SelectedRealtimeRealization) -> Result<G, E>,
1466) -> Result<PreparedRealtimeRealization<P, M, S, Q, G>, RealtimePreparationError<E>> {
1467 let selected = select_realtime_realization(requirements, request, capabilities)?;
1468 let payload = payload(&selected).map_err(RealtimePreparationError::Payload)?;
1469 let modules = modules(&selected, &payload).map_err(RealtimePreparationError::Modules)?;
1470 let state = state(&selected, &modules).map_err(RealtimePreparationError::State)?;
1471 let queue = queue(&selected).map_err(RealtimePreparationError::Queue)?;
1472 let group = if selected.topology().is_replicated() {
1473 None
1474 } else {
1475 Some(group(&selected).map_err(RealtimePreparationError::Group)?)
1476 };
1477 Ok(PreparedRealtimeRealization {
1478 selected,
1479 resources: ConstructedRealtimeResources {
1480 payload,
1481 modules,
1482 state,
1483 queue,
1484 group,
1485 },
1486 })
1487}
1488
1489#[derive(Debug, Clone, Eq, PartialEq, thiserror::Error)]
1491#[non_exhaustive]
1492pub enum RealtimeContractError {
1493 #[error("realtime identity must not be empty")]
1495 EmptyIdentity,
1496 #[error("realtime mechanism requirements must not be empty")]
1498 EmptyMechanismRequirements,
1499 #[error("realtime execution identities must not be empty")]
1501 EmptyExecutionIdentities,
1502 #[error("realtime execution graph and unit layout differ")]
1504 ExecutionGraphLayoutMismatch,
1505 #[error("realtime source parameter description differs from the retained graph or layout")]
1507 SourceParameterDescriptionMismatch,
1508 #[error(
1510 "realtime execution parameter description for {execution} differs from the retained graph or layout"
1511 )]
1512 ExecutionParameterDescriptionMismatch {
1513 execution: RealtimeIdentity,
1515 },
1516 #[error("realtime execution identities must be unique")]
1518 DuplicateExecutionIdentity,
1519 #[error("realtime execution weight lowerings must not be empty")]
1521 EmptyWeightLowerings,
1522 #[error("realtime weight component shape must contain only positive extents")]
1524 InvalidWeightComponentShape,
1525 #[error("realtime weight component recipe owner and identity must appear together")]
1527 IncompleteWeightComponentRecipe,
1528 #[error("realtime recipe-backed weight component sources must not be empty")]
1530 EmptyRecipeComponentSources,
1531 #[error("realtime weight component recipe does not match its sources and physical geometry")]
1533 WeightComponentRecipeMismatch,
1534 #[error("realtime generated weight component must not contain recipe sources")]
1536 GeneratedComponentHasSources,
1537 #[error("realtime weight lowering has no primary component")]
1539 MissingPrimaryWeightComponent,
1540 #[error("realtime weight lowering has more than one primary component")]
1542 DuplicatePrimaryWeightComponent,
1543 #[error("realtime primary weight component does not match its lowering target")]
1545 PrimaryWeightComponentTargetMismatch,
1546 #[error("realtime weight component targets must be unique within a lowering")]
1548 DuplicateWeightComponentTarget,
1549 #[error("realtime weight lowering repeats component role {role:?}")]
1551 DuplicateWeightComponentRole {
1552 role: RealtimeWeightComponentRole,
1554 },
1555 #[error("realtime execution weight lowering targets must be unique")]
1557 DuplicateWeightLoweringTarget,
1558 #[error("realtime weight lowerings differ from execution parameters")]
1560 WeightLoweringParameterMismatch,
1561 #[error("realtime weight components differ from execution parameters for {target}")]
1563 WeightComponentParameterMismatch {
1564 target: RealtimeIdentity,
1566 },
1567 #[error("realtime residencies must not be empty")]
1569 EmptyResidencies,
1570 #[error("realtime residencies must be unique")]
1572 DuplicateResidency,
1573 #[error("realtime topology policy must not be empty")]
1575 EmptyTopologyPolicy,
1576 #[error("pure tensor-parallel size {size} must be at least two")]
1578 InvalidPureTensorParallelSize {
1579 size: usize,
1581 },
1582}
1583
1584#[cfg(test)]
1585mod tests {
1586 use std::{cell::Cell, convert::Infallible, num::NonZeroUsize, rc::Rc, time::Duration};
1587
1588 use eredu_checkpoint::{LinearFormat, SourceTensorEncoding, StoredDtype};
1589 use eredu_core::{
1590 cache::LayerCachePolicy, AttentionPolicy, CompletionCancellationMode, LayerSchedule,
1591 RealtimeFrameConvention,
1592 };
1593
1594 use super::*;
1595 fn identity(value: &str) -> RealtimeIdentity {
1596 RealtimeIdentity::new(value).unwrap()
1597 }
1598
1599 fn execution_graph() -> ExecutionGraph {
1600 ExecutionGraph::chain(["temporal", "depth"]).unwrap()
1601 }
1602
1603 fn execution_units(graph: &ExecutionGraph) -> ExecutionUnitLayout {
1604 ExecutionUnitLayout::new(graph, [2, 2]).unwrap()
1605 }
1606
1607 fn parameter_description(
1608 graph: &ExecutionGraph,
1609 units: &ExecutionUnitLayout,
1610 ) -> ArchitectureParameterDescription {
1611 let group = crate::ParameterGroupSpec::new(
1612 "target",
1613 crate::ParameterRole::Replicated,
1614 [crate::ParameterMemberSpec::new(
1615 "target.weight",
1616 [8, 8],
1617 crate::MemberSharding::Replicated,
1618 )],
1619 )
1620 .unwrap();
1621 ArchitectureParameterDescription::new(
1622 graph,
1623 units,
1624 [group.clone()],
1625 [crate::OwnedParameterGroupSpec::new(
1626 crate::ParameterGroupOwner::static_role("target"),
1627 group,
1628 )],
1629 )
1630 .unwrap()
1631 }
1632
1633 fn required_operators() -> NeuralOperatorCapabilities {
1634 NeuralOperatorCapabilities::EXP.union(NeuralOperatorCapabilities::SOFTMAX_AXIS)
1635 }
1636
1637 fn state_layout() -> StateLayout {
1638 StateLayout::new(
1639 LayerSchedule::new(
1640 2,
1641 vec![
1642 LayerCachePolicy::key_value(AttentionPolicy::Full, 2, 8).unwrap(),
1643 LayerCachePolicy::key_value(AttentionPolicy::Full, 2, 8).unwrap(),
1644 ],
1645 )
1646 .unwrap(),
1647 )
1648 .unwrap()
1649 }
1650
1651 fn schedule() -> RealtimeSpeechConfig {
1652 RealtimeSpeechConfig::new(
1653 4,
1654 2,
1655 2,
1656 2,
1657 0,
1658 1,
1659 RealtimeFrameConvention::FeedbackAlignedHistory,
1660 vec![0, 1, 2, 1, 2],
1661 )
1662 .unwrap()
1663 }
1664
1665 fn required_mechanisms() -> Vec<RealtimeMechanism> {
1666 vec![
1667 RealtimeMechanism::TensorOperations,
1668 RealtimeMechanism::NeuralOperations,
1669 RealtimeMechanism::ParameterMaterialization,
1670 RealtimeMechanism::ParameterStorage,
1671 RealtimeMechanism::StateStorage,
1672 RealtimeMechanism::CoordinateStorage,
1673 RealtimeMechanism::Sampling,
1674 RealtimeMechanism::Randomness,
1675 RealtimeMechanism::HostConversion,
1676 RealtimeMechanism::ExactCompletion,
1677 RealtimeMechanism::ResourceRetention,
1678 RealtimeMechanism::Transfer,
1679 RealtimeMechanism::Collectives,
1680 ]
1681 }
1682
1683 fn completion_policy() -> crate::CommunicationCompletionPolicy {
1684 crate::CommunicationCompletionPolicy::new(
1685 Duration::from_secs(1),
1686 CompletionCancellationMode::QuarantineUntilComplete,
1687 )
1688 .unwrap()
1689 }
1690
1691 fn completion_capabilities() -> crate::CommunicationCompletionCapabilities {
1692 crate::CommunicationCompletionCapabilities::new([
1693 CompletionCancellationMode::QuarantineUntilComplete,
1694 ])
1695 .unwrap()
1696 }
1697
1698 fn lowering(executable: LinearFormat) -> WeightLoweringDescriptor {
1699 WeightLoweringDescriptor::new(
1700 SourceTensorEncoding::Safetensors(StoredDtype::F32),
1701 executable,
1702 vec![8, 8],
1703 vec![8, 8],
1704 Some(1),
1705 )
1706 .unwrap()
1707 }
1708
1709 fn lowering_requirement(
1710 target: &str,
1711 executable: LinearFormat,
1712 kind: WeightLoweringKind,
1713 ) -> RealtimeWeightLoweringRequirement {
1714 let primary = RealtimeWeightComponentRequirement::new(
1715 identity(target),
1716 Some(identity("canonical-owner")),
1717 Some(identity("recipe-v1")),
1718 Some(eredu_checkpoint::recipe::DerivedWeightRecipe::Concatenate {
1719 axis: 0,
1720 inputs: ["source-a", "source-b"]
1721 .into_iter()
1722 .map(|key| {
1723 eredu_checkpoint::recipe::DerivedWeightRecipe::source(
1724 key,
1725 eredu_checkpoint::store::TensorSelection::Full,
1726 )
1727 })
1728 .collect(),
1729 }),
1730 Some(eredu_checkpoint::recipe::RecipeMetadata {
1731 shape: vec![8, 8],
1732 dtype: eredu_checkpoint::recipe::RecipeDtype::F32,
1733 byte_len: 256,
1734 }),
1735 [identity("source-a"), identity("source-b")],
1736 [8, 8],
1737 RealtimeWeightComponentRole::Primary,
1738 )
1739 .unwrap();
1740 RealtimeWeightLoweringRequirement::new(
1741 identity(target),
1742 [primary],
1743 lowering(executable),
1744 kind,
1745 )
1746 .unwrap()
1747 }
1748
1749 fn execution(
1750 value: &str,
1751 parameters: ArchitectureParameterDescription,
1752 ) -> RealtimeExecutionRequirements {
1753 RealtimeExecutionRequirements::new(
1754 identity(value),
1755 parameters,
1756 vec![lowering_requirement(
1757 "target.weight",
1758 LinearFormat::Dense,
1759 WeightLoweringKind::Direct,
1760 )],
1761 )
1762 .unwrap()
1763 }
1764
1765 fn requirements() -> RealtimeArchitectureRequirements {
1766 let graph = execution_graph();
1767 let units = execution_units(&graph);
1768 let parameters = parameter_description(&graph, &units);
1769 requirements_with_model_contract(
1770 graph,
1771 units,
1772 parameters.clone(),
1773 vec![
1774 execution("execution-native", parameters.clone()),
1775 execution("execution-transformed", parameters),
1776 ],
1777 )
1778 .unwrap()
1779 }
1780
1781 fn requirements_with_model_contract(
1782 graph: ExecutionGraph,
1783 units: ExecutionUnitLayout,
1784 source_parameters: ArchitectureParameterDescription,
1785 executions: Vec<RealtimeExecutionRequirements>,
1786 ) -> Result<RealtimeArchitectureRequirements, RealtimeContractError> {
1787 RealtimeArchitectureRequirements::new(
1788 identity("architecture-7"),
1789 identity("artifact-4"),
1790 graph,
1791 units,
1792 source_parameters,
1793 executions,
1794 required_operators(),
1795 identity("schedule-3"),
1796 schedule(),
1797 identity("state-layout-8"),
1798 state_layout(),
1799 RealtimeMechanismRequirements::new(required_mechanisms()).unwrap(),
1800 RealtimeTopologyPolicy::new(identity("topology-plan-2"), true, [2]).unwrap(),
1801 [
1802 ExecutionResidency::FullyResident,
1803 ExecutionResidency::LayerwiseHost,
1804 ],
1805 )
1806 }
1807
1808 fn proof(requirements: &RealtimeArchitectureRequirements) -> RealtimeArchitectureProof {
1809 RealtimeArchitectureProof::new(
1810 requirements.architecture().clone(),
1811 requirements.speech_schedule_identity().clone(),
1812 requirements.state_layout_identity().clone(),
1813 requirements.topology().identity().clone(),
1814 ParallelTopology::new(2, 1, 1, 1).unwrap(),
1815 1,
1816 )
1817 }
1818
1819 fn request(requirements: &RealtimeArchitectureRequirements) -> RealtimeSelectionRequest {
1820 RealtimeSelectionRequest::new(
1821 requirements.source().clone(),
1822 identity("execution-transformed"),
1823 LayerWeightResidency::LayerwiseHost(crate::LayerwiseLoadOptions::default()),
1824 CacheResidencyPolicy::Device,
1825 Some(proof(requirements)),
1826 completion_policy(),
1827 RealtimeObservationRequirements::new(
1828 true,
1829 [identity("temporal.layer.0"), identity("depth.slice.0")],
1830 ),
1831 )
1832 }
1833
1834 fn state_capabilities() -> StateMechanismCapabilities {
1835 StateMechanismCapabilities::new((0..state_layout().len()).flat_map(|layer| {
1836 state_layout()
1837 .components(layer)
1838 .unwrap()
1839 .iter()
1840 .cloned()
1841 .map(move |component| {
1842 crate::StateComponentMechanism::new(
1843 layer,
1844 component,
1845 Some(crate::StateComponentPlacement::Device),
1846 None,
1847 )
1848 })
1849 .collect::<Vec<_>>()
1850 }))
1851 .with_transactions(true, true)
1852 .with_reset(true)
1853 .with_observation_retention(true)
1854 }
1855
1856 fn capabilities() -> RealtimeMechanismCapabilities {
1857 RealtimeMechanismCapabilities::new(
1858 required_operators(),
1859 required_mechanisms(),
1860 [
1861 ExecutionResidency::FullyResident,
1862 ExecutionResidency::LayerwiseHost,
1863 ],
1864 vec![WeightLoweringCapability::new(
1865 lowering(LinearFormat::Dense),
1866 WeightLoweringKind::Direct,
1867 )],
1868 state_capabilities(),
1869 NonZeroUsize::new(2).unwrap(),
1870 completion_capabilities(),
1871 SessionCapabilities::new(true, true, true),
1872 )
1873 .with_observation_identities([identity("temporal.layer.0"), identity("depth.slice.0")])
1874 }
1875
1876 struct SyntheticRealtimeSupport {
1877 supported: bool,
1878 lowering_queries: std::cell::Cell<usize>,
1879 state_queries: std::cell::Cell<usize>,
1880 }
1881
1882 impl crate::RealtimeMechanismSupport for SyntheticRealtimeSupport {
1883 fn facts(&self) -> crate::RealtimeMechanismFacts {
1884 let state = if self.supported {
1885 crate::StateLifecycleCapabilities::new()
1886 .with_transactions(true, true)
1887 .with_reset(true)
1888 .with_observation_retention(true)
1889 } else {
1890 crate::StateLifecycleCapabilities::new()
1891 };
1892 crate::RealtimeMechanismFacts::new(
1893 required_operators(),
1894 required_mechanisms(),
1895 [
1896 ExecutionResidency::FullyResident,
1897 ExecutionResidency::LayerwiseHost,
1898 ],
1899 NonZeroUsize::new(2).unwrap(),
1900 completion_capabilities(),
1901 SessionCapabilities::new(true, true, true),
1902 )
1903 .with_state_lifecycle(state)
1904 .with_observation_identities([identity("temporal.layer.0"), identity("depth.slice.0")])
1905 }
1906
1907 fn supports_lowering(
1908 &self,
1909 _descriptor: &WeightLoweringDescriptor,
1910 _kind: WeightLoweringKind,
1911 ) -> bool {
1912 self.lowering_queries.set(self.lowering_queries.get() + 1);
1913 self.supported
1914 }
1915
1916 fn state_component_placements(
1917 &self,
1918 _component: &eredu_core::cache::StateComponentPolicy,
1919 ) -> (
1920 Option<StateComponentPlacement>,
1921 Option<StateComponentPlacement>,
1922 ) {
1923 self.state_queries.set(self.state_queries.get() + 1);
1924 (
1925 self.supported.then_some(StateComponentPlacement::Device),
1926 None,
1927 )
1928 }
1929 }
1930
1931 #[test]
1932 fn realtime_synthesis_matches_independent_oracle_and_deduplicates_all_executions() {
1933 let requirements = requirements();
1934 let support = SyntheticRealtimeSupport {
1935 supported: true,
1936 lowering_queries: std::cell::Cell::new(0),
1937 state_queries: std::cell::Cell::new(0),
1938 };
1939 let synthesized = crate::synthesize_realtime_capabilities(&requirements, &support);
1940 assert_eq!(synthesized, capabilities());
1941 assert_eq!(support.lowering_queries.get(), 2);
1942 assert_eq!(synthesized.weight_lowerings().len(), 1);
1943 assert_eq!(
1944 support.state_queries.get(),
1945 synthesized.state().components().len()
1946 );
1947 assert_eq!(
1948 select_realtime_realization(&requirements, &request(&requirements), &synthesized),
1949 select_realtime_realization(&requirements, &request(&requirements), &capabilities()),
1950 );
1951 }
1952
1953 #[test]
1954 fn realtime_synthesis_preserves_kind_and_fails_closed_for_missing_mechanisms() {
1955 let mut requirements = requirements();
1956 requirements.executions[1].weight_lowerings = vec![lowering_requirement(
1957 "target.weight",
1958 LinearFormat::Dense,
1959 WeightLoweringKind::Derived,
1960 )];
1961 let mut support = SyntheticRealtimeSupport {
1962 supported: true,
1963 lowering_queries: std::cell::Cell::new(0),
1964 state_queries: std::cell::Cell::new(0),
1965 };
1966 let synthesized = crate::synthesize_realtime_capabilities(&requirements, &support);
1967 assert_eq!(synthesized.weight_lowerings().len(), 2);
1968 assert_eq!(
1969 synthesized.weight_lowerings()[0].kind(),
1970 WeightLoweringKind::Direct
1971 );
1972 assert_eq!(
1973 synthesized.weight_lowerings()[1].kind(),
1974 WeightLoweringKind::Derived
1975 );
1976 support.supported = false;
1977 let absent = crate::synthesize_realtime_capabilities(&requirements, &support);
1978 assert!(absent.weight_lowerings().is_empty());
1979 assert!(absent.state().components().is_empty());
1980 assert!(!absent.state().checkpoint());
1981 assert!(!absent.state().rollback());
1982 assert!(!absent.state().reset());
1983 assert!(!absent.state().observation_retention());
1984 assert!(
1985 select_realtime_realization(&requirements, &request(&requirements), &absent).is_err()
1986 );
1987 }
1988
1989 #[test]
1990 fn realtime_synthesis_rejects_portably_invalid_geometry_before_native_predicates() {
1991 let mut requirements = requirements();
1992 for execution in &mut requirements.executions {
1993 execution.weight_lowerings[0].descriptor = WeightLoweringDescriptor::new(
1994 SourceTensorEncoding::Safetensors(StoredDtype::F32),
1995 LinearFormat::Dense,
1996 vec![2, 3],
1997 vec![2, 4],
1998 None,
1999 )
2000 .unwrap();
2001 }
2002 let support = SyntheticRealtimeSupport {
2003 supported: true,
2004 lowering_queries: std::cell::Cell::new(0),
2005 state_queries: std::cell::Cell::new(0),
2006 };
2007 let capabilities = crate::synthesize_realtime_capabilities(&requirements, &support);
2008 assert!(capabilities.weight_lowerings().is_empty());
2009 assert_eq!(support.lowering_queries.get(), 0);
2010 assert!(
2011 select_realtime_realization(&requirements, &request(&requirements), &capabilities)
2012 .is_err()
2013 );
2014 }
2015
2016 #[test]
2017 fn realtime_session_requirements_do_not_depend_on_observation_requests() {
2018 let requirements = requirements();
2019 for required in [
2020 SessionCapabilities::new(false, true, false),
2021 SessionCapabilities::new(false, false, true),
2022 ] {
2023 let mut request = request(&requirements).with_required_session_capabilities(required);
2024 request.observations = RealtimeObservationRequirements::new(false, []);
2025 let mut capabilities = capabilities();
2026 capabilities.session = SessionCapabilities::new(true, false, false);
2027 let error =
2028 select_realtime_realization(&requirements, &request, &capabilities).unwrap_err();
2029 assert_eq!(
2030 error.issues(),
2031 if required.output_observation() {
2032 &[RealtimeSelectionIssue::MissingOutputObservation][..]
2033 } else {
2034 &[RealtimeSelectionIssue::MissingActivationInspection][..]
2035 }
2036 );
2037 capabilities.session = SessionCapabilities::new(true, true, true);
2038 capabilities.state = capabilities.state.with_observation_retention(false);
2039 let error =
2040 select_realtime_realization(&requirements, &request, &capabilities).unwrap_err();
2041 assert_eq!(
2042 error.issues(),
2043 &[RealtimeSelectionIssue::MissingStateObservationRetention]
2044 );
2045 }
2046 }
2047
2048 #[test]
2049 fn identities_and_requirement_sets_are_exact_and_nonempty() {
2050 assert_eq!(
2051 RealtimeIdentity::new(" \t"),
2052 Err(RealtimeContractError::EmptyIdentity)
2053 );
2054 let exact = RealtimeIdentity::new(" exact identity ").unwrap();
2055 assert_eq!(exact.as_str(), " exact identity ");
2056 assert_eq!(
2057 RealtimeMechanismRequirements::new([]),
2058 Err(RealtimeContractError::EmptyMechanismRequirements)
2059 );
2060 assert_eq!(
2061 RealtimeTopologyPolicy::new(identity("none"), false, []),
2062 Err(RealtimeContractError::EmptyTopologyPolicy)
2063 );
2064 let duplicate = lowering_requirement(
2065 "duplicate.target",
2066 LinearFormat::Dense,
2067 WeightLoweringKind::Direct,
2068 );
2069 let graph = execution_graph();
2070 let units = execution_units(&graph);
2071 assert_eq!(
2072 RealtimeExecutionRequirements::new(
2073 identity("duplicate-execution"),
2074 parameter_description(&graph, &units),
2075 vec![duplicate.clone(), duplicate],
2076 ),
2077 Err(RealtimeContractError::DuplicateWeightLoweringTarget)
2078 );
2079 }
2080
2081 #[test]
2082 fn model_construction_contract_requires_one_exact_graph_and_layout() {
2083 let graph = execution_graph();
2084 let units = execution_units(&graph);
2085 let parameters = parameter_description(&graph, &units);
2086 let other_graph = ExecutionGraph::chain(["other"]).unwrap();
2087 let other_units = ExecutionUnitLayout::new(&other_graph, [1]).unwrap();
2088 let other_parameters = parameter_description(&other_graph, &other_units);
2089
2090 assert_eq!(
2091 requirements_with_model_contract(
2092 graph.clone(),
2093 other_units,
2094 parameters.clone(),
2095 vec![execution("execution", parameters.clone())],
2096 ),
2097 Err(RealtimeContractError::ExecutionGraphLayoutMismatch)
2098 );
2099 assert_eq!(
2100 requirements_with_model_contract(
2101 graph.clone(),
2102 units.clone(),
2103 other_parameters.clone(),
2104 vec![execution("execution", parameters.clone())],
2105 ),
2106 Err(RealtimeContractError::SourceParameterDescriptionMismatch)
2107 );
2108 assert_eq!(
2109 requirements_with_model_contract(
2110 graph,
2111 units,
2112 parameters.clone(),
2113 vec![execution("execution", other_parameters)],
2114 ),
2115 Err(
2116 RealtimeContractError::ExecutionParameterDescriptionMismatch {
2117 execution: identity("execution"),
2118 }
2119 )
2120 );
2121 }
2122
2123 #[test]
2124 fn weight_components_enforce_recipe_geometry_and_atomic_family_invariants() {
2125 let component = |target: &str,
2126 owner: Option<&str>,
2127 recipe: Option<&str>,
2128 sources: Vec<&str>,
2129 shape: Vec<usize>,
2130 role| {
2131 let retained_recipe = (owner.is_some() && recipe.is_some()).then(|| {
2132 let mut inputs = sources
2133 .iter()
2134 .map(|key| {
2135 eredu_checkpoint::recipe::DerivedWeightRecipe::source(
2136 *key,
2137 eredu_checkpoint::store::TensorSelection::Full,
2138 )
2139 })
2140 .collect::<Vec<_>>();
2141 if inputs.len() == 1 {
2142 inputs.pop().expect("one source recipe")
2143 } else {
2144 eredu_checkpoint::recipe::DerivedWeightRecipe::Concatenate { axis: 0, inputs }
2145 }
2146 });
2147 let output = retained_recipe.as_ref().map(|_| {
2148 let elements = shape.iter().copied().product::<usize>();
2149 eredu_checkpoint::recipe::RecipeMetadata {
2150 shape: shape.clone(),
2151 dtype: eredu_checkpoint::recipe::RecipeDtype::F32,
2152 byte_len: u64::try_from(elements.saturating_mul(4)).unwrap(),
2153 }
2154 });
2155 RealtimeWeightComponentRequirement::new(
2156 identity(target),
2157 owner.map(identity),
2158 recipe.map(identity),
2159 retained_recipe,
2160 output,
2161 sources.into_iter().map(identity),
2162 shape,
2163 role,
2164 )
2165 };
2166 assert_eq!(
2167 component(
2168 "weight",
2169 Some("owner"),
2170 Some("recipe"),
2171 vec!["source"],
2172 vec![],
2173 RealtimeWeightComponentRole::Primary,
2174 ),
2175 Err(RealtimeContractError::InvalidWeightComponentShape)
2176 );
2177 assert_eq!(
2178 component(
2179 "weight",
2180 Some("owner"),
2181 Some("recipe"),
2182 vec!["source"],
2183 vec![8, 0],
2184 RealtimeWeightComponentRole::Primary,
2185 ),
2186 Err(RealtimeContractError::InvalidWeightComponentShape)
2187 );
2188 assert_eq!(
2189 component(
2190 "weight",
2191 Some("owner"),
2192 None,
2193 vec!["source"],
2194 vec![8, 8],
2195 RealtimeWeightComponentRole::Primary,
2196 ),
2197 Err(RealtimeContractError::IncompleteWeightComponentRecipe)
2198 );
2199 assert_eq!(
2200 component(
2201 "weight",
2202 None,
2203 Some("recipe"),
2204 vec!["source"],
2205 vec![8, 8],
2206 RealtimeWeightComponentRole::Primary,
2207 ),
2208 Err(RealtimeContractError::IncompleteWeightComponentRecipe)
2209 );
2210 assert_eq!(
2211 component(
2212 "weight",
2213 Some("owner"),
2214 Some("recipe"),
2215 vec![],
2216 vec![8, 8],
2217 RealtimeWeightComponentRole::Primary,
2218 ),
2219 Err(RealtimeContractError::EmptyRecipeComponentSources)
2220 );
2221 assert_eq!(
2222 component(
2223 "weight",
2224 None,
2225 None,
2226 vec!["source"],
2227 vec![8, 8],
2228 RealtimeWeightComponentRole::Primary,
2229 ),
2230 Err(RealtimeContractError::GeneratedComponentHasSources)
2231 );
2232
2233 let primary = component(
2234 "weight",
2235 Some("owner"),
2236 Some("recipe"),
2237 vec!["source", "source"],
2238 vec![8, 8],
2239 RealtimeWeightComponentRole::Primary,
2240 )
2241 .unwrap();
2242 let generated_scale = component(
2243 "scales",
2244 None,
2245 None,
2246 vec![],
2247 vec![8, 1],
2248 RealtimeWeightComponentRole::Scale,
2249 )
2250 .unwrap();
2251 let generated_bias = component(
2252 "biases",
2253 None,
2254 None,
2255 vec![],
2256 vec![8, 1],
2257 RealtimeWeightComponentRole::AffineBias,
2258 )
2259 .unwrap();
2260 let packed = RealtimeWeightLoweringRequirement::new(
2261 identity("weight"),
2262 [
2263 primary.clone(),
2264 generated_scale.clone(),
2265 generated_bias.clone(),
2266 ],
2267 lowering(LinearFormat::Dense),
2268 WeightLoweringKind::Transform,
2269 )
2270 .unwrap();
2271 assert_eq!(packed.primary(), &primary);
2272 assert_eq!(packed.scale(), Some(&generated_scale));
2273 assert_eq!(packed.affine_bias(), Some(&generated_bias));
2274 assert_eq!(
2275 packed.components(),
2276 &[
2277 primary.clone(),
2278 generated_scale.clone(),
2279 generated_bias.clone(),
2280 ]
2281 );
2282 assert_eq!(
2283 packed.primary().source_occurrences(),
2284 &[identity("source"), identity("source")]
2285 );
2286 assert!(!packed.scale().unwrap().is_recipe_backed());
2287 assert_eq!(packed.scale().unwrap().recipe_owner(), None);
2288 assert_eq!(packed.scale().unwrap().recipe_identity(), None);
2289 assert!(packed.scale().unwrap().source_occurrences().is_empty());
2290 assert_eq!(packed.scale().unwrap().physical_shape(), &[8, 1]);
2291
2292 assert_eq!(
2293 RealtimeWeightLoweringRequirement::new(
2294 identity("weight"),
2295 [generated_scale.clone()],
2296 lowering(LinearFormat::Dense),
2297 WeightLoweringKind::Transform,
2298 ),
2299 Err(RealtimeContractError::MissingPrimaryWeightComponent)
2300 );
2301 let other_primary = component(
2302 "other",
2303 Some("owner"),
2304 Some("recipe"),
2305 vec!["source"],
2306 vec![8, 8],
2307 RealtimeWeightComponentRole::Primary,
2308 )
2309 .unwrap();
2310 assert_eq!(
2311 RealtimeWeightLoweringRequirement::new(
2312 identity("weight"),
2313 [other_primary.clone()],
2314 lowering(LinearFormat::Dense),
2315 WeightLoweringKind::Direct,
2316 ),
2317 Err(RealtimeContractError::PrimaryWeightComponentTargetMismatch)
2318 );
2319 assert_eq!(
2320 RealtimeWeightLoweringRequirement::new(
2321 identity("weight"),
2322 [primary.clone(), other_primary],
2323 lowering(LinearFormat::Dense),
2324 WeightLoweringKind::Direct,
2325 ),
2326 Err(RealtimeContractError::DuplicatePrimaryWeightComponent)
2327 );
2328 let second_scale = component(
2329 "other.scales",
2330 None,
2331 None,
2332 vec![],
2333 vec![8, 1],
2334 RealtimeWeightComponentRole::Scale,
2335 )
2336 .unwrap();
2337 assert_eq!(
2338 RealtimeWeightLoweringRequirement::new(
2339 identity("weight"),
2340 [primary.clone(), generated_scale.clone(), second_scale],
2341 lowering(LinearFormat::Dense),
2342 WeightLoweringKind::Transform,
2343 ),
2344 Err(RealtimeContractError::DuplicateWeightComponentRole {
2345 role: RealtimeWeightComponentRole::Scale,
2346 })
2347 );
2348 let second_bias = component(
2349 "other.biases",
2350 None,
2351 None,
2352 vec![],
2353 vec![8, 1],
2354 RealtimeWeightComponentRole::AffineBias,
2355 )
2356 .unwrap();
2357 assert_eq!(
2358 RealtimeWeightLoweringRequirement::new(
2359 identity("weight"),
2360 [primary.clone(), generated_bias, second_bias],
2361 lowering(LinearFormat::Dense),
2362 WeightLoweringKind::Transform,
2363 ),
2364 Err(RealtimeContractError::DuplicateWeightComponentRole {
2365 role: RealtimeWeightComponentRole::AffineBias,
2366 })
2367 );
2368 let duplicate_target = component(
2369 "weight",
2370 None,
2371 None,
2372 vec![],
2373 vec![8, 1],
2374 RealtimeWeightComponentRole::Scale,
2375 )
2376 .unwrap();
2377 assert_eq!(
2378 RealtimeWeightLoweringRequirement::new(
2379 identity("weight"),
2380 [primary, duplicate_target],
2381 lowering(LinearFormat::Dense),
2382 WeightLoweringKind::Transform,
2383 ),
2384 Err(RealtimeContractError::DuplicateWeightComponentTarget)
2385 );
2386 }
2387
2388 #[test]
2389 fn exact_selection_closes_architecture_and_mechanism_choices_once() {
2390 let requirements = requirements();
2391 let selected =
2392 select_realtime_realization(&requirements, &request(&requirements), &capabilities())
2393 .unwrap();
2394
2395 assert_eq!(selected.requirements(), &requirements);
2396 assert_eq!(selected.source().as_str(), "artifact-4");
2397 assert_eq!(selected.execution().as_str(), "execution-transformed");
2398 assert_eq!(selected.execution_graph(), requirements.execution_graph());
2399 assert_eq!(selected.execution_units(), requirements.execution_units());
2400 assert_eq!(
2401 selected.source_parameters(),
2402 requirements.source_parameters()
2403 );
2404 assert_eq!(selected.operators(), required_operators());
2405 assert_eq!(capabilities().operators(), required_operators());
2406 assert_eq!(
2407 selected.execution_parameters(),
2408 requirements.executions()[1].parameters()
2409 );
2410 assert_eq!(
2411 selected.residency(),
2412 LayerWeightResidency::LayerwiseHost(crate::LayerwiseLoadOptions::default())
2413 );
2414 assert_eq!(
2415 selected.topology(),
2416 ParallelTopology::new(2, 1, 1, 1).unwrap()
2417 );
2418 assert_eq!(selected.rank(), 1);
2419 assert_eq!(selected.completion(), completion_policy());
2420 assert_eq!(
2421 selected.requirements().speech_schedule(),
2422 requirements.speech_schedule()
2423 );
2424 assert_eq!(
2425 selected.requirements().state_layout(),
2426 requirements.state_layout()
2427 );
2428 assert!(selected.observations().output());
2429 assert_eq!(
2430 selected
2431 .observations()
2432 .activations()
2433 .iter()
2434 .map(RealtimeIdentity::as_str)
2435 .collect::<Vec<_>>(),
2436 vec!["depth.slice.0", "temporal.layer.0"]
2437 );
2438 let selected_lowering = &selected.execution_requirements().weight_lowerings()[0];
2439 assert_eq!(selected_lowering.target().as_str(), "target.weight");
2440 let primary = selected_lowering.primary();
2441 assert_eq!(primary.target().as_str(), "target.weight");
2442 assert_eq!(primary.recipe_owner().unwrap().as_str(), "canonical-owner");
2443 assert_eq!(primary.recipe_identity().unwrap().as_str(), "recipe-v1");
2444 assert_eq!(
2445 primary
2446 .source_occurrences()
2447 .iter()
2448 .map(RealtimeIdentity::as_str)
2449 .collect::<Vec<_>>(),
2450 vec!["source-a", "source-b"]
2451 );
2452 assert_eq!(primary.physical_shape(), &[8, 8]);
2453 assert_eq!(primary.role(), RealtimeWeightComponentRole::Primary);
2454 assert_eq!(
2455 selected_lowering.components(),
2456 std::slice::from_ref(primary)
2457 );
2458 assert_eq!(selected_lowering.kind(), WeightLoweringKind::Direct);
2459 assert_eq!(selected_lowering.scale(), None);
2460 assert_eq!(selected_lowering.affine_bias(), None);
2461
2462 let selected_state = selected.state();
2463 assert_eq!(selected_state.layout(), requirements.state_layout());
2464 assert_eq!(selected_state.policy(), &CacheResidencyPolicy::Device);
2465 let expected_components = (0..requirements.state_layout().len()).flat_map(|layer| {
2466 requirements
2467 .state_layout()
2468 .components(layer)
2469 .unwrap()
2470 .iter()
2471 .map(move |component| (layer, component))
2472 });
2473 let expected_count = expected_components.clone().count();
2474 assert_eq!(selected_state.components().len(), expected_count);
2475 for (selected_component, (layer, component)) in
2476 selected_state.components().iter().zip(expected_components)
2477 {
2478 assert_eq!(selected_component.layer(), layer);
2479 assert_eq!(selected_component.component(), component);
2480 assert_eq!(
2481 selected_component.placement(),
2482 StateComponentPlacement::Device
2483 );
2484 }
2485 assert!(selected_state.checkpoint());
2486 assert!(selected_state.rollback());
2487 assert!(selected_state.reset());
2488 assert!(selected_state.observation_retention());
2489 }
2490
2491 #[test]
2492 fn selection_reports_stable_fail_closed_diagnostics() {
2493 let requirements = requirements();
2494 let request = RealtimeSelectionRequest::new(
2495 identity("wrong-artifact"),
2496 identity("unknown-execution"),
2497 LayerWeightResidency::DenseDiskStream(crate::DenseDiskStreamLoadOptions::default()),
2498 CacheResidencyPolicy::Device,
2499 None,
2500 completion_policy(),
2501 RealtimeObservationRequirements::new(true, [identity("temporal.layer.0")]),
2502 );
2503 let capabilities = RealtimeMechanismCapabilities::new(
2504 NeuralOperatorCapabilities::NONE,
2505 [RealtimeMechanism::NeuralOperations],
2506 [ExecutionResidency::FullyResident],
2507 Vec::new(),
2508 StateMechanismCapabilities::new([]),
2509 NonZeroUsize::new(1).unwrap(),
2510 crate::CommunicationCompletionCapabilities::new([
2511 CompletionCancellationMode::NativeCancel,
2512 ])
2513 .unwrap(),
2514 SessionCapabilities::default(),
2515 );
2516 let error =
2517 select_realtime_realization(&requirements, &request, &capabilities).unwrap_err();
2518
2519 assert_eq!(
2520 error.issues(),
2521 &[
2522 RealtimeSelectionIssue::MissingArchitectureProof,
2523 RealtimeSelectionIssue::SourceIdentityMismatch,
2524 RealtimeSelectionIssue::ExecutionIdentityNotAdmitted,
2525 RealtimeSelectionIssue::MissingNeuralOperator {
2526 operator: "exp".into(),
2527 },
2528 RealtimeSelectionIssue::MissingNeuralOperator {
2529 operator: "softmax_axis".into(),
2530 },
2531 RealtimeSelectionIssue::ResidencyNotAdmitted,
2532 RealtimeSelectionIssue::ResidencyUnavailable,
2533 RealtimeSelectionIssue::MissingMechanism(RealtimeMechanism::TensorOperations),
2534 RealtimeSelectionIssue::MissingMechanism(
2535 RealtimeMechanism::ParameterMaterialization,
2536 ),
2537 RealtimeSelectionIssue::MissingMechanism(RealtimeMechanism::ParameterStorage),
2538 RealtimeSelectionIssue::MissingMechanism(RealtimeMechanism::StateStorage),
2539 RealtimeSelectionIssue::MissingMechanism(RealtimeMechanism::CoordinateStorage),
2540 RealtimeSelectionIssue::MissingMechanism(RealtimeMechanism::Sampling),
2541 RealtimeSelectionIssue::MissingMechanism(RealtimeMechanism::Randomness),
2542 RealtimeSelectionIssue::MissingMechanism(RealtimeMechanism::HostConversion),
2543 RealtimeSelectionIssue::MissingMechanism(RealtimeMechanism::ExactCompletion),
2544 RealtimeSelectionIssue::MissingMechanism(RealtimeMechanism::ResourceRetention),
2545 RealtimeSelectionIssue::MissingMechanism(RealtimeMechanism::Transfer),
2546 RealtimeSelectionIssue::MissingMechanism(RealtimeMechanism::Collectives),
2547 RealtimeSelectionIssue::CompletionPolicyUnavailable,
2548 RealtimeSelectionIssue::StateComponentUnavailable {
2549 layer: 0,
2550 component: "attention.keys".into(),
2551 },
2552 RealtimeSelectionIssue::StateComponentUnavailable {
2553 layer: 0,
2554 component: "attention.values".into(),
2555 },
2556 RealtimeSelectionIssue::StateComponentUnavailable {
2557 layer: 1,
2558 component: "attention.keys".into(),
2559 },
2560 RealtimeSelectionIssue::StateComponentUnavailable {
2561 layer: 1,
2562 component: "attention.values".into(),
2563 },
2564 RealtimeSelectionIssue::MissingStateCheckpoint,
2565 RealtimeSelectionIssue::MissingStateRollback,
2566 RealtimeSelectionIssue::MissingStateReset,
2567 RealtimeSelectionIssue::MissingStateObservationRetention,
2568 RealtimeSelectionIssue::MissingPersistentState,
2569 RealtimeSelectionIssue::MissingOutputObservation,
2570 RealtimeSelectionIssue::MissingActivationInspection,
2571 RealtimeSelectionIssue::ObservationUnavailable {
2572 observation: identity("temporal.layer.0"),
2573 },
2574 ]
2575 );
2576 assert_eq!(
2577 error.to_string(),
2578 "realtime realization is unsupported: architecture topology and rank proof is missing; source artifact identity mismatch; execution identity is not architecture-admitted; required neural operator exp is unavailable; required neural operator softmax_axis is unavailable; residency is not architecture-admitted; residency mechanism is unavailable; required realtime mechanism TensorOperations is unavailable; required realtime mechanism ParameterMaterialization is unavailable; required realtime mechanism ParameterStorage is unavailable; required realtime mechanism StateStorage is unavailable; required realtime mechanism CoordinateStorage is unavailable; required realtime mechanism Sampling is unavailable; required realtime mechanism Randomness is unavailable; required realtime mechanism HostConversion is unavailable; required realtime mechanism ExactCompletion is unavailable; required realtime mechanism ResourceRetention is unavailable; required realtime mechanism Transfer is unavailable; required realtime mechanism Collectives is unavailable; requested completion timeout disposition is unavailable; state component attention.keys at layer 0 is unavailable; state component attention.values at layer 0 is unavailable; state component attention.keys at layer 1 is unavailable; state component attention.values at layer 1 is unavailable; state checkpoint mechanism is unavailable; state rollback mechanism is unavailable; state reset mechanism is unavailable; state observation retention is unavailable; persistent session state is unavailable; completed output observation is unavailable; activation inspection is unavailable; requested activation observation temporal.layer.0 is unavailable"
2579 );
2580 }
2581
2582 #[test]
2583 fn exact_weight_lowering_must_be_implemented_before_construction() {
2584 let requirements = requirements();
2585 let capabilities = RealtimeMechanismCapabilities::new(
2586 required_operators(),
2587 required_mechanisms(),
2588 [
2589 ExecutionResidency::FullyResident,
2590 ExecutionResidency::LayerwiseHost,
2591 ],
2592 vec![WeightLoweringCapability::new(
2593 lowering(LinearFormat::Dense),
2594 WeightLoweringKind::Transform,
2595 )],
2596 state_capabilities(),
2597 NonZeroUsize::new(2).unwrap(),
2598 completion_capabilities(),
2599 SessionCapabilities::new(true, true, true),
2600 )
2601 .with_observation_identities([identity("temporal.layer.0"), identity("depth.slice.0")]);
2602 let error =
2603 select_realtime_realization(&requirements, &request(&requirements), &capabilities)
2604 .unwrap_err();
2605
2606 assert_eq!(
2607 error.issues(),
2608 &[RealtimeSelectionIssue::WeightLoweringUnavailable {
2609 index: 0,
2610 target: identity("target.weight"),
2611 }]
2612 );
2613 }
2614
2615 #[test]
2616 fn topology_and_rank_are_fail_closed() {
2617 let requirements = requirements();
2618 let invalid_proof = RealtimeArchitectureProof::new(
2619 requirements.architecture().clone(),
2620 requirements.speech_schedule_identity().clone(),
2621 requirements.state_layout_identity().clone(),
2622 requirements.topology().identity().clone(),
2623 ParallelTopology::new(2, 2, 1, 1).unwrap(),
2624 4,
2625 );
2626 let request = RealtimeSelectionRequest::new(
2627 requirements.source().clone(),
2628 identity("execution-native"),
2629 LayerWeightResidency::FullyResident,
2630 CacheResidencyPolicy::Device,
2631 Some(invalid_proof),
2632 completion_policy(),
2633 RealtimeObservationRequirements::default(),
2634 );
2635 let error =
2636 select_realtime_realization(&requirements, &request, &capabilities()).unwrap_err();
2637 assert_eq!(
2638 error.issues(),
2639 &[
2640 RealtimeSelectionIssue::TopologyNotAdmitted,
2641 RealtimeSelectionIssue::RankOutOfRange,
2642 ]
2643 );
2644 }
2645
2646 #[test]
2647 fn selection_failure_prevents_every_construction_callback() {
2648 let requirements = requirements();
2649 let invalid_request = RealtimeSelectionRequest::new(
2650 identity("wrong-artifact"),
2651 identity("execution-native"),
2652 LayerWeightResidency::FullyResident,
2653 CacheResidencyPolicy::Device,
2654 Some(proof(&requirements)),
2655 completion_policy(),
2656 RealtimeObservationRequirements::default(),
2657 );
2658 let calls = Rc::new(Cell::new(0));
2659 let callback = || {
2660 let calls = Rc::clone(&calls);
2661 move |_: &SelectedRealtimeRealization| {
2662 calls.set(calls.get() + 1);
2663 Ok::<_, Infallible>(())
2664 }
2665 };
2666 let module_calls = Rc::clone(&calls);
2667 let state_calls = Rc::clone(&calls);
2668 let error = select_and_prepare_realtime_realization(
2669 &requirements,
2670 &invalid_request,
2671 &capabilities(),
2672 callback(),
2673 move |_, _| {
2674 module_calls.set(module_calls.get() + 1);
2675 Ok::<_, Infallible>(())
2676 },
2677 move |_, _| {
2678 state_calls.set(state_calls.get() + 1);
2679 Ok::<_, Infallible>(())
2680 },
2681 callback(),
2682 callback(),
2683 )
2684 .unwrap_err();
2685
2686 assert!(matches!(error, RealtimePreparationError::Selection(_)));
2687 assert_eq!(calls.get(), 0);
2688 }
2689
2690 #[test]
2691 fn successful_gate_constructs_in_dependency_order() {
2692 let requirements = requirements();
2693 let calls = Rc::new(Cell::new(0));
2694 let prepared = select_and_prepare_realtime_realization(
2695 &requirements,
2696 &request(&requirements),
2697 &capabilities(),
2698 {
2699 let calls = Rc::clone(&calls);
2700 move |_| {
2701 assert_eq!(calls.replace(1), 0);
2702 Ok::<_, Infallible>("payload")
2703 }
2704 },
2705 {
2706 let calls = Rc::clone(&calls);
2707 move |_, payload| {
2708 assert_eq!((*payload, calls.replace(2)), ("payload", 1));
2709 Ok::<_, Infallible>("modules")
2710 }
2711 },
2712 {
2713 let calls = Rc::clone(&calls);
2714 move |_, modules| {
2715 assert_eq!((*modules, calls.replace(3)), ("modules", 2));
2716 Ok::<_, Infallible>("state")
2717 }
2718 },
2719 {
2720 let calls = Rc::clone(&calls);
2721 move |_| {
2722 assert_eq!(calls.replace(4), 3);
2723 Ok::<_, Infallible>("queue")
2724 }
2725 },
2726 {
2727 let calls = Rc::clone(&calls);
2728 move |selected| {
2729 assert_eq!(selected.topology().tensor(), 2);
2730 assert_eq!(calls.replace(5), 4);
2731 Ok::<_, Infallible>("group")
2732 }
2733 },
2734 )
2735 .unwrap();
2736
2737 assert_eq!(calls.get(), 5);
2738 let (_, resources) = prepared.into_parts();
2739 assert_eq!(
2740 resources.into_parts(),
2741 ("payload", "modules", "state", "queue", Some("group"))
2742 );
2743 }
2744}