eredu_runtime/

realtime_selection.rs

//! Family-blind realtime requirements, selection, and construction gating.

use std::{collections::BTreeSet, num::NonZeroUsize};

use eredu_core::{ParallelTopology, RealtimeSpeechConfig, SessionCapabilities};
use eredu_nn::NeuralOperatorCapabilities;

use crate::{
    ArchitectureParameterDescription, CacheResidencyPolicy, ExecutionGraph, ExecutionResidency,
    ExecutionUnitLayout, LayerWeightResidency, StateComponentMechanism, StateComponentPlacement,
    StateLayout, StateMechanismCapabilities, WeightLoweringCapability, WeightLoweringDescriptor,
    WeightLoweringKind,
};

/// Exact nonempty identity used by neutral realtime contracts.
#[derive(Debug, Clone, Eq, Hash, Ord, PartialEq, PartialOrd)]
pub struct RealtimeIdentity(String);

impl RealtimeIdentity {
    /// Creates an identity while preserving its exact nonempty text.
    pub fn new(value: impl Into<String>) -> Result<Self, RealtimeContractError> {
        let value = value.into();
        if value.trim().is_empty() {
            return Err(RealtimeContractError::EmptyIdentity);
        }
        Ok(Self(value))
    }

    /// Returns the exact identity text.
    pub fn as_str(&self) -> &str {
        &self.0
    }
}

impl std::fmt::Display for RealtimeIdentity {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        formatter.write_str(&self.0)
    }
}

/// One generic mechanism that a neutral realtime executor may require.
#[derive(Debug, Clone, Copy, Eq, Hash, Ord, PartialEq, PartialOrd)]
#[non_exhaustive]
pub enum RealtimeMechanism {
    /// Generic tensor construction, indexing, stacking, and slicing.
    TensorOperations,
    /// Backend-neutral tensor operations and architecture neural operations.
    NeuralOperations,
    /// Exact selected checkpoint recipes and source-to-execution lowering.
    ParameterMaterialization,
    /// Resident or bounded immutable-parameter storage.
    ParameterStorage,
    /// Mutable model-state storage matching architecture geometry.
    StateStorage,
    /// Typed coordinate payload storage and retention.
    CoordinateStorage,
    /// Generic logits processing and token sampling.
    Sampling,
    /// Transactional random-state storage and advancement.
    Randomness,
    /// Portable-host to opaque-tensor conversion.
    HostConversion,
    /// Exact completion and resource-retention tracking.
    ExactCompletion,
    /// Submitted tensor, store, validation, queue, and collective retention.
    ResourceRetention,
    /// Generic device or rank-local transfer.
    Transfer,
    /// Named activation observation and intervention.
    Observation,
    /// Opaque collective communication.
    Collectives,
    /// Optional execution timing.
    Timing,
}

/// Semantic role of one physical component in an atomic matrix lowering.
#[derive(Debug, Clone, Copy, Eq, Hash, Ord, PartialEq, PartialOrd)]
pub enum RealtimeWeightComponentRole {
    /// Primary matrix weight.
    Primary,
    /// Scale values required by the executable format.
    Scale,
    /// Affine bias values required by the executable format.
    AffineBias,
}

/// One exact physical or transform-generated component of a matrix lowering.
#[derive(Debug, Clone, Eq, PartialEq)]
pub struct RealtimeWeightComponentRequirement {
    target: RealtimeIdentity,
    recipe_owner: Option<RealtimeIdentity>,
    recipe_identity: Option<RealtimeIdentity>,
    recipe: Option<eredu_checkpoint::recipe::DerivedWeightRecipe>,
    recipe_output: Option<eredu_checkpoint::recipe::RecipeMetadata>,
    source_occurrences: Vec<RealtimeIdentity>,
    physical_shape: Vec<usize>,
    role: RealtimeWeightComponentRole,
}

impl RealtimeWeightComponentRequirement {
    /// Creates one recipe-backed source component or one generated target component.
    #[allow(clippy::too_many_arguments)]
    pub fn new(
        target: RealtimeIdentity,
        recipe_owner: Option<RealtimeIdentity>,
        recipe_identity: Option<RealtimeIdentity>,
        recipe: Option<eredu_checkpoint::recipe::DerivedWeightRecipe>,
        recipe_output: Option<eredu_checkpoint::recipe::RecipeMetadata>,
        source_occurrences: impl IntoIterator<Item = RealtimeIdentity>,
        physical_shape: impl Into<Vec<usize>>,
        role: RealtimeWeightComponentRole,
    ) -> Result<Self, RealtimeContractError> {
        let source_occurrences = source_occurrences.into_iter().collect::<Vec<_>>();
        let physical_shape = physical_shape.into();
        if physical_shape.is_empty() || physical_shape.contains(&0) {
            return Err(RealtimeContractError::InvalidWeightComponentShape);
        }
        match (
            &recipe_owner,
            &recipe_identity,
            &recipe,
            &recipe_output,
            source_occurrences.is_empty(),
        ) {
            (Some(_), Some(_), Some(recipe), Some(output), false) => {
                let declared = source_occurrences
                    .iter()
                    .map(RealtimeIdentity::as_str)
                    .collect::<Vec<_>>();
                if recipe.source_occurrences() != declared || output.shape != physical_shape {
                    return Err(RealtimeContractError::WeightComponentRecipeMismatch);
                }
            }
            (Some(_), Some(_), Some(_), Some(_), true) => {
                return Err(RealtimeContractError::EmptyRecipeComponentSources);
            }
            (None, None, None, None, true) => {}
            (None, None, None, None, false) => {
                return Err(RealtimeContractError::GeneratedComponentHasSources);
            }
            _ => return Err(RealtimeContractError::IncompleteWeightComponentRecipe),
        }
        Ok(Self {
            target,
            recipe_owner,
            recipe_identity,
            recipe,
            recipe_output,
            source_occurrences,
            physical_shape,
            role,
        })
    }

    /// Returns the exact executable component target.
    pub const fn target(&self) -> &RealtimeIdentity {
        &self.target
    }

    /// Returns the canonical recipe owner for a source-backed component.
    pub const fn recipe_owner(&self) -> Option<&RealtimeIdentity> {
        self.recipe_owner.as_ref()
    }

    /// Returns the exact recipe identity for a source-backed component.
    pub const fn recipe_identity(&self) -> Option<&RealtimeIdentity> {
        self.recipe_identity.as_ref()
    }

    /// Returns the exact selected architecture recipe for a source-backed component.
    pub const fn recipe(&self) -> Option<&eredu_checkpoint::recipe::DerivedWeightRecipe> {
        self.recipe.as_ref()
    }

    /// Returns exact admission-time output metadata for a source-backed component.
    pub const fn recipe_output(&self) -> Option<&eredu_checkpoint::recipe::RecipeMetadata> {
        self.recipe_output.as_ref()
    }

    /// Returns source identities in recipe traversal order, retaining duplicates.
    pub fn source_occurrences(&self) -> &[RealtimeIdentity] {
        &self.source_occurrences
    }

    /// Returns exact component geometry before rank-local construction.
    pub fn physical_shape(&self) -> &[usize] {
        &self.physical_shape
    }

    /// Returns the component's semantic matrix-family role.
    pub const fn role(&self) -> RealtimeWeightComponentRole {
        self.role
    }

    /// Returns whether this component is produced by an architecture recipe.
    pub const fn is_recipe_backed(&self) -> bool {
        self.recipe_identity.is_some()
    }
}

/// One exact named source-to-execution weight lowering required by an architecture.
#[derive(Debug, Clone, Eq, PartialEq)]
pub struct RealtimeWeightLoweringRequirement {
    target: RealtimeIdentity,
    components: Vec<RealtimeWeightComponentRequirement>,
    descriptor: WeightLoweringDescriptor,
    kind: WeightLoweringKind,
}

impl RealtimeWeightLoweringRequirement {
    /// Binds one executable matrix target to its complete atomic component family.
    pub fn new(
        target: RealtimeIdentity,
        components: impl IntoIterator<Item = RealtimeWeightComponentRequirement>,
        descriptor: WeightLoweringDescriptor,
        kind: WeightLoweringKind,
    ) -> Result<Self, RealtimeContractError> {
        let components = components.into_iter().collect::<Vec<_>>();
        let primary = components
            .iter()
            .filter(|component| component.role == RealtimeWeightComponentRole::Primary)
            .collect::<Vec<_>>();
        match primary.as_slice() {
            [] => return Err(RealtimeContractError::MissingPrimaryWeightComponent),
            [primary] if primary.target != target => {
                return Err(RealtimeContractError::PrimaryWeightComponentTargetMismatch);
            }
            [_] => {}
            _ => return Err(RealtimeContractError::DuplicatePrimaryWeightComponent),
        }
        let targets = components
            .iter()
            .map(RealtimeWeightComponentRequirement::target)
            .collect::<BTreeSet<_>>();
        if targets.len() != components.len() {
            return Err(RealtimeContractError::DuplicateWeightComponentTarget);
        }
        for role in [
            RealtimeWeightComponentRole::Scale,
            RealtimeWeightComponentRole::AffineBias,
        ] {
            if components
                .iter()
                .filter(|component| component.role == role)
                .count()
                > 1
            {
                return Err(RealtimeContractError::DuplicateWeightComponentRole { role });
            }
        }
        Ok(Self {
            target,
            components,
            descriptor,
            kind,
        })
    }

    /// Returns the exact canonical executable parameter target.
    pub const fn target(&self) -> &RealtimeIdentity {
        &self.target
    }

    /// Returns the complete atomic component family in architecture order.
    pub fn components(&self) -> &[RealtimeWeightComponentRequirement] {
        &self.components
    }

    /// Returns the unique primary component.
    pub fn primary(&self) -> &RealtimeWeightComponentRequirement {
        self.component(RealtimeWeightComponentRole::Primary)
            .expect("validated realtime lowering contains one primary component")
    }

    /// Returns the optional scale companion.
    pub fn scale(&self) -> Option<&RealtimeWeightComponentRequirement> {
        self.component(RealtimeWeightComponentRole::Scale)
    }

    /// Returns the optional affine-bias companion.
    pub fn affine_bias(&self) -> Option<&RealtimeWeightComponentRequirement> {
        self.component(RealtimeWeightComponentRole::AffineBias)
    }

    fn component(
        &self,
        role: RealtimeWeightComponentRole,
    ) -> Option<&RealtimeWeightComponentRequirement> {
        self.components
            .iter()
            .find(|component| component.role == role)
    }

    /// Returns the exact geometry-bearing backend lowering query.
    pub const fn descriptor(&self) -> &WeightLoweringDescriptor {
        &self.descriptor
    }

    /// Returns the required direct, derived, transforming, or combined route.
    pub const fn kind(&self) -> WeightLoweringKind {
        self.kind
    }
}

/// One architecture-admitted execution configuration and its exact lowerings.
#[derive(Debug, Clone, Eq, PartialEq)]
pub struct RealtimeExecutionRequirements {
    identity: RealtimeIdentity,
    parameters: ArchitectureParameterDescription,
    weight_lowerings: Vec<RealtimeWeightLoweringRequirement>,
}

