tenferro_runtime/runtime/

snapshot.rs

use std::collections::BTreeMap;
use std::fmt;
use std::num::NonZeroU64;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, RwLock};

use tenferro_tensor::TensorRead;

use crate::graph::CompiledGraph;
use crate::program::FrozenProgram;

use super::cache::{PreparedPlanCacheLimits, RuntimeCacheSet};
use super::cache_owner::{FrozenCacheOwner, FrozenCacheOwnerKind};
use super::engine_registration::{CandidateRegistrationToken, EngineRegistrationState};
#[cfg(test)]
use super::extension::ExtensionSlotFullForTest;
use super::extension::{
    bind_candidate_module, configure_module, freeze_extension_slots, BoundCandidateModuleRecord,
    CandidateModuleRecord, CandidateRegistrationIdentity, ExtensionEngineSnapshotView,
    ExtensionFamilyId, FrozenExtensionSlots,
};
use super::preparation::{PreparedEntryKey, PreparedProgram, PreparedProgramResult};
use super::schedule::EventDomainId;
use super::{
    CoreCapabilityBundle, EngineId, EngineRegistration, ExecutionContextIdentity, ExecutionPolicy,
    ExtensionModule, ExtensionModuleError, ExtensionModuleId, FrozenTransferRegistry,
    HardwareClassId, InputSignature, PrepareCapability, PrepareOptions, ProviderDeviceIdentity,
    RegistrationIdentity, RegistrationKey, ResolvedTransferEndpoint, ResolvedTransferRoute,
    RuntimeCacheError, RuntimeCacheStats, RuntimeConfigError, RuntimeEpoch, RuntimeId,
    RuntimeReconfigureError, RuntimeStateError, StorageClass, TransferEndpoint, TransferProvider,
    TransferRoute,
};
use crate::{Error, ErrorPhase};

static NEXT_RUNTIME_ID: AtomicU64 = AtomicU64::new(1);
static NEXT_REGISTRATION_ISSUER: AtomicU64 = AtomicU64::new(1);
const INITIAL_REGISTRATION_ORDINAL: NonZeroU64 = NonZeroU64::MIN;

#[derive(Clone, Debug)]
struct CandidateEngineRecord {
    registration: EngineRegistration,
    identity: CandidateRegistrationIdentity,
}

#[derive(Clone, Debug)]
struct BoundCandidateEngineRecord {
    registration: EngineRegistration,
    identity: RegistrationIdentity,
}

#[derive(Clone, Debug)]
enum CandidateTransferBinding {
    /// A route registered before its complete candidate can be validated.
    New,
    /// A route carried forward from a frozen snapshot.
    Preserved {
        source: ProviderDeviceIdentity,
        destination: ProviderDeviceIdentity,
    },
}

#[derive(Clone, Debug)]
struct CandidateTransferRecord {
    provider: Arc<dyn TransferProvider>,
    binding: CandidateTransferBinding,
}

struct BoundCandidateTransferRecord {
    provider: Arc<dyn TransferProvider>,
    source: ProviderDeviceIdentity,
    destination: ProviderDeviceIdentity,
}

#[derive(Clone, Debug)]
struct CandidateConfig {
    policy: ExecutionPolicy,
    engines: BTreeMap<EngineId, CandidateEngineRecord>,
    modules: BTreeMap<ExtensionModuleId, CandidateModuleRecord>,
    transfers: BTreeMap<TransferRoute, CandidateTransferRecord>,
}

struct BoundCandidateConfig {
    policy: ExecutionPolicy,
    engines: BTreeMap<EngineId, BoundCandidateEngineRecord>,
    modules: BTreeMap<ExtensionModuleId, BoundCandidateModuleRecord>,
    transfers: BTreeMap<TransferRoute, BoundCandidateTransferRecord>,
}

impl CandidateConfig {
    fn empty() -> Self {
        Self {
            policy: default_execution_policy(),
            engines: BTreeMap::new(),
            modules: BTreeMap::new(),
            transfers: BTreeMap::new(),
        }
    }

    fn from_snapshot(snapshot: &RuntimeConfigSnapshot) -> Result<Self, RuntimeConfigError> {
        let engines = snapshot
            .engines
            .iter()
            .map(|slot| {
                let registration = slot.to_registration()?;
                Ok((
                    registration.engine_id().clone(),
                    CandidateEngineRecord {
                        registration,
                        identity: CandidateRegistrationIdentity::Preserved(
                            slot.metadata().identity,
                        ),
                    },
                ))
            })
            .collect::<Result<BTreeMap<_, _>, RuntimeConfigError>>()?;
        Ok(Self {
            policy: snapshot.policy.clone(),
            engines,
            modules: snapshot.extensions.to_candidate_modules(),
            transfers: snapshot
                .transfers
                .iter()
                .map(|(resolved_route, provider)| {
                    (
                        TransferRoute::new(
                            resolved_route.source().logical().clone(),
                            resolved_route.destination().logical().clone(),
                        ),
                        CandidateTransferRecord {
                            provider: Arc::clone(provider),
                            binding: CandidateTransferBinding::Preserved {
                                source: resolved_route.source().provider_device_identity().clone(),
                                destination: resolved_route
                                    .destination()
                                    .provider_device_identity()
                                    .clone(),
                            },
                        },
                    )
                })
                .collect(),
        })
    }
}

#[derive(Clone, Debug)]
struct FrozenEngineMetadata {
    candidate_token: Arc<CandidateRegistrationToken>,
    identity: RegistrationIdentity,
    event_domain_id: EventDomainId,
}

#[derive(Clone)]
struct PreparationOnlyEngineSnapshot {
    metadata: FrozenEngineMetadata,
    binding: super::ProviderPreparationBinding,
}

#[derive(Clone, Debug)]
pub(super) struct ExecutableEngineSnapshot {
    metadata: FrozenEngineMetadata,
    binding: super::ProviderExecutableBinding,
}

#[derive(Clone)]
enum FrozenEngineSlot {
    PreparationOnly(Arc<PreparationOnlyEngineSnapshot>),
    Executable(Arc<ExecutableEngineSnapshot>),
}

impl FrozenEngineSlot {
    fn metadata(&self) -> &FrozenEngineMetadata {
        match self {
            Self::PreparationOnly(snapshot) => &snapshot.metadata,
            Self::Executable(snapshot) => &snapshot.metadata,
        }
    }

    fn provider_device_identity(&self) -> &ProviderDeviceIdentity {
        match self {
            Self::PreparationOnly(snapshot) => snapshot.binding.provider_device_identity(),
            Self::Executable(snapshot) => snapshot.binding.contract().provider_device_identity(),
        }
    }

    fn engine_id(&self) -> &EngineId {
        match self {
            Self::PreparationOnly(snapshot) => snapshot.binding.engine_id(),
            Self::Executable(snapshot) => snapshot.binding.engine_id(),
        }
    }

    fn hardware_class(&self) -> &HardwareClassId {
        match self {
            Self::PreparationOnly(snapshot) => snapshot.binding.hardware_class(),
            Self::Executable(snapshot) => snapshot.binding.hardware_class(),
        }
    }

    fn storage_classes(&self) -> &[StorageClass] {
        match self {
            Self::PreparationOnly(snapshot) => snapshot.binding.storage_classes(),
            Self::Executable(snapshot) => snapshot.binding.storage_classes(),
        }
    }

    fn default_storage_class(&self) -> &StorageClass {
        match self {
            Self::PreparationOnly(snapshot) => snapshot.binding.default_storage_class(),
            Self::Executable(snapshot) => snapshot.binding.default_storage_class(),
        }
    }

    fn context_identity(&self) -> ExecutionContextIdentity {
        match self {
            Self::PreparationOnly(snapshot) => snapshot.binding.context_identity(),
            Self::Executable(snapshot) => snapshot.binding.contract().context_identity(),
        }
    }

    fn capabilities(&self) -> &CoreCapabilityBundle {
        match self {
            Self::PreparationOnly(snapshot) => snapshot.binding.capabilities(),
            Self::Executable(snapshot) => snapshot.binding.contract().capabilities(),
        }
    }

    fn executable(&self) -> Option<&Arc<ExecutableEngineSnapshot>> {
        match self {
            Self::PreparationOnly(_) => None,
            Self::Executable(snapshot) => Some(snapshot),
        }
    }
}

impl ExecutableEngineSnapshot {
    pub(super) fn engine_id(&self) -> &EngineId {
        self.binding.engine_id()
    }

    pub(super) fn event_domain_id(&self) -> EventDomainId {
        self.metadata.event_domain_id
    }

    pub(super) fn provider_device_identity(&self) -> &ProviderDeviceIdentity {
        self.binding.contract().provider_device_identity()
    }

    #[cfg(test)]
    pub(super) fn context_identity(&self) -> ExecutionContextIdentity {
        self.binding.contract().context_identity()
    }

    /// Fused-region input limit reported by this engine's elementwise
    /// capability; see [`super::ElementwiseRuntime::max_fused_region_inputs`].
    pub(super) fn max_fused_region_inputs(&self) -> Option<usize> {
        self.binding
            .contract()
            .capabilities()
            .elementwise()
            .and_then(|elementwise| elementwise.max_fused_region_inputs())
    }

    /// Whether this engine folds metadata-only operations into their reading
    /// consumer; see [`super::LayoutRuntime::folds_metadata_views`].
    pub(super) fn folds_metadata_views(&self) -> bool {
        self.binding
            .contract()
            .capabilities()
            .layout()
            .is_some_and(|layout| layout.folds_metadata_views())
    }

    pub(super) fn accepts_input_placement(
        &self,
        placement: &tenferro_tensor::Placement,
        storage_class: &StorageClass,
    ) -> bool {
        self.binding.storage_classes().contains(storage_class)
            && self
                .binding
                .contract()
                .accepts_input_placement(placement, storage_class)
    }

    pub(super) fn accepts_input_signature(
        &self,
        input: &super::InputSignatureEntry,
        storage_class: &StorageClass,
    ) -> bool {
        self.binding.storage_classes().contains(storage_class)
            && self
                .binding
                .contract()
                .accepts_input_signature(input, storage_class)
    }

    pub(super) fn accepts_runtime_input(
        &self,
        input: &tenferro_tensor::TensorRead<'_>,
        storage_class: &StorageClass,
    ) -> bool {
        self.binding.storage_classes().contains(storage_class)
            && self
                .binding
                .contract()
                .accepts_runtime_input(input, storage_class)
    }

    pub(super) fn owns_resident_tensor(
        &self,
        input: &tenferro_tensor::TensorRead<'_>,
        storage_class: &StorageClass,
    ) -> bool {
        self.binding.storage_classes().contains(storage_class)
            && self
                .binding
                .contract()
                .owns_resident_tensor(input, storage_class)
    }

    #[cfg(test)]
    pub(super) fn for_test(
        engine_id: EngineId,
        provider_device_identity: ProviderDeviceIdentity,
        event_domain_id: EventDomainId,
        storage_class: StorageClass,
    ) -> Arc<Self> {
        struct TestEngine;
        let ingress = super::InputIngressContract::new(
            super::InputPlacementContract::new(|_, _| true),
            super::InputSignatureContract::new(|_, _, _, _| true),
            super::RuntimeInputContract::new(|_, _| true),
            super::ResidentOutputContract::new(|_, _| true),
        );
        let contract = super::engine_registration::ExecutableEngineContract::new(
            provider_device_identity,
            ExecutionContextIdentity::of::<TestEngine>(),
            CoreCapabilityBundle::default(),
            ingress,
        );
        let binding = super::ProviderExecutableBinding::new(
            engine_id,
            HardwareClassId::new("tenferro.test.schedule.hardware").expect("test hardware class"),
            Arc::from(vec![storage_class.clone()]),
            storage_class,
            contract,
        )
        .expect("test executable binding");
        Arc::new(Self {
            metadata: FrozenEngineMetadata {
                candidate_token: Arc::new(CandidateRegistrationToken),
                identity: event_domain_id.registration_identity(),
                event_domain_id,
            },
            binding,
        })
    }
}

