Skip to main content

tenferro_runtime/runtime/
snapshot.rs

1use std::collections::BTreeMap;
2use std::fmt;
3use std::num::NonZeroU64;
4use std::sync::atomic::{AtomicU64, Ordering};
5use std::sync::{Arc, RwLock};
6
7use tenferro_tensor::{Tensor, TensorValue};
8
9use crate::graph::CompiledGraph;
10use crate::program::FrozenProgram;
11
12use super::cache::{PreparedPlanCacheLimits, RuntimeCacheSet};
13use super::cache_owner::{FrozenCacheOwner, FrozenCacheOwnerKind};
14use super::engine_registration::{CandidateRegistrationToken, EngineRegistrationState};
15use super::execution;
16#[cfg(test)]
17use super::extension::ExtensionSlotFullForTest;
18use super::extension::{
19    bind_candidate_module, configure_module, freeze_extension_slots, BoundCandidateModuleRecord,
20    CandidateModuleRecord, CandidateRegistrationIdentity, ExtensionEngineSnapshotView,
21    ExtensionFamilyId, FrozenExtensionSlots,
22};
23use super::preparation::{PreparedEntryKey, PreparedProgram, PreparedProgramResult};
24use super::schedule::EventDomainId;
25use super::{
26    CacheOwnerId, CoreCapabilityBundle, EngineId, EngineRegistration, ExecutionContextIdentity,
27    ExecutionPolicy, ExtensionModule, ExtensionModuleError, ExtensionModuleId,
28    FrozenTransferRegistry, HardwareClassId, InputSignature, PrepareCapability, PrepareOptions,
29    ProviderDeviceIdentity, RegistrationIdentity, RegistrationKey, ResolvedTransferEndpoint,
30    ResolvedTransferRoute, RuntimeCacheError, RuntimeCacheStats, RuntimeConfigError, RuntimeEpoch,
31    RuntimeId, RuntimeReconfigureError, RuntimeStateError, StorageClass, TransferEndpoint,
32    TransferProvider, TransferRoute,
33};
34use crate::{Error, ErrorPhase};
35
36static NEXT_RUNTIME_ID: AtomicU64 = AtomicU64::new(1);
37static NEXT_REGISTRATION_ISSUER: AtomicU64 = AtomicU64::new(1);
38const INITIAL_REGISTRATION_ORDINAL: NonZeroU64 = NonZeroU64::MIN;
39
40#[derive(Clone, Debug)]
41struct CandidateEngineRecord {
42    registration: EngineRegistration,
43    identity: CandidateRegistrationIdentity,
44}
45
46#[derive(Clone, Debug)]
47struct BoundCandidateEngineRecord {
48    registration: EngineRegistration,
49    identity: RegistrationIdentity,
50}
51
52#[derive(Clone, Debug)]
53enum CandidateTransferBinding {
54    /// A route registered before its complete candidate can be validated.
55    New,
56    /// A route carried forward from a frozen snapshot.
57    Preserved {
58        source: ProviderDeviceIdentity,
59        destination: ProviderDeviceIdentity,
60    },
61}
62
63#[derive(Clone, Debug)]
64struct CandidateTransferRecord {
65    provider: Arc<dyn TransferProvider>,
66    binding: CandidateTransferBinding,
67}
68
69struct BoundCandidateTransferRecord {
70    provider: Arc<dyn TransferProvider>,
71    source: ProviderDeviceIdentity,
72    destination: ProviderDeviceIdentity,
73}
74
75#[derive(Clone, Debug)]
76struct CandidateConfig {
77    policy: ExecutionPolicy,
78    engines: BTreeMap<EngineId, CandidateEngineRecord>,
79    modules: BTreeMap<ExtensionModuleId, CandidateModuleRecord>,
80    transfers: BTreeMap<TransferRoute, CandidateTransferRecord>,
81}
82
83struct BoundCandidateConfig {
84    policy: ExecutionPolicy,
85    engines: BTreeMap<EngineId, BoundCandidateEngineRecord>,
86    modules: BTreeMap<ExtensionModuleId, BoundCandidateModuleRecord>,
87    transfers: BTreeMap<TransferRoute, BoundCandidateTransferRecord>,
88}
89
90impl CandidateConfig {
91    fn empty() -> Self {
92        Self {
93            policy: default_execution_policy(),
94            engines: BTreeMap::new(),
95            modules: BTreeMap::new(),
96            transfers: BTreeMap::new(),
97        }
98    }
99
100    fn from_snapshot(snapshot: &RuntimeConfigSnapshot) -> Result<Self, RuntimeConfigError> {
101        let engines = snapshot
102            .engines
103            .iter()
104            .map(|slot| {
105                let registration = slot.to_registration()?;
106                Ok((
107                    registration.engine_id().clone(),
108                    CandidateEngineRecord {
109                        registration,
110                        identity: CandidateRegistrationIdentity::Preserved(
111                            slot.metadata().identity,
112                        ),
113                    },
114                ))
115            })
116            .collect::<Result<BTreeMap<_, _>, RuntimeConfigError>>()?;
117        Ok(Self {
118            policy: snapshot.policy.clone(),
119            engines,
120            modules: snapshot.extensions.to_candidate_modules(),
121            transfers: snapshot
122                .transfers
123                .iter()
124                .map(|(resolved_route, provider)| {
125                    (
126                        TransferRoute::new(
127                            resolved_route.source().logical().clone(),
128                            resolved_route.destination().logical().clone(),
129                        ),
130                        CandidateTransferRecord {
131                            provider: Arc::clone(provider),
132                            binding: CandidateTransferBinding::Preserved {
133                                source: resolved_route.source().provider_device_identity().clone(),
134                                destination: resolved_route
135                                    .destination()
136                                    .provider_device_identity()
137                                    .clone(),
138                            },
139                        },
140                    )
141                })
142                .collect(),
143        })
144    }
145}
146
147#[derive(Clone, Debug)]
148struct FrozenEngineMetadata {
149    candidate_token: Arc<CandidateRegistrationToken>,
150    identity: RegistrationIdentity,
151    event_domain_id: EventDomainId,
152}
153
154#[derive(Clone)]
155struct PreparationOnlyEngineSnapshot {
156    metadata: FrozenEngineMetadata,
157    binding: super::ProviderPreparationBinding,
158}
159
160#[derive(Clone, Debug)]
161pub(super) struct ExecutableEngineSnapshot {
162    metadata: FrozenEngineMetadata,
163    binding: super::ProviderExecutableBinding,
164}
165
166#[derive(Clone)]
167enum FrozenEngineSlot {
168    PreparationOnly(Arc<PreparationOnlyEngineSnapshot>),
169    Executable(Arc<ExecutableEngineSnapshot>),
170}
171
172impl FrozenEngineSlot {
173    fn metadata(&self) -> &FrozenEngineMetadata {
174        match self {
175            Self::PreparationOnly(snapshot) => &snapshot.metadata,
176            Self::Executable(snapshot) => &snapshot.metadata,
177        }
178    }
179
180    fn provider_device_identity(&self) -> &ProviderDeviceIdentity {
181        match self {
182            Self::PreparationOnly(snapshot) => snapshot.binding.provider_device_identity(),
183            Self::Executable(snapshot) => snapshot.binding.contract().provider_device_identity(),
184        }
185    }
186
187    fn engine_id(&self) -> &EngineId {
188        match self {
189            Self::PreparationOnly(snapshot) => snapshot.binding.engine_id(),
190            Self::Executable(snapshot) => snapshot.binding.engine_id(),
191        }
192    }
193
194    fn hardware_class(&self) -> &HardwareClassId {
195        match self {
196            Self::PreparationOnly(snapshot) => snapshot.binding.hardware_class(),
197            Self::Executable(snapshot) => snapshot.binding.hardware_class(),
198        }
199    }
200
201    fn storage_classes(&self) -> &[StorageClass] {
202        match self {
203            Self::PreparationOnly(snapshot) => snapshot.binding.storage_classes(),
204            Self::Executable(snapshot) => snapshot.binding.storage_classes(),
205        }
206    }
207
208    fn default_storage_class(&self) -> &StorageClass {
209        match self {
210            Self::PreparationOnly(snapshot) => snapshot.binding.default_storage_class(),
211            Self::Executable(snapshot) => snapshot.binding.default_storage_class(),
212        }
213    }
214
215    fn context_identity(&self) -> ExecutionContextIdentity {
216        match self {
217            Self::PreparationOnly(snapshot) => snapshot.binding.context_identity(),
218            Self::Executable(snapshot) => snapshot.binding.contract().context_identity(),
219        }
220    }
221
222    fn capabilities(&self) -> &CoreCapabilityBundle {
223        match self {
224            Self::PreparationOnly(snapshot) => snapshot.binding.capabilities(),
225            Self::Executable(snapshot) => snapshot.binding.contract().capabilities(),
226        }
227    }
228
229    fn executable(&self) -> Option<&Arc<ExecutableEngineSnapshot>> {
230        match self {
231            Self::PreparationOnly(_) => None,
232            Self::Executable(snapshot) => Some(snapshot),
233        }
234    }
235}
236
237impl ExecutableEngineSnapshot {
238    pub(super) fn engine_id(&self) -> &EngineId {
239        self.binding.engine_id()
240    }
241
242    pub(super) fn event_domain_id(&self) -> EventDomainId {
243        self.metadata.event_domain_id
244    }
245
246    pub(super) fn provider_device_identity(&self) -> &ProviderDeviceIdentity {
247        self.binding.contract().provider_device_identity()
248    }
249
250    #[cfg(test)]
251    pub(super) fn context_identity(&self) -> ExecutionContextIdentity {
252        self.binding.contract().context_identity()
253    }
254
255    pub(super) fn executor(&self) -> &Arc<dyn super::execution::ErasedTensorBackendExecutor> {
256        self.binding.contract().executor()
257    }
258
259    pub(super) fn event_domain_driver(&self) -> &Arc<dyn super::EventDomainDriver> {
260        self.binding.contract().event_domain_driver()
261    }
262
263    #[cfg(test)]
264    pub(super) fn has_executor(&self) -> bool {
265        true
266    }
267
268    #[cfg(test)]
269    pub(super) fn has_event_domain_driver(&self) -> bool {
270        true
271    }
272
273    pub(super) fn accepts_input_placement(
274        &self,
275        placement: &tenferro_tensor::Placement,
276        storage_class: &StorageClass,
277    ) -> bool {
278        self.binding.storage_classes().contains(storage_class)
279            && self
280                .binding
281                .contract()
282                .accepts_input_placement(placement, storage_class)
283    }
284
285    pub(super) fn accepts_input_signature(
286        &self,
287        input: &super::InputSignatureEntry,
288        storage_class: &StorageClass,
289    ) -> bool {
290        self.binding.storage_classes().contains(storage_class)
291            && self
292                .binding
293                .contract()
294                .accepts_input_signature(input, storage_class)
295    }
296
297    pub(super) fn accepts_runtime_input(
298        &self,
299        input: &tenferro_tensor::TensorRead<'_>,
300        storage_class: &StorageClass,
301    ) -> bool {
302        self.binding.storage_classes().contains(storage_class)
303            && self
304                .binding
305                .contract()
306                .accepts_runtime_input(input, storage_class)
307    }
308
309    pub(super) fn owns_resident_tensor(
310        &self,
311        input: &tenferro_tensor::TensorRead<'_>,
312        storage_class: &StorageClass,
313    ) -> bool {
314        self.binding.storage_classes().contains(storage_class)
315            && self
316                .binding
317                .contract()
318                .owns_resident_tensor(input, storage_class)
319    }
320
321    #[cfg(test)]
322    pub(super) fn for_test(
323        engine_id: EngineId,
324        provider_device_identity: ProviderDeviceIdentity,
325        event_domain_id: EventDomainId,
326        storage_class: StorageClass,
327    ) -> Arc<Self> {
328        Self::for_test_with_driver(
329            engine_id,
330            provider_device_identity,
331            event_domain_id,
332            storage_class,
333            Arc::new(super::ImmediateEventDomainDriver::new()),
334        )
335    }
336
337    #[cfg(test)]
338    pub(super) fn for_test_with_driver(
339        engine_id: EngineId,
340        provider_device_identity: ProviderDeviceIdentity,
341        event_domain_id: EventDomainId,
342        storage_class: StorageClass,
343        event_domain_driver: Arc<dyn super::EventDomainDriver>,
344    ) -> Arc<Self> {
345        let ingress = super::InputIngressContract::new(
346            super::InputPlacementContract::new(|_, _| true),
347            super::InputSignatureContract::new(|_, _, _, _| true),
348            super::RuntimeInputContract::new(|_, _| true),
349            super::ResidentOutputContract::new(|_, _| true),
350        );
351        let contract = super::ExecutableEngineContract::new(
352            provider_device_identity,
353            CoreCapabilityBundle::default(),
354            tenferro_cpu::CpuBackend::new(),
355            event_domain_driver,
356            ingress,
357            None,
358        );
359        let binding = super::ProviderExecutableBinding::new(
360            engine_id,
361            HardwareClassId::new("tenferro.test.schedule.hardware").expect("test hardware class"),
362            Arc::from(vec![storage_class.clone()]),
363            storage_class,
364            contract,
365        )
366        .expect("test executable binding");
367        Arc::new(Self {
368            metadata: FrozenEngineMetadata {
369                candidate_token: Arc::new(CandidateRegistrationToken),
370                identity: event_domain_id.registration_identity(),
371                event_domain_id,
372            },
373            binding,
374        })
375    }
376}
377
378impl FrozenEngineSlot {
379    fn to_registration(&self) -> Result<EngineRegistration, RuntimeConfigError> {
380        let metadata = self.metadata();
381        let registration = match self {
382            Self::PreparationOnly(snapshot) => {
383                EngineRegistration::from_state(EngineRegistrationState::PreparationOnly {
384                    binding: snapshot.binding.clone(),
385                })
386            }
387            Self::Executable(snapshot) => EngineRegistration::from_state(
388                EngineRegistrationState::Executable(snapshot.binding.clone()),
389            ),
390        };
391        Ok(registration.with_candidate_token(Arc::clone(&metadata.candidate_token)))
392    }
393}
394
395impl fmt::Debug for FrozenEngineSlot {
396    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
397        let metadata = self.metadata();
398        formatter
399            .debug_struct("FrozenEngineSlot")
400            .field("engine_id", self.engine_id())
401            .field("registration_identity", &metadata.identity)
402            .field("event_domain_id", &metadata.event_domain_id)
403            .field("context_identity", &self.context_identity())
404            .field("hardware_class", self.hardware_class())
405            .field(
406                "state",
407                &match self {
408                    Self::PreparationOnly(_) => "preparation-only",
409                    Self::Executable(_) => "executable",
410                },
411            )
412            .finish_non_exhaustive()
413    }
414}
415
416/// Immutable runtime configuration snapshot.
417///
418/// A snapshot is readable after later reconfiguration because the runtime
419/// publishes a new `Arc` instead of mutating existing slots.
420///
421/// # Examples
422///
423/// ```
424/// # fn main() -> Result<(), Box<dyn std::error::Error>> {
425/// use tenferro_runtime::RuntimeConfigBuilder;
426///
427/// let runtime = RuntimeConfigBuilder::new().build()?;
428/// let snapshot = runtime.snapshot()?;
429/// assert_eq!(snapshot.engine_count(), 0);
430/// # Ok(())
431/// # }
432/// ```
433#[derive(Clone)]
434pub struct RuntimeConfigSnapshot {
435    runtime_id: RuntimeId,
436    epoch: RuntimeEpoch,
437    policy: ExecutionPolicy,
438    engines: Arc<[FrozenEngineSlot]>,
439    engine_indices: BTreeMap<EngineId, usize>,
440    extensions: FrozenExtensionSlots,
441    transfers: FrozenTransferRegistry,
442    cache_owners: Arc<[FrozenCacheOwner]>,
443}
444
445impl RuntimeConfigSnapshot {
446    /// Return the runtime identity that published this snapshot.
447    pub fn runtime_id(&self) -> RuntimeId {
448        self.runtime_id
449    }
450
451    /// Return the epoch for this immutable snapshot.
452    pub fn epoch(&self) -> RuntimeEpoch {
453        self.epoch
454    }
455
456    /// Return the execution policy captured by this snapshot.
457    pub fn execution_policy(&self) -> &ExecutionPolicy {
458        &self.policy
459    }
460
461    /// Return the number of direct engine slots.
462    pub fn engine_count(&self) -> usize {
463        self.engines.len()
464    }
465
466    /// Return the number of installed extension modules.
467    pub fn extension_module_count(&self) -> usize {
468        self.extensions.module_count()
469    }
470
471    /// Return the number of registered transfer providers.
472    pub fn transfer_provider_count(&self) -> usize {
473        self.transfers.len()
474    }
475
476    /// Return whether this snapshot contains an extension engine for a family.
477    #[doc(hidden)]
478    pub fn has_extension_family(&self, family_id: &'static str) -> bool {
479        self.extensions.has_family(family_id)
480    }
481
482    /// Return whether this snapshot contains an extension engine for one exact
483    /// family/engine pair.
484    ///
485    /// This narrow, doc-hidden query is the cross-crate eager-extension
486    /// contract used by `tenferro-ad` to validate an owner-selected module
487    /// without exposing module internals or widening [`ExtensionModule`].
488    ///
489    /// # Examples
490    ///
491    /// ```rust
492    /// use tenferro_runtime::{EngineId, Runtime};
493    ///
494    /// let runtime = Runtime::builder().build()?;
495    /// let snapshot = runtime.snapshot()?;
496    /// let engine = EngineId::new("example.engine")?;
497    /// assert!(!snapshot.has_extension_engine("example.family.v1", &engine));
498    /// # Ok::<(), Box<dyn std::error::Error>>(())
499    /// ```
500    #[doc(hidden)]
501    pub fn has_extension_engine(&self, family_id: &'static str, engine_id: &EngineId) -> bool {
502        self.extensions.has_engine(family_id, engine_id)
503    }
504
505    /// Return whether the exact validated module instance is installed.
506    ///
507    /// This narrow, doc-hidden query is the eager direct-bridge steady-state
508    /// no-op contract: it mirrors the `module_identical` no-op predicate of the
509    /// replace-extension-module edit without building a candidate
510    /// configuration.
511    #[doc(hidden)]
512    pub fn has_extension_module_identical(&self, module: &Arc<dyn ExtensionModule>) -> bool {
513        self.extensions.has_module_identical(module)
514    }
515
516    /// Return whether the installed module with this exact module ID owns the
517    /// given family/engine registration.
518    ///
519    /// Mirrors the owner-scoped ensure-extension-module edit's no-op predicate
520    /// on the published snapshot. The module ID must be the one that owns the
521    /// registration; a different module registering the same family/engine
522    /// pair does not satisfy this query.
523    #[doc(hidden)]
524    pub fn has_extension_module_engine(
525        &self,
526        module_id: &ExtensionModuleId,
527        family_id: &'static str,
528        engine_id: &EngineId,
529    ) -> bool {
530        self.extensions
531            .has_module_engine(module_id, family_id, engine_id)
532    }
533
534    /// Return an immutable view of a registered engine slot.
535    pub fn engine(&self, id: &EngineId) -> Option<EngineSnapshotView<'_>> {
536        self.engine_indices
537            .get(id)
538            .map(|&index| EngineSnapshotView {
539                slot: &self.engines[index],
540            })
541    }
542
543    #[cfg(test)]
544    pub(crate) fn engine_ids_for_test(&self) -> impl Iterator<Item = &EngineId> {
545        self.engines.iter().map(FrozenEngineSlot::engine_id)
546    }
547
548    #[cfg(test)]
549    pub(crate) fn transfer_routes_for_test(&self) -> impl Iterator<Item = &ResolvedTransferRoute> {
550        self.transfers.iter().map(|(route, _)| route)
551    }
552
553    pub(super) fn engine_views_for_preparation(
554        &self,
555    ) -> impl Iterator<Item = EngineSnapshotView<'_>> + '_ {
556        self.engines.iter().map(|slot| EngineSnapshotView { slot })
557    }
558
559    pub(super) fn extension_slot_for_preparation(
560        &self,
561        family_id: ExtensionFamilyId,
562        engine_id: &EngineId,
563    ) -> Option<ExtensionEngineSnapshotView<'_>> {
564        self.extensions.slot_for_preparation(family_id, engine_id)
565    }
566
567    pub(super) fn transfer_registry_for_preparation(&self) -> FrozenTransferRegistry {
568        self.transfers.clone()
569    }
570
571    #[cfg(test)]
572    pub(crate) fn extension_slots_for_test(
573        &self,
574    ) -> impl Iterator<
575        Item = (
576            &ExtensionModuleId,
577            ExtensionFamilyId,
578            &EngineId,
579            RegistrationIdentity,
580        ),
581    > {
582        self.extensions.slots_for_test()
583    }
584
585    #[cfg(test)]
586    pub(crate) fn extension_slot_identity_for_test(
587        &self,
588        family_id: ExtensionFamilyId,
589        engine_id: &EngineId,
590    ) -> Option<RegistrationIdentity> {
591        self.extensions.slot_identity_for_test(family_id, engine_id)
592    }
593
594    #[cfg(test)]
595    pub(crate) fn extension_slot_full_for_test(
596        &self,
597        family_id: ExtensionFamilyId,
598        engine_id: &EngineId,
599    ) -> Option<ExtensionSlotFullForTest<'_>> {
600        self.extensions.slot_full_for_test(family_id, engine_id)
601    }
602
603    #[cfg(test)]
604    pub(super) fn cache_owners_for_test(&self) -> &[FrozenCacheOwner] {
605        &self.cache_owners
606    }
607
608    pub(super) fn cache_owners_for_runtime(&self) -> &[FrozenCacheOwner] {
609        &self.cache_owners
610    }
611}
612
613impl fmt::Debug for RuntimeConfigSnapshot {
614    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
615        formatter
616            .debug_struct("RuntimeConfigSnapshot")
617            .field("runtime_id", &self.runtime_id)
618            .field("epoch", &self.epoch)
619            .field("execution_policy", &self.policy)
620            .field("engine_count", &self.engines.len())
621            .field("extension_module_count", &self.extensions.module_count())
622            .field("extension_engine_count", &self.extensions.engine_count())
623            .field("transfer_provider_count", &self.transfers.len())
624            .field("cache_owner_count", &self.cache_owners.len())
625            .finish_non_exhaustive()
626    }
627}
628
629struct RuntimeState {
630    runtime_id: RuntimeId,
631    issuer: NonZeroU64,
632    next_registration_ordinal: AtomicU64,
633    active: RwLock<Arc<RuntimeConfigSnapshot>>,
634    published_epoch: AtomicU64,
635    caches: RuntimeCacheSet<PreparedEntryKey, PreparedProgram>,
636}
637
638/// Runtime owner for immutable configuration snapshots.
639///
640/// # Examples
641///
642/// ```
643/// # fn main() -> Result<(), Box<dyn std::error::Error>> {
644/// use tenferro_runtime::Runtime;
645///
646/// let runtime = Runtime::builder().build()?;
647/// assert_eq!(runtime.snapshot()?.engine_count(), 0);
648/// # Ok(())
649/// # }
650/// ```
651#[derive(Clone)]
652pub struct Runtime(Arc<RuntimeState>);
653
654impl Runtime {
655    /// Return a consuming runtime configuration builder.
656    pub fn builder() -> RuntimeConfigBuilder {
657        RuntimeConfigBuilder::new()
658    }
659
660    /// Return this runtime's opaque identity.
661    pub fn id(&self) -> RuntimeId {
662        self.0.runtime_id
663    }
664
665    /// Clone the current immutable runtime snapshot.
666    ///
667    /// # Errors
668    ///
669    /// Returns [`RuntimeStateError::Poisoned`] when the active snapshot lock was
670    /// poisoned by another thread.
671    pub fn snapshot(&self) -> Result<Arc<RuntimeConfigSnapshot>, RuntimeStateError> {
672        self.0
673            .active
674            .read()
675            .map(|snapshot| Arc::clone(&snapshot))
676            .map_err(|_| RuntimeStateError::Poisoned {
677                lock: "runtime.active",
678            })
679    }
680
681    /// Return the currently published runtime epoch without locking the active
682    /// snapshot.
683    ///
684    /// # Errors
685    ///
686    /// Returns [`RuntimeStateError`] only if runtime state invariants have been
687    /// violated internally.
688    pub fn epoch(&self) -> Result<RuntimeEpoch, RuntimeStateError> {
689        match NonZeroU64::new(self.0.published_epoch.load(Ordering::Acquire)) {
690            Some(value) => Ok(RuntimeEpoch::from_nonzero(value)),
691            None => Err(RuntimeStateError::Poisoned {
692                lock: "runtime.published_epoch",
693            }),
694        }
695    }
696
697    /// Return current prepared-plan cache limits.
698    ///
699    /// # Examples
700    ///
701    /// ```
702    /// # fn main() -> Result<(), Box<dyn std::error::Error>> {
703    /// use tenferro_runtime::Runtime;
704    ///
705    /// let runtime = Runtime::builder().build()?;
706    /// assert_eq!(runtime.prepared_cache_limits()?, Default::default());
707    /// # Ok(())
708    /// # }
709    /// ```
710    ///
711    /// # Errors
712    ///
713    /// Returns [`RuntimeStateError`] when the runtime-owned prepared cache state
714    /// cannot be accessed.
715    pub fn prepared_cache_limits(&self) -> Result<PreparedPlanCacheLimits, RuntimeStateError> {
716        self.0.caches.prepared().limits()
717    }
718
719    /// Replace current prepared-plan cache limits and evict retained entries
720    /// until the new limits are satisfied.
721    ///
722    /// # Examples
723    ///
724    /// ```
725    /// # fn main() -> Result<(), Box<dyn std::error::Error>> {
726    /// use std::num::NonZeroUsize;
727    /// use tenferro_runtime::{PreparedPlanCacheLimits, Runtime};
728    ///
729    /// let runtime = Runtime::builder().build()?;
730    /// runtime.set_prepared_cache_limits(PreparedPlanCacheLimits {
731    ///     max_entries: NonZeroUsize::new(1).unwrap(),
732    ///     max_retained_bytes: NonZeroUsize::new(1024).unwrap(),
733    ///     max_in_flight_entries: NonZeroUsize::new(1).unwrap(),
734    ///     max_queued_distinct_keys: NonZeroUsize::new(1).unwrap(),
735    /// })?;
736    /// assert_eq!(runtime.prepared_cache_limits()?.max_entries.get(), 1);
737    /// # Ok(())
738    /// # }
739    /// ```
740    ///
741    /// # Errors
742    ///
743    /// Returns [`RuntimeStateError`] when the runtime-owned prepared cache state
744    /// cannot be accessed.
745    pub fn set_prepared_cache_limits(
746        &self,
747        limits: PreparedPlanCacheLimits,
748    ) -> Result<(), RuntimeStateError> {
749        self.0.caches.prepared().set_limits(limits)
750    }
751
752    /// Clear the runtime-owned prepared-plan cache.
753    ///
754    /// # Examples
755    ///
756    /// ```
757    /// # fn main() -> Result<(), Box<dyn std::error::Error>> {
758    /// use tenferro_runtime::Runtime;
759    ///
760    /// let runtime = Runtime::builder().build()?;
761    /// runtime.clear_prepared_cache()?;
762    /// # Ok(())
763    /// # }
764    /// ```
765    ///
766    /// # Errors
767    ///
768    /// Returns [`RuntimeStateError`] when the runtime-owned prepared cache state
769    /// cannot be accessed.
770    pub fn clear_prepared_cache(&self) -> Result<(), RuntimeStateError> {
771        self.0.caches.prepared().clear()
772    }
773
774    /// Return aggregate cache statistics for the runtime and registered cache
775    /// owners.
776    ///
777    /// # Examples
778    ///
779    /// ```
780    /// # fn main() -> Result<(), Box<dyn std::error::Error>> {
781    /// use tenferro_runtime::Runtime;
782    ///
783    /// let runtime = Runtime::builder().build()?;
784    /// assert_eq!(runtime.cache_stats()?.prepared_plans.entries, 0);
785    /// # Ok(())
786    /// # }
787    /// ```
788    ///
789    /// # Errors
790    ///
791    /// Returns [`RuntimeCacheError`] when the runtime cache or a registered
792    /// cache owner cannot report statistics.
793    pub fn cache_stats(&self) -> Result<RuntimeCacheStats, RuntimeCacheError> {
794        super::preparation::cache_stats(self, &self.0.caches)
795    }
796
797    /// Clear runtime-owned prepared plans and all registered engine/extension
798    /// cache owners.
799    ///
800    /// # Examples
801    ///
802    /// ```
803    /// # fn main() -> Result<(), Box<dyn std::error::Error>> {
804    /// use tenferro_runtime::Runtime;
805    ///
806    /// let runtime = Runtime::builder().build()?;
807    /// runtime.clear_caches()?;
808    /// # Ok(())
809    /// # }
810    /// ```
811    ///
812    /// # Errors
813    ///
814    /// Returns [`RuntimeCacheError`] when the runtime cache or a registered
815    /// cache owner cannot be cleared.
816    pub fn clear_caches(&self) -> Result<(), RuntimeCacheError> {
817        super::preparation::clear_caches(self, &self.0.caches)
818    }
819
820    #[allow(
821        dead_code,
822        reason = "Phase 5 graph execution consumes crate-private prepared programs"
823    )]
824    pub(crate) fn prepare_for(
825        &self,
826        frozen: &FrozenProgram,
827        signature: &InputSignature,
828        options: &PrepareOptions,
829    ) -> PreparedProgramResult<Arc<PreparedProgram>> {
830        super::preparation::prepare_for(self, &self.0.caches, frozen, signature, options)
831    }
832
833    pub(crate) fn prepare_compiled_for(
834        &self,
835        program: &CompiledGraph,
836        signature: &InputSignature,
837        options: &PrepareOptions,
838    ) -> PreparedProgramResult<Arc<PreparedProgram>> {
839        super::preparation::prepare_compiled_for(self, &self.0.caches, program, signature, options)
840    }
841
842    /// Resolve and prepare a single extension operation for immediate eager
843    /// execution against one exact engine, bypassing SemanticProgram planning
844    /// and the prepared-program cache.
845    ///
846    /// Provider selection is pinned to `engine_id` (the eager context's exact
847    /// engine); the op is prepared only when that engine is executable, owns
848    /// the family's extension slot, and accepts every input signature entry.
849    /// Capability resolution happens before any planning fields are built.
850    /// Returns [`PrepareCapability::Unsupported`] when `engine_id` is not
851    /// registered, is not executable, or has no extension slot for the op's
852    /// family so callers may fall back to the compiled path.
853    ///
854    /// # Examples
855    ///
856    /// ```
857    /// # fn main() -> Result<(), Box<dyn std::error::Error>> {
858    /// use std::any::Any;
859    /// use std::hash::Hasher;
860    /// use std::sync::Arc;
861    ///
862    /// use tenferro_cpu::CpuBackend;
863    /// use tenferro_ops::ExtensionShapeContext;
864    /// use tenferro_runtime::extension::ExtensionOp;
865    /// use tenferro_runtime::{
866    ///     ExtensionEngine, ExtensionModule, ExtensionModuleError, ExtensionModuleId,
867    ///     ExtensionModuleRegistrar, ExtensionPlanningConfig, InputSignature, PrepareCapability,
868    ///     PrepareError, PrepareOptions, Runtime, UnsupportedReason, Tensor, TensorRead,
869    /// };
870    /// use tenferro_tensor::DType;
871    ///
872    /// #[derive(Debug)]
873    /// struct Probe;
874    ///
875    /// impl ExtensionOp for Probe {
876    ///     fn family_id(&self) -> &'static str { "tenferro-tests.immediate-probe.v1" }
877    ///     fn payload_hash(&self, _hasher: &mut dyn Hasher) {}
878    ///     fn payload_eq(&self, other: &dyn ExtensionOp) -> bool { other.as_any().downcast_ref::<Self>().is_some() }
879    ///     fn clone_arc(&self) -> Arc<dyn ExtensionOp> { Arc::new(Self) }
880    ///     fn as_any(&self) -> &dyn Any { self }
881    ///     fn input_count(&self) -> usize { 1 }
882    ///     fn output_count(&self) -> usize { 1 }
883    ///     fn infer_output_meta(&self, ctx: &mut ExtensionShapeContext<'_>) -> tenferro_tensor::Result<Vec<(DType, Vec<tenferro_ops::SymDim>)>> {
884    ///         Ok(vec![(ctx.input_dtype(0)?, ctx.input_shape(0)?.to_vec())])
885    ///     }
886    /// }
887    ///
888    /// #[derive(Debug)]
889    /// struct Config;
890    ///
891    /// impl ExtensionPlanningConfig for Config {
892    ///     fn family_id(&self) -> &'static str { Probe.family_id() }
893    ///     fn as_any(&self) -> &dyn Any { self }
894    ///     fn payload_hash(&self, _state: &mut dyn Hasher) {}
895    ///     fn payload_eq(&self, other: &dyn ExtensionPlanningConfig) -> bool { other.as_any().downcast_ref::<Self>().is_some() }
896    ///     fn retained_bytes(&self) -> usize { 0 }
897    /// }
898    ///
899    /// #[derive(Debug)]
900    /// struct Module(ExtensionModuleId);
901    ///
902    /// impl ExtensionModule for Module {
903    ///     fn module_id(&self) -> &ExtensionModuleId { &self.0 }
904    ///     fn configure(&self, registrar: &mut ExtensionModuleRegistrar<'_>) -> Result<(), ExtensionModuleError> {
905    ///         let Ok(engine_id) = tenferro_cpu::runtime_engine_id() else {
906    ///             // Best-effort: an unresolvable engine id keeps the op unsupported.
907    ///             return Ok(());
908    ///         };
909    ///         registrar.register_engine(Arc::new(NoopEngine { engine_id: engine_id.clone() }))?;
910    ///         registrar.register_planning_config(engine_id, Arc::new(Config))?;
911    ///         Ok(())
912    ///     }
913    /// }
914    ///
915    /// #[derive(Debug)]
916    /// struct NoopEngine { engine_id: tenferro_runtime::EngineId }
917    ///
918    /// impl ExtensionEngine for NoopEngine {
919    ///     fn family_id(&self) -> &'static str { Probe.family_id() }
920    ///     fn engine_id(&self) -> &tenferro_runtime::EngineId { &self.engine_id }
921    ///     fn context_identity(&self) -> tenferro_runtime::ExecutionContextIdentity { tenferro_runtime::ExecutionContextIdentity::of::<CpuBackend>() }
922    ///     fn prepare(&self, _request: tenferro_runtime::ExtensionPrepareRequest<'_>) -> Result<PrepareCapability, PrepareError> {
923    ///         Ok(PrepareCapability::Unsupported(UnsupportedReason::Operation { operation: Probe.family_id() }))
924    ///     }
925    /// }
926    ///
927    /// let backend = CpuBackend::new();
928    /// let mut builder = Runtime::builder();
929    /// builder.register_engine(tenferro_cpu::runtime_engine_registration(&backend)?)?;
930    /// builder.install_extension_module(Arc::new(Module(ExtensionModuleId::new("tenferro-tests.immediate-probe.module")?)))?;
931    /// let runtime = builder.build()?;
932    /// let tensor = Tensor::from_vec_col_major(vec![1], vec![1.0_f64])?;
933    /// let signature = InputSignature::from_reads(&[TensorRead::from_tensor(&tensor)])?;
934    /// let engine_id = tenferro_cpu::runtime_engine_id()?;
935    /// let capability = runtime.prepare_extension_immediate(&engine_id, &Probe, &signature)?;
936    /// assert!(matches!(capability, PrepareCapability::Unsupported { .. }));
937    /// # Ok(())
938    /// # }
939    /// ```
940    ///
941    /// # Errors
942    ///
943    /// Returns [`crate::Error::RuntimeState`] when the snapshot cannot be read
944    /// or an engine's preparation fails, with the typed
945    /// [`PrepareError`]/[`RuntimeStateError`] source retained.
946    pub fn prepare_extension_immediate(
947        &self,
948        engine_id: &EngineId,
949        op: &dyn tenferro_ops::ext_op::ExtensionOp,
950        signature: &InputSignature,
951    ) -> crate::Result<PrepareCapability> {
952        super::preparation::prepare_extension_immediate(self, engine_id, op, signature).map_err(
953            |source| {
954                Error::runtime_state_source(
955                    "Runtime::prepare_extension_immediate",
956                    ErrorPhase::Execution,
957                    source,
958                )
959            },
960        )
961    }
962
963    /// Run a compiled graph synchronously with borrowed tensor inputs.
964    ///
965    /// The borrows remain valid until this call returns; this surface never
966    /// detaches work. Asynchronous [`Self::submit`] accepts only the owning
967    /// [`super::execution::ExecutionInputs`] package.
968    ///
969    /// # Examples
970    ///
971    /// ```
972    /// # fn main() -> Result<(), Box<dyn std::error::Error>> {
973    /// use tenferro_runtime::{Runtime, TracedTensor, GraphCompiler};
974    ///
975    /// let runtime = Runtime::builder().build()?;
976    /// let x = TracedTensor::from_vec_col_major(vec![1], vec![1.0_f64])?;
977    /// let program = GraphCompiler::new().compile(&x)?;
978    /// let error = runtime.run_compiled(&program, &[]).unwrap_err();
979    /// assert!(error.to_string().contains("no eligible engine"));
980    /// # Ok(())
981    /// # }
982    /// ```
983    ///
984    /// # Errors
985    ///
986    /// Returns [`crate::Error::UnboundPlaceholder`] when no explicit inputs are
987    /// supplied and a semantic input has no bound default tensor.
988    /// Returns [`crate::Error::GraphInputCountMismatch`],
989    /// [`crate::Error::PlaceholderDtypeMismatch`],
990    /// [`crate::Error::PlaceholderRankMismatch`],
991    /// [`crate::Error::PlaceholderShapeMismatch`], or
992    /// [`crate::Error::PlaceholderShapeBoundExceeded`] when ordered runtime
993    /// inputs do not match the compiled graph metadata.
994    /// Returns [`crate::Error::RuntimeState`] when runtime preparation, schedule
995    /// validation, snapshot access, stale epoch checks, or execution-bridge
996    /// resolution fails, including [`crate::PrepareError::NoInputIngress`]
997    /// when no engine accepts an input's physical backend/allocation domain,
998    /// [`crate::PrepareError::MissingTransferProvider`] when ingress cannot
999    /// reach its first scheduled consumer, a runtime with no eligible engine,
1000    /// or no execution bridge for the prepared engine. Backend execution may
1001    /// also return concrete backend variants such as
1002    /// [`crate::Error::Unsupported`], [`crate::Error::Validation`], or
1003    /// [`crate::Error::Extension`].
1004    pub fn run_compiled(
1005        &self,
1006        program: &CompiledGraph,
1007        inputs: &[&Tensor],
1008    ) -> crate::Result<Vec<Tensor>> {
1009        super::execution::run_compiled(self, program, inputs)
1010    }
1011
1012    /// Execute borrowed read-only inputs synchronously through retirement.
1013    ///
1014    /// Host/CPU providers may complete this call. Asynchronous device
1015    /// providers reject before admission and return the unchanged borrowed
1016    /// package through [`crate::ScopedSubmitRejected`].
1017    ///
1018    /// # Errors
1019    ///
1020    /// Returns [`crate::Error::Unsupported`] when the selected asynchronous
1021    /// provider cannot execute borrowed inputs synchronously, or
1022    /// [`crate::ScopedSubmitRejected`] when pre-admission validation fails.
1023    /// Provider execution failures are reported as
1024    /// [`crate::runtime::execution::ScopedExecutionOutcome::RetiredFailed`].
1025    pub fn execute_scoped_read_only<'env>(
1026        &self,
1027        program: &CompiledGraph,
1028        inputs: super::execution::ScopedReadInputs<'env>,
1029    ) -> std::result::Result<
1030        super::execution::ScopedExecutionOutcome<'env>,
1031        super::execution::ScopedSubmitRejected<'env>,
1032    > {
1033        super::execution::execute_scoped_read_only(self, program, inputs)
1034    }
1035
1036    /// Prepare a compiled graph for repeated execution with the same runtime.
1037    ///
1038    /// Preparation validates the supplied input metadata, selects a runtime
1039    /// engine, and caches the staged execution plan. Use [`Self::run_prepared`]
1040    /// for steady-state execution when the same compiled graph is run many
1041    /// times.
1042    ///
1043    /// # Errors
1044    ///
1045    /// Returns [`crate::Error::UnboundPlaceholder`] when no explicit inputs are
1046    /// supplied and a semantic input has no bound default tensor.
1047    /// Returns [`crate::Error::GraphInputCountMismatch`],
1048    /// [`crate::Error::PlaceholderDtypeMismatch`],
1049    /// [`crate::Error::PlaceholderRankMismatch`],
1050    /// [`crate::Error::PlaceholderShapeMismatch`], or
1051    /// [`crate::Error::PlaceholderShapeBoundExceeded`] when ordered runtime
1052    /// inputs do not match the compiled graph metadata.
1053    /// Returns [`crate::Error::RuntimeState`] when runtime preparation, schedule
1054    /// validation, snapshot access, stale epoch checks, or execution-bridge
1055    /// resolution fails, including [`crate::PrepareError::NoInputIngress`]
1056    /// when no engine accepts an input's physical backend/allocation domain,
1057    /// [`crate::PrepareError::MissingTransferProvider`] when ingress cannot
1058    /// reach its first scheduled consumer, a runtime with no eligible engine,
1059    /// or no execution bridge for the prepared engine.
1060    pub fn prepare_compiled(
1061        &self,
1062        program: &CompiledGraph,
1063        inputs: &[&Tensor],
1064    ) -> crate::Result<super::execution::PreparedCompiledGraph> {
1065        super::execution::prepare_compiled(self, program, inputs)
1066    }
1067
1068    /// Run a graph previously prepared by [`Self::prepare_compiled`].
1069    ///
1070    /// # Errors
1071    ///
1072    /// Returns metadata validation errors for incompatible inputs, a runtime
1073    /// state error with [`crate::InputIngressContractError`] as its typed source
1074    /// when an input's physical residency does not match the prepared ingress,
1075    /// or a runtime state error if the prepared handle belongs to a different
1076    /// runtime or a stale runtime epoch.
1077    pub fn run_prepared(
1078        &self,
1079        prepared: &super::execution::PreparedCompiledGraph,
1080        inputs: &[&Tensor],
1081    ) -> crate::Result<Vec<Tensor>> {
1082        super::execution::run_prepared(self, prepared, inputs)
1083    }
1084
1085    /// Submit a compiled graph for asynchronous runtime-owned execution.
1086    ///
1087    /// Dropping the returned handle detaches the observer without blocking.
1088    /// Use [`super::execution::ExecutionHandle::wait`] to observe completion.
1089    ///
1090    /// # Errors
1091    ///
1092    /// Returns the same [`crate::PrepareError::InputSignature`],
1093    /// [`crate::PrepareError::Specialization`],
1094    /// [`crate::PrepareError::NoEligibleEngine`],
1095    /// [`crate::PrepareError::NoInputIngress`], and
1096    /// [`crate::PrepareError::MissingTransferProvider`] failures as
1097    /// [`Self::run_compiled`] before the worker is submitted. Returns a runtime
1098    /// state error with [`crate::SubmissionError`] as its typed source if the
1099    /// operating system rejects worker creation after admission.
1100    pub fn submit(
1101        &self,
1102        program: &CompiledGraph,
1103        inputs: super::execution::ExecutionInputs,
1104    ) -> std::result::Result<super::execution::ExecutionHandle, super::execution::SubmitError> {
1105        super::execution::submit(self, program, inputs)
1106    }
1107
1108    /// Run a compiled graph and preserve lazy owned output views.
1109    ///
1110    /// # Examples
1111    ///
1112    /// ```
1113    /// # fn main() -> Result<(), Box<dyn std::error::Error>> {
1114    /// use tenferro_runtime::{Runtime, TracedTensor, GraphCompiler};
1115    ///
1116    /// let runtime = Runtime::builder().build()?;
1117    /// let x = TracedTensor::from_vec_col_major(vec![1], vec![1.0_f64])?;
1118    /// let program = GraphCompiler::new().compile(&x)?;
1119    /// let error = runtime.run_compiled_values(&program, &[]).unwrap_err();
1120    /// assert!(error.to_string().contains("no eligible engine"));
1121    /// # Ok(())
1122    /// # }
1123    /// ```
1124    ///
1125    /// # Errors
1126    ///
1127    /// Returns [`crate::Error::UnboundPlaceholder`] when no explicit inputs are
1128    /// supplied and a semantic input has no bound default tensor.
1129    /// Returns [`crate::Error::GraphInputCountMismatch`],
1130    /// [`crate::Error::PlaceholderDtypeMismatch`],
1131    /// [`crate::Error::PlaceholderRankMismatch`],
1132    /// [`crate::Error::PlaceholderShapeMismatch`], or
1133    /// [`crate::Error::PlaceholderShapeBoundExceeded`] when ordered runtime
1134    /// inputs do not match the compiled graph metadata.
1135    /// Returns [`crate::Error::RuntimeState`] when runtime preparation, schedule
1136    /// validation, snapshot access, stale epoch checks, or execution-bridge
1137    /// resolution fails, including [`crate::PrepareError::NoInputIngress`]
1138    /// when no engine accepts an input's physical backend/allocation domain,
1139    /// [`crate::PrepareError::MissingTransferProvider`] when ingress cannot
1140    /// reach its first scheduled consumer, a runtime with no eligible engine,
1141    /// or no execution bridge for the prepared engine. Backend execution may
1142    /// also return concrete backend variants such as
1143    /// [`crate::Error::Unsupported`], [`crate::Error::Validation`], or
1144    /// [`crate::Error::Extension`].
1145    pub fn run_compiled_values(
1146        &self,
1147        program: &CompiledGraph,
1148        inputs: &[&Tensor],
1149    ) -> crate::Result<Vec<TensorValue>> {
1150        super::execution::run_compiled_values(self, program, inputs)
1151    }
1152
1153    /// Transactionally edit and publish runtime configuration.
1154    ///
1155    /// No user callback runs while the publication lock is held. If another
1156    /// writer publishes over the same base snapshot, this call returns
1157    /// [`RuntimeReconfigureError::ConcurrentReconfiguration`] and publishes
1158    /// nothing.
1159    ///
1160    /// # Errors
1161    ///
1162    /// Returns [`RuntimeReconfigureError`] when state access, edit validation,
1163    /// identity allocation, epoch advancement, or compare-and-publish fails.
1164    /// Invalid transfer endpoints are reported as the typed
1165    /// [`RuntimeConfigError::UnknownTransferEndpointEngine`] or
1166    /// [`RuntimeConfigError::UnsupportedTransferEndpointStorage`] source of
1167    /// [`RuntimeReconfigureError::Edit`].
1168    pub fn reconfigure(
1169        &self,
1170        edit: impl FnOnce(&mut RuntimeReconfiguration<'_>) -> Result<(), RuntimeConfigError>,
1171    ) -> Result<RuntimeEpoch, RuntimeReconfigureError> {
1172        let base = self
1173            .snapshot()
1174            .map_err(|source| RuntimeReconfigureError::State { source })?;
1175        let mut candidate = CandidateConfig::from_snapshot(&base)
1176            .map_err(|source| RuntimeReconfigureError::Edit { source })?;
1177        let mut changed = false;
1178        {
1179            let mut reconfiguration = RuntimeReconfiguration {
1180                candidate: &mut candidate,
1181                changed: &mut changed,
1182            };
1183            edit(&mut reconfiguration)
1184                .map_err(|source| RuntimeReconfigureError::Edit { source })?;
1185        }
1186
1187        if !changed {
1188            return Ok(base.epoch());
1189        }
1190        let next_identity_ordinal = NonZeroU64::new(
1191            self.0.next_registration_ordinal.load(Ordering::SeqCst),
1192        )
1193        .ok_or(RuntimeReconfigureError::Edit {
1194            source: RuntimeConfigError::IdentityExhausted,
1195        })?;
1196        let (bound_candidate, post_ordinal) =
1197            validate_candidate(candidate, self.0.issuer, next_identity_ordinal)
1198                .map_err(|source| RuntimeReconfigureError::Edit { source })?;
1199
1200        let next_epoch =
1201            base.epoch()
1202                .checked_next()
1203                .ok_or(RuntimeReconfigureError::EpochExhausted {
1204                    current: base.epoch(),
1205                })?;
1206
1207        let mut guard = self
1208            .0
1209            .active
1210            .write()
1211            .map_err(|_| RuntimeReconfigureError::State {
1212                source: RuntimeStateError::Poisoned {
1213                    lock: "runtime.active",
1214                },
1215            })?;
1216        if !Arc::ptr_eq(&*guard, &base) {
1217            return Err(RuntimeReconfigureError::ConcurrentReconfiguration {
1218                base: base.epoch(),
1219                current: guard.epoch(),
1220            });
1221        }
1222
1223        let next_snapshot = Arc::new(
1224            freeze_candidate(self.0.runtime_id, next_epoch, bound_candidate)
1225                .map_err(|source| RuntimeReconfigureError::Edit { source })?,
1226        );
1227
1228        self.0
1229            .next_registration_ordinal
1230            .store(post_ordinal.get(), Ordering::SeqCst);
1231        *guard = next_snapshot;
1232        self.0
1233            .published_epoch
1234            .store(next_epoch.get().get(), Ordering::Release);
1235        Ok(next_epoch)
1236    }
1237
1238    #[cfg(test)]
1239    pub(crate) fn force_epoch_for_test(&self, epoch: RuntimeEpoch) {
1240        let mut guard = self.0.active.write().expect("test runtime lock");
1241        let mut replacement = (**guard).clone();
1242        replacement.epoch = epoch;
1243        *guard = Arc::new(replacement);
1244        self.0
1245            .published_epoch
1246            .store(epoch.get().get(), Ordering::Release);
1247    }
1248
1249    #[cfg(test)]
1250    pub(crate) fn force_next_registration_ordinal_for_test(&self, next: NonZeroU64) {
1251        self.0
1252            .next_registration_ordinal
1253            .store(next.get(), Ordering::SeqCst);
1254    }
1255
1256    #[cfg(test)]
1257    pub(crate) fn poison_active_lock_for_test(&self) {
1258        let state = Arc::clone(&self.0);
1259        let _ = std::panic::catch_unwind(std::panic::AssertUnwindSafe(move || {
1260            let _guard = state.active.write().expect("test runtime lock");
1261            panic!("poison runtime.active for test");
1262        }));
1263    }
1264}
1265
1266impl fmt::Debug for Runtime {
1267    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
1268        formatter
1269            .debug_struct("Runtime")
1270            .field("runtime_id", &self.0.runtime_id)
1271            .field("published_epoch", &self.epoch().ok())
1272            .finish_non_exhaustive()
1273    }
1274}
1275
1276/// Consuming builder for an immutable runtime configuration.
1277///
1278/// # Examples
1279///
1280/// ```
1281/// # fn main() -> Result<(), Box<dyn std::error::Error>> {
1282/// use tenferro_runtime::RuntimeConfigBuilder;
1283///
1284/// let runtime = RuntimeConfigBuilder::new().build()?;
1285/// assert_eq!(runtime.snapshot()?.engine_count(), 0);
1286/// # Ok(())
1287/// # }
1288/// ```
1289pub struct RuntimeConfigBuilder {
1290    candidate: CandidateConfig,
1291}
1292
1293impl RuntimeConfigBuilder {
1294    /// Create an empty runtime builder.
1295    pub fn new() -> Self {
1296        Self {
1297            candidate: CandidateConfig::empty(),
1298        }
1299    }
1300
1301    /// Replace the execution policy in the candidate configuration.
1302    pub fn execution_policy(&mut self, value: ExecutionPolicy) -> &mut Self {
1303        self.candidate.policy = value;
1304        self
1305    }
1306
1307    /// Register a new engine candidate.
1308    ///
1309    /// # Errors
1310    ///
1311    /// Returns [`RuntimeConfigError::DuplicateEngine`] if a different candidate
1312    /// with the same engine ID is already present.
1313    pub fn register_engine(
1314        &mut self,
1315        value: EngineRegistration,
1316    ) -> Result<&mut Self, RuntimeConfigError> {
1317        let mut changed = false;
1318        register_engine_candidate(&mut self.candidate, value, &mut changed)?;
1319        Ok(self)
1320    }
1321
1322    /// Explicitly replace an existing engine candidate.
1323    ///
1324    /// # Errors
1325    ///
1326    /// Returns [`RuntimeConfigError::MissingEngine`] if the engine ID is absent.
1327    pub fn replace_engine(
1328        &mut self,
1329        value: EngineRegistration,
1330    ) -> Result<&mut Self, RuntimeConfigError> {
1331        let mut changed = false;
1332        replace_engine_candidate(&mut self.candidate, value, &mut changed)?;
1333        Ok(self)
1334    }
1335
1336    /// Remove an existing engine candidate.
1337    ///
1338    /// # Errors
1339    ///
1340    /// Returns [`RuntimeConfigError::MissingEngine`] if the engine ID is absent.
1341    pub fn remove_engine(&mut self, id: &EngineId) -> Result<&mut Self, RuntimeConfigError> {
1342        let mut changed = false;
1343        remove_engine_candidate(&mut self.candidate, id, &mut changed)?;
1344        Ok(self)
1345    }
1346
1347    /// Install an extension module transaction.
1348    ///
1349    /// # Errors
1350    ///
1351    /// Returns [`RuntimeConfigError::ExtensionModule`] when module configuration
1352    /// fails or a distinct module already uses the same module ID.
1353    pub fn install_extension_module(
1354        &mut self,
1355        value: Arc<dyn ExtensionModule>,
1356    ) -> Result<&mut Self, RuntimeConfigError> {
1357        let mut changed = false;
1358        install_extension_module_candidate(&mut self.candidate, value, &mut changed)?;
1359        Ok(self)
1360    }
1361
1362    /// Register a transfer provider keyed by source and destination endpoints.
1363    ///
1364    /// # Examples
1365    ///
1366    /// ```
1367    /// # fn main() -> Result<(), Box<dyn std::error::Error>> {
1368    /// use std::sync::Arc;
1369    /// use tenferro_runtime::{
1370    ///     assemble_preparation_only_engine_registration, CoreCapabilityBundle, EngineId,
1371    ///     EngineRegistration, EngineRegistrationMetadata, Error, ExecutionContextIdentity,
1372    ///     HardwareClassId, PreparationOnlyEngineRegistrationConfig, ProviderDeviceIdentity,
1373    ///     ProviderId, Runtime, StorageClass, TransferEndpoint, TransferProvider, TransferRequest,
1374    /// };
1375    ///
1376    /// #[derive(Debug)]
1377    /// struct ExampleProvider;
1378    ///
1379    /// impl TransferProvider for ExampleProvider {
1380    ///     fn transfer_blocking(
1381    ///         &self,
1382    ///         _request: TransferRequest<'_>,
1383    ///     ) -> tenferro_runtime::Result<tenferro_tensor::Tensor> {
1384    ///         Err(Error::Internal("the example does not execute a transfer".into()))
1385    ///     }
1386    /// }
1387    ///
1388    /// fn registration(
1389    ///     id: EngineId,
1390    ///     target: &str,
1391    ///     storage: &StorageClass,
1392    /// ) -> Result<EngineRegistration, tenferro_runtime::RuntimeConfigError> {
1393    ///     let metadata = EngineRegistrationMetadata::new(
1394    ///         id,
1395    ///         ProviderDeviceIdentity::new(ProviderId::new("example.provider")?, target)?,
1396    ///         HardwareClassId::new("example.hardware")?,
1397    ///         Arc::from([storage.clone()]),
1398    ///         storage.clone(),
1399    ///         CoreCapabilityBundle::default(),
1400    ///     );
1401    ///     assemble_preparation_only_engine_registration(
1402    ///         PreparationOnlyEngineRegistrationConfig::new(
1403    ///             metadata,
1404    ///             ExecutionContextIdentity::of::<()>(),
1405    ///         ),
1406    ///     )
1407    /// }
1408    ///
1409    /// let storage = StorageClass::new("example.storage.host")?;
1410    /// let source_id = EngineId::new("example.engine.source")?;
1411    /// let destination_id = EngineId::new("example.engine.destination")?;
1412    /// let source = registration(
1413    ///     source_id.clone(),
1414    ///     "source-0",
1415    ///     &storage,
1416    /// )?;
1417    /// let destination = registration(
1418    ///     destination_id.clone(),
1419    ///     "destination-0",
1420    ///     &storage,
1421    /// )?;
1422    /// let mut builder = Runtime::builder();
1423    /// builder.register_engine(source)?;
1424    /// builder.register_engine(destination)?;
1425    /// builder.register_transfer_provider(
1426    ///     TransferEndpoint::new(source_id, storage.clone()),
1427    ///     TransferEndpoint::new(destination_id, storage),
1428    ///     Arc::new(ExampleProvider),
1429    /// )?;
1430    /// let runtime = builder.build()?;
1431    /// assert_eq!(runtime.snapshot()?.transfer_provider_count(), 1);
1432    /// # Ok(())
1433    /// # }
1434    /// ```
1435    ///
1436    /// # Errors
1437    ///
1438    /// Returns [`RuntimeConfigError::ConflictingRegistration`] if a different
1439    /// provider is already registered for the same endpoint pair. The complete
1440    /// endpoint pair is validated when [`Self::build`] freezes the candidate.
1441    pub fn register_transfer_provider(
1442        &mut self,
1443        source: TransferEndpoint,
1444        destination: TransferEndpoint,
1445        provider: Arc<dyn TransferProvider>,
1446    ) -> Result<&mut Self, RuntimeConfigError> {
1447        let mut changed = false;
1448        register_transfer_provider_candidate(
1449            &mut self.candidate,
1450            source,
1451            destination,
1452            provider,
1453            &mut changed,
1454        )?;
1455        Ok(self)
1456    }
1457
1458    /// Remove the transfer provider for an endpoint pair.
1459    ///
1460    /// This explicit removal is required before changing an engine's physical
1461    /// binding. The route can then be registered again against the replacement
1462    /// binding in the same candidate transaction.
1463    ///
1464    /// # Errors
1465    ///
1466    /// Returns [`RuntimeConfigError::MissingTransferProvider`] when the exact
1467    /// endpoint pair is not registered.
1468    pub fn remove_transfer_provider(
1469        &mut self,
1470        source: TransferEndpoint,
1471        destination: TransferEndpoint,
1472    ) -> Result<&mut Self, RuntimeConfigError> {
1473        let mut changed = false;
1474        remove_transfer_provider_candidate(&mut self.candidate, source, destination, &mut changed)?;
1475        Ok(self)
1476    }
1477
1478    /// Replace an extension module transaction, installing it when absent.
1479    ///
1480    /// # Errors
1481    ///
1482    /// Returns [`RuntimeConfigError::ExtensionModule`] when module configuration
1483    /// fails.
1484    pub fn replace_extension_module(
1485        &mut self,
1486        value: Arc<dyn ExtensionModule>,
1487    ) -> Result<&mut Self, RuntimeConfigError> {
1488        let mut changed = false;
1489        replace_extension_module_candidate(&mut self.candidate, value, &mut changed)?;
1490        Ok(self)
1491    }
1492
1493    /// Remove an extension module candidate if present.
1494    ///
1495    /// # Errors
1496    ///
1497    /// This method currently has no failing absent-module path; it returns
1498    /// [`RuntimeConfigError`] only for future validated module removal failures.
1499    pub fn remove_extension_module(
1500        &mut self,
1501        id: &ExtensionModuleId,
1502    ) -> Result<&mut Self, RuntimeConfigError> {
1503        let mut changed = false;
1504        remove_extension_module_candidate(&mut self.candidate, id, &mut changed)?;
1505        Ok(self)
1506    }
1507
1508    /// Build and publish the initial runtime snapshot.
1509    ///
1510    /// # Errors
1511    ///
1512    /// Returns [`RuntimeConfigError::IdentityExhausted`] if runtime or
1513    /// registration identity allocation would wrap, or
1514    /// [`RuntimeConfigError::UnknownTransferEndpointEngine`] or
1515    /// [`RuntimeConfigError::UnsupportedTransferEndpointStorage`] if a
1516    /// registered transfer endpoint is invalid for the complete candidate.
1517    pub fn build(self) -> Result<Runtime, RuntimeConfigError> {
1518        let runtime_id = RuntimeId::from_nonzero(allocate_nonzero(&NEXT_RUNTIME_ID)?);
1519        let issuer = allocate_nonzero(&NEXT_REGISTRATION_ISSUER)?;
1520        let (bound_candidate, post_ordinal) =
1521            validate_candidate(self.candidate, issuer, INITIAL_REGISTRATION_ORDINAL)?;
1522        let epoch = RuntimeEpoch::one();
1523        let snapshot = Arc::new(freeze_candidate(runtime_id, epoch, bound_candidate)?);
1524        let state = RuntimeState {
1525            runtime_id,
1526            issuer,
1527            next_registration_ordinal: AtomicU64::new(post_ordinal.get()),
1528            active: RwLock::new(snapshot),
1529            published_epoch: AtomicU64::new(epoch.get().get()),
1530            caches: RuntimeCacheSet::new(PreparedPlanCacheLimits::default()),
1531        };
1532        Ok(Runtime(Arc::new(state)))
1533    }
1534}
1535
1536impl Default for RuntimeConfigBuilder {
1537    fn default() -> Self {
1538        Self::new()
1539    }
1540}
1541
1542impl fmt::Debug for RuntimeConfigBuilder {
1543    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
1544        formatter
1545            .debug_struct("RuntimeConfigBuilder")
1546            .field("execution_policy", &self.candidate.policy)
1547            .field("engine_count", &self.candidate.engines.len())
1548            .field("extension_module_count", &self.candidate.modules.len())
1549            .field("transfer_provider_count", &self.candidate.transfers.len())
1550            .finish_non_exhaustive()
1551    }
1552}
1553
1554/// Non-escapable reconfiguration edit view.
1555pub struct RuntimeReconfiguration<'a> {
1556    candidate: &'a mut CandidateConfig,
1557    changed: &'a mut bool,
1558}
1559
1560impl RuntimeReconfiguration<'_> {
1561    /// Replace the candidate execution policy.
1562    pub fn execution_policy(&mut self, policy: ExecutionPolicy) -> &mut Self {
1563        if self.candidate.policy != policy {
1564            self.candidate.policy = policy;
1565            *self.changed = true;
1566        }
1567        self
1568    }
1569
1570    /// Register a new engine in this reconfiguration.
1571    ///
1572    /// # Errors
1573    ///
1574    /// Returns [`RuntimeConfigError::DuplicateEngine`] if a different candidate
1575    /// with the same engine ID is already present.
1576    pub fn register_engine(
1577        &mut self,
1578        value: EngineRegistration,
1579    ) -> Result<&mut Self, RuntimeConfigError> {
1580        register_engine_candidate(self.candidate, value, self.changed)?;
1581        Ok(self)
1582    }
1583
1584    /// Replace an existing engine in this reconfiguration.
1585    ///
1586    /// # Errors
1587    ///
1588    /// Returns [`RuntimeConfigError::MissingEngine`] if the engine ID is absent.
1589    pub fn replace_engine(
1590        &mut self,
1591        value: EngineRegistration,
1592    ) -> Result<&mut Self, RuntimeConfigError> {
1593        replace_engine_candidate(self.candidate, value, self.changed)?;
1594        Ok(self)
1595    }
1596
1597    /// Remove an existing engine in this reconfiguration.
1598    ///
1599    /// # Errors
1600    ///
1601    /// Returns [`RuntimeConfigError::MissingEngine`] if the engine ID is absent.
1602    pub fn remove_engine(&mut self, id: &EngineId) -> Result<&mut Self, RuntimeConfigError> {
1603        remove_engine_candidate(self.candidate, id, self.changed)?;
1604        Ok(self)
1605    }
1606
1607    /// Install an extension module in this reconfiguration.
1608    ///
1609    /// # Errors
1610    ///
1611    /// Returns [`RuntimeConfigError::ExtensionModule`] when module configuration
1612    /// fails or a distinct module already uses the same module ID.
1613    pub fn install_extension_module(
1614        &mut self,
1615        value: Arc<dyn ExtensionModule>,
1616    ) -> Result<&mut Self, RuntimeConfigError> {
1617        install_extension_module_candidate(self.candidate, value, self.changed)?;
1618        Ok(self)
1619    }
1620
1621    /// Register a transfer provider in this reconfiguration by endpoint pair.
1622    ///
1623    /// # Examples
1624    ///
1625    /// ```
1626    /// # fn main() -> Result<(), Box<dyn std::error::Error>> {
1627    /// use std::sync::Arc;
1628    /// use tenferro_runtime::{
1629    ///     assemble_preparation_only_engine_registration, CoreCapabilityBundle, EngineId,
1630    ///     EngineRegistration, EngineRegistrationMetadata, Error, ExecutionContextIdentity,
1631    ///     HardwareClassId, PreparationOnlyEngineRegistrationConfig, ProviderDeviceIdentity,
1632    ///     ProviderId, Runtime, StorageClass, TransferEndpoint, TransferProvider, TransferRequest,
1633    /// };
1634    ///
1635    /// #[derive(Debug)]
1636    /// struct ExampleProvider;
1637    ///
1638    /// impl TransferProvider for ExampleProvider {
1639    ///     fn transfer_blocking(
1640    ///         &self,
1641    ///         _request: TransferRequest<'_>,
1642    ///     ) -> tenferro_runtime::Result<tenferro_tensor::Tensor> {
1643    ///         Err(Error::Internal("the example does not execute a transfer".into()))
1644    ///     }
1645    /// }
1646    ///
1647    /// fn registration(
1648    ///     id: &str,
1649    ///     storage: &StorageClass,
1650    /// ) -> Result<EngineRegistration, tenferro_runtime::RuntimeConfigError> {
1651    ///     let metadata = EngineRegistrationMetadata::new(
1652    ///         EngineId::new(id)?,
1653    ///         ProviderDeviceIdentity::new(
1654    ///             ProviderId::new("example.provider")?,
1655    ///             format!("engine:{id}"),
1656    ///         )?,
1657    ///         HardwareClassId::new("example.hardware")?,
1658    ///         Arc::from([storage.clone()]),
1659    ///         storage.clone(),
1660    ///         CoreCapabilityBundle::default(),
1661    ///     );
1662    ///     Ok(assemble_preparation_only_engine_registration(
1663    ///         PreparationOnlyEngineRegistrationConfig::new(
1664    ///             metadata,
1665    ///             ExecutionContextIdentity::of::<()>(),
1666    ///         ),
1667    ///     )?)
1668    /// }
1669    ///
1670    /// let storage = StorageClass::new("example.storage.host")?;
1671    /// let source_id = EngineId::new("example.engine.source")?;
1672    /// let destination_id = EngineId::new("example.engine.destination")?;
1673    /// let mut builder = Runtime::builder();
1674    /// builder.register_engine(registration(source_id.as_str(), &storage)?)?;
1675    /// builder.register_engine(registration(destination_id.as_str(), &storage)?)?;
1676    /// let runtime = builder.build()?;
1677    /// runtime.reconfigure(|edit| {
1678    ///     edit.register_transfer_provider(
1679    ///         TransferEndpoint::new(source_id, storage.clone()),
1680    ///         TransferEndpoint::new(destination_id, storage),
1681    ///         Arc::new(ExampleProvider),
1682    ///     )?;
1683    ///     Ok(())
1684    /// })?;
1685    /// assert_eq!(runtime.snapshot()?.transfer_provider_count(), 1);
1686    /// # Ok(())
1687    /// # }
1688    /// ```
1689    ///
1690    /// # Errors
1691    ///
1692    /// Returns [`RuntimeConfigError::ConflictingRegistration`] if a different
1693    /// provider is already registered for the same endpoint pair. The complete
1694    /// endpoint pair is validated before this reconfiguration is published.
1695    pub fn register_transfer_provider(
1696        &mut self,
1697        source: TransferEndpoint,
1698        destination: TransferEndpoint,
1699        provider: Arc<dyn TransferProvider>,
1700    ) -> Result<&mut Self, RuntimeConfigError> {
1701        register_transfer_provider_candidate(
1702            self.candidate,
1703            source,
1704            destination,
1705            provider,
1706            self.changed,
1707        )?;
1708        Ok(self)
1709    }
1710
1711    /// Remove the transfer provider for an endpoint pair in this
1712    /// reconfiguration.
1713    ///
1714    /// # Errors
1715    ///
1716    /// Returns [`RuntimeConfigError::MissingTransferProvider`] when the exact
1717    /// endpoint pair is not registered.
1718    pub fn remove_transfer_provider(
1719        &mut self,
1720        source: TransferEndpoint,
1721        destination: TransferEndpoint,
1722    ) -> Result<&mut Self, RuntimeConfigError> {
1723        remove_transfer_provider_candidate(self.candidate, source, destination, self.changed)?;
1724        Ok(self)
1725    }
1726
1727    /// Replace an extension module in this reconfiguration, installing when
1728    /// absent.
1729    ///
1730    /// # Errors
1731    ///
1732    /// Returns [`RuntimeConfigError::ExtensionModule`] when module configuration
1733    /// fails.
1734    pub fn replace_extension_module(
1735        &mut self,
1736        value: Arc<dyn ExtensionModule>,
1737    ) -> Result<&mut Self, RuntimeConfigError> {
1738        replace_extension_module_candidate(self.candidate, value, self.changed)?;
1739        Ok(self)
1740    }
1741
1742    /// Ensure an extension module owns the selected family and engine
1743    /// registration, installing it when absent.
1744    ///
1745    /// If the candidate already contains the incoming module ID with the exact
1746    /// `(family_id, engine_id)` registration, this owner-scoped operation is a
1747    /// no-op even when the incoming module is a fresh `Arc`. Otherwise it
1748    /// validates the incoming module before [`Runtime::reconfigure`] can
1749    /// publish it. A failed validation therefore leaves the previously
1750    /// published module and snapshot untouched.
1751    ///
1752    /// # Examples
1753    ///
1754    /// ```rust
1755    /// use std::sync::Arc;
1756    ///
1757    /// use tenferro_runtime::{
1758    ///     EngineId, ExtensionModule, ExtensionModuleError, ExtensionModuleId,
1759    ///     ExtensionModuleRegistrar, Runtime,
1760    /// };
1761    ///
1762    /// #[derive(Debug)]
1763    /// struct ExampleModule {
1764    ///     id: ExtensionModuleId,
1765    /// }
1766    ///
1767    /// impl ExtensionModule for ExampleModule {
1768    ///     fn module_id(&self) -> &ExtensionModuleId {
1769    ///         &self.id
1770    ///     }
1771    ///
1772    ///     fn configure(
1773    ///         &self,
1774    ///         _registrar: &mut ExtensionModuleRegistrar<'_>,
1775    ///     ) -> Result<(), ExtensionModuleError> {
1776    ///         Ok(())
1777    ///     }
1778    /// }
1779    ///
1780    /// let runtime = Runtime::builder().build()?;
1781    /// let engine = EngineId::new("example.engine.v1")?;
1782    /// let error = runtime
1783    ///     .reconfigure(|edit| {
1784    ///         edit.ensure_extension_module_for_engine(
1785    ///             Arc::new(ExampleModule {
1786    ///                 id: ExtensionModuleId::new("example.module.v1")?,
1787    ///             }),
1788    ///             "example.family.v1",
1789    ///             &engine,
1790    ///         )?;
1791    ///         Ok(())
1792    ///     })
1793    ///     .unwrap_err();
1794    /// let source = std::error::Error::source(&error).unwrap();
1795    /// assert!(source.to_string().contains("example.module.v1"));
1796    /// # Ok::<(), Box<dyn std::error::Error>>(())
1797    /// ```
1798    ///
1799    /// # Errors
1800    ///
1801    /// Returns [`RuntimeConfigError::MissingExtensionEngine`] when the
1802    /// configured module does not register `family_id` for `engine_id`, or
1803    /// [`RuntimeConfigError::ExtensionModule`] when module configuration fails.
1804    #[doc(hidden)]
1805    pub fn ensure_extension_module_for_engine(
1806        &mut self,
1807        value: Arc<dyn ExtensionModule>,
1808        family_id: &'static str,
1809        engine_id: &EngineId,
1810    ) -> Result<&mut Self, RuntimeConfigError> {
1811        let module_id = value.module_id().clone();
1812        if let Some(existing) = self.candidate.modules.get(&module_id) {
1813            if existing
1814                .engines
1815                .contains_key(&(family_id, engine_id.clone()))
1816            {
1817                return Ok(self);
1818            }
1819        }
1820
1821        let record = configure_module(Arc::clone(&value))
1822            .map_err(|source| RuntimeConfigError::ExtensionModule { source })?;
1823        if !record.engines.contains_key(&(family_id, engine_id.clone())) {
1824            return Err(RuntimeConfigError::MissingExtensionEngine {
1825                module_id,
1826                family_id,
1827                engine_id: engine_id.clone(),
1828            });
1829        }
1830        self.candidate.modules.insert(module_id, record);
1831        *self.changed = true;
1832        Ok(self)
1833    }
1834
1835    /// Replace one extension module only when it owns the selected family and
1836    /// engine registration.
1837    ///
1838    /// This owner-scoped operation validates the configured candidate before
1839    /// [`Runtime::reconfigure`] can publish it. A failed validation therefore
1840    /// leaves the previously published module and snapshot untouched.
1841    ///
1842    /// # Examples
1843    ///
1844    /// ```rust
1845    /// use std::sync::Arc;
1846    ///
1847    /// use tenferro_runtime::{
1848    ///     EngineId, ExtensionModule, ExtensionModuleError, ExtensionModuleId,
1849    ///     ExtensionModuleRegistrar, Runtime,
1850    /// };
1851    ///
1852    /// #[derive(Debug)]
1853    /// struct ExampleModule {
1854    ///     id: ExtensionModuleId,
1855    /// }
1856    ///
1857    /// impl ExtensionModule for ExampleModule {
1858    ///     fn module_id(&self) -> &ExtensionModuleId {
1859    ///         &self.id
1860    ///     }
1861    ///
1862    ///     fn configure(
1863    ///         &self,
1864    ///         _registrar: &mut ExtensionModuleRegistrar<'_>,
1865    ///     ) -> Result<(), ExtensionModuleError> {
1866    ///         Ok(())
1867    ///     }
1868    /// }
1869    ///
1870    /// let runtime = Runtime::builder().build()?;
1871    /// let engine = EngineId::new("example.engine.v1")?;
1872    /// let error = runtime
1873    ///     .reconfigure(|edit| {
1874    ///         edit.replace_extension_module_for_engine(
1875    ///             Arc::new(ExampleModule {
1876    ///                 id: ExtensionModuleId::new("example.module.v1")?,
1877    ///             }),
1878    ///             "example.family.v1",
1879    ///             &engine,
1880    ///         )?;
1881    ///         Ok(())
1882    ///     })
1883    ///     .unwrap_err();
1884    /// let source = std::error::Error::source(&error).unwrap();
1885    /// assert!(source.to_string().contains("example.module.v1"));
1886    /// # Ok::<(), Box<dyn std::error::Error>>(())
1887    /// ```
1888    ///
1889    /// # Errors
1890    ///
1891    /// Returns [`RuntimeConfigError::MissingExtensionEngine`] when the
1892    /// configured module does not register `family_id` for `engine_id`, or
1893    /// [`RuntimeConfigError::ExtensionModule`] when module configuration fails.
1894    #[doc(hidden)]
1895    pub fn replace_extension_module_for_engine(
1896        &mut self,
1897        value: Arc<dyn ExtensionModule>,
1898        family_id: &'static str,
1899        engine_id: &EngineId,
1900    ) -> Result<&mut Self, RuntimeConfigError> {
1901        let module_id = value.module_id().clone();
1902        if let Some(existing) = self.candidate.modules.get(&module_id) {
1903            if existing.module_identical(&value) {
1904                if existing
1905                    .engines
1906                    .contains_key(&(family_id, engine_id.clone()))
1907                {
1908                    return Ok(self);
1909                }
1910                return Err(RuntimeConfigError::MissingExtensionEngine {
1911                    module_id,
1912                    family_id,
1913                    engine_id: engine_id.clone(),
1914                });
1915            }
1916        }
1917
1918        let record = configure_module(Arc::clone(&value))
1919            .map_err(|source| RuntimeConfigError::ExtensionModule { source })?;
1920        if !record.engines.contains_key(&(family_id, engine_id.clone())) {
1921            return Err(RuntimeConfigError::MissingExtensionEngine {
1922                module_id,
1923                family_id,
1924                engine_id: engine_id.clone(),
1925            });
1926        }
1927        self.candidate.modules.insert(module_id, record);
1928        *self.changed = true;
1929        Ok(self)
1930    }
1931
1932    /// Remove an extension module if present.
1933    ///
1934    /// # Errors
1935    ///
1936    /// This method currently has no failing absent-module path; it returns
1937    /// [`RuntimeConfigError`] only for future validated module removal failures.
1938    pub fn remove_extension_module(
1939        &mut self,
1940        id: &ExtensionModuleId,
1941    ) -> Result<&mut Self, RuntimeConfigError> {
1942        remove_extension_module_candidate(self.candidate, id, self.changed)?;
1943        Ok(self)
1944    }
1945}
1946
1947impl fmt::Debug for RuntimeReconfiguration<'_> {
1948    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
1949        formatter
1950            .debug_struct("RuntimeReconfiguration")
1951            .field("engine_count", &self.candidate.engines.len())
1952            .field("extension_module_count", &self.candidate.modules.len())
1953            .field("transfer_provider_count", &self.candidate.transfers.len())
1954            .field("changed", &*self.changed)
1955            .finish_non_exhaustive()
1956    }
1957}
1958
1959/// Borrowed immutable view of one engine slot in a runtime snapshot.
1960///
1961/// # Examples
1962///
1963/// ```
1964/// use std::fmt::Debug;
1965/// use tenferro_runtime::EngineSnapshotView;
1966///
1967/// fn requires_debug<T: Debug>() {}
1968/// requires_debug::<EngineSnapshotView<'_>>();
1969/// ```
1970#[derive(Clone, Copy)]
1971pub struct EngineSnapshotView<'a> {
1972    slot: &'a FrozenEngineSlot,
1973}
1974
1975impl<'a> EngineSnapshotView<'a> {
1976    /// Return the engine ID for this slot.
1977    pub fn engine_id(&self) -> &'a EngineId {
1978        self.slot.engine_id()
1979    }
1980
1981    /// Return the immutable provider/device binding for this engine slot.
1982    ///
1983    /// # Examples
1984    ///
1985    /// ```
1986    /// # fn inspect(view: tenferro_runtime::EngineSnapshotView<'_>) {
1987    /// let _ = view.provider_device_identity();
1988    /// # }
1989    /// ```
1990    pub fn provider_device_identity(&self) -> &'a super::ProviderDeviceIdentity {
1991        self.slot.provider_device_identity()
1992    }
1993
1994    /// Return the runtime-local registration identity for this slot.
1995    pub fn registration_identity(&self) -> RegistrationIdentity {
1996        self.slot.metadata().identity
1997    }
1998
1999    /// Return the execution-context identity required by this slot.
2000    pub fn context_identity(&self) -> ExecutionContextIdentity {
2001        self.slot.context_identity()
2002    }
2003
2004    /// Return this engine's runtime event domain.
2005    pub fn event_domain_id(&self) -> EventDomainId {
2006        self.slot.metadata().event_domain_id
2007    }
2008
2009    pub(super) fn executable_witness(&self) -> Option<&'a Arc<ExecutableEngineSnapshot>> {
2010        self.slot.executable()
2011    }
2012
2013    /// Return the hardware class for this slot.
2014    pub fn hardware_class(&self) -> &'a HardwareClassId {
2015        self.slot.hardware_class()
2016    }
2017
2018    /// Return direct core capability slots for this engine.
2019    pub fn capabilities(&self) -> &'a CoreCapabilityBundle {
2020        self.slot.capabilities()
2021    }
2022
2023    pub(super) fn storage_classes(&self) -> &'a [StorageClass] {
2024        self.slot.storage_classes()
2025    }
2026
2027    pub(super) fn default_storage_class(&self) -> &'a StorageClass {
2028        self.slot.default_storage_class()
2029    }
2030
2031    /// Return whether this engine's registered ingress accepts one physical
2032    /// input signature across its advertised storage classes.
2033    ///
2034    /// This narrow, doc-hidden query is used by eager extension owners before
2035    /// invoking a module factory. It includes placement, backend family, and
2036    /// allocation-domain admission rather than checking only tensor shape or
2037    /// dtype.
2038    ///
2039    /// # Examples
2040    ///
2041    /// ```rust
2042    /// use tenferro_runtime::{InputSignature, Runtime, Tensor, TensorRead};
2043    ///
2044    /// let backend = tenferro_cpu::CpuBackend::new();
2045    /// let mut builder = Runtime::builder();
2046    /// builder.register_engine(tenferro_cpu::runtime_engine_registration(&backend)?)?;
2047    /// let runtime = builder.build()?;
2048    /// let engine_id = tenferro_cpu::runtime_engine_id()?;
2049    /// let snapshot = runtime.snapshot()?;
2050    /// let engine = snapshot.engine(&engine_id).unwrap();
2051    /// let tensor = Tensor::from_vec_col_major(vec![1], vec![1.0_f64])?;
2052    /// let reads = [TensorRead::from_tensor(&tensor)];
2053    /// let signature = InputSignature::from_reads(&reads)?;
2054    /// assert!(engine.accepts_input_signature(&signature.entries()[0]));
2055    /// # Ok::<(), Box<dyn std::error::Error>>(())
2056    /// ```
2057    #[doc(hidden)]
2058    pub fn accepts_input_signature(&self, input: &super::InputSignatureEntry) -> bool {
2059        self.slot
2060            .storage_classes()
2061            .iter()
2062            .any(|storage_class| self.accepts_input_signature_for_storage(input, storage_class))
2063    }
2064
2065    pub(super) fn accepts_input_signature_for_storage(
2066        &self,
2067        input: &super::InputSignatureEntry,
2068        storage_class: &StorageClass,
2069    ) -> bool {
2070        self.slot
2071            .executable()
2072            .is_some_and(|snapshot| snapshot.accepts_input_signature(input, storage_class))
2073    }
2074
2075    #[cfg(test)]
2076    pub(crate) fn has_execution_engine_for_test(&self) -> bool {
2077        self.slot.executable().is_some()
2078    }
2079}
2080
2081impl fmt::Debug for EngineSnapshotView<'_> {
2082    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
2083        formatter
2084            .debug_struct("EngineSnapshotView")
2085            .field("engine_id", self.engine_id())
2086            .field("registration_identity", &self.registration_identity())
2087            .field("context_identity", &self.context_identity())
2088            .field("hardware_class", self.hardware_class())
2089            .field("capabilities", self.capabilities())
2090            .finish()
2091    }
2092}
2093
2094fn default_execution_policy() -> ExecutionPolicy {
2095    ExecutionPolicy::new(super::Determinism::Fast, None, 0)
2096}
2097
2098fn register_engine_candidate(
2099    candidate: &mut CandidateConfig,
2100    registration: EngineRegistration,
2101    changed: &mut bool,
2102) -> Result<(), RuntimeConfigError> {
2103    let engine_id = registration.engine_id().clone();
2104    match candidate.engines.get(&engine_id) {
2105        Some(existing) if existing.registration.candidate_identical(&registration) => Ok(()),
2106        Some(_) => Err(RuntimeConfigError::DuplicateEngine { engine_id }),
2107        None => {
2108            ensure_unique_provider_device_target(candidate, &registration)?;
2109            candidate.engines.insert(
2110                engine_id,
2111                CandidateEngineRecord {
2112                    registration,
2113                    identity: CandidateRegistrationIdentity::New,
2114                },
2115            );
2116            *changed = true;
2117            Ok(())
2118        }
2119    }
2120}
2121
2122fn replace_engine_candidate(
2123    candidate: &mut CandidateConfig,
2124    registration: EngineRegistration,
2125    changed: &mut bool,
2126) -> Result<(), RuntimeConfigError> {
2127    let engine_id = registration.engine_id().clone();
2128    let Some(existing) = candidate.engines.get(&engine_id) else {
2129        return Err(RuntimeConfigError::MissingEngine { engine_id });
2130    };
2131    if existing.registration.candidate_identical(&registration) {
2132        return Ok(());
2133    }
2134    if existing.registration.provider_device_identity() != registration.provider_device_identity() {
2135        return Err(RuntimeConfigError::EngineTargetRebind {
2136            engine_id,
2137            current: existing.registration.provider_device_identity().clone(),
2138            replacement: registration.provider_device_identity().clone(),
2139        });
2140    }
2141    ensure_unique_provider_device_target_except(candidate, &registration, &engine_id)?;
2142    candidate.engines.insert(
2143        engine_id,
2144        CandidateEngineRecord {
2145            registration,
2146            identity: CandidateRegistrationIdentity::New,
2147        },
2148    );
2149    *changed = true;
2150    Ok(())
2151}
2152
2153fn remove_engine_candidate(
2154    candidate: &mut CandidateConfig,
2155    id: &EngineId,
2156    changed: &mut bool,
2157) -> Result<(), RuntimeConfigError> {
2158    match candidate.engines.remove(id) {
2159        Some(_) => {
2160            *changed = true;
2161            Ok(())
2162        }
2163        None => Err(RuntimeConfigError::MissingEngine {
2164            engine_id: id.clone(),
2165        }),
2166    }
2167}
2168
2169fn install_extension_module_candidate(
2170    candidate: &mut CandidateConfig,
2171    module: Arc<dyn ExtensionModule>,
2172    changed: &mut bool,
2173) -> Result<(), RuntimeConfigError> {
2174    let module_id = module.module_id().clone();
2175    match candidate.modules.get(&module_id) {
2176        Some(existing) if existing.module_identical(&module) => Ok(()),
2177        Some(_) => Err(RuntimeConfigError::ExtensionModule {
2178            source: ExtensionModuleError::ConflictingModule { module_id },
2179        }),
2180        None => {
2181            let record = configure_module(module)
2182                .map_err(|source| RuntimeConfigError::ExtensionModule { source })?;
2183            candidate.modules.insert(module_id, record);
2184            *changed = true;
2185            Ok(())
2186        }
2187    }
2188}
2189
2190fn replace_extension_module_candidate(
2191    candidate: &mut CandidateConfig,
2192    module: Arc<dyn ExtensionModule>,
2193    changed: &mut bool,
2194) -> Result<(), RuntimeConfigError> {
2195    let module_id = module.module_id().clone();
2196    match candidate.modules.get(&module_id) {
2197        Some(existing) if existing.module_identical(&module) => Ok(()),
2198        _ => {
2199            let record = configure_module(module)
2200                .map_err(|source| RuntimeConfigError::ExtensionModule { source })?;
2201            candidate.modules.insert(module_id, record);
2202            *changed = true;
2203            Ok(())
2204        }
2205    }
2206}
2207
2208fn remove_extension_module_candidate(
2209    candidate: &mut CandidateConfig,
2210    id: &ExtensionModuleId,
2211    changed: &mut bool,
2212) -> Result<(), RuntimeConfigError> {
2213    if candidate.modules.remove(id).is_some() {
2214        *changed = true;
2215    }
2216    Ok(())
2217}
2218
2219fn register_transfer_provider_candidate(
2220    candidate: &mut CandidateConfig,
2221    source: TransferEndpoint,
2222    destination: TransferEndpoint,
2223    provider: Arc<dyn TransferProvider>,
2224    changed: &mut bool,
2225) -> Result<(), RuntimeConfigError> {
2226    let key = TransferRoute::new(source, destination);
2227    match candidate.transfers.get(&key) {
2228        Some(existing) if Arc::ptr_eq(&existing.provider, &provider) => Ok(()),
2229        Some(_) => Err(RuntimeConfigError::ConflictingRegistration {
2230            key: RegistrationKey::TransferProvider {
2231                source: key.source().clone(),
2232                destination: key.destination().clone(),
2233            },
2234        }),
2235        None => {
2236            candidate.transfers.insert(
2237                key,
2238                CandidateTransferRecord {
2239                    provider,
2240                    binding: CandidateTransferBinding::New,
2241                },
2242            );
2243            *changed = true;
2244            Ok(())
2245        }
2246    }
2247}
2248
2249fn remove_transfer_provider_candidate(
2250    candidate: &mut CandidateConfig,
2251    source: TransferEndpoint,
2252    destination: TransferEndpoint,
2253    changed: &mut bool,
2254) -> Result<(), RuntimeConfigError> {
2255    let key = TransferRoute::new(source, destination);
2256    if candidate.transfers.remove(&key).is_none() {
2257        return Err(RuntimeConfigError::MissingTransferProvider {
2258            source_endpoint: key.source().clone(),
2259            destination: key.destination().clone(),
2260        });
2261    }
2262    *changed = true;
2263    Ok(())
2264}
2265
2266fn ensure_unique_provider_device_target(
2267    candidate: &CandidateConfig,
2268    registration: &EngineRegistration,
2269) -> Result<(), RuntimeConfigError> {
2270    ensure_unique_provider_device_target_except(candidate, registration, registration.engine_id())
2271}
2272
2273fn ensure_unique_provider_device_target_except(
2274    candidate: &CandidateConfig,
2275    registration: &EngineRegistration,
2276    ignored_engine_id: &EngineId,
2277) -> Result<(), RuntimeConfigError> {
2278    if let Some((first_engine_id, _)) = candidate.engines.iter().find(|(engine_id, record)| {
2279        *engine_id != ignored_engine_id
2280            && record.registration.provider_device_identity()
2281                == registration.provider_device_identity()
2282    }) {
2283        return Err(RuntimeConfigError::DuplicateProviderDeviceTarget {
2284            provider_device_identity: registration.provider_device_identity().clone(),
2285            first_engine_id: first_engine_id.clone(),
2286            duplicate_engine_id: registration.engine_id().clone(),
2287        });
2288    }
2289    Ok(())
2290}
2291
2292fn validate_candidate(
2293    candidate: CandidateConfig,
2294    issuer: NonZeroU64,
2295    next_ordinal: NonZeroU64,
2296) -> Result<(BoundCandidateConfig, NonZeroU64), RuntimeConfigError> {
2297    let mut seen_targets = BTreeMap::<ProviderDeviceIdentity, EngineId>::new();
2298    for (engine_id, record) in &candidate.engines {
2299        if let Some(first_engine_id) = seen_targets.insert(
2300            record.registration.provider_device_identity().clone(),
2301            engine_id.clone(),
2302        ) {
2303            return Err(RuntimeConfigError::DuplicateProviderDeviceTarget {
2304                provider_device_identity: record.registration.provider_device_identity().clone(),
2305                first_engine_id,
2306                duplicate_engine_id: engine_id.clone(),
2307            });
2308        }
2309    }
2310
2311    let mut bound_transfers = BTreeMap::new();
2312    for (route, record) in &candidate.transfers {
2313        let source_binding = validate_transfer_endpoint(&candidate, route.source())?;
2314        let destination_binding = validate_transfer_endpoint(&candidate, route.destination())?;
2315        let preserved = match &record.binding {
2316            CandidateTransferBinding::New => None,
2317            CandidateTransferBinding::Preserved {
2318                source,
2319                destination,
2320            } => Some((source, destination)),
2321        };
2322        if let Some((registered_source, registered_destination)) = preserved {
2323            if registered_source != &source_binding {
2324                return Err(RuntimeConfigError::StaleTransferRoute {
2325                    source_endpoint: route.source().clone(),
2326                    destination: route.destination().clone(),
2327                    endpoint: route.source().clone(),
2328                    registered: Box::new(registered_source.clone()),
2329                    current: Box::new(source_binding.clone()),
2330                });
2331            }
2332            if registered_destination != &destination_binding {
2333                return Err(RuntimeConfigError::StaleTransferRoute {
2334                    source_endpoint: route.source().clone(),
2335                    destination: route.destination().clone(),
2336                    endpoint: route.destination().clone(),
2337                    registered: Box::new(registered_destination.clone()),
2338                    current: Box::new(destination_binding.clone()),
2339                });
2340            }
2341        }
2342        bound_transfers.insert(
2343            route.clone(),
2344            BoundCandidateTransferRecord {
2345                provider: Arc::clone(&record.provider),
2346                source: source_binding,
2347                destination: destination_binding,
2348            },
2349        );
2350    }
2351    let mut seen = BTreeMap::<(ExtensionFamilyId, EngineId), ExtensionModuleId>::new();
2352    for (module_id, module) in &candidate.modules {
2353        for family_engine in module.engines.keys() {
2354            if seen
2355                .insert(
2356                    (family_engine.0, family_engine.1.clone()),
2357                    module_id.clone(),
2358                )
2359                .is_some()
2360            {
2361                return Err(RuntimeConfigError::ConflictingRegistration {
2362                    key: RegistrationKey::ExtensionEngine {
2363                        family: family_engine.0,
2364                        engine: family_engine.1.clone(),
2365                    },
2366                });
2367            }
2368        }
2369    }
2370
2371    let CandidateConfig {
2372        policy,
2373        engines,
2374        modules,
2375        transfers: _,
2376    } = candidate;
2377    let mut allocator = RegistrationIdentityAllocator::new(issuer, next_ordinal);
2378    let engines = engines
2379        .into_iter()
2380        .map(|(engine_id, record)| {
2381            let identity = match record.identity {
2382                CandidateRegistrationIdentity::New => allocator.allocate()?,
2383                CandidateRegistrationIdentity::Preserved(identity) => identity,
2384            };
2385            Ok((
2386                engine_id,
2387                BoundCandidateEngineRecord {
2388                    registration: record.registration,
2389                    identity,
2390                },
2391            ))
2392        })
2393        .collect::<Result<BTreeMap<_, _>, RuntimeConfigError>>()?;
2394    let modules = modules
2395        .into_iter()
2396        .map(|(module_id, module)| {
2397            let mut allocate = || allocator.allocate();
2398            Ok((module_id, bind_candidate_module(module, &mut allocate)?))
2399        })
2400        .collect::<Result<BTreeMap<_, _>, RuntimeConfigError>>()?;
2401    Ok((
2402        BoundCandidateConfig {
2403            policy,
2404            engines,
2405            modules,
2406            transfers: bound_transfers,
2407        },
2408        allocator.next_ordinal(),
2409    ))
2410}
2411
2412fn validate_transfer_endpoint(
2413    candidate: &CandidateConfig,
2414    endpoint: &TransferEndpoint,
2415) -> Result<ProviderDeviceIdentity, RuntimeConfigError> {
2416    let Some(engine) = candidate.engines.get(endpoint.engine_id()) else {
2417        return Err(RuntimeConfigError::UnknownTransferEndpointEngine {
2418            endpoint: endpoint.clone(),
2419        });
2420    };
2421    if !engine
2422        .registration
2423        .storage_classes()
2424        .contains(endpoint.storage_class())
2425    {
2426        return Err(RuntimeConfigError::UnsupportedTransferEndpointStorage {
2427            endpoint: endpoint.clone(),
2428        });
2429    }
2430    Ok(engine.registration.provider_device_identity().clone())
2431}
2432
2433struct RegistrationIdentityAllocator {
2434    issuer: NonZeroU64,
2435    next: NonZeroU64,
2436}
2437
2438impl RegistrationIdentityAllocator {
2439    fn new(issuer: NonZeroU64, next: NonZeroU64) -> Self {
2440        Self { issuer, next }
2441    }
2442
2443    fn allocate(&mut self) -> Result<RegistrationIdentity, RuntimeConfigError> {
2444        let identity = RegistrationIdentity::new(self.issuer, self.next);
2445        let next = self
2446            .next
2447            .get()
2448            .checked_add(1)
2449            .and_then(NonZeroU64::new)
2450            .ok_or(RuntimeConfigError::IdentityExhausted)?;
2451        self.next = next;
2452        Ok(identity)
2453    }
2454
2455    fn next_ordinal(&self) -> NonZeroU64 {
2456        self.next
2457    }
2458}
2459
2460fn freeze_candidate(
2461    runtime_id: RuntimeId,
2462    epoch: RuntimeEpoch,
2463    candidate: BoundCandidateConfig,
2464) -> Result<RuntimeConfigSnapshot, RuntimeConfigError> {
2465    let mut engines = Vec::with_capacity(candidate.engines.len());
2466    let mut engine_indices = BTreeMap::new();
2467    let mut engine_locations = BTreeMap::new();
2468    let mut cache_owners = Vec::new();
2469    for (index, (engine_id, record)) in candidate.engines.into_iter().enumerate() {
2470        let BoundCandidateEngineRecord {
2471            registration,
2472            identity,
2473        } = record;
2474        let event_domain_id = EventDomainId::new(runtime_id, epoch, identity);
2475        let (state, candidate_token) = registration.into_state_and_token();
2476        let provider_device_identity = state.provider_device_identity().clone();
2477        let metadata = FrozenEngineMetadata {
2478            candidate_token,
2479            identity,
2480            event_domain_id,
2481        };
2482        let frozen = match state {
2483            EngineRegistrationState::PreparationOnly { binding } => {
2484                FrozenEngineSlot::PreparationOnly(Arc::new(PreparationOnlyEngineSnapshot {
2485                    metadata,
2486                    binding,
2487                }))
2488            }
2489            EngineRegistrationState::Executable(binding) => {
2490                if let Some(owner) = binding.contract().cache_owner().cloned() {
2491                    cache_owners.push(FrozenCacheOwner {
2492                        id: engine_cache_owner_id(&engine_id),
2493                        kind: FrozenCacheOwnerKind::Engine,
2494                        owner,
2495                    });
2496                }
2497                cache_owners.push(FrozenCacheOwner {
2498                    id: engine_extension_cache_owner_id(&engine_id),
2499                    kind: FrozenCacheOwnerKind::Extension,
2500                    owner: execution::extension_cache_owner(binding.contract().executor().clone()),
2501                });
2502                FrozenEngineSlot::Executable(Arc::new(ExecutableEngineSnapshot {
2503                    metadata,
2504                    binding,
2505                }))
2506            }
2507        };
2508        engine_locations.insert(
2509            engine_id.clone(),
2510            (provider_device_identity, event_domain_id),
2511        );
2512        engine_indices.insert(engine_id, index);
2513        engines.push(frozen);
2514    }
2515    let extensions = freeze_extension_slots(candidate.modules)?;
2516    for (id, owner) in extensions.cache_owner_records() {
2517        cache_owners.push(FrozenCacheOwner {
2518            id,
2519            kind: FrozenCacheOwnerKind::Extension,
2520            owner,
2521        });
2522    }
2523    let mut transfers = BTreeMap::new();
2524    for (route, record) in candidate.transfers {
2525        let BoundCandidateTransferRecord {
2526            provider,
2527            source: source_binding,
2528            destination: destination_binding,
2529        } = record;
2530        let (_, source_event_domain_id) = bound_engine_location(&engine_locations, route.source())?;
2531        let (_, destination_event_domain_id) =
2532            bound_engine_location(&engine_locations, route.destination())?;
2533        let resolved_route = ResolvedTransferRoute::new(
2534            ResolvedTransferEndpoint::new(
2535                route.source().clone(),
2536                source_binding,
2537                *source_event_domain_id,
2538            ),
2539            ResolvedTransferEndpoint::new(
2540                route.destination().clone(),
2541                destination_binding,
2542                *destination_event_domain_id,
2543            ),
2544        );
2545        transfers.insert(resolved_route, provider);
2546    }
2547    Ok(RuntimeConfigSnapshot {
2548        runtime_id,
2549        epoch,
2550        policy: candidate.policy,
2551        engines: engines.into(),
2552        engine_indices,
2553        extensions,
2554        transfers: FrozenTransferRegistry::new(transfers),
2555        cache_owners: cache_owners.into(),
2556    })
2557}
2558
2559fn bound_engine_location<'a>(
2560    locations: &'a BTreeMap<EngineId, (ProviderDeviceIdentity, EventDomainId)>,
2561    endpoint: &TransferEndpoint,
2562) -> Result<&'a (ProviderDeviceIdentity, EventDomainId), RuntimeConfigError> {
2563    locations
2564        .get(endpoint.engine_id())
2565        .ok_or_else(|| RuntimeConfigError::BoundCandidateInvariant {
2566            endpoint: endpoint.clone(),
2567        })
2568}
2569
2570fn engine_cache_owner_id(engine_id: &EngineId) -> CacheOwnerId {
2571    let id = engine_id.as_str();
2572    CacheOwnerId::from_canonical_owner_id(Arc::<str>::from(format!("engine[{}]:{id}", id.len())))
2573}
2574
2575fn engine_extension_cache_owner_id(engine_id: &EngineId) -> CacheOwnerId {
2576    let id = engine_id.as_str();
2577    CacheOwnerId::from_canonical_owner_id(Arc::<str>::from(format!(
2578        "extension-executor[{}]:{id}",
2579        id.len()
2580    )))
2581}
2582
2583fn allocate_nonzero(counter: &AtomicU64) -> Result<NonZeroU64, RuntimeConfigError> {
2584    let value = counter
2585        .fetch_update(Ordering::SeqCst, Ordering::SeqCst, |next| {
2586            next.checked_add(1)
2587        })
2588        .map_err(|_| RuntimeConfigError::IdentityExhausted)?;
2589    NonZeroU64::new(value).ok_or(RuntimeConfigError::IdentityExhausted)
2590}
2591
2592#[cfg(test)]
2593mod freeze_tests {
2594    use crate::{ProviderId, TransferRequest};
2595
2596    use super::*;
2597
2598    #[derive(Debug)]
2599    struct FreezeTestContext;
2600
2601    #[derive(Debug)]
2602    struct FreezeTestProvider;
2603
2604    impl TransferProvider for FreezeTestProvider {
2605        fn transfer_blocking(
2606            &self,
2607            _request: TransferRequest<'_>,
2608        ) -> crate::Result<tenferro_tensor::Tensor> {
2609            Err(crate::Error::Internal("freeze test provider".into()))
2610        }
2611    }
2612
2613    fn registration(
2614        engine_id: &str,
2615        target: &str,
2616    ) -> Result<EngineRegistration, RuntimeConfigError> {
2617        let engine_id = EngineId::new(engine_id).map_err(RuntimeConfigError::from)?;
2618        let storage =
2619            StorageClass::new("tenferro.test.freeze.storage").map_err(RuntimeConfigError::from)?;
2620        Ok(EngineRegistration::preparation_only(
2621            super::super::ProviderPreparationBinding::new(
2622                engine_id,
2623                ProviderDeviceIdentity::new(
2624                    ProviderId::new("tenferro.test.freeze.provider")
2625                        .map_err(RuntimeConfigError::from)?,
2626                    target,
2627                )
2628                .map_err(RuntimeConfigError::from)?,
2629                ExecutionContextIdentity::of::<FreezeTestContext>(),
2630                HardwareClassId::new("tenferro.test.freeze.hardware")
2631                    .map_err(RuntimeConfigError::from)?,
2632                Arc::from(vec![storage.clone()]),
2633                storage,
2634                CoreCapabilityBundle::default(),
2635            )?,
2636        ))
2637    }
2638
2639    fn candidate(binding: CandidateTransferBinding) -> Result<CandidateConfig, RuntimeConfigError> {
2640        let source_id =
2641            EngineId::new("tenferro.test.freeze.source").map_err(RuntimeConfigError::from)?;
2642        let destination_id =
2643            EngineId::new("tenferro.test.freeze.destination").map_err(RuntimeConfigError::from)?;
2644        let storage =
2645            StorageClass::new("tenferro.test.freeze.storage").map_err(RuntimeConfigError::from)?;
2646        let source_endpoint = TransferEndpoint::new(source_id.clone(), storage.clone());
2647        let destination_endpoint = TransferEndpoint::new(destination_id.clone(), storage);
2648        let mut candidate = CandidateConfig::empty();
2649        let mut changed = false;
2650        register_engine_candidate(
2651            &mut candidate,
2652            registration(source_id.as_str(), "freeze-source")?,
2653            &mut changed,
2654        )?;
2655        register_engine_candidate(
2656            &mut candidate,
2657            registration(destination_id.as_str(), "freeze-destination")?,
2658            &mut changed,
2659        )?;
2660        register_transfer_provider_candidate(
2661            &mut candidate,
2662            source_endpoint.clone(),
2663            destination_endpoint.clone(),
2664            Arc::new(FreezeTestProvider),
2665            &mut changed,
2666        )?;
2667        candidate
2668            .transfers
2669            .get_mut(&TransferRoute::new(source_endpoint, destination_endpoint))
2670            .expect("registered route")
2671            .binding = binding;
2672        Ok(candidate)
2673    }
2674
2675    #[test]
2676    fn validation_owns_stale_route_rejection_and_bound_freeze_is_total() {
2677        let wrong_source = ProviderDeviceIdentity::new(
2678            ProviderId::new("tenferro.test.freeze.provider").unwrap(),
2679            "different-source",
2680        )
2681        .unwrap();
2682        let preserved = CandidateTransferBinding::Preserved {
2683            source: wrong_source,
2684            destination: ProviderDeviceIdentity::new(
2685                ProviderId::new("tenferro.test.freeze.provider").unwrap(),
2686                "freeze-destination",
2687            )
2688            .unwrap(),
2689        };
2690        let result = validate_candidate(
2691            candidate(preserved).unwrap(),
2692            NonZeroU64::new(1).unwrap(),
2693            NonZeroU64::new(1).unwrap(),
2694        );
2695        let error = match result {
2696            Ok(_) => panic!("candidate validation must reject stale preserved bindings"),
2697            Err(error) => error,
2698        };
2699        assert!(matches!(
2700            error,
2701            RuntimeConfigError::StaleTransferRoute { .. }
2702        ));
2703
2704        let (bound, _) = validate_candidate(
2705            candidate(CandidateTransferBinding::New).unwrap(),
2706            NonZeroU64::new(1).unwrap(),
2707            NonZeroU64::new(1).unwrap(),
2708        )
2709        .expect("validation must produce a complete bound candidate");
2710        freeze_candidate(
2711            RuntimeId::from_nonzero(NonZeroU64::new(1).unwrap()),
2712            RuntimeEpoch::one(),
2713            bound,
2714        )
2715        .expect("a bound candidate must freeze without semantic route revalidation");
2716    }
2717
2718    #[test]
2719    fn frozen_engine_slots_are_arc_sized() {
2720        let slot_size = std::mem::size_of::<FrozenEngineSlot>();
2721        let arc_size = std::mem::size_of::<Arc<()>>();
2722
2723        assert!(
2724            slot_size <= 2 * arc_size,
2725            "frozen engine slots should keep immutable snapshot payloads behind Arc: slot_size={slot_size}, arc_size={arc_size}",
2726        );
2727    }
2728}