impl RealtimeExecutionRequirements {
    /// Binds a stable execution identity to every geometry-bearing weight lowering.
    pub fn new(
        identity: RealtimeIdentity,
        parameters: ArchitectureParameterDescription,
        weight_lowerings: Vec<RealtimeWeightLoweringRequirement>,
    ) -> Result<Self, RealtimeContractError> {
        if weight_lowerings.is_empty() {
            return Err(RealtimeContractError::EmptyWeightLowerings);
        }
        let targets = weight_lowerings
            .iter()
            .map(RealtimeWeightLoweringRequirement::target)
            .collect::<BTreeSet<_>>();
        if targets.len() != weight_lowerings.len() {
            return Err(RealtimeContractError::DuplicateWeightLoweringTarget);
        }
        let members = parameters
            .groups()
            .iter()
            .flat_map(|group| group.group().members())
            .collect::<Vec<_>>();
        let expected_targets = members
            .iter()
            .filter(|member| member.linear_companion().is_none())
            .map(|member| member.target())
            .collect::<BTreeSet<_>>();
        let actual_targets = weight_lowerings
            .iter()
            .map(|lowering| lowering.target().as_str())
            .collect::<BTreeSet<_>>();
        if expected_targets != actual_targets {
            return Err(RealtimeContractError::WeightLoweringParameterMismatch);
        }
        for lowering in &weight_lowerings {
            let expected_components = members
                .iter()
                .filter(|member| {
                    member.target() == lowering.target().as_str()
                        || member.linear_companion_of() == Some(lowering.target().as_str())
                })
                .map(|member| member.target())
                .collect::<BTreeSet<_>>();
            let actual_components = lowering
                .components()
                .iter()
                .map(|component| component.target().as_str())
                .collect::<BTreeSet<_>>();
            if expected_components != actual_components {
                return Err(RealtimeContractError::WeightComponentParameterMismatch {
                    target: lowering.target().clone(),
                });
            }
        }
        Ok(Self {
            identity,
            parameters,
            weight_lowerings,
        })
    }

    /// Returns the stable selected execution identity.
    pub const fn identity(&self) -> &RealtimeIdentity {
        &self.identity
    }

    /// Returns exact parameter ownership and geometry for this execution format.
    pub const fn parameters(&self) -> &ArchitectureParameterDescription {
        &self.parameters
    }

    /// Returns exact source-to-execution lowerings for every materialized tensor.
    pub fn weight_lowerings(&self) -> &[RealtimeWeightLoweringRequirement] {
        &self.weight_lowerings
    }
}

/// Exact generic mechanisms required by an architecture.
#[derive(Debug, Clone, Eq, PartialEq)]
pub struct RealtimeMechanismRequirements {
    mechanisms: BTreeSet<RealtimeMechanism>,
}

impl RealtimeMechanismRequirements {
    /// Creates a nonempty requirement set.
    pub fn new(
        mechanisms: impl IntoIterator<Item = RealtimeMechanism>,
    ) -> Result<Self, RealtimeContractError> {
        let mechanisms = mechanisms.into_iter().collect::<BTreeSet<_>>();
        if mechanisms.is_empty() {
            return Err(RealtimeContractError::EmptyMechanismRequirements);
        }
        Ok(Self { mechanisms })
    }

    /// Returns required mechanisms in stable order.
    pub const fn mechanisms(&self) -> &BTreeSet<RealtimeMechanism> {
        &self.mechanisms
    }

    /// Returns whether this contract requires a mechanism.
    pub fn requires(&self, mechanism: RealtimeMechanism) -> bool {
        self.mechanisms.contains(&mechanism)
    }
}

/// Architecture-admitted replicated and pure tensor-parallel topology policy.
#[derive(Debug, Clone, Eq, PartialEq)]
pub struct RealtimeTopologyPolicy {
    identity: RealtimeIdentity,
    replicated: bool,
    pure_tensor_parallel_sizes: BTreeSet<usize>,
}

impl RealtimeTopologyPolicy {
    /// Creates an exact fail-closed topology policy.
    pub fn new(
        identity: RealtimeIdentity,
        replicated: bool,
        pure_tensor_parallel_sizes: impl IntoIterator<Item = usize>,
    ) -> Result<Self, RealtimeContractError> {
        let pure_tensor_parallel_sizes = pure_tensor_parallel_sizes
            .into_iter()
            .collect::<BTreeSet<_>>();
        if let Some(size) = pure_tensor_parallel_sizes
            .iter()
            .copied()
            .find(|size| *size < 2)
        {
            return Err(RealtimeContractError::InvalidPureTensorParallelSize { size });
        }
        if !replicated && pure_tensor_parallel_sizes.is_empty() {
            return Err(RealtimeContractError::EmptyTopologyPolicy);
        }
        Ok(Self {
            identity,
            replicated,
            pure_tensor_parallel_sizes,
        })
    }

    /// Returns the stable architecture topology-policy identity.
    pub const fn identity(&self) -> &RealtimeIdentity {
        &self.identity
    }

    /// Returns whether replicated execution is admitted.
    pub const fn admits_replicated(&self) -> bool {
        self.replicated
    }

    /// Returns admitted pure tensor-parallel sizes.
    pub const fn pure_tensor_parallel_sizes(&self) -> &BTreeSet<usize> {
        &self.pure_tensor_parallel_sizes
    }

    /// Returns whether an exact topology is architecture-admitted.
    pub fn admits(&self, topology: ParallelTopology) -> bool {
        if topology.is_replicated() {
            return self.replicated;
        }
        topology.pipeline() == 1
            && topology.expert() == 1
            && topology.data() == 1
            && self.pure_tensor_parallel_sizes.contains(&topology.tensor())
    }
}

/// Exact architecture requirements used for realtime selection.
#[derive(Debug, Clone, Eq, PartialEq)]
pub struct RealtimeArchitectureRequirements {
    architecture: RealtimeIdentity,
    source: RealtimeIdentity,
    execution_graph: ExecutionGraph,
    execution_units: ExecutionUnitLayout,
    source_parameters: ArchitectureParameterDescription,
    executions: Vec<RealtimeExecutionRequirements>,
    operators: NeuralOperatorCapabilities,
    speech_schedule_identity: RealtimeIdentity,
    speech_schedule: RealtimeSpeechConfig,
    state_layout_identity: RealtimeIdentity,
    state_layout: StateLayout,
    mechanisms: RealtimeMechanismRequirements,
    topology: RealtimeTopologyPolicy,
    residencies: Vec<ExecutionResidency>,
}

impl RealtimeArchitectureRequirements {
    /// Creates one exact architecture-owned realtime requirement contract.
    #[allow(clippy::too_many_arguments)]
    pub fn new(
        architecture: RealtimeIdentity,
        source: RealtimeIdentity,
        execution_graph: ExecutionGraph,
        execution_units: ExecutionUnitLayout,
        source_parameters: ArchitectureParameterDescription,
        executions: impl IntoIterator<Item = RealtimeExecutionRequirements>,
        operators: NeuralOperatorCapabilities,
        speech_schedule_identity: RealtimeIdentity,
        speech_schedule: RealtimeSpeechConfig,
        state_layout_identity: RealtimeIdentity,
        state_layout: StateLayout,
        mechanisms: RealtimeMechanismRequirements,
        topology: RealtimeTopologyPolicy,
        residencies: impl IntoIterator<Item = ExecutionResidency>,
    ) -> Result<Self, RealtimeContractError> {
        if execution_graph.groups().len() != execution_units.group_count()
            || execution_graph
                .groups()
                .iter()
                .enumerate()
                .any(|(index, group)| {
                    execution_units
                        .group_id(index)
                        .is_none_or(|identity| identity.as_str() != group.id())
                })
        {
            return Err(RealtimeContractError::ExecutionGraphLayoutMismatch);
        }
        if source_parameters.graph() != &execution_graph
            || source_parameters.unit_layout() != &execution_units
        {
            return Err(RealtimeContractError::SourceParameterDescriptionMismatch);
        }
        let executions = executions.into_iter().collect::<Vec<_>>();
        if executions.is_empty() {
            return Err(RealtimeContractError::EmptyExecutionIdentities);
        }
        let execution_identities = executions
            .iter()
            .map(RealtimeExecutionRequirements::identity)
            .collect::<BTreeSet<_>>();
        if execution_identities.len() != executions.len() {
            return Err(RealtimeContractError::DuplicateExecutionIdentity);
        }
        if let Some(execution) = executions.iter().find(|execution| {
            execution.parameters.graph() != &execution_graph
                || execution.parameters.unit_layout() != &execution_units
        }) {
            return Err(
                RealtimeContractError::ExecutionParameterDescriptionMismatch {
                    execution: execution.identity.clone(),
                },
            );
        }
        let residencies = residencies.into_iter().collect::<Vec<_>>();
        if residencies.is_empty() {
            return Err(RealtimeContractError::EmptyResidencies);
        }
        if residencies
            .iter()
            .enumerate()
            .any(|(index, residency)| residencies[..index].contains(residency))
        {
            return Err(RealtimeContractError::DuplicateResidency);
        }
        Ok(Self {
            architecture,
            source,
            execution_graph,
            execution_units,
            source_parameters,
            executions,
            operators,
            speech_schedule_identity,
            speech_schedule,
            state_layout_identity,
            state_layout,
            mechanisms,
            topology,
            residencies,
        })
    }

    /// Returns the exact architecture identity.
    pub const fn architecture(&self) -> &RealtimeIdentity {
        &self.architecture
    }

    /// Returns the exact admitted source artifact identity.
    pub const fn source(&self) -> &RealtimeIdentity {
        &self.source
    }

    /// Returns the exact temporal/depth execution dependency graph.
    pub const fn execution_graph(&self) -> &ExecutionGraph {
        &self.execution_graph
    }

    /// Returns the exact temporal/depth execution-unit layout.
    pub const fn execution_units(&self) -> &ExecutionUnitLayout {
        &self.execution_units
    }

    /// Returns exact architecture-owned source parameter geometry and ownership.
    pub const fn source_parameters(&self) -> &ArchitectureParameterDescription {
        &self.source_parameters
    }

    /// Returns architecture-admitted execution identities.
    pub fn executions(&self) -> &[RealtimeExecutionRequirements] {
        &self.executions
    }

    /// Returns optional neural operators required by the architecture equations.
    pub const fn operators(&self) -> NeuralOperatorCapabilities {
        self.operators
    }

    /// Returns the stable speech schedule identity.
    pub const fn speech_schedule_identity(&self) -> &RealtimeIdentity {
        &self.speech_schedule_identity
    }

    /// Returns exact portable speech geometry and delays.
    pub const fn speech_schedule(&self) -> &RealtimeSpeechConfig {
        &self.speech_schedule
    }

    /// Returns the stable state-layout identity.
    pub const fn state_layout_identity(&self) -> &RealtimeIdentity {
        &self.state_layout_identity
    }

    /// Returns complete architecture-owned mutable-state geometry.
    pub const fn state_layout(&self) -> &StateLayout {
        &self.state_layout
    }

    /// Returns required generic mechanisms.
    pub const fn mechanisms(&self) -> &RealtimeMechanismRequirements {
        &self.mechanisms
    }

    /// Returns architecture-admitted topology policy.
    pub const fn topology(&self) -> &RealtimeTopologyPolicy {
        &self.topology
    }

    /// Returns architecture-admitted residency policies.
    pub fn residencies(&self) -> &[ExecutionResidency] {
        &self.residencies
    }
}

/// Family-blind backend capability report for realtime construction.
#[derive(Debug, Clone, Eq, PartialEq)]
pub struct RealtimeMechanismCapabilities {
    operators: NeuralOperatorCapabilities,
    mechanisms: BTreeSet<RealtimeMechanism>,
    residencies: Vec<ExecutionResidency>,
    weight_lowerings: Vec<WeightLoweringCapability>,
    observation_identities: BTreeSet<RealtimeIdentity>,
    state: StateMechanismCapabilities,
    maximum_tensor_parallel_size: NonZeroUsize,
    completion: crate::CommunicationCompletionCapabilities,
    session: SessionCapabilities,
}