impl FrozenEngineSlot {
    fn to_registration(&self) -> Result<EngineRegistration, RuntimeConfigError> {
        let metadata = self.metadata();
        let registration = match self {
            Self::PreparationOnly(snapshot) => {
                EngineRegistration::from_state(EngineRegistrationState::PreparationOnly {
                    binding: snapshot.binding.clone(),
                })
            }
            Self::Executable(snapshot) => EngineRegistration::from_state(
                EngineRegistrationState::Executable(snapshot.binding.clone()),
            ),
        };
        Ok(registration.with_candidate_token(Arc::clone(&metadata.candidate_token)))
    }
}

impl fmt::Debug for FrozenEngineSlot {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        let metadata = self.metadata();
        formatter
            .debug_struct("FrozenEngineSlot")
            .field("engine_id", self.engine_id())
            .field("registration_identity", &metadata.identity)
            .field("event_domain_id", &metadata.event_domain_id)
            .field("context_identity", &self.context_identity())
            .field("hardware_class", self.hardware_class())
            .field(
                "state",
                &match self {
                    Self::PreparationOnly(_) => "preparation-only",
                    Self::Executable(_) => "executable",
                },
            )
            .finish_non_exhaustive()
    }
}

/// Immutable runtime configuration snapshot.
///
/// A snapshot is readable after later reconfiguration because the runtime
/// publishes a new `Arc` instead of mutating existing slots.
///
/// # Examples
///
/// ```
/// # fn main() -> Result<(), Box<dyn std::error::Error>> {
/// use tenferro_runtime::RuntimeConfigBuilder;
///
/// let runtime = RuntimeConfigBuilder::new().build()?;
/// let snapshot = runtime.snapshot()?;
/// assert_eq!(snapshot.engine_count(), 0);
/// # Ok(())
/// # }
/// ```
#[derive(Clone)]
pub struct RuntimeConfigSnapshot {
    runtime_id: RuntimeId,
    epoch: RuntimeEpoch,
    policy: ExecutionPolicy,
    engines: Arc<[FrozenEngineSlot]>,
    engine_indices: BTreeMap<EngineId, usize>,
    extensions: FrozenExtensionSlots,
    transfers: FrozenTransferRegistry,
    cache_owners: Arc<[FrozenCacheOwner]>,
}

impl RuntimeConfigSnapshot {
    /// Return the runtime identity that published this snapshot.
    pub fn runtime_id(&self) -> RuntimeId {
        self.runtime_id
    }

    /// Return the epoch for this immutable snapshot.
    pub fn epoch(&self) -> RuntimeEpoch {
        self.epoch
    }

    /// Return the execution policy captured by this snapshot.
    pub fn execution_policy(&self) -> &ExecutionPolicy {
        &self.policy
    }

    /// Return the number of direct engine slots.
    pub fn engine_count(&self) -> usize {
        self.engines.len()
    }

    /// Return the number of installed extension modules.
    pub fn extension_module_count(&self) -> usize {
        self.extensions.module_count()
    }

    /// Return the number of registered transfer providers.
    pub fn transfer_provider_count(&self) -> usize {
        self.transfers.len()
    }

    /// Return whether this snapshot contains an extension engine for a family.
    #[doc(hidden)]
    pub fn has_extension_family(&self, family_id: &'static str) -> bool {
        self.extensions.has_family(family_id)
    }

    /// Return whether this snapshot contains an extension engine for one exact
    /// family/engine pair.
    ///
    /// This narrow, doc-hidden query is the cross-crate eager-extension
    /// contract used by `tenferro-ad` to validate an owner-selected module
    /// without exposing module internals or widening [`ExtensionModule`].
    ///
    /// # Examples
    ///
    /// ```rust
    /// use tenferro_runtime::{EngineId, Runtime};
    ///
    /// let runtime = Runtime::builder().build()?;
    /// let snapshot = runtime.snapshot()?;
    /// let engine = EngineId::new("example.engine")?;
    /// assert!(!snapshot.has_extension_engine("example.family.v1", &engine));
    /// # Ok::<(), Box<dyn std::error::Error>>(())
    /// ```
    #[doc(hidden)]
    pub fn has_extension_engine(&self, family_id: &'static str, engine_id: &EngineId) -> bool {
        self.extensions.has_engine(family_id, engine_id)
    }

    /// Return whether the exact validated module instance is installed.
    ///
    /// This narrow, doc-hidden query is the eager direct-bridge steady-state
    /// no-op contract: it mirrors the `module_identical` no-op predicate of the
    /// replace-extension-module edit without building a candidate
    /// configuration.
    #[doc(hidden)]
    pub fn has_extension_module_identical(&self, module: &Arc<dyn ExtensionModule>) -> bool {
        self.extensions.has_module_identical(module)
    }

    /// Return whether the installed module with this exact module ID owns the
    /// given family/engine registration.
    ///
    /// Mirrors the owner-scoped ensure-extension-module edit's no-op predicate
    /// on the published snapshot. The module ID must be the one that owns the
    /// registration; a different module registering the same family/engine
    /// pair does not satisfy this query.
    #[doc(hidden)]
    pub fn has_extension_module_engine(
        &self,
        module_id: &ExtensionModuleId,
        family_id: &'static str,
        engine_id: &EngineId,
    ) -> bool {
        self.extensions
            .has_module_engine(module_id, family_id, engine_id)
    }

    /// Return an immutable view of a registered engine slot.
    pub fn engine(&self, id: &EngineId) -> Option<EngineSnapshotView<'_>> {
        self.engine_indices
            .get(id)
            .map(|&index| EngineSnapshotView {
                slot: &self.engines[index],
            })
    }

    #[cfg(test)]
    pub(crate) fn engine_ids_for_test(&self) -> impl Iterator<Item = &EngineId> {
        self.engines.iter().map(FrozenEngineSlot::engine_id)
    }

    #[cfg(test)]
    pub(crate) fn transfer_routes_for_test(&self) -> impl Iterator<Item = &ResolvedTransferRoute> {
        self.transfers.iter().map(|(route, _)| route)
    }

    pub(super) fn engine_views_for_preparation(
        &self,
    ) -> impl Iterator<Item = EngineSnapshotView<'_>> + '_ {
        self.engines.iter().map(|slot| EngineSnapshotView { slot })
    }

    pub(super) fn extension_slot_for_preparation(
        &self,
        family_id: ExtensionFamilyId,
        engine_id: &EngineId,
    ) -> Option<ExtensionEngineSnapshotView<'_>> {
        self.extensions.slot_for_preparation(family_id, engine_id)
    }

    pub(super) fn transfer_registry_for_preparation(&self) -> FrozenTransferRegistry {
        self.transfers.clone()
    }

    #[cfg(test)]
    pub(crate) fn extension_slots_for_test(
        &self,
    ) -> impl Iterator<
        Item = (
            &ExtensionModuleId,
            ExtensionFamilyId,
            &EngineId,
            RegistrationIdentity,
        ),
    > {
        self.extensions.slots_for_test()
    }

    #[cfg(test)]
    pub(crate) fn extension_slot_identity_for_test(
        &self,
        family_id: ExtensionFamilyId,
        engine_id: &EngineId,
    ) -> Option<RegistrationIdentity> {
        self.extensions.slot_identity_for_test(family_id, engine_id)
    }

    #[cfg(test)]
    pub(crate) fn extension_slot_full_for_test(
        &self,
        family_id: ExtensionFamilyId,
        engine_id: &EngineId,
    ) -> Option<ExtensionSlotFullForTest<'_>> {
        self.extensions.slot_full_for_test(family_id, engine_id)
    }

    #[cfg(test)]
    pub(super) fn cache_owners_for_test(&self) -> &[FrozenCacheOwner] {
        &self.cache_owners
    }

    pub(super) fn cache_owners_for_runtime(&self) -> &[FrozenCacheOwner] {
        &self.cache_owners
    }
}

impl fmt::Debug for RuntimeConfigSnapshot {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter
            .debug_struct("RuntimeConfigSnapshot")
            .field("runtime_id", &self.runtime_id)
            .field("epoch", &self.epoch)
            .field("execution_policy", &self.policy)
            .field("engine_count", &self.engines.len())
            .field("extension_module_count", &self.extensions.module_count())
            .field("extension_engine_count", &self.extensions.engine_count())
            .field("transfer_provider_count", &self.transfers.len())
            .field("cache_owner_count", &self.cache_owners.len())
            .finish_non_exhaustive()
    }
}

struct RuntimeState {
    runtime_id: RuntimeId,
    issuer: NonZeroU64,
    next_registration_ordinal: AtomicU64,
    active: RwLock<Arc<RuntimeConfigSnapshot>>,
    published_epoch: AtomicU64,
    caches: RuntimeCacheSet<PreparedEntryKey, PreparedProgram>,
}

/// Runtime owner for immutable configuration snapshots.
///
/// # Examples
///
/// ```
/// # fn main() -> Result<(), Box<dyn std::error::Error>> {
/// use tenferro_runtime::Runtime;
///
/// let runtime = Runtime::builder().build()?;
/// assert_eq!(runtime.snapshot()?.engine_count(), 0);
/// # Ok(())
/// # }
/// ```
#[derive(Clone)]
pub struct Runtime(Arc<RuntimeState>);

impl Runtime {
    /// Return a consuming runtime configuration builder.
    pub fn builder() -> RuntimeConfigBuilder {
        RuntimeConfigBuilder::new()
    }

    /// Return this runtime's opaque identity.
    pub fn id(&self) -> RuntimeId {
        self.0.runtime_id
    }

    /// Clone the current immutable runtime snapshot.
    ///
    /// # Errors
    ///
    /// Returns [`RuntimeStateError::Poisoned`] when the active snapshot lock was
    /// poisoned by another thread.
    pub fn snapshot(&self) -> Result<Arc<RuntimeConfigSnapshot>, RuntimeStateError> {
        self.0
            .active
            .read()
            .map(|snapshot| Arc::clone(&snapshot))
            .map_err(|_| RuntimeStateError::Poisoned {
                lock: "runtime.active",
            })
    }

    /// Return the currently published runtime epoch without locking the active
    /// snapshot.
    ///
    /// # Errors
    ///
    /// Returns [`RuntimeStateError`] only if runtime state invariants have been
    /// violated internally.
    pub fn epoch(&self) -> Result<RuntimeEpoch, RuntimeStateError> {
        match NonZeroU64::new(self.0.published_epoch.load(Ordering::Acquire)) {
            Some(value) => Ok(RuntimeEpoch::from_nonzero(value)),
            None => Err(RuntimeStateError::Poisoned {
                lock: "runtime.published_epoch",
            }),
        }
    }

    /// Return current prepared-plan cache limits.
    ///
    /// # Examples
    ///
    /// ```
    /// # fn main() -> Result<(), Box<dyn std::error::Error>> {
    /// use tenferro_runtime::Runtime;
    ///
    /// let runtime = Runtime::builder().build()?;
    /// assert_eq!(runtime.prepared_cache_limits()?, Default::default());
    /// # Ok(())
    /// # }
    /// ```
    ///
    /// # Errors
    ///
    /// Returns [`RuntimeStateError`] when the runtime-owned prepared cache state
    /// cannot be accessed.
    pub fn prepared_cache_limits(&self) -> Result<PreparedPlanCacheLimits, RuntimeStateError> {
        self.0.caches.prepared().limits()
    }

    /// Replace current prepared-plan cache limits and evict retained entries
    /// until the new limits are satisfied.
    ///
    /// # Examples
    ///
    /// ```
    /// # fn main() -> Result<(), Box<dyn std::error::Error>> {
    /// use std::num::NonZeroUsize;
    /// use tenferro_runtime::{PreparedPlanCacheLimits, Runtime};
    ///
    /// let runtime = Runtime::builder().build()?;
    /// runtime.set_prepared_cache_limits(PreparedPlanCacheLimits {
    ///     max_entries: NonZeroUsize::new(1).unwrap(),
    ///     max_retained_bytes: NonZeroUsize::new(1024).unwrap(),
    ///     max_in_flight_entries: NonZeroUsize::new(1).unwrap(),
    ///     max_queued_distinct_keys: NonZeroUsize::new(1).unwrap(),
    /// })?;
    /// assert_eq!(runtime.prepared_cache_limits()?.max_entries.get(), 1);
    /// # Ok(())
    /// # }
    /// ```
    ///
    /// # Errors
    ///
    /// Returns [`RuntimeStateError`] when the runtime-owned prepared cache state
    /// cannot be accessed.
    pub fn set_prepared_cache_limits(
        &self,
        limits: PreparedPlanCacheLimits,
    ) -> Result<(), RuntimeStateError> {
        self.0.caches.prepared().set_limits(limits)
    }

    /// Clear the runtime-owned prepared-plan cache.
    ///
    /// # Examples
    ///
    /// ```
    /// # fn main() -> Result<(), Box<dyn std::error::Error>> {
    /// use tenferro_runtime::Runtime;
    ///
    /// let runtime = Runtime::builder().build()?;
    /// runtime.clear_prepared_cache()?;
    /// # Ok(())
    /// # }
    /// ```
    ///
    /// # Errors
    ///
    /// Returns [`RuntimeStateError`] when the runtime-owned prepared cache state
    /// cannot be accessed.
    pub fn clear_prepared_cache(&self) -> Result<(), RuntimeStateError> {
        self.0.caches.prepared().clear()
    }

    /// Return aggregate cache statistics for the runtime and registered cache
    /// owners.
    ///
    /// # Examples
    ///
    /// ```
    /// # fn main() -> Result<(), Box<dyn std::error::Error>> {
    /// use tenferro_runtime::Runtime;
    ///
    /// let runtime = Runtime::builder().build()?;
    /// assert_eq!(runtime.cache_stats()?.prepared_plans.entries, 0);
    /// # Ok(())
    /// # }
    /// ```
    ///
    /// # Errors
    ///
    /// Returns [`RuntimeCacheError`] when the runtime cache or a registered
    /// cache owner cannot report statistics.
    pub fn cache_stats(&self) -> Result<RuntimeCacheStats, RuntimeCacheError> {
        super::preparation::cache_stats(self, &self.0.caches)
    }

    /// Clear runtime-owned prepared plans and all registered engine/extension
    /// cache owners.
    ///
    /// # Examples
    ///
    /// ```
    /// # fn main() -> Result<(), Box<dyn std::error::Error>> {
    /// use tenferro_runtime::Runtime;
    ///
    /// let runtime = Runtime::builder().build()?;
    /// runtime.clear_caches()?;
    /// # Ok(())
    /// # }
    /// ```
    ///
    /// # Errors
    ///
    /// Returns [`RuntimeCacheError`] when the runtime cache or a registered
    /// cache owner cannot be cleared.
    pub fn clear_caches(&self) -> Result<(), RuntimeCacheError> {
        super::preparation::clear_caches(self, &self.0.caches)
    }