impl RealtimeMechanismCapabilities {
    /// Creates one exact generic mechanism capability report.
    #[allow(clippy::too_many_arguments)]
    pub fn new(
        operators: NeuralOperatorCapabilities,
        mechanisms: impl IntoIterator<Item = RealtimeMechanism>,
        residencies: impl IntoIterator<Item = ExecutionResidency>,
        weight_lowerings: Vec<WeightLoweringCapability>,
        state: StateMechanismCapabilities,
        maximum_tensor_parallel_size: NonZeroUsize,
        completion: crate::CommunicationCompletionCapabilities,
        session: SessionCapabilities,
    ) -> Self {
        Self {
            operators,
            mechanisms: mechanisms.into_iter().collect(),
            residencies: residencies.into_iter().collect(),
            weight_lowerings,
            observation_identities: BTreeSet::new(),
            state,
            maximum_tensor_parallel_size,
            completion,
            session,
        }
    }

    /// Returns optional neural operators implemented by this backend path.
    pub const fn operators(&self) -> NeuralOperatorCapabilities {
        self.operators
    }

    /// Returns whether a generic mechanism is implemented.
    pub fn supports(&self, mechanism: RealtimeMechanism) -> bool {
        self.mechanisms.contains(&mechanism)
    }

    /// Returns supported residency policies.
    pub fn residencies(&self) -> &[ExecutionResidency] {
        &self.residencies
    }

    /// Returns exact geometry-bearing source-to-execution lowering mechanisms.
    pub fn weight_lowerings(&self) -> &[WeightLoweringCapability] {
        &self.weight_lowerings
    }

    /// Adds exact named activation observations implemented by this backend path.
    pub fn with_observation_identities(
        mut self,
        observations: impl IntoIterator<Item = RealtimeIdentity>,
    ) -> Self {
        self.observation_identities = observations.into_iter().collect();
        self
    }

    /// Returns exact named activation observations implemented by this backend path.
    pub const fn observation_identities(&self) -> &BTreeSet<RealtimeIdentity> {
        &self.observation_identities
    }

    /// Returns whether one exact named activation observation is implemented.
    pub fn supports_observation(&self, observation: &RealtimeIdentity) -> bool {
        self.observation_identities.contains(observation)
    }

    /// Returns exact mutable-state component and transaction mechanisms.
    pub const fn state(&self) -> &StateMechanismCapabilities {
        &self.state
    }

    /// Returns the maximum supported pure tensor-parallel size.
    pub const fn maximum_tensor_parallel_size(&self) -> NonZeroUsize {
        self.maximum_tensor_parallel_size
    }

    /// Returns supported exact-completion timeout dispositions.
    pub const fn completion(&self) -> &crate::CommunicationCompletionCapabilities {
        &self.completion
    }

    /// Returns generic prepared-session capabilities.
    pub const fn session(&self) -> SessionCapabilities {
        self.session
    }
}

/// Architecture-issued proof for the exact topology and rank request.
#[derive(Debug, Clone, Eq, PartialEq)]
pub struct RealtimeArchitectureProof {
    architecture: RealtimeIdentity,
    speech_schedule: RealtimeIdentity,
    state_layout: RealtimeIdentity,
    topology_policy: RealtimeIdentity,
    topology: ParallelTopology,
    rank: usize,
}

impl RealtimeArchitectureProof {
    /// Binds exact architecture contracts to one requested topology and rank.
    pub fn new(
        architecture: RealtimeIdentity,
        speech_schedule: RealtimeIdentity,
        state_layout: RealtimeIdentity,
        topology_policy: RealtimeIdentity,
        topology: ParallelTopology,
        rank: usize,
    ) -> Self {
        Self {
            architecture,
            speech_schedule,
            state_layout,
            topology_policy,
            topology,
            rank,
        }
    }

    /// Returns the proven topology.
    pub const fn topology(&self) -> ParallelTopology {
        self.topology
    }

    /// Returns the proven global rank.
    pub const fn rank(&self) -> usize {
        self.rank
    }
}

/// Host-visible observation facilities requested at construction time.
#[derive(Debug, Clone, Default, Eq, PartialEq)]
pub struct RealtimeObservationRequirements {
    output: bool,
    activations: BTreeSet<RealtimeIdentity>,
}

impl RealtimeObservationRequirements {
    /// Creates exact observation requirements.
    pub fn new(output: bool, activations: impl IntoIterator<Item = RealtimeIdentity>) -> Self {
        Self {
            output,
            activations: activations.into_iter().collect(),
        }
    }

    /// Returns whether completed host output is required.
    pub const fn output(&self) -> bool {
        self.output
    }

    /// Returns exact requested activation observations in stable identity order.
    pub const fn activations(&self) -> &BTreeSet<RealtimeIdentity> {
        &self.activations
    }

    /// Returns whether any named activation observation is required.
    pub fn requires_activations(&self) -> bool {
        !self.activations.is_empty()
    }
}

/// Exact source, execution, placement, and observation selection request.
#[derive(Debug, Clone, Eq, PartialEq)]
pub struct RealtimeSelectionRequest {
    max_cached_shards: usize,
    required_session_capabilities: SessionCapabilities,
    source: RealtimeIdentity,
    execution: RealtimeIdentity,
    residency: LayerWeightResidency,
    state: CacheResidencyPolicy,
    architecture_proof: Option<RealtimeArchitectureProof>,
    completion: crate::CommunicationCompletionPolicy,
    observations: RealtimeObservationRequirements,
}

impl RealtimeSelectionRequest {
    /// Requires exact portable session facilities independently of requested observations.
    pub const fn with_required_session_capabilities(
        mut self,
        required: SessionCapabilities,
    ) -> Self {
        self.required_session_capabilities = required;
        self
    }
    /// Sets the source reader-cache limit independently of executable residency.
    pub const fn with_max_cached_shards(mut self, maximum: std::num::NonZeroUsize) -> Self {
        self.max_cached_shards = maximum.get();
        self
    }
    /// Creates one complete fail-closed selection request.
    pub fn new(
        source: RealtimeIdentity,
        execution: RealtimeIdentity,
        residency: LayerWeightResidency,
        state: CacheResidencyPolicy,
        architecture_proof: Option<RealtimeArchitectureProof>,
        completion: crate::CommunicationCompletionPolicy,
        observations: RealtimeObservationRequirements,
    ) -> Self {
        Self {
            source,
            execution,
            max_cached_shards: residency.max_cached_shards(),
            required_session_capabilities: SessionCapabilities::default(),
            residency,
            state,
            architecture_proof,
            completion,
            observations,
        }
    }
}

/// One exact mutable-state component selected before native allocation.
#[derive(Debug, Clone, Eq, PartialEq)]
pub struct SelectedRealtimeStateComponentRealization {
    mechanism: StateComponentMechanism,
    placement: StateComponentPlacement,
}

impl SelectedRealtimeStateComponentRealization {
    /// Returns the architecture-global state layer.
    pub const fn layer(&self) -> usize {
        self.mechanism.layer()
    }

    /// Returns the exact architecture-declared semantic component.
    pub const fn component(&self) -> &eredu_core::cache::StateComponentPolicy {
        self.mechanism.component()
    }

    /// Returns the exact placement resolved during selection.
    pub const fn placement(&self) -> StateComponentPlacement {
        self.placement
    }
}

/// Authoritative mutable-state realization selected before native allocation.
#[derive(Debug, Clone, Eq, PartialEq)]
pub struct SelectedRealtimeStateRealization {
    layout: StateLayout,
    policy: CacheResidencyPolicy,
    components: Vec<SelectedRealtimeStateComponentRealization>,
    checkpoint: bool,
    rollback: bool,
    reset: bool,
    observation_retention: bool,
}

impl SelectedRealtimeStateRealization {
    /// Returns the exact architecture-owned mutable-state layout.
    pub const fn layout(&self) -> &StateLayout {
        &self.layout
    }

    /// Returns the exact selected state residency policy.
    pub const fn policy(&self) -> &CacheResidencyPolicy {
        &self.policy
    }

    /// Returns state components in architecture layer/component order.
    pub fn components(&self) -> &[SelectedRealtimeStateComponentRealization] {
        &self.components
    }

    /// Returns whether state checkpoints are guaranteed.
    pub const fn checkpoint(&self) -> bool {
        self.checkpoint
    }

    /// Returns whether checkpoint rollback is guaranteed.
    pub const fn rollback(&self) -> bool {
        self.rollback
    }

    /// Returns whether complete state reset is guaranteed.
    pub const fn reset(&self) -> bool {
        self.reset
    }

    /// Returns whether observed submissions retain every live state component.
    pub const fn observation_retention(&self) -> bool {
        self.observation_retention
    }
}

/// One immutable realtime realization selected before native construction.
#[derive(Debug, Clone, Eq, PartialEq)]
pub struct SelectedRealtimeRealization {
    max_cached_shards: usize,
    requirements: RealtimeArchitectureRequirements,
    source: RealtimeIdentity,
    execution: RealtimeExecutionRequirements,
    residency: LayerWeightResidency,
    state: SelectedRealtimeStateRealization,
    topology: ParallelTopology,
    rank: usize,
    completion: crate::CommunicationCompletionPolicy,
    observations: RealtimeObservationRequirements,
}

impl SelectedRealtimeRealization {
    /// Returns the exact source reader-cache bound retained during selection.
    pub const fn max_cached_shards(&self) -> usize {
        self.max_cached_shards
    }
    /// Returns the complete architecture requirements selected once.
    pub const fn requirements(&self) -> &RealtimeArchitectureRequirements {
        &self.requirements
    }

    /// Returns the exact source identity.
    pub const fn source(&self) -> &RealtimeIdentity {
        &self.source
    }

    /// Returns the exact selected temporal/depth execution graph.
    pub const fn execution_graph(&self) -> &ExecutionGraph {
        self.requirements.execution_graph()
    }

    /// Returns the exact selected temporal/depth execution-unit layout.
    pub const fn execution_units(&self) -> &ExecutionUnitLayout {
        self.requirements.execution_units()
    }

    /// Returns exact source parameter ownership and geometry.
    pub const fn source_parameters(&self) -> &ArchitectureParameterDescription {
        self.requirements.source_parameters()
    }

    /// Returns optional neural operators required by the selected architecture.
    pub const fn operators(&self) -> NeuralOperatorCapabilities {
        self.requirements.operators()
    }

    /// Returns the selected execution identity.
    pub const fn execution(&self) -> &RealtimeIdentity {
        self.execution.identity()
    }

    /// Returns the exact execution contract selected for materialization.
    pub const fn execution_requirements(&self) -> &RealtimeExecutionRequirements {
        &self.execution
    }

    /// Returns exact selected execution parameter ownership and geometry.
    pub const fn execution_parameters(&self) -> &ArchitectureParameterDescription {
        self.execution.parameters()
    }

    /// Returns the exact named lowering requirements retained by this selection.
    pub fn weight_lowerings(&self) -> &[RealtimeWeightLoweringRequirement] {
        self.execution.weight_lowerings()
    }

    /// Returns selected residency.
    pub const fn residency(&self) -> LayerWeightResidency {
        self.residency
    }

    /// Returns the exact selected mutable-state realization.
    pub const fn state(&self) -> &SelectedRealtimeStateRealization {
        &self.state
    }

    /// Returns the architecture-proven topology.
    pub const fn topology(&self) -> ParallelTopology {
        self.topology
    }

    /// Returns the architecture-proven global rank.
    pub const fn rank(&self) -> usize {
        self.rank
    }

    /// Returns the one selected bounded-wait and timeout disposition policy.
    pub const fn completion(&self) -> crate::CommunicationCompletionPolicy {
        self.completion
    }

    /// Returns selected observation requirements.
    pub const fn observations(&self) -> &RealtimeObservationRequirements {
        &self.observations
    }
}

/// One stable fail-closed realtime selection issue.
#[derive(Debug, Clone, Eq, PartialEq, thiserror::Error)]
#[non_exhaustive]
pub enum RealtimeSelectionIssue {
    /// No architecture proof was supplied.
    #[error("architecture topology and rank proof is missing")]
    MissingArchitectureProof,
    /// Proof names another architecture.
    #[error("architecture proof identity mismatch")]
    ArchitectureIdentityMismatch,
    /// Proof names another schedule.
    #[error("speech schedule proof identity mismatch")]
    SpeechScheduleIdentityMismatch,
    /// Proof names another state layout.
    #[error("state layout proof identity mismatch")]
    StateLayoutIdentityMismatch,
    /// Proof names another topology policy.
    #[error("topology policy proof identity mismatch")]
    TopologyPolicyIdentityMismatch,
    /// Request names another source artifact.
    #[error("source artifact identity mismatch")]
    SourceIdentityMismatch,
    /// Requested execution configuration was not admitted.
    #[error("execution identity is not architecture-admitted")]
    ExecutionIdentityNotAdmitted,
    /// A required optional neural operator is unavailable.
    #[error("required neural operator {operator} is unavailable")]
    MissingNeuralOperator {
        /// Stable generic operator name.
        operator: String,
    },
    /// A required named geometry-bearing weight lowering is unavailable.
    #[error("execution weight lowering {index} for target {target} is unavailable")]
    WeightLoweringUnavailable {
        /// Stable ordinal in the architecture execution contract.
        index: usize,
        /// Exact canonical executable target.
        target: RealtimeIdentity,
    },
    /// Requested residency was not admitted by the architecture.
    #[error("residency is not architecture-admitted")]
    ResidencyNotAdmitted,
    /// Backend cannot implement requested residency.
    #[error("residency mechanism is unavailable")]
    ResidencyUnavailable,
    /// Architecture did not admit the proven topology.
    #[error("topology is not architecture-admitted as replicated or pure tensor parallel")]
    TopologyNotAdmitted,
    /// Proven rank lies outside the topology.
    #[error("rank lies outside the proven topology")]
    RankOutOfRange,
    /// Backend cannot realize the requested tensor-parallel width.
    #[error("tensor-parallel width exceeds backend capability")]
    TensorParallelWidthUnavailable,
    /// A required generic mechanism is unavailable.
    #[error("required realtime mechanism {0:?} is unavailable")]
    MissingMechanism(RealtimeMechanism),
    /// Backend cannot safely apply the requested completion timeout disposition.
    #[error("requested completion timeout disposition is unavailable")]
    CompletionPolicyUnavailable,
    /// No unique compatible mechanism exists for one architecture state component.
    #[error("state component {component} at layer {layer} is unavailable")]
    StateComponentUnavailable {
        /// Architecture-global state layer.
        layer: usize,
        /// Stable component role.
        component: String,
    },
    /// State checkpointing is required for unpublished realtime branches.
    #[error("state checkpoint mechanism is unavailable")]
    MissingStateCheckpoint,
    /// State rollback is required for discarded realtime branches.
    #[error("state rollback mechanism is unavailable")]
    MissingStateRollback,
    /// State reset is required at architecture-declared frame seams.
    #[error("state reset mechanism is unavailable")]
    MissingStateReset,
    /// Observation must retain every referenced state component.
    #[error("state observation retention is unavailable")]
    MissingStateObservationRetention,
    /// Persistent model-state storage is unavailable.
    #[error("persistent session state is unavailable")]
    MissingPersistentState,
    /// Requested completed-output observation is unavailable.
    #[error("completed output observation is unavailable")]
    MissingOutputObservation,
    /// Requested activation inspection is unavailable.
    #[error("activation inspection is unavailable")]
    MissingActivationInspection,
    /// One exact requested activation observation is unavailable.
    #[error("requested activation observation {observation} is unavailable")]
    ObservationUnavailable {
        /// Exact architecture-owned observation identity.
        observation: RealtimeIdentity,
    },
}

/// Fail-closed diagnostic containing issues in stable validation order.
#[derive(Debug, Clone, Eq, PartialEq, thiserror::Error)]
#[error("realtime realization is unsupported: {message}", message = selection_message(.issues))]
pub struct RealtimeSelectionError {
    issues: Vec<RealtimeSelectionIssue>,
}

impl RealtimeSelectionError {
    /// Returns every mismatch in stable validation order.
    pub fn issues(&self) -> &[RealtimeSelectionIssue] {
        &self.issues
    }
}

fn selection_message(issues: &[RealtimeSelectionIssue]) -> String {
    issues
        .iter()
        .map(ToString::to_string)
        .collect::<Vec<_>>()
        .join("; ")
}

/// Selects the complete realtime realization without invoking native construction.
pub fn select_realtime_realization(
    requirements: &RealtimeArchitectureRequirements,
    request: &RealtimeSelectionRequest,
    capabilities: &RealtimeMechanismCapabilities,
) -> Result<SelectedRealtimeRealization, RealtimeSelectionError> {
    let mut issues = Vec::new();
    let proof = request.architecture_proof.as_ref();
    match proof {
        Some(proof) => {
            if proof.architecture != requirements.architecture {
                issues.push(RealtimeSelectionIssue::ArchitectureIdentityMismatch);
            }
            if proof.speech_schedule != requirements.speech_schedule_identity {
                issues.push(RealtimeSelectionIssue::SpeechScheduleIdentityMismatch);
            }
            if proof.state_layout != requirements.state_layout_identity {
                issues.push(RealtimeSelectionIssue::StateLayoutIdentityMismatch);
            }
            if proof.topology_policy != *requirements.topology.identity() {
                issues.push(RealtimeSelectionIssue::TopologyPolicyIdentityMismatch);
            }
        }
        None => issues.push(RealtimeSelectionIssue::MissingArchitectureProof),
    }
    if request.source != requirements.source {
        issues.push(RealtimeSelectionIssue::SourceIdentityMismatch);
    }
    let execution = requirements
        .executions
        .iter()
        .find(|execution| execution.identity == request.execution);
    if execution.is_none() {
        issues.push(RealtimeSelectionIssue::ExecutionIdentityNotAdmitted);
    }
    for operator in capabilities
        .operators
        .missing_capability_names(requirements.operators)
    {
        issues.push(RealtimeSelectionIssue::MissingNeuralOperator {
            operator: operator.to_owned(),
        });
    }
    if let Some(execution) = execution {
        for (index, required) in execution.weight_lowerings().iter().enumerate() {
            if !capabilities.weight_lowerings.iter().any(|available| {
                available.descriptor() == required.descriptor()
                    && available.kind() == required.kind()
            }) {
                issues.push(RealtimeSelectionIssue::WeightLoweringUnavailable {
                    index,
                    target: required.target().clone(),
                });
            }
        }
    }
    let execution_residency = request.residency.execution_residency();
    if !requirements.residencies.contains(&execution_residency) {
        issues.push(RealtimeSelectionIssue::ResidencyNotAdmitted);
    }
    if !capabilities.residencies.contains(&execution_residency) {
        issues.push(RealtimeSelectionIssue::ResidencyUnavailable);
    }
    if let Some(proof) = proof {
        if !requirements.topology.admits(proof.topology) {
            issues.push(RealtimeSelectionIssue::TopologyNotAdmitted);
        }
        if proof.rank >= proof.topology.world_size() {
            issues.push(RealtimeSelectionIssue::RankOutOfRange);
        }
        if proof.topology.tensor() > capabilities.maximum_tensor_parallel_size.get() {
            issues.push(RealtimeSelectionIssue::TensorParallelWidthUnavailable);
        }
    }
    for mechanism in requirements.mechanisms.mechanisms() {
        if !capabilities.supports(*mechanism) {
            issues.push(RealtimeSelectionIssue::MissingMechanism(*mechanism));
        }
    }
    if !capabilities.completion.supports(request.completion) {
        issues.push(RealtimeSelectionIssue::CompletionPolicyUnavailable);
    }
    let mut state_components = Vec::new();
    for layer in 0..requirements.state_layout.len() {
        for component in requirements
            .state_layout
            .components(layer)
            .expect("state layout exposes every validated layer")
        {
            let matches = capabilities
                .state
                .components()
                .iter()
                .filter(|mechanism| {
                    mechanism.layer() == layer && mechanism.component() == component
                })
                .cloned()
                .collect::<Vec<_>>();
            match matches.as_slice() {
                [mechanism] => match mechanism.placement(&request.state) {
                    Some(placement)
                        if crate::replicated_text::placement_is_compatible(
                            component,
                            &request.state,
                            placement,
                        ) =>
                    {
                        state_components.push(SelectedRealtimeStateComponentRealization {
                            mechanism: mechanism.clone(),
                            placement,
                        });
                    }
                    _ => issues.push(RealtimeSelectionIssue::StateComponentUnavailable {
                        layer,
                        component: component.role().stable_name().to_owned(),
                    }),
                },
                _ => issues.push(RealtimeSelectionIssue::StateComponentUnavailable {
                    layer,
                    component: component.role().stable_name().to_owned(),
                }),
            }
        }
    }
    if !capabilities.state.checkpoint() {
        issues.push(RealtimeSelectionIssue::MissingStateCheckpoint);
    }
    if !capabilities.state.rollback() {
        issues.push(RealtimeSelectionIssue::MissingStateRollback);
    }
    if !capabilities.state.reset() {
        issues.push(RealtimeSelectionIssue::MissingStateReset);
    }
    if (request.observations.output()
        || request.observations.requires_activations()
        || request.required_session_capabilities.output_observation()
        || request
            .required_session_capabilities
            .activation_inspection())
        && !capabilities.state.observation_retention()
    {
        issues.push(RealtimeSelectionIssue::MissingStateObservationRetention);
    }
    if let Some(proof) = proof {
        if !proof.topology.is_replicated()
            && !capabilities.supports(RealtimeMechanism::Collectives)
            && !requirements
                .mechanisms
                .requires(RealtimeMechanism::Collectives)
        {
            issues.push(RealtimeSelectionIssue::MissingMechanism(
                RealtimeMechanism::Collectives,
            ));
        }
    }
    if !capabilities.session.persistent_cache() {
        issues.push(RealtimeSelectionIssue::MissingPersistentState);
    }
    if (request.observations.output() || request.required_session_capabilities.output_observation())
        && !capabilities.session.output_observation()
    {
        issues.push(RealtimeSelectionIssue::MissingOutputObservation);
    }
    if (request.observations.requires_activations()
        || request
            .required_session_capabilities
            .activation_inspection())
        && !capabilities.session.activation_inspection()
    {
        issues.push(RealtimeSelectionIssue::MissingActivationInspection);
    }
    for observation in request.observations.activations() {
        if !capabilities.supports_observation(observation) {
            issues.push(RealtimeSelectionIssue::ObservationUnavailable {
                observation: observation.clone(),
            });
        }
    }
    if !issues.is_empty() {
        return Err(RealtimeSelectionError { issues });
    }
    let proof = proof.expect("successful realtime selection has architecture proof");
    let state = SelectedRealtimeStateRealization {
        layout: requirements.state_layout.clone(),
        policy: request.state.clone(),
        components: state_components,
        checkpoint: capabilities.state.checkpoint(),
        rollback: capabilities.state.rollback(),
        reset: capabilities.state.reset(),
        observation_retention: capabilities.state.observation_retention(),
    };
    Ok(SelectedRealtimeRealization {
        max_cached_shards: request.max_cached_shards,
        requirements: requirements.clone(),
        source: request.source.clone(),
        execution: execution
            .expect("successful realtime selection has admitted execution")
            .clone(),
        residency: request.residency,
        state,
        topology: proof.topology,
        rank: proof.rank,
        completion: request.completion,
        observations: request.observations.clone(),
    })
}