    #[allow(
        dead_code,
        reason = "Phase 5 graph execution consumes crate-private prepared programs"
    )]
    pub(crate) fn prepare_for(
        &self,
        frozen: &FrozenProgram,
        signature: &InputSignature,
        options: &PrepareOptions,
    ) -> PreparedProgramResult<Arc<PreparedProgram>> {
        super::preparation::prepare_for(self, &self.0.caches, frozen, signature, options)
    }

    pub(crate) fn prepare_compiled_for(
        &self,
        program: &CompiledGraph,
        signature: &InputSignature,
        options: &PrepareOptions,
    ) -> PreparedProgramResult<Arc<PreparedProgram>> {
        super::preparation::prepare_compiled_for(self, &self.0.caches, program, signature, options)
    }

    /// Resolve and prepare a single extension operation for immediate eager
    /// execution against one exact engine, bypassing SemanticProgram planning
    /// and the prepared-program cache.
    ///
    /// Provider selection is pinned to `engine_id` (the eager context's exact
    /// engine); the op is prepared only when that engine is executable, owns
    /// the family's extension slot, and accepts every input signature entry.
    /// Capability resolution happens before any planning fields are built.
    /// Returns [`PrepareCapability::Unsupported`] when `engine_id` is not
    /// registered, is not executable, or has no extension slot for the op's
    /// family so callers may fall back to the compiled path.
    ///
    /// # Examples
    ///
    /// ```
    /// # fn main() -> Result<(), Box<dyn std::error::Error>> {
    /// use std::any::Any;
    /// use std::hash::Hasher;
    /// use std::sync::Arc;
    ///
    /// use tenferro_cpu::CpuEngine;
    /// use tenferro_ops::ExtensionShapeContext;
    /// use tenferro_runtime::extension::ExtensionOp;
    /// use tenferro_runtime::{
    ///     ExtensionEngine, ExtensionModule, ExtensionModuleError, ExtensionModuleId,
    ///     ExtensionModuleRegistrar, ExtensionPlanningConfig, InputSignature, PrepareCapability,
    ///     PrepareError, PrepareOptions, Runtime, UnsupportedReason, Tensor, TensorRead,
    /// };
    /// use tenferro_tensor::DType;
    ///
    /// #[derive(Debug)]
    /// struct Probe;
    ///
    /// impl ExtensionOp for Probe {
    ///     fn family_id(&self) -> &'static str { "tenferro-tests.immediate-probe.v1" }
    ///     fn payload_hash(&self, _hasher: &mut dyn Hasher) {}
    ///     fn payload_eq(&self, other: &dyn ExtensionOp) -> bool { other.as_any().downcast_ref::<Self>().is_some() }
    ///     fn clone_arc(&self) -> Arc<dyn ExtensionOp> { Arc::new(Self) }
    ///     fn as_any(&self) -> &dyn Any { self }
    ///     fn input_count(&self) -> usize { 1 }
    ///     fn output_count(&self) -> usize { 1 }
    ///     fn infer_output_meta(&self, ctx: &mut ExtensionShapeContext<'_>) -> tenferro_tensor::Result<Vec<(DType, Vec<tenferro_ops::SymDim>)>> {
    ///         Ok(vec![(ctx.input_dtype(0)?, ctx.input_shape(0)?.to_vec())])
    ///     }
    /// }
    ///
    /// #[derive(Debug)]
    /// struct Config;
    ///
    /// impl ExtensionPlanningConfig for Config {
    ///     fn family_id(&self) -> &'static str { Probe.family_id() }
    ///     fn as_any(&self) -> &dyn Any { self }
    ///     fn payload_hash(&self, _state: &mut dyn Hasher) {}
    ///     fn payload_eq(&self, other: &dyn ExtensionPlanningConfig) -> bool { other.as_any().downcast_ref::<Self>().is_some() }
    ///     fn retained_bytes(&self) -> usize { 0 }
    /// }
    ///
    /// #[derive(Debug)]
    /// struct Module(ExtensionModuleId);
    ///
    /// impl ExtensionModule for Module {
    ///     fn module_id(&self) -> &ExtensionModuleId { &self.0 }
    ///     fn configure(&self, registrar: &mut ExtensionModuleRegistrar<'_>) -> Result<(), ExtensionModuleError> {
    ///         let Ok(engine_id) = tenferro_cpu::runtime_engine_id() else {
    ///             // Best-effort: an unresolvable engine id keeps the op unsupported.
    ///             return Ok(());
    ///         };
    ///         registrar.register_engine(Arc::new(NoopEngine { engine_id: engine_id.clone() }))?;
    ///         registrar.register_planning_config(engine_id, Arc::new(Config))?;
    ///         Ok(())
    ///     }
    /// }
    ///
    /// #[derive(Debug)]
    /// struct NoopEngine { engine_id: tenferro_runtime::EngineId }
    ///
    /// impl ExtensionEngine for NoopEngine {
    ///     fn family_id(&self) -> &'static str { Probe.family_id() }
    ///     fn engine_id(&self) -> &tenferro_runtime::EngineId { &self.engine_id }
    ///     fn context_identity(&self) -> tenferro_runtime::ExecutionContextIdentity { tenferro_runtime::ExecutionContextIdentity::of::<CpuEngine>() }
    ///     fn prepare(&self, _request: tenferro_runtime::ExtensionPrepareRequest<'_>) -> Result<PrepareCapability, PrepareError> {
    ///         Ok(PrepareCapability::Unsupported(UnsupportedReason::Operation { operation: Probe.family_id() }))
    ///     }
    /// }
    ///
    /// let engine = CpuEngine::serial();
    /// let mut builder = Runtime::builder();
    /// builder.register_engine(tenferro_cpu::runtime_engine_registration(&engine)?)?;
    /// builder.install_extension_module(Arc::new(Module(ExtensionModuleId::new("tenferro-tests.immediate-probe.module")?)))?;
    /// let runtime = builder.build()?;
    /// let tensor = Tensor::from_vec_col_major(vec![1], vec![1.0_f64])?;
    /// let signature = InputSignature::from_reads(&[TensorRead::from_tensor(&tensor)])?;
    /// let engine_id = tenferro_cpu::runtime_engine_id()?;
    /// let capability = runtime.prepare_extension_immediate(&engine_id, &Probe, &signature)?;
    /// assert!(matches!(capability, PrepareCapability::Unsupported { .. }));
    /// # Ok(())
    /// # }
    /// ```
    ///
    /// # Errors
    ///
    /// Returns [`crate::Error::RuntimeState`] when the snapshot cannot be read
    /// or an engine's preparation fails, with the typed
    /// [`PrepareError`](super::PrepareError)/[`RuntimeStateError`] source retained.
    pub fn prepare_extension_immediate(
        &self,
        engine_id: &EngineId,
        op: &dyn tenferro_ops::ext_op::ExtensionOp,
        signature: &InputSignature,
    ) -> crate::Result<PrepareCapability> {
        super::preparation::prepare_extension_immediate(self, engine_id, op, signature).map_err(
            |source| {
                Error::runtime_state_source(
                    "Runtime::prepare_extension_immediate",
                    ErrorPhase::Execution,
                    source,
                )
            },
        )
    }

    /// Prepare `program` to run on engines described by `targets`.
    ///
    /// Preparation does the expensive checks and decisions once: it places
    /// each part of the program on one of the targets, plans fusion for those
    /// targets, and records what each target slot requires. The prepared
    /// program holds no engine; [`Self::run`] binds slot `i` to the `i`-th
    /// engine it receives. Pass the inputs (or none, to use the graph's bound
    /// tensors) whose shapes, types, layouts and memory the program will run on.
    ///
    /// # Examples
    ///
    /// ```
    /// use tenferro_runtime::{EngineTarget, Prepared, Runtime, TracedTensor, GraphCompiler};
    ///
    /// fn prepare_for(
    ///     runtime: &Runtime,
    ///     x: &TracedTensor,
    ///     target: EngineTarget,
    /// ) -> tenferro_runtime::Result<Prepared> {
    ///     let program = GraphCompiler::new().compile(x)?;
    ///     runtime.prepare(&program, &[], &[target])
    /// }
    /// ```
    ///
    /// # Errors
    ///
    /// Returns [`crate::Error::InvalidArgument`] when `targets` is empty or names
    /// an engine twice, [`crate::Error::RuntimeState`] when a target names an
    /// engine this runtime has not registered or reports a different context
    /// identity than its registration, metadata validation errors for inputs
    /// that do not match the graph, and preparation failures (no eligible
    /// engine, missing ingress or transfer route) with the typed
    /// [`crate::PrepareError`] source retained.
    pub fn prepare(
        &self,
        program: &CompiledGraph,
        inputs: &[TensorRead<'_>],
        targets: &[super::EngineTarget],
    ) -> crate::Result<super::Prepared> {
        super::execution::prepare(self, program, inputs, targets)
    }

    /// Run a prepared program on the engines passed in.
    ///
    /// Slot `i` of `prepared` runs on `engines[i]`; every engine must report
    /// the target the program was prepared for. On a CPU engine the call returns
    /// when the work is done. On a GPU engine it returns once the work is queued
    /// and no longer reads caller memory that could change; later work on the
    /// same engine runs after it in stream order. An output that is one of the
    /// inputs, or a view of one, is returned borrowed; every other output is a
    /// new compact tensor.
    ///
    /// # Examples
    ///
    /// ```
    /// use tenferro_runtime::{ExecutionEngine, Output, Prepared, Runtime};
    /// use tenferro_tensor::TensorRead;
    ///
    /// fn run_on<'a>(
    ///     runtime: &Runtime,
    ///     engine: &mut dyn ExecutionEngine,
    ///     prepared: &'a Prepared,
    ///     inputs: &[TensorRead<'a>],
    /// ) -> tenferro_runtime::Result<Vec<Output<'a>>> {
    ///     runtime.run(&mut [engine], prepared, inputs)
    /// }
    /// ```
    ///
    /// # Errors
    ///
    /// Returns [`crate::Error::RuntimeState`] when `prepared` belongs to another
    /// runtime or a stale epoch, [`crate::Error::InvalidArgument`] when the
    /// engines do not match the prepared targets, metadata validation errors for
    /// mismatched inputs, and the engines' own execution errors.
    pub fn run<'a>(
        &self,
        engines: &mut [&mut dyn super::ExecutionEngine],
        prepared: &'a super::Prepared,
        inputs: &[TensorRead<'a>],
    ) -> crate::Result<Vec<super::Output<'a>>> {
        super::execution::run(self, engines, prepared, inputs)
    }

    /// Pay `prepared`'s first execution now: run it once and wait for it.
    ///
    /// A prepared program is specialized to its input signature, so one
    /// execution compiles exactly the kernels it will use — fused regions,
    /// reduction strategies and any vendor plan — and initializes whatever the
    /// engine creates lazily. [`Self::run`] is otherwise unchanged: this is one
    /// ordinary run whose outputs are dropped, followed by
    /// [`ExecutionEngine::synchronize`], so the cost is not still queued when
    /// the caller's measured region starts.
    ///
    /// # Examples
    ///
    /// ```
    /// use tenferro_runtime::{ExecutionEngine, Prepared, Runtime};
    /// use tenferro_tensor::TensorRead;
    ///
    /// fn warm_up<'a>(
    ///     runtime: &Runtime,
    ///     engine: &mut dyn ExecutionEngine,
    ///     prepared: &'a Prepared,
    ///     inputs: &[TensorRead<'a>],
    /// ) -> tenferro_runtime::Result<()> {
    ///     runtime.warm_up(&mut [engine], prepared, inputs)
    /// }
    /// ```
    ///
    /// # Errors
    ///
    /// [`crate::Error::RuntimeState`] when `prepared` belongs to another runtime
    /// or a stale epoch, [`crate::Error::InvalidArgument`] when the engines do
    /// not match the prepared targets or the input metadata does not match the
    /// program, and [`crate::Error::TensorRuntime`] when an engine fails the run
    /// or cannot synchronize.
    pub fn warm_up<'a>(
        &self,
        engines: &mut [&mut dyn super::ExecutionEngine],
        prepared: &'a super::Prepared,
        inputs: &[TensorRead<'a>],
    ) -> crate::Result<()> {
        let outputs = self.run(engines, prepared, inputs)?;
        drop(outputs);
        for engine in engines.iter_mut() {
            engine.synchronize()?;
        }
        Ok(())
    }

    /// Transactionally edit and publish runtime configuration.
    ///
    /// No user callback runs while the publication lock is held. If another
    /// writer publishes over the same base snapshot, this call returns
    /// [`RuntimeReconfigureError::ConcurrentReconfiguration`] and publishes
    /// nothing.
    ///
    /// # Errors
    ///
    /// Returns [`RuntimeReconfigureError`] when state access, edit validation,
    /// identity allocation, epoch advancement, or compare-and-publish fails.
    /// Invalid transfer endpoints are reported as the typed
    /// [`RuntimeConfigError::UnknownTransferEndpointEngine`] or
    /// [`RuntimeConfigError::UnsupportedTransferEndpointStorage`] source of
    /// [`RuntimeReconfigureError::Edit`].
    pub fn reconfigure(
        &self,
        edit: impl FnOnce(&mut RuntimeReconfiguration<'_>) -> Result<(), RuntimeConfigError>,
    ) -> Result<RuntimeEpoch, RuntimeReconfigureError> {
        let base = self
            .snapshot()
            .map_err(|source| RuntimeReconfigureError::State { source })?;
        let mut candidate = CandidateConfig::from_snapshot(&base)
            .map_err(|source| RuntimeReconfigureError::Edit { source })?;
        let mut changed = false;
        {
            let mut reconfiguration = RuntimeReconfiguration {
                candidate: &mut candidate,
                changed: &mut changed,
            };
            edit(&mut reconfiguration)
                .map_err(|source| RuntimeReconfigureError::Edit { source })?;
        }

        if !changed {
            return Ok(base.epoch());
        }
        let next_identity_ordinal = NonZeroU64::new(
            self.0.next_registration_ordinal.load(Ordering::SeqCst),
        )
        .ok_or(RuntimeReconfigureError::Edit {
            source: RuntimeConfigError::IdentityExhausted,
        })?;
        let (bound_candidate, post_ordinal) =
            validate_candidate(candidate, self.0.issuer, next_identity_ordinal)
                .map_err(|source| RuntimeReconfigureError::Edit { source })?;

        let next_epoch =
            base.epoch()
                .checked_next()
                .ok_or(RuntimeReconfigureError::EpochExhausted {
                    current: base.epoch(),
                })?;

        let mut guard = self
            .0
            .active
            .write()
            .map_err(|_| RuntimeReconfigureError::State {
                source: RuntimeStateError::Poisoned {
                    lock: "runtime.active",
                },
            })?;
        if !Arc::ptr_eq(&*guard, &base) {
            return Err(RuntimeReconfigureError::ConcurrentReconfiguration {
                base: base.epoch(),
                current: guard.epoch(),
            });
        }

        let next_snapshot = Arc::new(
            freeze_candidate(self.0.runtime_id, next_epoch, bound_candidate)
                .map_err(|source| RuntimeReconfigureError::Edit { source })?,
        );

        self.0
            .next_registration_ordinal
            .store(post_ordinal.get(), Ordering::SeqCst);
        *guard = next_snapshot;
        self.0
            .published_epoch
            .store(next_epoch.get().get(), Ordering::Release);
        Ok(next_epoch)
    }

    #[cfg(test)]
    pub(crate) fn force_epoch_for_test(&self, epoch: RuntimeEpoch) {
        let mut guard = self.0.active.write().expect("test runtime lock");
        let mut replacement = (**guard).clone();
        replacement.epoch = epoch;
        *guard = Arc::new(replacement);
        self.0
            .published_epoch
            .store(epoch.get().get(), Ordering::Release);
    }

    #[cfg(test)]
    pub(crate) fn force_next_registration_ordinal_for_test(&self, next: NonZeroU64) {
        self.0
            .next_registration_ordinal
            .store(next.get(), Ordering::SeqCst);
    }

    #[cfg(test)]
    pub(crate) fn poison_active_lock_for_test(&self) {
        let state = Arc::clone(&self.0);
        let _ = std::panic::catch_unwind(std::panic::AssertUnwindSafe(move || {
            let _guard = state.active.write().expect("test runtime lock");
            panic!("poison runtime.active for test");
        }));
    }
}

impl fmt::Debug for Runtime {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter
            .debug_struct("Runtime")
            .field("runtime_id", &self.0.runtime_id)
            .field("published_epoch", &self.epoch().ok())
            .finish_non_exhaustive()
    }
}

/// Consuming builder for an immutable runtime configuration.
///
/// # Examples
///
/// ```
/// # fn main() -> Result<(), Box<dyn std::error::Error>> {
/// use tenferro_runtime::RuntimeConfigBuilder;
///
/// let runtime = RuntimeConfigBuilder::new().build()?;
/// assert_eq!(runtime.snapshot()?.engine_count(), 0);
/// # Ok(())
/// # }
/// ```
pub struct RuntimeConfigBuilder {
    candidate: CandidateConfig,
}

impl RuntimeConfigBuilder {
    /// Create an empty runtime builder.
    pub fn new() -> Self {
        Self {
            candidate: CandidateConfig::empty(),
        }
    }

    /// Replace the execution policy in the candidate configuration.
    pub fn execution_policy(&mut self, value: ExecutionPolicy) -> &mut Self {
        self.candidate.policy = value;
        self
    }

    /// Register a new engine candidate.
    ///
    /// # Errors
    ///
    /// Returns [`RuntimeConfigError::DuplicateEngine`] if a different candidate
    /// with the same engine ID is already present.
    pub fn register_engine(
        &mut self,
        value: EngineRegistration,
    ) -> Result<&mut Self, RuntimeConfigError> {
        let mut changed = false;
        register_engine_candidate(&mut self.candidate, value, &mut changed)?;
        Ok(self)
    }

    /// Explicitly replace an existing engine candidate.
    ///
    /// # Errors
    ///
    /// Returns [`RuntimeConfigError::MissingEngine`] if the engine ID is absent.
    pub fn replace_engine(
        &mut self,
        value: EngineRegistration,
    ) -> Result<&mut Self, RuntimeConfigError> {
        let mut changed = false;
        replace_engine_candidate(&mut self.candidate, value, &mut changed)?;
        Ok(self)
    }

    /// Remove an existing engine candidate.
    ///
    /// # Errors
    ///
    /// Returns [`RuntimeConfigError::MissingEngine`] if the engine ID is absent.
    pub fn remove_engine(&mut self, id: &EngineId) -> Result<&mut Self, RuntimeConfigError> {
        let mut changed = false;
        remove_engine_candidate(&mut self.candidate, id, &mut changed)?;
        Ok(self)
    }

    /// Install an extension module transaction.
    ///
    /// # Errors
    ///
    /// Returns [`RuntimeConfigError::ExtensionModule`] when module configuration
    /// fails or a distinct module already uses the same module ID.
    pub fn install_extension_module(
        &mut self,
        value: Arc<dyn ExtensionModule>,
    ) -> Result<&mut Self, RuntimeConfigError> {
        let mut changed = false;
        install_extension_module_candidate(&mut self.candidate, value, &mut changed)?;
        Ok(self)
    }

    /// Register a transfer provider keyed by source and destination endpoints.
    ///
    /// # Examples
    ///
    /// ```
    /// # fn main() -> Result<(), Box<dyn std::error::Error>> {
    /// use std::sync::Arc;
    /// use tenferro_runtime::{
    ///     assemble_preparation_only_engine_registration, CoreCapabilityBundle, EngineId,
    ///     EngineRegistration, EngineRegistrationMetadata, Error, ExecutionContextIdentity,
    ///     HardwareClassId, PreparationOnlyEngineRegistrationConfig, ProviderDeviceIdentity,
    ///     ProviderId, Runtime, StorageClass, TransferEndpoint, TransferProvider, TransferRequest,
    /// };
    ///
    /// #[derive(Debug)]
    /// struct ExampleProvider;
    ///
    /// impl TransferProvider for ExampleProvider {
    ///     fn transfer_blocking(
    ///         &self,
    ///         _request: TransferRequest<'_>,
    ///     ) -> tenferro_runtime::Result<tenferro_tensor::Tensor> {
    ///         Err(Error::Internal("the example does not execute a transfer".into()))
    ///     }
    /// }
    ///
    /// fn registration(
    ///     id: EngineId,
    ///     target: &str,
    ///     storage: &StorageClass,
    /// ) -> Result<EngineRegistration, tenferro_runtime::RuntimeConfigError> {
    ///     let metadata = EngineRegistrationMetadata::new(
    ///         id,
    ///         ProviderDeviceIdentity::new(ProviderId::new("example.provider")?, target)?,
    ///         HardwareClassId::new("example.hardware")?,
    ///         Arc::from([storage.clone()]),
    ///         storage.clone(),
    ///         CoreCapabilityBundle::default(),
    ///     );
    ///     assemble_preparation_only_engine_registration(
    ///         PreparationOnlyEngineRegistrationConfig::new(
    ///             metadata,
    ///             ExecutionContextIdentity::of::<()>(),
    ///         ),
    ///     )
    /// }
    ///
    /// let storage = StorageClass::new("example.storage.host")?;
    /// let source_id = EngineId::new("example.engine.source")?;
    /// let destination_id = EngineId::new("example.engine.destination")?;
    /// let source = registration(
    ///     source_id.clone(),
    ///     "source-0",
    ///     &storage,
    /// )?;
    /// let destination = registration(
    ///     destination_id.clone(),
    ///     "destination-0",
    ///     &storage,
    /// )?;
    /// let mut builder = Runtime::builder();
    /// builder.register_engine(source)?;
    /// builder.register_engine(destination)?;
    /// builder.register_transfer_provider(
    ///     TransferEndpoint::new(source_id, storage.clone()),
    ///     TransferEndpoint::new(destination_id, storage),
    ///     Arc::new(ExampleProvider),
    /// )?;
    /// let runtime = builder.build()?;
    /// assert_eq!(runtime.snapshot()?.transfer_provider_count(), 1);
    /// # Ok(())
    /// # }
    /// ```
    ///
    /// # Errors
    ///
    /// Returns [`RuntimeConfigError::ConflictingRegistration`] if a different
    /// provider is already registered for the same endpoint pair. The complete
    /// endpoint pair is validated when [`Self::build`] freezes the candidate.
    pub fn register_transfer_provider(
        &mut self,
        source: TransferEndpoint,
        destination: TransferEndpoint,
        provider: Arc<dyn TransferProvider>,
    ) -> Result<&mut Self, RuntimeConfigError> {
        let mut changed = false;
        register_transfer_provider_candidate(
            &mut self.candidate,
            source,
            destination,
            provider,
            &mut changed,
        )?;
        Ok(self)
    }

    /// Remove the transfer provider for an endpoint pair.
    ///
    /// This explicit removal is required before changing an engine's physical
    /// binding. The route can then be registered again against the replacement
    /// binding in the same candidate transaction.
    ///
    /// # Errors
    ///
    /// Returns [`RuntimeConfigError::MissingTransferProvider`] when the exact
    /// endpoint pair is not registered.
    pub fn remove_transfer_provider(
        &mut self,
        source: TransferEndpoint,
        destination: TransferEndpoint,
    ) -> Result<&mut Self, RuntimeConfigError> {
        let mut changed = false;
        remove_transfer_provider_candidate(&mut self.candidate, source, destination, &mut changed)?;
        Ok(self)
    }

    /// Replace an extension module transaction, installing it when absent.
    ///
    /// # Errors
    ///
    /// Returns [`RuntimeConfigError::ExtensionModule`] when module configuration
    /// fails.
    pub fn replace_extension_module(
        &mut self,
        value: Arc<dyn ExtensionModule>,
    ) -> Result<&mut Self, RuntimeConfigError> {
        let mut changed = false;
        replace_extension_module_candidate(&mut self.candidate, value, &mut changed)?;
        Ok(self)
    }

    /// Remove an extension module candidate if present.
    ///
    /// # Errors
    ///
    /// This method currently has no failing absent-module path; it returns
    /// [`RuntimeConfigError`] only for future validated module removal failures.
    pub fn remove_extension_module(
        &mut self,
        id: &ExtensionModuleId,
    ) -> Result<&mut Self, RuntimeConfigError> {
        let mut changed = false;
        remove_extension_module_candidate(&mut self.candidate, id, &mut changed)?;
        Ok(self)
    }

    /// Build and publish the initial runtime snapshot.
    ///
    /// # Errors
    ///
    /// Returns [`RuntimeConfigError::IdentityExhausted`] if runtime or
    /// registration identity allocation would wrap, or
    /// [`RuntimeConfigError::UnknownTransferEndpointEngine`] or
    /// [`RuntimeConfigError::UnsupportedTransferEndpointStorage`] if a
    /// registered transfer endpoint is invalid for the complete candidate.
    pub fn build(self) -> Result<Runtime, RuntimeConfigError> {
        let runtime_id = RuntimeId::from_nonzero(allocate_nonzero(&NEXT_RUNTIME_ID)?);
        let issuer = allocate_nonzero(&NEXT_REGISTRATION_ISSUER)?;
        let (bound_candidate, post_ordinal) =
            validate_candidate(self.candidate, issuer, INITIAL_REGISTRATION_ORDINAL)?;
        let epoch = RuntimeEpoch::one();
        let snapshot = Arc::new(freeze_candidate(runtime_id, epoch, bound_candidate)?);
        let state = RuntimeState {
            runtime_id,
            issuer,
            next_registration_ordinal: AtomicU64::new(post_ordinal.get()),
            active: RwLock::new(snapshot),
            published_epoch: AtomicU64::new(epoch.get().get()),
            caches: RuntimeCacheSet::new(PreparedPlanCacheLimits::default()),
        };
        Ok(Runtime(Arc::new(state)))
    }
}

impl Default for RuntimeConfigBuilder {
    fn default() -> Self {
        Self::new()
    }
}

impl fmt::Debug for RuntimeConfigBuilder {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter
            .debug_struct("RuntimeConfigBuilder")
            .field("execution_policy", &self.candidate.policy)
            .field("engine_count", &self.candidate.engines.len())
            .field("extension_module_count", &self.candidate.modules.len())
            .field("transfer_provider_count", &self.candidate.transfers.len())
            .finish_non_exhaustive()
    }
}

/// Non-escapable reconfiguration edit view.
pub struct RuntimeReconfiguration<'a> {
    candidate: &'a mut CandidateConfig,
    changed: &'a mut bool,
}