/// Resources constructed only after realtime selection succeeds.
#[derive(Debug)]
pub struct ConstructedRealtimeResources<P, M, S, Q, G> {
    payload: P,
    modules: M,
    state: S,
    queue: Q,
    group: Option<G>,
}

impl<P, M, S, Q, G> ConstructedRealtimeResources<P, M, S, Q, G> {
    /// Consumes resources in construction order.
    pub fn into_parts(self) -> (P, M, S, Q, Option<G>) {
        (
            self.payload,
            self.modules,
            self.state,
            self.queue,
            self.group,
        )
    }
}

/// A selected realtime realization paired with native mechanism resources.
#[derive(Debug)]
pub struct PreparedRealtimeRealization<P, M, S, Q, G> {
    selected: SelectedRealtimeRealization,
    resources: ConstructedRealtimeResources<P, M, S, Q, G>,
}

impl<P, M, S, Q, G> PreparedRealtimeRealization<P, M, S, Q, G> {
    /// Returns the authoritative neutral selection.
    pub const fn selected(&self) -> &SelectedRealtimeRealization {
        &self.selected
    }

    /// Consumes the prepared realization into selection and resources.
    pub fn into_parts(
        self,
    ) -> (
        SelectedRealtimeRealization,
        ConstructedRealtimeResources<P, M, S, Q, G>,
    ) {
        (self.selected, self.resources)
    }
}