impl RuntimeReconfiguration<'_> {
    /// Replace the candidate execution policy.
    pub fn execution_policy(&mut self, policy: ExecutionPolicy) -> &mut Self {
        if self.candidate.policy != policy {
            self.candidate.policy = policy;
            *self.changed = true;
        }
        self
    }

    /// Register a new engine in this reconfiguration.
    ///
    /// # Errors
    ///
    /// Returns [`RuntimeConfigError::DuplicateEngine`] if a different candidate
    /// with the same engine ID is already present.
    pub fn register_engine(
        &mut self,
        value: EngineRegistration,
    ) -> Result<&mut Self, RuntimeConfigError> {
        register_engine_candidate(self.candidate, value, self.changed)?;
        Ok(self)
    }

    /// Replace an existing engine in this reconfiguration.
    ///
    /// # Errors
    ///
    /// Returns [`RuntimeConfigError::MissingEngine`] if the engine ID is absent.
    pub fn replace_engine(
        &mut self,
        value: EngineRegistration,
    ) -> Result<&mut Self, RuntimeConfigError> {
        replace_engine_candidate(self.candidate, value, self.changed)?;
        Ok(self)
    }

    /// Remove an existing engine in this reconfiguration.
    ///
    /// # Errors
    ///
    /// Returns [`RuntimeConfigError::MissingEngine`] if the engine ID is absent.
    pub fn remove_engine(&mut self, id: &EngineId) -> Result<&mut Self, RuntimeConfigError> {
        remove_engine_candidate(self.candidate, id, self.changed)?;
        Ok(self)
    }

    /// Install an extension module in this reconfiguration.
    ///
    /// # Errors
    ///
    /// Returns [`RuntimeConfigError::ExtensionModule`] when module configuration
    /// fails or a distinct module already uses the same module ID.
    pub fn install_extension_module(
        &mut self,
        value: Arc<dyn ExtensionModule>,
    ) -> Result<&mut Self, RuntimeConfigError> {
        install_extension_module_candidate(self.candidate, value, self.changed)?;
        Ok(self)
    }

    /// Register a transfer provider in this reconfiguration by endpoint pair.
    ///
    /// # Examples
    ///
    /// ```
    /// # fn main() -> Result<(), Box<dyn std::error::Error>> {
    /// use std::sync::Arc;
    /// use tenferro_runtime::{
    ///     assemble_preparation_only_engine_registration, CoreCapabilityBundle, EngineId,
    ///     EngineRegistration, EngineRegistrationMetadata, Error, ExecutionContextIdentity,
    ///     HardwareClassId, PreparationOnlyEngineRegistrationConfig, ProviderDeviceIdentity,
    ///     ProviderId, Runtime, StorageClass, TransferEndpoint, TransferProvider, TransferRequest,
    /// };
    ///
    /// #[derive(Debug)]
    /// struct ExampleProvider;
    ///
    /// impl TransferProvider for ExampleProvider {
    ///     fn transfer_blocking(
    ///         &self,
    ///         _request: TransferRequest<'_>,
    ///     ) -> tenferro_runtime::Result<tenferro_tensor::Tensor> {
    ///         Err(Error::Internal("the example does not execute a transfer".into()))
    ///     }
    /// }
    ///
    /// fn registration(
    ///     id: &str,
    ///     storage: &StorageClass,
    /// ) -> Result<EngineRegistration, tenferro_runtime::RuntimeConfigError> {
    ///     let metadata = EngineRegistrationMetadata::new(
    ///         EngineId::new(id)?,
    ///         ProviderDeviceIdentity::new(
    ///             ProviderId::new("example.provider")?,
    ///             format!("engine:{id}"),
    ///         )?,
    ///         HardwareClassId::new("example.hardware")?,
    ///         Arc::from([storage.clone()]),
    ///         storage.clone(),
    ///         CoreCapabilityBundle::default(),
    ///     );
    ///     Ok(assemble_preparation_only_engine_registration(
    ///         PreparationOnlyEngineRegistrationConfig::new(
    ///             metadata,
    ///             ExecutionContextIdentity::of::<()>(),
    ///         ),
    ///     )?)
    /// }
    ///
    /// let storage = StorageClass::new("example.storage.host")?;
    /// let source_id = EngineId::new("example.engine.source")?;
    /// let destination_id = EngineId::new("example.engine.destination")?;
    /// let mut builder = Runtime::builder();
    /// builder.register_engine(registration(source_id.as_str(), &storage)?)?;
    /// builder.register_engine(registration(destination_id.as_str(), &storage)?)?;
    /// let runtime = builder.build()?;
    /// runtime.reconfigure(|edit| {
    ///     edit.register_transfer_provider(
    ///         TransferEndpoint::new(source_id, storage.clone()),
    ///         TransferEndpoint::new(destination_id, storage),
    ///         Arc::new(ExampleProvider),
    ///     )?;
    ///     Ok(())
    /// })?;
    /// assert_eq!(runtime.snapshot()?.transfer_provider_count(), 1);
    /// # Ok(())
    /// # }
    /// ```
    ///
    /// # Errors
    ///
    /// Returns [`RuntimeConfigError::ConflictingRegistration`] if a different
    /// provider is already registered for the same endpoint pair. The complete
    /// endpoint pair is validated before this reconfiguration is published.
    pub fn register_transfer_provider(
        &mut self,
        source: TransferEndpoint,
        destination: TransferEndpoint,
        provider: Arc<dyn TransferProvider>,
    ) -> Result<&mut Self, RuntimeConfigError> {
        register_transfer_provider_candidate(
            self.candidate,
            source,
            destination,
            provider,
            self.changed,
        )?;
        Ok(self)
    }

    /// Remove the transfer provider for an endpoint pair in this
    /// reconfiguration.
    ///
    /// # Errors
    ///
    /// Returns [`RuntimeConfigError::MissingTransferProvider`] when the exact
    /// endpoint pair is not registered.
    pub fn remove_transfer_provider(
        &mut self,
        source: TransferEndpoint,
        destination: TransferEndpoint,
    ) -> Result<&mut Self, RuntimeConfigError> {
        remove_transfer_provider_candidate(self.candidate, source, destination, self.changed)?;
        Ok(self)
    }

    /// Replace an extension module in this reconfiguration, installing when
    /// absent.
    ///
    /// # Errors
    ///
    /// Returns [`RuntimeConfigError::ExtensionModule`] when module configuration
    /// fails.
    pub fn replace_extension_module(
        &mut self,
        value: Arc<dyn ExtensionModule>,
    ) -> Result<&mut Self, RuntimeConfigError> {
        replace_extension_module_candidate(self.candidate, value, self.changed)?;
        Ok(self)
    }

    /// Ensure an extension module owns the selected family and engine
    /// registration, installing it when absent.
    ///
    /// If the candidate already contains the incoming module ID with the exact
    /// `(family_id, engine_id)` registration, this owner-scoped operation is a
    /// no-op even when the incoming module is a fresh `Arc`. Otherwise it
    /// validates the incoming module before [`Runtime::reconfigure`] can
    /// publish it. A failed validation therefore leaves the previously
    /// published module and snapshot untouched.
    ///
    /// # Examples
    ///
    /// ```rust
    /// use std::sync::Arc;
    ///
    /// use tenferro_runtime::{
    ///     EngineId, ExtensionModule, ExtensionModuleError, ExtensionModuleId,
    ///     ExtensionModuleRegistrar, Runtime,
    /// };
    ///
    /// #[derive(Debug)]
    /// struct ExampleModule {
    ///     id: ExtensionModuleId,
    /// }
    ///
    /// impl ExtensionModule for ExampleModule {
    ///     fn module_id(&self) -> &ExtensionModuleId {
    ///         &self.id
    ///     }
    ///
    ///     fn configure(
    ///         &self,
    ///         _registrar: &mut ExtensionModuleRegistrar<'_>,
    ///     ) -> Result<(), ExtensionModuleError> {
    ///         Ok(())
    ///     }
    /// }
    ///
    /// let runtime = Runtime::builder().build()?;
    /// let engine = EngineId::new("example.engine.v1")?;
    /// let error = runtime
    ///     .reconfigure(|edit| {
    ///         edit.ensure_extension_module_for_engine(
    ///             Arc::new(ExampleModule {
    ///                 id: ExtensionModuleId::new("example.module.v1")?,
    ///             }),
    ///             "example.family.v1",
    ///             &engine,
    ///         )?;
    ///         Ok(())
    ///     })
    ///     .unwrap_err();
    /// let source = std::error::Error::source(&error).unwrap();
    /// assert!(source.to_string().contains("example.module.v1"));
    /// # Ok::<(), Box<dyn std::error::Error>>(())
    /// ```
    ///
    /// # Errors
    ///
    /// Returns [`RuntimeConfigError::MissingExtensionEngine`] when the
    /// configured module does not register `family_id` for `engine_id`, or
    /// [`RuntimeConfigError::ExtensionModule`] when module configuration fails.
    #[doc(hidden)]
    pub fn ensure_extension_module_for_engine(
        &mut self,
        value: Arc<dyn ExtensionModule>,
        family_id: &'static str,
        engine_id: &EngineId,
    ) -> Result<&mut Self, RuntimeConfigError> {
        let module_id = value.module_id().clone();
        if let Some(existing) = self.candidate.modules.get(&module_id)
            && existing
                .engines
                .contains_key(&(family_id, engine_id.clone()))
        {
            return Ok(self);
        }

        let record = configure_module(Arc::clone(&value))
            .map_err(|source| RuntimeConfigError::ExtensionModule { source })?;
        if !record.engines.contains_key(&(family_id, engine_id.clone())) {
            return Err(RuntimeConfigError::MissingExtensionEngine {
                module_id,
                family_id,
                engine_id: engine_id.clone(),
            });
        }
        self.candidate.modules.insert(module_id, record);
        *self.changed = true;
        Ok(self)
    }

    /// Replace one extension module only when it owns the selected family and
    /// engine registration.
    ///
    /// This owner-scoped operation validates the configured candidate before
    /// [`Runtime::reconfigure`] can publish it. A failed validation therefore
    /// leaves the previously published module and snapshot untouched.
    ///
    /// # Examples
    ///
    /// ```rust
    /// use std::sync::Arc;
    ///
    /// use tenferro_runtime::{
    ///     EngineId, ExtensionModule, ExtensionModuleError, ExtensionModuleId,
    ///     ExtensionModuleRegistrar, Runtime,
    /// };
    ///
    /// #[derive(Debug)]
    /// struct ExampleModule {
    ///     id: ExtensionModuleId,
    /// }
    ///
    /// impl ExtensionModule for ExampleModule {
    ///     fn module_id(&self) -> &ExtensionModuleId {
    ///         &self.id
    ///     }
    ///
    ///     fn configure(
    ///         &self,
    ///         _registrar: &mut ExtensionModuleRegistrar<'_>,
    ///     ) -> Result<(), ExtensionModuleError> {
    ///         Ok(())
    ///     }
    /// }
    ///
    /// let runtime = Runtime::builder().build()?;
    /// let engine = EngineId::new("example.engine.v1")?;
    /// let error = runtime
    ///     .reconfigure(|edit| {
    ///         edit.replace_extension_module_for_engine(
    ///             Arc::new(ExampleModule {
    ///                 id: ExtensionModuleId::new("example.module.v1")?,
    ///             }),
    ///             "example.family.v1",
    ///             &engine,
    ///         )?;
    ///         Ok(())
    ///     })
    ///     .unwrap_err();
    /// let source = std::error::Error::source(&error).unwrap();
    /// assert!(source.to_string().contains("example.module.v1"));
    /// # Ok::<(), Box<dyn std::error::Error>>(())
    /// ```
    ///
    /// # Errors
    ///
    /// Returns [`RuntimeConfigError::MissingExtensionEngine`] when the
    /// configured module does not register `family_id` for `engine_id`, or
    /// [`RuntimeConfigError::ExtensionModule`] when module configuration fails.
    #[doc(hidden)]
    pub fn replace_extension_module_for_engine(
        &mut self,
        value: Arc<dyn ExtensionModule>,
        family_id: &'static str,
        engine_id: &EngineId,
    ) -> Result<&mut Self, RuntimeConfigError> {
        let module_id = value.module_id().clone();
        if let Some(existing) = self.candidate.modules.get(&module_id)
            && existing.module_identical(&value)
        {
            if existing
                .engines
                .contains_key(&(family_id, engine_id.clone()))
            {
                return Ok(self);
            }
            return Err(RuntimeConfigError::MissingExtensionEngine {
                module_id,
                family_id,
                engine_id: engine_id.clone(),
            });
        }

        let record = configure_module(Arc::clone(&value))
            .map_err(|source| RuntimeConfigError::ExtensionModule { source })?;
        if !record.engines.contains_key(&(family_id, engine_id.clone())) {
            return Err(RuntimeConfigError::MissingExtensionEngine {
                module_id,
                family_id,
                engine_id: engine_id.clone(),
            });
        }
        self.candidate.modules.insert(module_id, record);
        *self.changed = true;
        Ok(self)
    }

    /// Remove an extension module if present.
    ///
    /// # Errors
    ///
    /// This method currently has no failing absent-module path; it returns
    /// [`RuntimeConfigError`] only for future validated module removal failures.
    pub fn remove_extension_module(
        &mut self,
        id: &ExtensionModuleId,
    ) -> Result<&mut Self, RuntimeConfigError> {
        remove_extension_module_candidate(self.candidate, id, self.changed)?;
        Ok(self)
    }
}

impl fmt::Debug for RuntimeReconfiguration<'_> {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter
            .debug_struct("RuntimeReconfiguration")
            .field("engine_count", &self.candidate.engines.len())
            .field("extension_module_count", &self.candidate.modules.len())
            .field("transfer_provider_count", &self.candidate.transfers.len())
            .field("changed", &*self.changed)
            .finish_non_exhaustive()
    }
}

/// Borrowed immutable view of one engine slot in a runtime snapshot.
///
/// # Examples
///
/// ```
/// use std::fmt::Debug;
/// use tenferro_runtime::EngineSnapshotView;
///
/// fn requires_debug<T: Debug>() {}
/// requires_debug::<EngineSnapshotView<'_>>();
/// ```
#[derive(Clone, Copy)]
pub struct EngineSnapshotView<'a> {
    slot: &'a FrozenEngineSlot,
}

impl<'a> EngineSnapshotView<'a> {
    /// Return the engine ID for this slot.
    pub fn engine_id(&self) -> &'a EngineId {
        self.slot.engine_id()
    }

    /// Return the immutable provider/device binding for this engine slot.
    ///
    /// # Examples
    ///
    /// ```
    /// # fn inspect(view: tenferro_runtime::EngineSnapshotView<'_>) {
    /// let _ = view.provider_device_identity();
    /// # }
    /// ```
    pub fn provider_device_identity(&self) -> &'a super::ProviderDeviceIdentity {
        self.slot.provider_device_identity()
    }

    /// Return the runtime-local registration identity for this slot.
    pub fn registration_identity(&self) -> RegistrationIdentity {
        self.slot.metadata().identity
    }

    /// Return the execution-context identity required by this slot.
    pub fn context_identity(&self) -> ExecutionContextIdentity {
        self.slot.context_identity()
    }

    /// Return this engine's runtime event domain.
    pub fn event_domain_id(&self) -> EventDomainId {
        self.slot.metadata().event_domain_id
    }

    pub(super) fn executable_witness(&self) -> Option<&'a Arc<ExecutableEngineSnapshot>> {
        self.slot.executable()
    }

    /// Return the hardware class for this slot.
    pub fn hardware_class(&self) -> &'a HardwareClassId {
        self.slot.hardware_class()
    }

    /// Return direct core capability slots for this engine.
    pub fn capabilities(&self) -> &'a CoreCapabilityBundle {
        self.slot.capabilities()
    }

    pub(super) fn storage_classes(&self) -> &'a [StorageClass] {
        self.slot.storage_classes()
    }

    pub(super) fn default_storage_class(&self) -> &'a StorageClass {
        self.slot.default_storage_class()
    }

    /// Return whether this engine's registered ingress accepts one physical
    /// input signature across its advertised storage classes.
    ///
    /// This narrow, doc-hidden query is used by eager extension owners before
    /// invoking a module factory. It includes placement, backend family, and
    /// allocation-domain admission rather than checking only tensor shape or
    /// dtype.
    ///
    /// # Examples
    ///
    /// ```rust
    /// use tenferro_runtime::{InputSignature, Runtime, Tensor, TensorRead};
    ///
    /// let engine = tenferro_cpu::CpuEngine::serial();
    /// let mut builder = Runtime::builder();
    /// builder.register_engine(tenferro_cpu::runtime_engine_registration(&engine)?)?;
    /// let runtime = builder.build()?;
    /// let engine_id = tenferro_cpu::runtime_engine_id()?;
    /// let snapshot = runtime.snapshot()?;
    /// let engine = snapshot.engine(&engine_id).unwrap();
    /// let tensor = Tensor::from_vec_col_major(vec![1], vec![1.0_f64])?;
    /// let reads = [TensorRead::from_tensor(&tensor)];
    /// let signature = InputSignature::from_reads(&reads)?;
    /// assert!(engine.accepts_input_signature(&signature.entries()[0]));
    /// # Ok::<(), Box<dyn std::error::Error>>(())
    /// ```
    #[doc(hidden)]
    pub fn accepts_input_signature(&self, input: &super::InputSignatureEntry) -> bool {
        self.slot
            .storage_classes()
            .iter()
            .any(|storage_class| self.accepts_input_signature_for_storage(input, storage_class))
    }

    pub(super) fn accepts_input_signature_for_storage(
        &self,
        input: &super::InputSignatureEntry,
        storage_class: &StorageClass,
    ) -> bool {
        self.slot
            .executable()
            .is_some_and(|snapshot| snapshot.accepts_input_signature(input, storage_class))
    }

    #[cfg(test)]
    pub(crate) fn has_execution_engine_for_test(&self) -> bool {
        self.slot.executable().is_some()
    }
}

impl fmt::Debug for EngineSnapshotView<'_> {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter
            .debug_struct("EngineSnapshotView")
            .field("engine_id", self.engine_id())
            .field("registration_identity", &self.registration_identity())
            .field("context_identity", &self.context_identity())
            .field("hardware_class", self.hardware_class())
            .field("capabilities", self.capabilities())
            .finish()
    }
}

fn default_execution_policy() -> ExecutionPolicy {
    ExecutionPolicy::new(super::Determinism::Fast, None, 0)
}

fn register_engine_candidate(
    candidate: &mut CandidateConfig,
    registration: EngineRegistration,
    changed: &mut bool,
) -> Result<(), RuntimeConfigError> {
    let engine_id = registration.engine_id().clone();
    match candidate.engines.get(&engine_id) {
        Some(existing) if existing.registration.candidate_identical(&registration) => Ok(()),
        Some(_) => Err(RuntimeConfigError::DuplicateEngine { engine_id }),
        None => {
            ensure_unique_provider_device_target(candidate, &registration)?;
            candidate.engines.insert(
                engine_id,
                CandidateEngineRecord {
                    registration,
                    identity: CandidateRegistrationIdentity::New,
                },
            );
            *changed = true;
            Ok(())
        }
    }
}

fn replace_engine_candidate(
    candidate: &mut CandidateConfig,
    registration: EngineRegistration,
    changed: &mut bool,
) -> Result<(), RuntimeConfigError> {
    let engine_id = registration.engine_id().clone();
    let Some(existing) = candidate.engines.get(&engine_id) else {
        return Err(RuntimeConfigError::MissingEngine { engine_id });
    };
    if existing.registration.candidate_identical(&registration) {
        return Ok(());
    }
    if existing.registration.provider_device_identity() != registration.provider_device_identity() {
        return Err(RuntimeConfigError::EngineTargetRebind {
            engine_id,
            current: existing.registration.provider_device_identity().clone(),
            replacement: registration.provider_device_identity().clone(),
        });
    }
    ensure_unique_provider_device_target_except(candidate, &registration, &engine_id)?;
    candidate.engines.insert(
        engine_id,
        CandidateEngineRecord {
            registration,
            identity: CandidateRegistrationIdentity::New,
        },
    );
    *changed = true;
    Ok(())
}

fn remove_engine_candidate(
    candidate: &mut CandidateConfig,
    id: &EngineId,
    changed: &mut bool,
) -> Result<(), RuntimeConfigError> {
    match candidate.engines.remove(id) {
        Some(_) => {
            *changed = true;
            Ok(())
        }
        None => Err(RuntimeConfigError::MissingEngine {
            engine_id: id.clone(),
        }),
    }
}

fn install_extension_module_candidate(
    candidate: &mut CandidateConfig,
    module: Arc<dyn ExtensionModule>,
    changed: &mut bool,
) -> Result<(), RuntimeConfigError> {
    let module_id = module.module_id().clone();
    match candidate.modules.get(&module_id) {
        Some(existing) if existing.module_identical(&module) => Ok(()),
        Some(_) => Err(RuntimeConfigError::ExtensionModule {
            source: ExtensionModuleError::ConflictingModule { module_id },
        }),
        None => {
            let record = configure_module(module)
                .map_err(|source| RuntimeConfigError::ExtensionModule { source })?;
            candidate.modules.insert(module_id, record);
            *changed = true;
            Ok(())
        }
    }
}

fn replace_extension_module_candidate(
    candidate: &mut CandidateConfig,
    module: Arc<dyn ExtensionModule>,
    changed: &mut bool,
) -> Result<(), RuntimeConfigError> {
    let module_id = module.module_id().clone();
    match candidate.modules.get(&module_id) {
        Some(existing) if existing.module_identical(&module) => Ok(()),
        _ => {
            let record = configure_module(module)
                .map_err(|source| RuntimeConfigError::ExtensionModule { source })?;
            candidate.modules.insert(module_id, record);
            *changed = true;
            Ok(())
        }
    }
}

fn remove_extension_module_candidate(
    candidate: &mut CandidateConfig,
    id: &ExtensionModuleId,
    changed: &mut bool,
) -> Result<(), RuntimeConfigError> {
    if candidate.modules.remove(id).is_some() {
        *changed = true;
    }
    Ok(())
}

fn register_transfer_provider_candidate(
    candidate: &mut CandidateConfig,
    source: TransferEndpoint,
    destination: TransferEndpoint,
    provider: Arc<dyn TransferProvider>,
    changed: &mut bool,
) -> Result<(), RuntimeConfigError> {
    let key = TransferRoute::new(source, destination);
    match candidate.transfers.get(&key) {
        Some(existing) if Arc::ptr_eq(&existing.provider, &provider) => Ok(()),
        Some(_) => Err(RuntimeConfigError::ConflictingRegistration {
            key: RegistrationKey::TransferProvider {
                source: key.source().clone(),
                destination: key.destination().clone(),
            },
        }),
        None => {
            candidate.transfers.insert(
                key,
                CandidateTransferRecord {
                    provider,
                    binding: CandidateTransferBinding::New,
                },
            );
            *changed = true;
            Ok(())
        }
    }
}

fn remove_transfer_provider_candidate(
    candidate: &mut CandidateConfig,
    source: TransferEndpoint,
    destination: TransferEndpoint,
    changed: &mut bool,
) -> Result<(), RuntimeConfigError> {
    let key = TransferRoute::new(source, destination);
    if candidate.transfers.remove(&key).is_none() {
        return Err(RuntimeConfigError::MissingTransferProvider {
            source_endpoint: key.source().clone(),
            destination: key.destination().clone(),
        });
    }
    *changed = true;
    Ok(())
}

fn ensure_unique_provider_device_target(
    candidate: &CandidateConfig,
    registration: &EngineRegistration,
) -> Result<(), RuntimeConfigError> {
    ensure_unique_provider_device_target_except(candidate, registration, registration.engine_id())
}

fn ensure_unique_provider_device_target_except(
    candidate: &CandidateConfig,
    registration: &EngineRegistration,
    ignored_engine_id: &EngineId,
) -> Result<(), RuntimeConfigError> {
    if let Some((first_engine_id, _)) = candidate.engines.iter().find(|(engine_id, record)| {
        *engine_id != ignored_engine_id
            && record.registration.provider_device_identity()
                == registration.provider_device_identity()
    }) {
        return Err(RuntimeConfigError::DuplicateProviderDeviceTarget {
            provider_device_identity: registration.provider_device_identity().clone(),
            first_engine_id: first_engine_id.clone(),
            duplicate_engine_id: registration.engine_id().clone(),
        });
    }
    Ok(())
}

fn validate_candidate(
    candidate: CandidateConfig,
    issuer: NonZeroU64,
    next_ordinal: NonZeroU64,
) -> Result<(BoundCandidateConfig, NonZeroU64), RuntimeConfigError> {
    let mut seen_targets = BTreeMap::<ProviderDeviceIdentity, EngineId>::new();
    for (engine_id, record) in &candidate.engines {
        if let Some(first_engine_id) = seen_targets.insert(
            record.registration.provider_device_identity().clone(),
            engine_id.clone(),
        ) {
            return Err(RuntimeConfigError::DuplicateProviderDeviceTarget {
                provider_device_identity: record.registration.provider_device_identity().clone(),
                first_engine_id,
                duplicate_engine_id: engine_id.clone(),
            });
        }
    }

    let mut bound_transfers = BTreeMap::new();
    for (route, record) in &candidate.transfers {
        let source_binding = validate_transfer_endpoint(&candidate, route.source())?;
        let destination_binding = validate_transfer_endpoint(&candidate, route.destination())?;
        let preserved = match &record.binding {
            CandidateTransferBinding::New => None,
            CandidateTransferBinding::Preserved {
                source,
                destination,
            } => Some((source, destination)),
        };
        if let Some((registered_source, registered_destination)) = preserved {
            if registered_source != &source_binding {
                return Err(RuntimeConfigError::StaleTransferRoute {
                    source_endpoint: route.source().clone(),
                    destination: route.destination().clone(),
                    endpoint: route.source().clone(),
                    registered: Box::new(registered_source.clone()),
                    current: Box::new(source_binding.clone()),
                });
            }
            if registered_destination != &destination_binding {
                return Err(RuntimeConfigError::StaleTransferRoute {
                    source_endpoint: route.source().clone(),
                    destination: route.destination().clone(),
                    endpoint: route.destination().clone(),
                    registered: Box::new(registered_destination.clone()),
                    current: Box::new(destination_binding.clone()),
                });
            }
        }
        bound_transfers.insert(
            route.clone(),
            BoundCandidateTransferRecord {
                provider: Arc::clone(&record.provider),
                source: source_binding,
                destination: destination_binding,
            },
        );
    }
    let mut seen = BTreeMap::<(ExtensionFamilyId, EngineId), ExtensionModuleId>::new();
    for (module_id, module) in &candidate.modules {
        for family_engine in module.engines.keys() {
            if seen
                .insert(
                    (family_engine.0, family_engine.1.clone()),
                    module_id.clone(),
                )
                .is_some()
            {
                return Err(RuntimeConfigError::ConflictingRegistration {
                    key: RegistrationKey::ExtensionEngine {
                        family: family_engine.0,
                        engine: family_engine.1.clone(),
                    },
                });
            }
        }
    }

    let CandidateConfig {
        policy,
        engines,
        modules,
        transfers: _,
    } = candidate;
    let mut allocator = RegistrationIdentityAllocator::new(issuer, next_ordinal);
    let engines = engines
        .into_iter()
        .map(|(engine_id, record)| {
            let identity = match record.identity {
                CandidateRegistrationIdentity::New => allocator.allocate()?,
                CandidateRegistrationIdentity::Preserved(identity) => identity,
            };
            Ok((
                engine_id,
                BoundCandidateEngineRecord {
                    registration: record.registration,
                    identity,
                },
            ))
        })
        .collect::<Result<BTreeMap<_, _>, RuntimeConfigError>>()?;
    let modules = modules
        .into_iter()
        .map(|(module_id, module)| {
            let mut allocate = || allocator.allocate();
            Ok((module_id, bind_candidate_module(module, &mut allocate)?))
        })
        .collect::<Result<BTreeMap<_, _>, RuntimeConfigError>>()?;
    Ok((
        BoundCandidateConfig {
            policy,
            engines,
            modules,
            transfers: bound_transfers,
        },
        allocator.next_ordinal(),
    ))
}