/// Failure during selection or one post-selection construction stage.
#[derive(Debug, thiserror::Error)]
pub enum RealtimePreparationError<E> {
    /// Neutral selection rejected before every construction callback.
    #[error(transparent)]
    Selection(#[from] RealtimeSelectionError),
    /// Checkpoint payload construction failed.
    #[error("realtime payload construction failed")]
    Payload(#[source] E),
    /// Architecture module construction failed.
    #[error("realtime module construction failed")]
    Modules(#[source] E),
    /// Mutable model-state construction failed.
    #[error("realtime state construction failed")]
    State(#[source] E),
    /// Execution queue construction failed.
    #[error("realtime queue construction failed")]
    Queue(#[source] E),
    /// Communication group construction failed.
    #[error("realtime communication group construction failed")]
    Group(#[source] E),
}

/// Selects once, then constructs family-blind resources in dependency order.
#[allow(clippy::too_many_arguments)]
pub fn select_and_prepare_realtime_realization<P, M, S, Q, G, E>(
    requirements: &RealtimeArchitectureRequirements,
    request: &RealtimeSelectionRequest,
    capabilities: &RealtimeMechanismCapabilities,
    payload: impl FnOnce(&SelectedRealtimeRealization) -> Result<P, E>,
    modules: impl FnOnce(&SelectedRealtimeRealization, &P) -> Result<M, E>,
    state: impl FnOnce(&SelectedRealtimeRealization, &M) -> Result<S, E>,
    queue: impl FnOnce(&SelectedRealtimeRealization) -> Result<Q, E>,
    group: impl FnOnce(&SelectedRealtimeRealization) -> Result<G, E>,
) -> Result<PreparedRealtimeRealization<P, M, S, Q, G>, RealtimePreparationError<E>> {
    let selected = select_realtime_realization(requirements, request, capabilities)?;
    let payload = payload(&selected).map_err(RealtimePreparationError::Payload)?;
    let modules = modules(&selected, &payload).map_err(RealtimePreparationError::Modules)?;
    let state = state(&selected, &modules).map_err(RealtimePreparationError::State)?;
    let queue = queue(&selected).map_err(RealtimePreparationError::Queue)?;
    let group = if selected.topology().is_replicated() {
        None
    } else {
        Some(group(&selected).map_err(RealtimePreparationError::Group)?)
    };
    Ok(PreparedRealtimeRealization {
        selected,
        resources: ConstructedRealtimeResources {
            payload,
            modules,
            state,
            queue,
            group,
        },
    })
}

/// Invalid neutral realtime contract.
#[derive(Debug, Clone, Eq, PartialEq, thiserror::Error)]
#[non_exhaustive]
pub enum RealtimeContractError {
    /// Stable identities must contain non-whitespace text.
    #[error("realtime identity must not be empty")]
    EmptyIdentity,
    /// A realtime architecture must require generic mechanisms.
    #[error("realtime mechanism requirements must not be empty")]
    EmptyMechanismRequirements,
    /// At least one execution identity must be admitted.
    #[error("realtime execution identities must not be empty")]
    EmptyExecutionIdentities,
    /// The retained execution graph and execution-unit layout must name the same groups.
    #[error("realtime execution graph and unit layout differ")]
    ExecutionGraphLayoutMismatch,
    /// Source parameters must be described against the exact retained graph and layout.
    #[error("realtime source parameter description differs from the retained graph or layout")]
    SourceParameterDescriptionMismatch,
    /// Execution parameters must be described against the exact retained graph and layout.
    #[error(
        "realtime execution parameter description for {execution} differs from the retained graph or layout"
    )]
    ExecutionParameterDescriptionMismatch {
        /// Admitted execution whose description is incoherent.
        execution: RealtimeIdentity,
    },
    /// Execution identities must be unique within one architecture contract.
    #[error("realtime execution identities must be unique")]
    DuplicateExecutionIdentity,
    /// Every admitted execution must describe its exact checkpoint lowerings.
    #[error("realtime execution weight lowerings must not be empty")]
    EmptyWeightLowerings,
    /// Physical component geometry must have positive nonempty extents.
    #[error("realtime weight component shape must contain only positive extents")]
    InvalidWeightComponentShape,
    /// Recipe owner and recipe identity must either both be present or both be absent.
    #[error("realtime weight component recipe owner and identity must appear together")]
    IncompleteWeightComponentRecipe,
    /// A recipe-backed physical component must retain at least one source occurrence.
    #[error("realtime recipe-backed weight component sources must not be empty")]
    EmptyRecipeComponentSources,
    /// A retained recipe or its exact output metadata differs from the component declaration.
    #[error("realtime weight component recipe does not match its sources and physical geometry")]
    WeightComponentRecipeMismatch,
    /// A transform-generated component cannot claim physical recipe sources.
    #[error("realtime generated weight component must not contain recipe sources")]
    GeneratedComponentHasSources,
    /// Every matrix lowering must contain its primary physical component.
    #[error("realtime weight lowering has no primary component")]
    MissingPrimaryWeightComponent,
    /// Every matrix lowering must contain exactly one primary physical component.
    #[error("realtime weight lowering has more than one primary component")]
    DuplicatePrimaryWeightComponent,
    /// The primary component target must equal the matrix lowering target.
    #[error("realtime primary weight component does not match its lowering target")]
    PrimaryWeightComponentTargetMismatch,
    /// Atomic component targets must be unique within one matrix lowering.
    #[error("realtime weight component targets must be unique within a lowering")]
    DuplicateWeightComponentTarget,
    /// Scale and affine-bias roles may each occur at most once.
    #[error("realtime weight lowering repeats component role {role:?}")]
    DuplicateWeightComponentRole {
        /// Repeated companion role.
        role: RealtimeWeightComponentRole,
    },
    /// One execution cannot bind the same canonical target more than once.
    #[error("realtime execution weight lowering targets must be unique")]
    DuplicateWeightLoweringTarget,
    /// Lowering targets must exactly equal primary execution parameters.
    #[error("realtime weight lowerings differ from execution parameters")]
    WeightLoweringParameterMismatch,
    /// One lowering's components must exactly equal its execution parameter family.
    #[error("realtime weight components differ from execution parameters for {target}")]
    WeightComponentParameterMismatch {
        /// Affected primary lowering target.
        target: RealtimeIdentity,
    },
    /// At least one residency must be admitted.
    #[error("realtime residencies must not be empty")]
    EmptyResidencies,
    /// Residency declarations must not be duplicated.
    #[error("realtime residencies must be unique")]
    DuplicateResidency,
    /// A topology policy must admit replicated or pure tensor-parallel execution.
    #[error("realtime topology policy must not be empty")]
    EmptyTopologyPolicy,
    /// Pure tensor-parallel sizes must contain more than one rank.
    #[error("pure tensor-parallel size {size} must be at least two")]
    InvalidPureTensorParallelSize {
        /// Invalid requested size.
        size: usize,
    },
}

#[cfg(test)]
mod tests {
    use std::{cell::Cell, convert::Infallible, num::NonZeroUsize, rc::Rc, time::Duration};

    use eredu_checkpoint::{LinearFormat, SourceTensorEncoding, StoredDtype};
    use eredu_core::{
        cache::LayerCachePolicy, AttentionPolicy, CompletionCancellationMode, LayerSchedule,
        RealtimeFrameConvention,
    };

    use super::*;
    fn identity(value: &str) -> RealtimeIdentity {
        RealtimeIdentity::new(value).unwrap()
    }

    fn execution_graph() -> ExecutionGraph {
        ExecutionGraph::chain(["temporal", "depth"]).unwrap()
    }

    fn execution_units(graph: &ExecutionGraph) -> ExecutionUnitLayout {
        ExecutionUnitLayout::new(graph, [2, 2]).unwrap()
    }

    fn parameter_description(
        graph: &ExecutionGraph,
        units: &ExecutionUnitLayout,
    ) -> ArchitectureParameterDescription {
        let group = crate::ParameterGroupSpec::new(
            "target",
            crate::ParameterRole::Replicated,
            [crate::ParameterMemberSpec::new(
                "target.weight",
                [8, 8],
                crate::MemberSharding::Replicated,
            )],
        )
        .unwrap();
        ArchitectureParameterDescription::new(
            graph,
            units,
            [group.clone()],
            [crate::OwnedParameterGroupSpec::new(
                crate::ParameterGroupOwner::static_role("target"),
                group,
            )],
        )
        .unwrap()
    }

    fn required_operators() -> NeuralOperatorCapabilities {
        NeuralOperatorCapabilities::EXP.union(NeuralOperatorCapabilities::SOFTMAX_AXIS)
    }

    fn state_layout() -> StateLayout {
        StateLayout::new(
            LayerSchedule::new(
                2,
                vec![
                    LayerCachePolicy::key_value(AttentionPolicy::Full, 2, 8).unwrap(),
                    LayerCachePolicy::key_value(AttentionPolicy::Full, 2, 8).unwrap(),
                ],
            )
            .unwrap(),
        )
        .unwrap()
    }

    fn schedule() -> RealtimeSpeechConfig {
        RealtimeSpeechConfig::new(
            4,
            2,
            2,
            2,
            0,
            1,
            RealtimeFrameConvention::FeedbackAlignedHistory,
            vec![0, 1, 2, 1, 2],
        )
        .unwrap()
    }

    fn required_mechanisms() -> Vec<RealtimeMechanism> {
        vec![
            RealtimeMechanism::TensorOperations,
            RealtimeMechanism::NeuralOperations,
            RealtimeMechanism::ParameterMaterialization,
            RealtimeMechanism::ParameterStorage,
            RealtimeMechanism::StateStorage,
            RealtimeMechanism::CoordinateStorage,
            RealtimeMechanism::Sampling,
            RealtimeMechanism::Randomness,
            RealtimeMechanism::HostConversion,
            RealtimeMechanism::ExactCompletion,
            RealtimeMechanism::ResourceRetention,
            RealtimeMechanism::Transfer,
            RealtimeMechanism::Collectives,
        ]
    }

    fn completion_policy() -> crate::CommunicationCompletionPolicy {
        crate::CommunicationCompletionPolicy::new(
            Duration::from_secs(1),
            CompletionCancellationMode::QuarantineUntilComplete,
        )
        .unwrap()
    }

    fn completion_capabilities() -> crate::CommunicationCompletionCapabilities {
        crate::CommunicationCompletionCapabilities::new([
            CompletionCancellationMode::QuarantineUntilComplete,
        ])
        .unwrap()
    }

    fn lowering(executable: LinearFormat) -> WeightLoweringDescriptor {
        WeightLoweringDescriptor::new(
            SourceTensorEncoding::Safetensors(StoredDtype::F32),
            executable,
            vec![8, 8],
            vec![8, 8],
            Some(1),
        )
        .unwrap()
    }

    fn lowering_requirement(
        target: &str,
        executable: LinearFormat,
        kind: WeightLoweringKind,
    ) -> RealtimeWeightLoweringRequirement {
        let primary = RealtimeWeightComponentRequirement::new(
            identity(target),
            Some(identity("canonical-owner")),
            Some(identity("recipe-v1")),
            Some(eredu_checkpoint::recipe::DerivedWeightRecipe::Concatenate {
                axis: 0,
                inputs: ["source-a", "source-b"]
                    .into_iter()
                    .map(|key| {
                        eredu_checkpoint::recipe::DerivedWeightRecipe::source(
                            key,
                            eredu_checkpoint::store::TensorSelection::Full,
                        )
                    })
                    .collect(),
            }),
            Some(eredu_checkpoint::recipe::RecipeMetadata {
                shape: vec![8, 8],
                dtype: eredu_checkpoint::recipe::RecipeDtype::F32,
                byte_len: 256,
            }),
            [identity("source-a"), identity("source-b")],
            [8, 8],
            RealtimeWeightComponentRole::Primary,
        )
        .unwrap();
        RealtimeWeightLoweringRequirement::new(
            identity(target),
            [primary],
            lowering(executable),
            kind,
        )
        .unwrap()
    }

    fn execution(
        value: &str,
        parameters: ArchitectureParameterDescription,
    ) -> RealtimeExecutionRequirements {
        RealtimeExecutionRequirements::new(
            identity(value),
            parameters,
            vec![lowering_requirement(
                "target.weight",
                LinearFormat::Dense,
                WeightLoweringKind::Direct,
            )],
        )
        .unwrap()
    }

    fn requirements() -> RealtimeArchitectureRequirements {
        let graph = execution_graph();
        let units = execution_units(&graph);
        let parameters = parameter_description(&graph, &units);
        requirements_with_model_contract(
            graph,
            units,
            parameters.clone(),
            vec![
                execution("execution-native", parameters.clone()),
                execution("execution-transformed", parameters),
            ],
        )
        .unwrap()
    }

    fn requirements_with_model_contract(
        graph: ExecutionGraph,
        units: ExecutionUnitLayout,
        source_parameters: ArchitectureParameterDescription,
        executions: Vec<RealtimeExecutionRequirements>,
    ) -> Result<RealtimeArchitectureRequirements, RealtimeContractError> {
        RealtimeArchitectureRequirements::new(
            identity("architecture-7"),
            identity("artifact-4"),
            graph,
            units,
            source_parameters,
            executions,
            required_operators(),
            identity("schedule-3"),
            schedule(),
            identity("state-layout-8"),
            state_layout(),
            RealtimeMechanismRequirements::new(required_mechanisms()).unwrap(),
            RealtimeTopologyPolicy::new(identity("topology-plan-2"), true, [2]).unwrap(),
            [
                ExecutionResidency::FullyResident,
                ExecutionResidency::LayerwiseHost,
            ],
        )
    }

    fn proof(requirements: &RealtimeArchitectureRequirements) -> RealtimeArchitectureProof {
        RealtimeArchitectureProof::new(
            requirements.architecture().clone(),
            requirements.speech_schedule_identity().clone(),
            requirements.state_layout_identity().clone(),
            requirements.topology().identity().clone(),
            ParallelTopology::new(2, 1, 1, 1).unwrap(),
            1,
        )
    }

    fn request(requirements: &RealtimeArchitectureRequirements) -> RealtimeSelectionRequest {
        RealtimeSelectionRequest::new(
            requirements.source().clone(),
            identity("execution-transformed"),
            LayerWeightResidency::LayerwiseHost(crate::LayerwiseLoadOptions::default()),
            CacheResidencyPolicy::Device,
            Some(proof(requirements)),
            completion_policy(),
            RealtimeObservationRequirements::new(
                true,
                [identity("temporal.layer.0"), identity("depth.slice.0")],
            ),
        )
    }

    fn state_capabilities() -> StateMechanismCapabilities {
        StateMechanismCapabilities::new((0..state_layout().len()).flat_map(|layer| {
            state_layout()
                .components(layer)
                .unwrap()
                .iter()
                .cloned()
                .map(move |component| {
                    crate::StateComponentMechanism::new(
                        layer,
                        component,
                        Some(crate::StateComponentPlacement::Device),
                        None,
                    )
                })
                .collect::<Vec<_>>()
        }))
        .with_transactions(true, true)
        .with_reset(true)
        .with_observation_retention(true)
    }

    fn capabilities() -> RealtimeMechanismCapabilities {
        RealtimeMechanismCapabilities::new(
            required_operators(),
            required_mechanisms(),
            [
                ExecutionResidency::FullyResident,
                ExecutionResidency::LayerwiseHost,
            ],
            vec![WeightLoweringCapability::new(
                lowering(LinearFormat::Dense),
                WeightLoweringKind::Direct,
            )],
            state_capabilities(),
            NonZeroUsize::new(2).unwrap(),
            completion_capabilities(),
            SessionCapabilities::new(true, true, true),
        )
        .with_observation_identities([identity("temporal.layer.0"), identity("depth.slice.0")])
    }

    struct SyntheticRealtimeSupport {
        supported: bool,
        lowering_queries: std::cell::Cell<usize>,
        state_queries: std::cell::Cell<usize>,
    }

    impl crate::RealtimeMechanismSupport for SyntheticRealtimeSupport {
        fn facts(&self) -> crate::RealtimeMechanismFacts {
            let state = if self.supported {
                crate::StateLifecycleCapabilities::new()
                    .with_transactions(true, true)
                    .with_reset(true)
                    .with_observation_retention(true)
            } else {
                crate::StateLifecycleCapabilities::new()
            };
            crate::RealtimeMechanismFacts::new(
                required_operators(),
                required_mechanisms(),
                [
                    ExecutionResidency::FullyResident,
                    ExecutionResidency::LayerwiseHost,
                ],
                NonZeroUsize::new(2).unwrap(),
                completion_capabilities(),
                SessionCapabilities::new(true, true, true),
            )
            .with_state_lifecycle(state)
            .with_observation_identities([identity("temporal.layer.0"), identity("depth.slice.0")])
        }

        fn supports_lowering(
            &self,
            _descriptor: &WeightLoweringDescriptor,
            _kind: WeightLoweringKind,
        ) -> bool {
            self.lowering_queries.set(self.lowering_queries.get() + 1);
            self.supported
        }

        fn state_component_placements(
            &self,
            _component: &eredu_core::cache::StateComponentPolicy,
        ) -> (
            Option<StateComponentPlacement>,
            Option<StateComponentPlacement>,
        ) {
            self.state_queries.set(self.state_queries.get() + 1);
            (
                self.supported.then_some(StateComponentPlacement::Device),
                None,
            )
        }
    }

    #[test]
    fn realtime_synthesis_matches_independent_oracle_and_deduplicates_all_executions() {
        let requirements = requirements();
        let support = SyntheticRealtimeSupport {
            supported: true,
            lowering_queries: std::cell::Cell::new(0),
            state_queries: std::cell::Cell::new(0),
        };
        let synthesized = crate::synthesize_realtime_capabilities(&requirements, &support);
        assert_eq!(synthesized, capabilities());
        assert_eq!(support.lowering_queries.get(), 2);
        assert_eq!(synthesized.weight_lowerings().len(), 1);
        assert_eq!(
            support.state_queries.get(),
            synthesized.state().components().len()
        );
        assert_eq!(
            select_realtime_realization(&requirements, &request(&requirements), &synthesized),
            select_realtime_realization(&requirements, &request(&requirements), &capabilities()),
        );
    }

    #[test]
    fn realtime_synthesis_preserves_kind_and_fails_closed_for_missing_mechanisms() {
        let mut requirements = requirements();
        requirements.executions[1].weight_lowerings = vec![lowering_requirement(
            "target.weight",
            LinearFormat::Dense,
            WeightLoweringKind::Derived,
        )];
        let mut support = SyntheticRealtimeSupport {
            supported: true,
            lowering_queries: std::cell::Cell::new(0),
            state_queries: std::cell::Cell::new(0),
        };
        let synthesized = crate::synthesize_realtime_capabilities(&requirements, &support);
        assert_eq!(synthesized.weight_lowerings().len(), 2);
        assert_eq!(
            synthesized.weight_lowerings()[0].kind(),
            WeightLoweringKind::Direct
        );
        assert_eq!(
            synthesized.weight_lowerings()[1].kind(),
            WeightLoweringKind::Derived
        );
        support.supported = false;
        let absent = crate::synthesize_realtime_capabilities(&requirements, &support);
        assert!(absent.weight_lowerings().is_empty());
        assert!(absent.state().components().is_empty());
        assert!(!absent.state().checkpoint());
        assert!(!absent.state().rollback());
        assert!(!absent.state().reset());
        assert!(!absent.state().observation_retention());
        assert!(
            select_realtime_realization(&requirements, &request(&requirements), &absent).is_err()
        );
    }

    #[test]
    fn realtime_synthesis_rejects_portably_invalid_geometry_before_native_predicates() {
        let mut requirements = requirements();
        for execution in &mut requirements.executions {
            execution.weight_lowerings[0].descriptor = WeightLoweringDescriptor::new(
                SourceTensorEncoding::Safetensors(StoredDtype::F32),
                LinearFormat::Dense,
                vec![2, 3],
                vec![2, 4],
                None,
            )
            .unwrap();
        }
        let support = SyntheticRealtimeSupport {
            supported: true,
            lowering_queries: std::cell::Cell::new(0),
            state_queries: std::cell::Cell::new(0),
        };
        let capabilities = crate::synthesize_realtime_capabilities(&requirements, &support);
        assert!(capabilities.weight_lowerings().is_empty());
        assert_eq!(support.lowering_queries.get(), 0);
        assert!(
            select_realtime_realization(&requirements, &request(&requirements), &capabilities)
                .is_err()
        );
    }

    #[test]
    fn realtime_session_requirements_do_not_depend_on_observation_requests() {
        let requirements = requirements();
        for required in [
            SessionCapabilities::new(false, true, false),
            SessionCapabilities::new(false, false, true),
        ] {
            let mut request = request(&requirements).with_required_session_capabilities(required);
            request.observations = RealtimeObservationRequirements::new(false, []);
            let mut capabilities = capabilities();
            capabilities.session = SessionCapabilities::new(true, false, false);
            let error =
                select_realtime_realization(&requirements, &request, &capabilities).unwrap_err();
            assert_eq!(
                error.issues(),
                if required.output_observation() {
                    &[RealtimeSelectionIssue::MissingOutputObservation][..]
                } else {
                    &[RealtimeSelectionIssue::MissingActivationInspection][..]
                }
            );
            capabilities.session = SessionCapabilities::new(true, true, true);
            capabilities.state = capabilities.state.with_observation_retention(false);
            let error =
                select_realtime_realization(&requirements, &request, &capabilities).unwrap_err();
            assert_eq!(
                error.issues(),
                &[RealtimeSelectionIssue::MissingStateObservationRetention]
            );
        }
    }

    #[test]
    fn identities_and_requirement_sets_are_exact_and_nonempty() {
        assert_eq!(
            RealtimeIdentity::new(" \t"),
            Err(RealtimeContractError::EmptyIdentity)
        );
        let exact = RealtimeIdentity::new("  exact identity  ").unwrap();
        assert_eq!(exact.as_str(), "  exact identity  ");
        assert_eq!(
            RealtimeMechanismRequirements::new([]),
            Err(RealtimeContractError::EmptyMechanismRequirements)
        );
        assert_eq!(
            RealtimeTopologyPolicy::new(identity("none"), false, []),
            Err(RealtimeContractError::EmptyTopologyPolicy)
        );
        let duplicate = lowering_requirement(
            "duplicate.target",
            LinearFormat::Dense,
            WeightLoweringKind::Direct,
        );
        let graph = execution_graph();
        let units = execution_units(&graph);
        assert_eq!(
            RealtimeExecutionRequirements::new(
                identity("duplicate-execution"),
                parameter_description(&graph, &units),
                vec![duplicate.clone(), duplicate],
            ),
            Err(RealtimeContractError::DuplicateWeightLoweringTarget)
        );
    }

    #[test]
    fn model_construction_contract_requires_one_exact_graph_and_layout() {
        let graph = execution_graph();
        let units = execution_units(&graph);
        let parameters = parameter_description(&graph, &units);
        let other_graph = ExecutionGraph::chain(["other"]).unwrap();
        let other_units = ExecutionUnitLayout::new(&other_graph, [1]).unwrap();
        let other_parameters = parameter_description(&other_graph, &other_units);

        assert_eq!(
            requirements_with_model_contract(
                graph.clone(),
                other_units,
                parameters.clone(),
                vec![execution("execution", parameters.clone())],
            ),
            Err(RealtimeContractError::ExecutionGraphLayoutMismatch)
        );
        assert_eq!(
            requirements_with_model_contract(
                graph.clone(),
                units.clone(),
                other_parameters.clone(),
                vec![execution("execution", parameters.clone())],
            ),
            Err(RealtimeContractError::SourceParameterDescriptionMismatch)
        );
        assert_eq!(
            requirements_with_model_contract(
                graph,
                units,
                parameters.clone(),
                vec![execution("execution", other_parameters)],
            ),
            Err(
                RealtimeContractError::ExecutionParameterDescriptionMismatch {
                    execution: identity("execution"),
                }
            )
        );
    }

    #[test]
    fn weight_components_enforce_recipe_geometry_and_atomic_family_invariants() {
        let component = |target: &str,
                         owner: Option<&str>,
                         recipe: Option<&str>,
                         sources: Vec<&str>,
                         shape: Vec<usize>,
                         role| {
            let retained_recipe = (owner.is_some() && recipe.is_some()).then(|| {
                let mut inputs = sources
                    .iter()
                    .map(|key| {
                        eredu_checkpoint::recipe::DerivedWeightRecipe::source(
                            *key,
                            eredu_checkpoint::store::TensorSelection::Full,
                        )
                    })
                    .collect::<Vec<_>>();
                if inputs.len() == 1 {
                    inputs.pop().expect("one source recipe")
                } else {
                    eredu_checkpoint::recipe::DerivedWeightRecipe::Concatenate { axis: 0, inputs }
                }
            });
            let output = retained_recipe.as_ref().map(|_| {
                let elements = shape.iter().copied().product::<usize>();
                eredu_checkpoint::recipe::RecipeMetadata {
                    shape: shape.clone(),
                    dtype: eredu_checkpoint::recipe::RecipeDtype::F32,
                    byte_len: u64::try_from(elements.saturating_mul(4)).unwrap(),
                }
            });
            RealtimeWeightComponentRequirement::new(
                identity(target),
                owner.map(identity),
                recipe.map(identity),
                retained_recipe,
                output,
                sources.into_iter().map(identity),
                shape,
                role,
            )
        };
        assert_eq!(
            component(
                "weight",
                Some("owner"),
                Some("recipe"),
                vec!["source"],
                vec![],
                RealtimeWeightComponentRole::Primary,
            ),
            Err(RealtimeContractError::InvalidWeightComponentShape)
        );
        assert_eq!(
            component(
                "weight",
                Some("owner"),
                Some("recipe"),
                vec!["source"],
                vec![8, 0],
                RealtimeWeightComponentRole::Primary,
            ),
            Err(RealtimeContractError::InvalidWeightComponentShape)
        );
        assert_eq!(
            component(
                "weight",
                Some("owner"),
                None,
                vec!["source"],
                vec![8, 8],
                RealtimeWeightComponentRole::Primary,
            ),
            Err(RealtimeContractError::IncompleteWeightComponentRecipe)
        );
        assert_eq!(
            component(
                "weight",
                None,
                Some("recipe"),
                vec!["source"],
                vec![8, 8],
                RealtimeWeightComponentRole::Primary,
            ),
            Err(RealtimeContractError::IncompleteWeightComponentRecipe)
        );
        assert_eq!(
            component(
                "weight",
                Some("owner"),
                Some("recipe"),
                vec![],
                vec![8, 8],
                RealtimeWeightComponentRole::Primary,
            ),
            Err(RealtimeContractError::EmptyRecipeComponentSources)
        );
        assert_eq!(
            component(
                "weight",
                None,
                None,
                vec!["source"],
                vec![8, 8],
                RealtimeWeightComponentRole::Primary,
            ),
            Err(RealtimeContractError::GeneratedComponentHasSources)
        );

        let primary = component(
            "weight",
            Some("owner"),
            Some("recipe"),
            vec!["source", "source"],
            vec![8, 8],
            RealtimeWeightComponentRole::Primary,
        )
        .unwrap();
        let generated_scale = component(
            "scales",
            None,
            None,
            vec![],
            vec![8, 1],
            RealtimeWeightComponentRole::Scale,
        )
        .unwrap();
        let generated_bias = component(
            "biases",
            None,
            None,
            vec![],
            vec![8, 1],
            RealtimeWeightComponentRole::AffineBias,
        )
        .unwrap();
        let packed = RealtimeWeightLoweringRequirement::new(
            identity("weight"),
            [
                primary.clone(),
                generated_scale.clone(),
                generated_bias.clone(),
            ],
            lowering(LinearFormat::Dense),
            WeightLoweringKind::Transform,
        )
        .unwrap();
        assert_eq!(packed.primary(), &primary);
        assert_eq!(packed.scale(), Some(&generated_scale));
        assert_eq!(packed.affine_bias(), Some(&generated_bias));
        assert_eq!(
            packed.components(),
            &[
                primary.clone(),
                generated_scale.clone(),
                generated_bias.clone(),
            ]
        );
        assert_eq!(
            packed.primary().source_occurrences(),
            &[identity("source"), identity("source")]
        );
        assert!(!packed.scale().unwrap().is_recipe_backed());
        assert_eq!(packed.scale().unwrap().recipe_owner(), None);
        assert_eq!(packed.scale().unwrap().recipe_identity(), None);
        assert!(packed.scale().unwrap().source_occurrences().is_empty());
        assert_eq!(packed.scale().unwrap().physical_shape(), &[8, 1]);

        assert_eq!(
            RealtimeWeightLoweringRequirement::new(
                identity("weight"),
                [generated_scale.clone()],
                lowering(LinearFormat::Dense),
                WeightLoweringKind::Transform,
            ),
            Err(RealtimeContractError::MissingPrimaryWeightComponent)
        );
        let other_primary = component(
            "other",
            Some("owner"),
            Some("recipe"),
            vec!["source"],
            vec![8, 8],
            RealtimeWeightComponentRole::Primary,
        )
        .unwrap();
        assert_eq!(
            RealtimeWeightLoweringRequirement::new(
                identity("weight"),
                [other_primary.clone()],
                lowering(LinearFormat::Dense),
                WeightLoweringKind::Direct,
            ),
            Err(RealtimeContractError::PrimaryWeightComponentTargetMismatch)
        );
        assert_eq!(
            RealtimeWeightLoweringRequirement::new(
                identity("weight"),
                [primary.clone(), other_primary],
                lowering(LinearFormat::Dense),
                WeightLoweringKind::Direct,
            ),
            Err(RealtimeContractError::DuplicatePrimaryWeightComponent)
        );
        let second_scale = component(
            "other.scales",
            None,
            None,
            vec![],
            vec![8, 1],
            RealtimeWeightComponentRole::Scale,
        )
        .unwrap();
        assert_eq!(
            RealtimeWeightLoweringRequirement::new(
                identity("weight"),
                [primary.clone(), generated_scale.clone(), second_scale],
                lowering(LinearFormat::Dense),
                WeightLoweringKind::Transform,
            ),
            Err(RealtimeContractError::DuplicateWeightComponentRole {
                role: RealtimeWeightComponentRole::Scale,
            })
        );
        let second_bias = component(
            "other.biases",
            None,
            None,
            vec![],
            vec![8, 1],
            RealtimeWeightComponentRole::AffineBias,
        )
        .unwrap();
        assert_eq!(
            RealtimeWeightLoweringRequirement::new(
                identity("weight"),
                [primary.clone(), generated_bias, second_bias],
                lowering(LinearFormat::Dense),
                WeightLoweringKind::Transform,
            ),
            Err(RealtimeContractError::DuplicateWeightComponentRole {
                role: RealtimeWeightComponentRole::AffineBias,
            })
        );
        let duplicate_target = component(
            "weight",
            None,
            None,
            vec![],
            vec![8, 1],
            RealtimeWeightComponentRole::Scale,
        )
        .unwrap();
        assert_eq!(
            RealtimeWeightLoweringRequirement::new(
                identity("weight"),
                [primary, duplicate_target],
                lowering(LinearFormat::Dense),
                WeightLoweringKind::Transform,
            ),
            Err(RealtimeContractError::DuplicateWeightComponentTarget)
        );
    }

    #[test]
    fn exact_selection_closes_architecture_and_mechanism_choices_once() {
        let requirements = requirements();
        let selected =
            select_realtime_realization(&requirements, &request(&requirements), &capabilities())
                .unwrap();

        assert_eq!(selected.requirements(), &requirements);
        assert_eq!(selected.source().as_str(), "artifact-4");
        assert_eq!(selected.execution().as_str(), "execution-transformed");
        assert_eq!(selected.execution_graph(), requirements.execution_graph());
        assert_eq!(selected.execution_units(), requirements.execution_units());
        assert_eq!(
            selected.source_parameters(),
            requirements.source_parameters()
        );
        assert_eq!(selected.operators(), required_operators());
        assert_eq!(capabilities().operators(), required_operators());
        assert_eq!(
            selected.execution_parameters(),
            requirements.executions()[1].parameters()
        );
        assert_eq!(
            selected.residency(),
            LayerWeightResidency::LayerwiseHost(crate::LayerwiseLoadOptions::default())
        );
        assert_eq!(
            selected.topology(),
            ParallelTopology::new(2, 1, 1, 1).unwrap()
        );
        assert_eq!(selected.rank(), 1);
        assert_eq!(selected.completion(), completion_policy());
        assert_eq!(
            selected.requirements().speech_schedule(),
            requirements.speech_schedule()
        );
        assert_eq!(
            selected.requirements().state_layout(),
            requirements.state_layout()
        );
        assert!(selected.observations().output());
        assert_eq!(
            selected
                .observations()
                .activations()
                .iter()
                .map(RealtimeIdentity::as_str)
                .collect::<Vec<_>>(),
            vec!["depth.slice.0", "temporal.layer.0"]
        );
        let selected_lowering = &selected.execution_requirements().weight_lowerings()[0];
        assert_eq!(selected_lowering.target().as_str(), "target.weight");
        let primary = selected_lowering.primary();
        assert_eq!(primary.target().as_str(), "target.weight");
        assert_eq!(primary.recipe_owner().unwrap().as_str(), "canonical-owner");
        assert_eq!(primary.recipe_identity().unwrap().as_str(), "recipe-v1");
        assert_eq!(
            primary
                .source_occurrences()
                .iter()
                .map(RealtimeIdentity::as_str)
                .collect::<Vec<_>>(),
            vec!["source-a", "source-b"]
        );
        assert_eq!(primary.physical_shape(), &[8, 8]);
        assert_eq!(primary.role(), RealtimeWeightComponentRole::Primary);
        assert_eq!(
            selected_lowering.components(),
            std::slice::from_ref(primary)
        );
        assert_eq!(selected_lowering.kind(), WeightLoweringKind::Direct);
        assert_eq!(selected_lowering.scale(), None);
        assert_eq!(selected_lowering.affine_bias(), None);

        let selected_state = selected.state();
        assert_eq!(selected_state.layout(), requirements.state_layout());
        assert_eq!(selected_state.policy(), &CacheResidencyPolicy::Device);
        let expected_components = (0..requirements.state_layout().len()).flat_map(|layer| {
            requirements
                .state_layout()
                .components(layer)
                .unwrap()
                .iter()
                .map(move |component| (layer, component))
        });
        let expected_count = expected_components.clone().count();
        assert_eq!(selected_state.components().len(), expected_count);
        for (selected_component, (layer, component)) in
            selected_state.components().iter().zip(expected_components)
        {
            assert_eq!(selected_component.layer(), layer);
            assert_eq!(selected_component.component(), component);
            assert_eq!(
                selected_component.placement(),
                StateComponentPlacement::Device
            );
        }
        assert!(selected_state.checkpoint());
        assert!(selected_state.rollback());
        assert!(selected_state.reset());
        assert!(selected_state.observation_retention());
    }

    #[test]
    fn selection_reports_stable_fail_closed_diagnostics() {
        let requirements = requirements();
        let request = RealtimeSelectionRequest::new(
            identity("wrong-artifact"),
            identity("unknown-execution"),
            LayerWeightResidency::DenseDiskStream(crate::DenseDiskStreamLoadOptions::default()),
            CacheResidencyPolicy::Device,
            None,
            completion_policy(),
            RealtimeObservationRequirements::new(true, [identity("temporal.layer.0")]),
        );
        let capabilities = RealtimeMechanismCapabilities::new(
            NeuralOperatorCapabilities::NONE,
            [RealtimeMechanism::NeuralOperations],
            [ExecutionResidency::FullyResident],
            Vec::new(),
            StateMechanismCapabilities::new([]),
            NonZeroUsize::new(1).unwrap(),
            crate::CommunicationCompletionCapabilities::new([
                CompletionCancellationMode::NativeCancel,
            ])
            .unwrap(),
            SessionCapabilities::default(),
        );
        let error =
            select_realtime_realization(&requirements, &request, &capabilities).unwrap_err();

        assert_eq!(
            error.issues(),
            &[
                RealtimeSelectionIssue::MissingArchitectureProof,
                RealtimeSelectionIssue::SourceIdentityMismatch,
                RealtimeSelectionIssue::ExecutionIdentityNotAdmitted,
                RealtimeSelectionIssue::MissingNeuralOperator {
                    operator: "exp".into(),
                },
                RealtimeSelectionIssue::MissingNeuralOperator {
                    operator: "softmax_axis".into(),
                },
                RealtimeSelectionIssue::ResidencyNotAdmitted,
                RealtimeSelectionIssue::ResidencyUnavailable,
                RealtimeSelectionIssue::MissingMechanism(RealtimeMechanism::TensorOperations),
                RealtimeSelectionIssue::MissingMechanism(
                    RealtimeMechanism::ParameterMaterialization,
                ),
                RealtimeSelectionIssue::MissingMechanism(RealtimeMechanism::ParameterStorage),
                RealtimeSelectionIssue::MissingMechanism(RealtimeMechanism::StateStorage),
                RealtimeSelectionIssue::MissingMechanism(RealtimeMechanism::CoordinateStorage),
                RealtimeSelectionIssue::MissingMechanism(RealtimeMechanism::Sampling),
                RealtimeSelectionIssue::MissingMechanism(RealtimeMechanism::Randomness),
                RealtimeSelectionIssue::MissingMechanism(RealtimeMechanism::HostConversion),
                RealtimeSelectionIssue::MissingMechanism(RealtimeMechanism::ExactCompletion),
                RealtimeSelectionIssue::MissingMechanism(RealtimeMechanism::ResourceRetention),
                RealtimeSelectionIssue::MissingMechanism(RealtimeMechanism::Transfer),
                RealtimeSelectionIssue::MissingMechanism(RealtimeMechanism::Collectives),
                RealtimeSelectionIssue::CompletionPolicyUnavailable,
                RealtimeSelectionIssue::StateComponentUnavailable {
                    layer: 0,
                    component: "attention.keys".into(),
                },
                RealtimeSelectionIssue::StateComponentUnavailable {
                    layer: 0,
                    component: "attention.values".into(),
                },
                RealtimeSelectionIssue::StateComponentUnavailable {
                    layer: 1,
                    component: "attention.keys".into(),
                },
                RealtimeSelectionIssue::StateComponentUnavailable {
                    layer: 1,
                    component: "attention.values".into(),
                },
                RealtimeSelectionIssue::MissingStateCheckpoint,
                RealtimeSelectionIssue::MissingStateRollback,
                RealtimeSelectionIssue::MissingStateReset,
                RealtimeSelectionIssue::MissingStateObservationRetention,
                RealtimeSelectionIssue::MissingPersistentState,
                RealtimeSelectionIssue::MissingOutputObservation,
                RealtimeSelectionIssue::MissingActivationInspection,
                RealtimeSelectionIssue::ObservationUnavailable {
                    observation: identity("temporal.layer.0"),
                },
            ]
        );
        assert_eq!(
            error.to_string(),
            "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"
        );
    }

    #[test]
    fn exact_weight_lowering_must_be_implemented_before_construction() {
        let requirements = requirements();
        let capabilities = RealtimeMechanismCapabilities::new(
            required_operators(),
            required_mechanisms(),
            [
                ExecutionResidency::FullyResident,
                ExecutionResidency::LayerwiseHost,
            ],
            vec![WeightLoweringCapability::new(
                lowering(LinearFormat::Dense),
                WeightLoweringKind::Transform,
            )],
            state_capabilities(),
            NonZeroUsize::new(2).unwrap(),
            completion_capabilities(),
            SessionCapabilities::new(true, true, true),
        )
        .with_observation_identities([identity("temporal.layer.0"), identity("depth.slice.0")]);
        let error =
            select_realtime_realization(&requirements, &request(&requirements), &capabilities)
                .unwrap_err();

        assert_eq!(
            error.issues(),
            &[RealtimeSelectionIssue::WeightLoweringUnavailable {
                index: 0,
                target: identity("target.weight"),
            }]
        );
    }

    #[test]
    fn topology_and_rank_are_fail_closed() {
        let requirements = requirements();
        let invalid_proof = RealtimeArchitectureProof::new(
            requirements.architecture().clone(),
            requirements.speech_schedule_identity().clone(),
            requirements.state_layout_identity().clone(),
            requirements.topology().identity().clone(),
            ParallelTopology::new(2, 2, 1, 1).unwrap(),
            4,
        );
        let request = RealtimeSelectionRequest::new(
            requirements.source().clone(),
            identity("execution-native"),
            LayerWeightResidency::FullyResident,
            CacheResidencyPolicy::Device,
            Some(invalid_proof),
            completion_policy(),
            RealtimeObservationRequirements::default(),
        );
        let error =
            select_realtime_realization(&requirements, &request, &capabilities()).unwrap_err();
        assert_eq!(
            error.issues(),
            &[
                RealtimeSelectionIssue::TopologyNotAdmitted,
                RealtimeSelectionIssue::RankOutOfRange,
            ]
        );
    }

    #[test]
    fn selection_failure_prevents_every_construction_callback() {
        let requirements = requirements();
        let invalid_request = RealtimeSelectionRequest::new(
            identity("wrong-artifact"),
            identity("execution-native"),
            LayerWeightResidency::FullyResident,
            CacheResidencyPolicy::Device,
            Some(proof(&requirements)),
            completion_policy(),
            RealtimeObservationRequirements::default(),
        );
        let calls = Rc::new(Cell::new(0));
        let callback = || {
            let calls = Rc::clone(&calls);
            move |_: &SelectedRealtimeRealization| {
                calls.set(calls.get() + 1);
                Ok::<_, Infallible>(())
            }
        };
        let module_calls = Rc::clone(&calls);
        let state_calls = Rc::clone(&calls);
        let error = select_and_prepare_realtime_realization(
            &requirements,
            &invalid_request,
            &capabilities(),
            callback(),
            move |_, _| {
                module_calls.set(module_calls.get() + 1);
                Ok::<_, Infallible>(())
            },
            move |_, _| {
                state_calls.set(state_calls.get() + 1);
                Ok::<_, Infallible>(())
            },
            callback(),
            callback(),
        )
        .unwrap_err();

        assert!(matches!(error, RealtimePreparationError::Selection(_)));
        assert_eq!(calls.get(), 0);
    }

    #[test]
    fn successful_gate_constructs_in_dependency_order() {
        let requirements = requirements();
        let calls = Rc::new(Cell::new(0));
        let prepared = select_and_prepare_realtime_realization(
            &requirements,
            &request(&requirements),
            &capabilities(),
            {
                let calls = Rc::clone(&calls);
                move |_| {
                    assert_eq!(calls.replace(1), 0);
                    Ok::<_, Infallible>("payload")
                }
            },
            {
                let calls = Rc::clone(&calls);
                move |_, payload| {
                    assert_eq!((*payload, calls.replace(2)), ("payload", 1));
                    Ok::<_, Infallible>("modules")
                }
            },
            {
                let calls = Rc::clone(&calls);
                move |_, modules| {
                    assert_eq!((*modules, calls.replace(3)), ("modules", 2));
                    Ok::<_, Infallible>("state")
                }
            },
            {
                let calls = Rc::clone(&calls);
                move |_| {
                    assert_eq!(calls.replace(4), 3);
                    Ok::<_, Infallible>("queue")
                }
            },
            {
                let calls = Rc::clone(&calls);
                move |selected| {
                    assert_eq!(selected.topology().tensor(), 2);
                    assert_eq!(calls.replace(5), 4);
                    Ok::<_, Infallible>("group")
                }
            },
        )
        .unwrap();

        assert_eq!(calls.get(), 5);
        let (_, resources) = prepared.into_parts();
        assert_eq!(
            resources.into_parts(),
            ("payload", "modules", "state", "queue", Some("group"))
        );
    }
}