fn validate_transfer_endpoint(
    candidate: &CandidateConfig,
    endpoint: &TransferEndpoint,
) -> Result<ProviderDeviceIdentity, RuntimeConfigError> {
    let Some(engine) = candidate.engines.get(endpoint.engine_id()) else {
        return Err(RuntimeConfigError::UnknownTransferEndpointEngine {
            endpoint: endpoint.clone(),
        });
    };
    if !engine
        .registration
        .storage_classes()
        .contains(endpoint.storage_class())
    {
        return Err(RuntimeConfigError::UnsupportedTransferEndpointStorage {
            endpoint: endpoint.clone(),
        });
    }
    Ok(engine.registration.provider_device_identity().clone())
}

struct RegistrationIdentityAllocator {
    issuer: NonZeroU64,
    next: NonZeroU64,
}

impl RegistrationIdentityAllocator {
    fn new(issuer: NonZeroU64, next: NonZeroU64) -> Self {
        Self { issuer, next }
    }

    fn allocate(&mut self) -> Result<RegistrationIdentity, RuntimeConfigError> {
        let identity = RegistrationIdentity::new(self.issuer, self.next);
        let next = self
            .next
            .get()
            .checked_add(1)
            .and_then(NonZeroU64::new)
            .ok_or(RuntimeConfigError::IdentityExhausted)?;
        self.next = next;
        Ok(identity)
    }

    fn next_ordinal(&self) -> NonZeroU64 {
        self.next
    }
}

fn freeze_candidate(
    runtime_id: RuntimeId,
    epoch: RuntimeEpoch,
    candidate: BoundCandidateConfig,
) -> Result<RuntimeConfigSnapshot, RuntimeConfigError> {
    let mut engines = Vec::with_capacity(candidate.engines.len());
    let mut engine_indices = BTreeMap::new();
    let mut engine_locations = BTreeMap::new();
    let mut cache_owners = Vec::new();
    for (index, (engine_id, record)) in candidate.engines.into_iter().enumerate() {
        let BoundCandidateEngineRecord {
            registration,
            identity,
        } = record;
        let event_domain_id = EventDomainId::new(runtime_id, epoch, identity);
        let (state, candidate_token) = registration.into_state_and_token();
        let provider_device_identity = state.provider_device_identity().clone();
        let metadata = FrozenEngineMetadata {
            candidate_token,
            identity,
            event_domain_id,
        };
        let frozen = match state {
            EngineRegistrationState::PreparationOnly { binding } => {
                FrozenEngineSlot::PreparationOnly(Arc::new(PreparationOnlyEngineSnapshot {
                    metadata,
                    binding,
                }))
            }
            EngineRegistrationState::Executable(binding) => {
                FrozenEngineSlot::Executable(Arc::new(ExecutableEngineSnapshot {
                    metadata,
                    binding,
                }))
            }
        };
        engine_locations.insert(
            engine_id.clone(),
            (provider_device_identity, event_domain_id),
        );
        engine_indices.insert(engine_id, index);
        engines.push(frozen);
    }
    let extensions = freeze_extension_slots(candidate.modules)?;
    for (id, owner) in extensions.cache_owner_records() {
        cache_owners.push(FrozenCacheOwner {
            id,
            kind: FrozenCacheOwnerKind::Extension,
            owner,
        });
    }
    let mut transfers = BTreeMap::new();
    for (route, record) in candidate.transfers {
        let BoundCandidateTransferRecord {
            provider,
            source: source_binding,
            destination: destination_binding,
        } = record;
        let (_, source_event_domain_id) = bound_engine_location(&engine_locations, route.source())?;
        let (_, destination_event_domain_id) =
            bound_engine_location(&engine_locations, route.destination())?;
        let resolved_route = ResolvedTransferRoute::new(
            ResolvedTransferEndpoint::new(
                route.source().clone(),
                source_binding,
                *source_event_domain_id,
            ),
            ResolvedTransferEndpoint::new(
                route.destination().clone(),
                destination_binding,
                *destination_event_domain_id,
            ),
        );
        transfers.insert(resolved_route, provider);
    }
    Ok(RuntimeConfigSnapshot {
        runtime_id,
        epoch,
        policy: candidate.policy,
        engines: engines.into(),
        engine_indices,
        extensions,
        transfers: FrozenTransferRegistry::new(transfers),
        cache_owners: cache_owners.into(),
    })
}

fn bound_engine_location<'a>(
    locations: &'a BTreeMap<EngineId, (ProviderDeviceIdentity, EventDomainId)>,
    endpoint: &TransferEndpoint,
) -> Result<&'a (ProviderDeviceIdentity, EventDomainId), RuntimeConfigError> {
    locations
        .get(endpoint.engine_id())
        .ok_or_else(|| RuntimeConfigError::BoundCandidateInvariant {
            endpoint: endpoint.clone(),
        })
}

fn allocate_nonzero(counter: &AtomicU64) -> Result<NonZeroU64, RuntimeConfigError> {
    let value = counter
        .try_update(Ordering::SeqCst, Ordering::SeqCst, |next| {
            next.checked_add(1)
        })
        .map_err(|_| RuntimeConfigError::IdentityExhausted)?;
    NonZeroU64::new(value).ok_or(RuntimeConfigError::IdentityExhausted)
}

#[cfg(test)]
mod freeze_tests {
    use crate::{ProviderId, TransferRequest};

    use super::*;

    #[derive(Debug)]
    struct FreezeTestContext;

    #[derive(Debug)]
    struct FreezeTestProvider;

    impl TransferProvider for FreezeTestProvider {
        fn transfer_blocking(
            &self,
            _request: TransferRequest<'_>,
        ) -> crate::Result<tenferro_tensor::Tensor> {
            Err(crate::Error::Internal("freeze test provider".into()))
        }
    }

    fn registration(
        engine_id: &str,
        target: &str,
    ) -> Result<EngineRegistration, RuntimeConfigError> {
        let engine_id = EngineId::new(engine_id).map_err(RuntimeConfigError::from)?;
        let storage =
            StorageClass::new("tenferro.test.freeze.storage").map_err(RuntimeConfigError::from)?;
        Ok(EngineRegistration::preparation_only(
            super::super::ProviderPreparationBinding::new(
                engine_id,
                ProviderDeviceIdentity::new(
                    ProviderId::new("tenferro.test.freeze.provider")
                        .map_err(RuntimeConfigError::from)?,
                    target,
                )
                .map_err(RuntimeConfigError::from)?,
                ExecutionContextIdentity::of::<FreezeTestContext>(),
                HardwareClassId::new("tenferro.test.freeze.hardware")
                    .map_err(RuntimeConfigError::from)?,
                Arc::from(vec![storage.clone()]),
                storage,
                CoreCapabilityBundle::default(),
            )?,
        ))
    }

    fn candidate(binding: CandidateTransferBinding) -> Result<CandidateConfig, RuntimeConfigError> {
        let source_id =
            EngineId::new("tenferro.test.freeze.source").map_err(RuntimeConfigError::from)?;
        let destination_id =
            EngineId::new("tenferro.test.freeze.destination").map_err(RuntimeConfigError::from)?;
        let storage =
            StorageClass::new("tenferro.test.freeze.storage").map_err(RuntimeConfigError::from)?;
        let source_endpoint = TransferEndpoint::new(source_id.clone(), storage.clone());
        let destination_endpoint = TransferEndpoint::new(destination_id.clone(), storage);
        let mut candidate = CandidateConfig::empty();
        let mut changed = false;
        register_engine_candidate(
            &mut candidate,
            registration(source_id.as_str(), "freeze-source")?,
            &mut changed,
        )?;
        register_engine_candidate(
            &mut candidate,
            registration(destination_id.as_str(), "freeze-destination")?,
            &mut changed,
        )?;
        register_transfer_provider_candidate(
            &mut candidate,
            source_endpoint.clone(),
            destination_endpoint.clone(),
            Arc::new(FreezeTestProvider),
            &mut changed,
        )?;
        candidate
            .transfers
            .get_mut(&TransferRoute::new(source_endpoint, destination_endpoint))
            .expect("registered route")
            .binding = binding;
        Ok(candidate)
    }

    #[test]
    fn validation_owns_stale_route_rejection_and_bound_freeze_is_total() {
        let wrong_source = ProviderDeviceIdentity::new(
            ProviderId::new("tenferro.test.freeze.provider").unwrap(),
            "different-source",
        )
        .unwrap();
        let preserved = CandidateTransferBinding::Preserved {
            source: wrong_source,
            destination: ProviderDeviceIdentity::new(
                ProviderId::new("tenferro.test.freeze.provider").unwrap(),
                "freeze-destination",
            )
            .unwrap(),
        };
        let result = validate_candidate(
            candidate(preserved).unwrap(),
            NonZeroU64::new(1).unwrap(),
            NonZeroU64::new(1).unwrap(),
        );
        let error = match result {
            Ok(_) => panic!("candidate validation must reject stale preserved bindings"),
            Err(error) => error,
        };
        assert!(matches!(
            error,
            RuntimeConfigError::StaleTransferRoute { .. }
        ));

        let (bound, _) = validate_candidate(
            candidate(CandidateTransferBinding::New).unwrap(),
            NonZeroU64::new(1).unwrap(),
            NonZeroU64::new(1).unwrap(),
        )
        .expect("validation must produce a complete bound candidate");
        freeze_candidate(
            RuntimeId::from_nonzero(NonZeroU64::new(1).unwrap()),
            RuntimeEpoch::one(),
            bound,
        )
        .expect("a bound candidate must freeze without semantic route revalidation");
    }

    #[test]
    fn frozen_engine_slots_are_arc_sized() {
        let slot_size = std::mem::size_of::<FrozenEngineSlot>();
        let arc_size = std::mem::size_of::<Arc<()>>();

        assert!(
            slot_size <= 2 * arc_size,
            "frozen engine slots should keep immutable snapshot payloads behind Arc: slot_size={slot_size}, arc_size={arc_size}",
        );
    }
}