Skip to main content

atman_runtime/
provider_lifecycle.rs

1use std::collections::{BTreeSet, HashMap};
2use std::path::{Path, PathBuf};
3use std::sync::{Arc, LazyLock, Mutex, Weak};
4
5use crate::auth_store::{ProviderKind, StoredProvider};
6use crate::config_hub::{
7    AuthModelCacheCommit, AuthModelCacheUpdate, AuthProviderInsertCommit,
8    AuthProviderRuntimeCommit, AuthProviderRuntimeState, ConfigError, ConfigHub,
9    ProviderConfigUpdate, ProviderConfigWriteMode,
10};
11use crate::model_registry::{
12    CatalogDelta, CatalogError, PreparedProviderCatalog, ProviderDescriptor,
13    commit_prepared_provider_catalog, prepare_provider_catalog, provider_catalog_namespace,
14    remove_provider_catalog, shortest_unique_provider_id,
15};
16use crate::provider::{
17    ModelDiscoveryError, Provider, ProviderRegistry, ReasoningWireProfile, WeakProviderRegistry,
18};
19
20#[derive(Debug, thiserror::Error)]
21#[non_exhaustive]
22pub enum ProviderLifecycleError {
23    #[error(transparent)]
24    Config(#[from] ConfigError),
25    #[error(transparent)]
26    Catalog(#[from] CatalogError),
27    #[error(transparent)]
28    Discovery(#[from] ModelDiscoveryError),
29    #[error("auth provider `{id}` does not exist")]
30    ProviderNotFound { id: String },
31    #[error("auth provider `{id}` is disabled")]
32    ProviderDisabled { id: String },
33    #[error("auth provider `{id}` has no live provider")]
34    LiveProviderMissing { id: String },
35    #[error("live provider name `{actual}` does not match auth provider id `{expected}`")]
36    LiveProviderNameMismatch { expected: String, actual: String },
37    #[error("auth provider `{id}` changed while the operation was running")]
38    Stale { id: String },
39}
40
41#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
42pub struct ProviderStateChange {
43    pub auth_changed: bool,
44    pub live_changed: bool,
45    pub catalog_changed: bool,
46}
47
48#[derive(Debug, Clone, Default, PartialEq, Eq)]
49pub struct ProviderLifecycleOutcome {
50    pub state: ProviderStateChange,
51    pub catalog: Option<CatalogDelta>,
52}
53
54#[derive(Debug, Clone, Default, PartialEq, Eq)]
55pub struct ProviderReconcileOutcome {
56    pub providers: Vec<(String, ProviderStateChange)>,
57}
58
59/// Result of a conditional provider catalog refresh.
60#[derive(Debug, Clone, Copy, PartialEq, Eq)]
61#[non_exhaustive]
62pub enum ProviderCatalogRefreshOutcome {
63    /// The persisted catalog is already current.
64    NotNeeded,
65    /// Another task in this process is refreshing the same provider.
66    AlreadyInFlight,
67    /// The in-memory catalog was updated from discovery or a newer persisted cache.
68    CatalogUpdated(CatalogDelta),
69}
70
71#[derive(Clone)]
72pub struct ProviderLifecycle {
73    hub: ConfigHub,
74    providers: ProviderRegistry,
75    state: Arc<Mutex<LifecycleState>>,
76    config_state: Arc<Mutex<ConfigProviderLifecycleState>>,
77}
78
79#[derive(Clone)]
80pub(crate) struct ProviderLifecycleOwner {
81    hub: ConfigHub,
82    state: Arc<Mutex<LifecycleState>>,
83    config_state: Arc<Mutex<ConfigProviderLifecycleState>>,
84}
85
86#[derive(Default)]
87struct LifecycleState {
88    generations: HashMap<String, u64>,
89    operation_locks: HashMap<String, Arc<tokio::sync::Mutex<()>>>,
90    refresh_locks: HashMap<String, Arc<tokio::sync::Mutex<()>>>,
91    live_providers: HashMap<String, LiveProviderEntry>,
92    registries: Vec<WeakProviderRegistry>,
93}
94
95#[derive(Default)]
96struct ConfigProviderLifecycleState {
97    config_providers: HashMap<String, crate::model_registry::ProviderEntry>,
98    registries: Vec<WeakProviderRegistry>,
99}
100
101#[derive(Clone)]
102struct LiveProviderEntry {
103    kind: ProviderKind,
104    provider: Arc<dyn Provider>,
105}
106
107struct LiveProviderRegistration {
108    changed: bool,
109    replaced: Vec<Arc<dyn Provider>>,
110}
111
112struct LiveProviderRemoval {
113    changed: bool,
114    removed: Vec<Arc<dyn Provider>>,
115}
116
117struct ProviderOperationFence {
118    operation: Option<tokio::sync::OwnedMutexGuard<()>>,
119    providers: Vec<Arc<dyn Provider>>,
120}
121
122struct ProviderRefreshAttempt {
123    generation: u64,
124    runtime: AuthProviderRuntimeState,
125    live_provider: Arc<dyn Provider>,
126}
127
128enum ProviderRefreshPreparation {
129    Hydrated { delta: CatalogDelta, changed: bool },
130    Discover(Box<ProviderRefreshAttempt>),
131}
132
133type ProviderRefreshPreparationResult = (
134    Result<ProviderRefreshPreparation, ProviderLifecycleError>,
135    Vec<Arc<dyn Provider>>,
136);
137
138type ProviderRefreshAttemptResult = (
139    Result<ProviderRefreshAttempt, ProviderLifecycleError>,
140    Vec<Arc<dyn Provider>>,
141);
142
143type ProviderCatalogHydrationResult = (
144    Result<(CatalogDelta, bool), ProviderLifecycleError>,
145    Vec<Arc<dyn Provider>>,
146);
147
148impl ProviderOperationFence {
149    fn new(operation: tokio::sync::OwnedMutexGuard<()>) -> Self {
150        Self {
151            operation: Some(operation),
152            providers: Vec::new(),
153        }
154    }
155
156    fn protect(&mut self, provider: Arc<dyn Provider>) {
157        self.providers.push(provider);
158    }
159
160    fn extend(&mut self, providers: Vec<Arc<dyn Provider>>) {
161        self.providers.extend(providers);
162    }
163}
164
165impl Drop for ProviderOperationFence {
166    fn drop(&mut self) {
167        drop(self.operation.take());
168        self.providers.clear();
169    }
170}
171
172static LIFECYCLE_COORDINATORS: LazyLock<Mutex<HashMap<PathBuf, Weak<Mutex<LifecycleState>>>>> =
173    LazyLock::new(|| Mutex::new(HashMap::new()));
174static CONFIG_PROVIDER_COORDINATORS: LazyLock<
175    Mutex<HashMap<PathBuf, Weak<Mutex<ConfigProviderLifecycleState>>>>,
176> = LazyLock::new(|| Mutex::new(HashMap::new()));
177
178impl ProviderLifecycle {
179    pub fn new(hub: ConfigHub, providers: ProviderRegistry) -> Self {
180        let (lifecycle, replaced) = Self::new_deferred(hub, providers);
181        drop(replaced);
182        lifecycle
183    }
184
185    pub(crate) fn new_deferred(
186        hub: ConfigHub,
187        providers: ProviderRegistry,
188    ) -> (Self, Vec<Arc<dyn Provider>>) {
189        let state = shared_lifecycle_state(&hub);
190        let config_state = shared_config_provider_state(&hub);
191        let mut replaced = Vec::new();
192        {
193            let mut shared = state
194                .lock()
195                .unwrap_or_else(std::sync::PoisonError::into_inner);
196            let mut registered = false;
197            shared.registries.retain(|registry| {
198                let Some(registry) = registry.upgrade() else {
199                    return false;
200                };
201                registered |= registry.shares_storage_with(&providers);
202                true
203            });
204            if !registered {
205                for (provider_id, live) in &shared.live_providers {
206                    replaced.extend(
207                        providers.register_named(provider_id.clone(), live.provider.clone()),
208                    );
209                }
210                shared.registries.push(providers.downgrade());
211            }
212        }
213        {
214            let mut shared = config_state
215                .lock()
216                .unwrap_or_else(std::sync::PoisonError::into_inner);
217            let mut registered = false;
218            shared.registries.retain(|registry| {
219                let Some(registry) = registry.upgrade() else {
220                    return false;
221                };
222                registered |= registry.shares_storage_with(&providers);
223                true
224            });
225            if !registered {
226                let mut config_providers = shared
227                    .config_providers
228                    .iter()
229                    .map(|(name, entry)| (name.clone(), entry.clone()))
230                    .collect::<Vec<_>>();
231                config_providers.sort_by(|left, right| left.0.cmp(&right.0));
232                for (name, entry) in config_providers {
233                    let (_, previous) = crate::config_provider::reconcile_config_provider_deferred(
234                        &providers, &name, &entry,
235                    );
236                    replaced.extend(previous);
237                }
238                shared.registries.push(providers.downgrade());
239            }
240        }
241        (
242            Self {
243                hub,
244                providers,
245                state,
246                config_state,
247            },
248            replaced,
249        )
250    }
251
252    pub fn config_hub(&self) -> &ConfigHub {
253        &self.hub
254    }
255
256    pub fn provider_registry(&self) -> &ProviderRegistry {
257        &self.providers
258    }
259
260    pub(crate) fn owner(&self) -> ProviderLifecycleOwner {
261        ProviderLifecycleOwner {
262            hub: self.hub.clone(),
263            state: self.state.clone(),
264            config_state: self.config_state.clone(),
265        }
266    }
267
268    pub(crate) fn from_owner(owner: ProviderLifecycleOwner, providers: ProviderRegistry) -> Self {
269        Self {
270            hub: owner.hub,
271            providers,
272            state: owner.state,
273            config_state: owner.config_state,
274        }
275    }
276
277    /// Reload config-backed providers and reconcile every attached live registry.
278    pub fn reload_config_providers(&self) -> Result<(), ProviderLifecycleError> {
279        reload_config_providers_for_hub(&self.hub)?;
280        Ok(())
281    }
282
283    /// Create one config-backed provider and reconcile every attached live registry.
284    pub fn create_config_provider(
285        &self,
286        update: ProviderConfigUpdate<'_>,
287    ) -> Result<(), ProviderLifecycleError> {
288        self.commit_config_provider(update, ProviderConfigWriteMode::Create)
289    }
290
291    /// Update one config-backed provider and reconcile every attached live registry.
292    pub fn update_config_provider(
293        &self,
294        update: ProviderConfigUpdate<'_>,
295    ) -> Result<(), ProviderLifecycleError> {
296        self.commit_config_provider(update, ProviderConfigWriteMode::Update)
297    }
298
299    fn commit_config_provider(
300        &self,
301        update: ProviderConfigUpdate<'_>,
302        mode: ProviderConfigWriteMode,
303    ) -> Result<(), ProviderLifecycleError> {
304        mutate_config_provider_for_hub(&self.hub, update, mode)?;
305        Ok(())
306    }
307
308    pub async fn install_provider(
309        &self,
310        provider_record: StoredProvider,
311        live_provider: Arc<dyn Provider>,
312    ) -> Result<CatalogDelta, ProviderLifecycleError> {
313        self.validate_new_provider(&provider_record, &live_provider)?;
314        let provider_id = provider_record.id.clone();
315        let operation_lock = self.operation_lock(&provider_id);
316        let mut fence = ProviderOperationFence::new(operation_lock.lock_owned().await);
317        fence.protect(live_provider.clone());
318        let generation = self.generation(&provider_id);
319        let result = match live_provider.try_discover_models().await {
320            Ok(models) => {
321                self.commit_new_provider(provider_record, live_provider, &models, generation)
322            }
323            Err(error) => Err(error.into()),
324        };
325        match result {
326            Ok((delta, replaced)) => {
327                fence.extend(replaced);
328                drop(fence);
329                Ok(delta)
330            }
331            Err(error) => {
332                drop(fence);
333                Err(error)
334            }
335        }
336    }
337
338    pub async fn install_pre_discovered_provider(
339        &self,
340        provider_record: StoredProvider,
341        live_provider: Arc<dyn Provider>,
342        models: Vec<crate::provider::DiscoveredModelDetails>,
343    ) -> Result<CatalogDelta, ProviderLifecycleError> {
344        self.validate_new_provider(&provider_record, &live_provider)?;
345        let provider_id = provider_record.id.clone();
346        let operation_lock = self.operation_lock(&provider_id);
347        let mut fence = ProviderOperationFence::new(operation_lock.lock_owned().await);
348        fence.protect(live_provider.clone());
349        let generation = self.generation(&provider_id);
350        let result = self.commit_new_provider(provider_record, live_provider, &models, generation);
351        match result {
352            Ok((delta, replaced)) => {
353                fence.extend(replaced);
354                drop(fence);
355                Ok(delta)
356            }
357            Err(error) => {
358                drop(fence);
359                Err(error)
360            }
361        }
362    }
363
364    fn commit_new_provider(
365        &self,
366        provider_record: StoredProvider,
367        live_provider: Arc<dyn Provider>,
368        models: &[crate::provider::DiscoveredModelDetails],
369        generation: u64,
370    ) -> Result<(CatalogDelta, Vec<Arc<dyn Provider>>), ProviderLifecycleError> {
371        let provider_id = provider_record.id.clone();
372        let mut state = self.lock_state();
373        self.ensure_generation(&state, &provider_id, generation)?;
374        let mut store = self.hub.load_auth()?;
375        if store
376            .providers
377            .iter()
378            .any(|provider| provider.id == provider_id)
379        {
380            return Err(ConfigError::Invalid(format!(
381                "auth provider id {provider_id:?} already exists"
382            ))
383            .into());
384        }
385        let mut expected_provider_ids = store
386            .providers
387            .iter()
388            .map(|provider| provider.id.clone())
389            .collect::<Vec<_>>();
390        expected_provider_ids.sort();
391        store.providers.push(provider_record.clone());
392        let descriptor = self.provider_descriptor(&provider_record, &store.providers)?;
393        let prepared = prepare_provider_catalog(descriptor, models)?;
394        let namespace = prepared.namespace().to_string();
395        let provider_kind = provider_record.kind.clone();
396        let (commit, applied) = self
397            .hub
398            .add_auth_provider_with_model_cache_details_if_provider_ids_and_then(
399                provider_record,
400                Some(&expected_provider_ids),
401                &namespace,
402                chrono::Utc::now().timestamp(),
403                models,
404                || {
405                    let registration = Self::register_live_provider(
406                        &mut state,
407                        &provider_id,
408                        provider_kind,
409                        live_provider,
410                    );
411                    (
412                        commit_prepared_provider_catalog(prepared),
413                        registration.replaced,
414                    )
415                },
416            )?;
417        match (commit, applied) {
418            (AuthProviderInsertCommit::Inserted, Some((delta, replaced_providers))) => {
419                Self::bump_generation(&mut state, &provider_id);
420                drop(state);
421                Ok((delta, replaced_providers))
422            }
423            (AuthProviderInsertCommit::Changed, _) => {
424                drop(state);
425                Err(ProviderLifecycleError::Stale { id: provider_id })
426            }
427            (AuthProviderInsertCommit::Inserted, None) => {
428                unreachable!("provider insertion callback was not run")
429            }
430        }
431    }
432
433    /// Return all in-memory live providers that must be reconciled with storage.
434    pub fn catalog_refresh_plan(&self) -> Vec<String> {
435        let state = self.lock_state();
436        let mut plan = state.live_providers.keys().cloned().collect::<Vec<_>>();
437        plan.sort();
438        plan
439    }
440
441    pub async fn refresh_models(
442        &self,
443        provider_id: &str,
444    ) -> Result<CatalogDelta, ProviderLifecycleError> {
445        let refresh_lock = self.refresh_lock(provider_id);
446        let _refresh = refresh_lock.lock_owned().await;
447        match self.run_catalog_refresh(provider_id, false).await? {
448            ProviderCatalogRefreshOutcome::CatalogUpdated(delta) => Ok(delta),
449            ProviderCatalogRefreshOutcome::NotNeeded
450            | ProviderCatalogRefreshOutcome::AlreadyInFlight => {
451                unreachable!("forced catalog refresh did not run")
452            }
453        }
454    }
455
456    /// Refresh one stale catalog without waiting behind an equivalent in-process refresh.
457    pub async fn refresh_models_if_stale(
458        &self,
459        provider_id: &str,
460    ) -> Result<ProviderCatalogRefreshOutcome, ProviderLifecycleError> {
461        let refresh_lock = self.refresh_lock(provider_id);
462        let Ok(_refresh) = refresh_lock.try_lock_owned() else {
463            return Ok(ProviderCatalogRefreshOutcome::AlreadyInFlight);
464        };
465        self.run_catalog_refresh(provider_id, true).await
466    }
467
468    async fn run_catalog_refresh(
469        &self,
470        provider_id: &str,
471        only_if_stale: bool,
472    ) -> Result<ProviderCatalogRefreshOutcome, ProviderLifecycleError> {
473        let operation_lock = self.operation_lock(provider_id);
474        let mut fence = ProviderOperationFence::new(operation_lock.lock_owned().await);
475        let (preparation, deferred) = self.prepare_catalog_refresh(
476            provider_id,
477            chrono::Utc::now().timestamp(),
478            only_if_stale,
479            None,
480            None,
481        );
482        fence.extend(deferred);
483        let attempt = match preparation? {
484            ProviderRefreshPreparation::Hydrated { delta, changed } => {
485                drop(fence);
486                return Ok(if changed {
487                    ProviderCatalogRefreshOutcome::CatalogUpdated(delta)
488                } else {
489                    ProviderCatalogRefreshOutcome::NotNeeded
490                });
491            }
492            ProviderRefreshPreparation::Discover(attempt) => attempt,
493        };
494        drop(fence);
495
496        let discovery = attempt.live_provider.try_discover_models().await;
497        let operation_lock = self.operation_lock(provider_id);
498        let mut fence = ProviderOperationFence::new(operation_lock.lock_owned().await);
499        fence.protect(attempt.live_provider.clone());
500        let (result, deferred) = match discovery {
501            Ok(models) => self.finish_catalog_refresh(provider_id, &attempt, &models),
502            Err(error) => {
503                self.reconcile_refresh_failure(provider_id, &attempt, error.into(), Vec::new())
504            }
505        };
506        fence.extend(deferred);
507        drop(fence);
508        result.map(ProviderCatalogRefreshOutcome::CatalogUpdated)
509    }
510
511    fn prepare_catalog_refresh(
512        &self,
513        provider_id: &str,
514        now: i64,
515        hydrate_if_fresh: bool,
516        expected_live_provider: Option<&Arc<dyn Provider>>,
517        expected_catalog_snapshot: Option<&crate::auth_store::AuthProviderCatalogSnapshot>,
518    ) -> ProviderRefreshPreparationResult {
519        for _ in 0..3 {
520            let (attempt, mut deferred) = self.load_catalog_refresh_attempt(provider_id, now);
521            let attempt = match attempt {
522                Ok(attempt) => attempt,
523                Err(error) => return (Err(error), deferred),
524            };
525            if expected_live_provider
526                .is_some_and(|expected| !Arc::ptr_eq(expected, &attempt.live_provider))
527            {
528                return (
529                    Err(ProviderLifecycleError::Stale {
530                        id: provider_id.to_string(),
531                    }),
532                    deferred,
533                );
534            }
535            if !hydrate_if_fresh
536                || attempt.runtime.model_cache_freshness
537                    != crate::auth_store::ModelCacheFreshness::Fresh
538                || expected_catalog_snapshot
539                    .is_some_and(|expected| expected == &attempt.runtime.catalog_snapshot)
540            {
541                return (
542                    Ok(ProviderRefreshPreparation::Discover(Box::new(attempt))),
543                    deferred,
544                );
545            }
546            let (hydrated, hydrate_deferred) =
547                self.hydrate_cached_catalog_if_current(provider_id, &attempt);
548            deferred.extend(hydrate_deferred);
549            match hydrated {
550                Ok((delta, changed)) => {
551                    return (
552                        Ok(ProviderRefreshPreparation::Hydrated { delta, changed }),
553                        deferred,
554                    );
555                }
556                Err(ProviderLifecycleError::Stale { .. }) => continue,
557                Err(error) => return (Err(error), deferred),
558            }
559        }
560        let (authoritative, deferred) = self.load_catalog_refresh_attempt(provider_id, now);
561        match authoritative {
562            Ok(attempt)
563                if expected_live_provider
564                    .is_none_or(|expected| Arc::ptr_eq(expected, &attempt.live_provider)) =>
565            {
566                (
567                    Err(ProviderLifecycleError::Stale {
568                        id: provider_id.to_string(),
569                    }),
570                    deferred,
571                )
572            }
573            Ok(_) => (
574                Err(ProviderLifecycleError::Stale {
575                    id: provider_id.to_string(),
576                }),
577                deferred,
578            ),
579            Err(error) => (Err(error), deferred),
580        }
581    }
582
583    fn load_catalog_refresh_attempt(
584        &self,
585        provider_id: &str,
586        now: i64,
587    ) -> ProviderRefreshAttemptResult {
588        let mut state = self.lock_state();
589        let runtime = self
590            .hub
591            .load_or_create_auth_provider_runtime_state_at(provider_id, now);
592        let runtime = match runtime {
593            Ok(Some(runtime)) => runtime,
594            Ok(None) => {
595                Self::bump_generation(&mut state, provider_id);
596                let (_, removed) = Self::remove_runtime_provider(&mut state, provider_id, false);
597                drop(state);
598                return (
599                    Err(ProviderLifecycleError::ProviderNotFound {
600                        id: provider_id.to_string(),
601                    }),
602                    removed,
603                );
604            }
605            Err(error) => {
606                Self::bump_generation(&mut state, provider_id);
607                let (_, removed) = Self::remove_runtime_provider(&mut state, provider_id, false);
608                drop(state);
609                return (Err(error.into()), removed);
610            }
611        };
612        if !runtime.provider.enabled {
613            Self::bump_generation(&mut state, provider_id);
614            let (_, removed) = Self::remove_runtime_provider(&mut state, provider_id, false);
615            drop(state);
616            return (
617                Err(ProviderLifecycleError::ProviderDisabled {
618                    id: provider_id.to_string(),
619                }),
620                removed,
621            );
622        }
623        let Some(live) = state.live_providers.get(provider_id) else {
624            Self::bump_generation(&mut state, provider_id);
625            let (_, removed) = Self::remove_runtime_provider(&mut state, provider_id, false);
626            drop(state);
627            return (
628                Err(ProviderLifecycleError::LiveProviderMissing {
629                    id: provider_id.to_string(),
630                }),
631                removed,
632            );
633        };
634        if live.kind != runtime.provider.kind {
635            Self::bump_generation(&mut state, provider_id);
636            let (_, removed) = Self::remove_runtime_provider(&mut state, provider_id, false);
637            drop(state);
638            return (
639                Err(ProviderLifecycleError::Stale {
640                    id: provider_id.to_string(),
641                }),
642                removed,
643            );
644        }
645        let attempt = ProviderRefreshAttempt {
646            generation: Self::generation_in(&state, provider_id),
647            runtime,
648            live_provider: live.provider.clone(),
649        };
650        (Ok(attempt), Vec::new())
651    }
652
653    fn hydrate_cached_catalog_if_current(
654        &self,
655        provider_id: &str,
656        attempt: &ProviderRefreshAttempt,
657    ) -> ProviderCatalogHydrationResult {
658        let mut state = self.lock_state();
659        if let Err(error) = self.ensure_generation(&state, provider_id, attempt.generation) {
660            return (Err(error), Vec::new());
661        }
662        let Some(current_live) = state.live_providers.get(provider_id) else {
663            return (
664                Err(ProviderLifecycleError::Stale {
665                    id: provider_id.to_string(),
666                }),
667                Vec::new(),
668            );
669        };
670        if current_live.kind != attempt.runtime.provider.kind
671            || !Arc::ptr_eq(&attempt.live_provider, &current_live.provider)
672        {
673            return (
674                Err(ProviderLifecycleError::Stale {
675                    id: provider_id.to_string(),
676                }),
677                Vec::new(),
678            );
679        }
680        let Some(models) = attempt.runtime.model_cache.as_deref() else {
681            return (
682                Err(ProviderLifecycleError::Stale {
683                    id: provider_id.to_string(),
684                }),
685                Vec::new(),
686            );
687        };
688        let descriptor = match self.provider_descriptor_from_state(
689            &attempt.runtime.provider,
690            &attempt.runtime.provider_ids,
691            attempt.runtime.model_namespace.as_deref(),
692        ) {
693            Ok(descriptor) => descriptor,
694            Err(error) => return (Err(error), Vec::new()),
695        };
696        let prepared = match prepare_provider_catalog(descriptor, models) {
697            Ok(prepared) => prepared,
698            Err(error) => return (Err(error.into()), Vec::new()),
699        };
700        let namespace = prepared.namespace().to_string();
701        let revision = crate::model_registry::model_catalog_revision();
702        let expected_provider_ids = attempt
703            .runtime
704            .model_namespace
705            .is_none()
706            .then_some(attempt.runtime.provider_ids.as_slice());
707        let commit = self.hub.commit_auth_provider_runtime_if_current_and_then(
708            &attempt.runtime.provider,
709            &attempt.runtime.catalog_snapshot,
710            false,
711            Some(&namespace),
712            expected_provider_ids,
713            || {
714                let delta = commit_prepared_provider_catalog(prepared);
715                let changed = crate::model_registry::model_catalog_revision() != revision;
716                (delta, changed)
717            },
718        );
719        let (commit, hydrated) = match commit {
720            Ok(result) => result,
721            Err(error) => return (Err(error.into()), Vec::new()),
722        };
723        match commit {
724            AuthProviderRuntimeCommit::Applied { .. } => {
725                drop(state);
726                (
727                    Ok(hydrated.expect("catalog hydration callback was not run")),
728                    Vec::new(),
729                )
730            }
731            AuthProviderRuntimeCommit::Missing => {
732                Self::bump_generation(&mut state, provider_id);
733                let (_, removed) = Self::remove_runtime_provider(&mut state, provider_id, false);
734                drop(state);
735                (
736                    Err(ProviderLifecycleError::ProviderNotFound {
737                        id: provider_id.to_string(),
738                    }),
739                    removed,
740                )
741            }
742            AuthProviderRuntimeCommit::Disabled => {
743                Self::bump_generation(&mut state, provider_id);
744                let (_, removed) = Self::remove_runtime_provider(&mut state, provider_id, false);
745                drop(state);
746                (
747                    Err(ProviderLifecycleError::ProviderDisabled {
748                        id: provider_id.to_string(),
749                    }),
750                    removed,
751                )
752            }
753            AuthProviderRuntimeCommit::Changed => {
754                drop(state);
755                (
756                    Err(ProviderLifecycleError::Stale {
757                        id: provider_id.to_string(),
758                    }),
759                    Vec::new(),
760                )
761            }
762        }
763    }
764
765    fn finish_catalog_refresh(
766        &self,
767        provider_id: &str,
768        attempt: &ProviderRefreshAttempt,
769        models: &[crate::provider::DiscoveredModelDetails],
770    ) -> (
771        Result<CatalogDelta, ProviderLifecycleError>,
772        Vec<Arc<dyn Provider>>,
773    ) {
774        let ProviderRefreshAttempt {
775            generation,
776            runtime,
777            live_provider,
778        } = attempt;
779        let deferred = Vec::new();
780        let state = self.lock_state();
781        if let Err(error) = self.ensure_generation(&state, provider_id, *generation) {
782            drop(state);
783            return self.finish_refresh_failure(
784                provider_id,
785                &runtime.provider.kind,
786                *generation,
787                live_provider,
788                error,
789                deferred,
790            );
791        }
792        let Some(current_live) = state.live_providers.get(provider_id) else {
793            drop(state);
794            return self.finish_refresh_failure(
795                provider_id,
796                &runtime.provider.kind,
797                *generation,
798                live_provider,
799                ProviderLifecycleError::Stale {
800                    id: provider_id.to_string(),
801                },
802                deferred,
803            );
804        };
805        if current_live.kind != runtime.provider.kind
806            || !Arc::ptr_eq(live_provider, &current_live.provider)
807        {
808            drop(state);
809            return self.finish_refresh_failure(
810                provider_id,
811                &runtime.provider.kind,
812                *generation,
813                live_provider,
814                ProviderLifecycleError::Stale {
815                    id: provider_id.to_string(),
816                },
817                deferred,
818            );
819        }
820
821        let descriptor = match self.provider_descriptor_from_state(
822            &runtime.provider,
823            &runtime.provider_ids,
824            runtime.model_namespace.as_deref(),
825        ) {
826            Ok(descriptor) => descriptor,
827            Err(error) => {
828                drop(state);
829                return self.finish_refresh_failure(
830                    provider_id,
831                    &runtime.provider.kind,
832                    *generation,
833                    live_provider,
834                    error,
835                    deferred,
836                );
837            }
838        };
839        let prepared = match prepare_provider_catalog(descriptor, models) {
840            Ok(prepared) => prepared,
841            Err(error) => {
842                drop(state);
843                return self.finish_refresh_failure(
844                    provider_id,
845                    &runtime.provider.kind,
846                    *generation,
847                    live_provider,
848                    error.into(),
849                    deferred,
850                );
851            }
852        };
853        let expected_provider_ids = runtime
854            .model_namespace
855            .is_none()
856            .then_some(runtime.provider_ids.as_slice());
857        let result = self.commit_cache_and_catalog_if_current(
858            &runtime.provider,
859            &runtime.catalog_snapshot,
860            expected_provider_ids,
861            chrono::Utc::now().timestamp(),
862            models,
863            prepared,
864        );
865        drop(state);
866        match result {
867            Ok(delta) => (Ok(delta), deferred),
868            Err(error @ ProviderLifecycleError::Stale { .. }) => {
869                self.reconcile_refresh_failure(provider_id, attempt, error, deferred)
870            }
871            Err(error) => self.finish_refresh_failure(
872                provider_id,
873                &runtime.provider.kind,
874                *generation,
875                live_provider,
876                error,
877                deferred,
878            ),
879        }
880    }
881
882    fn reconcile_refresh_failure(
883        &self,
884        provider_id: &str,
885        attempt: &ProviderRefreshAttempt,
886        original: ProviderLifecycleError,
887        mut deferred: Vec<Arc<dyn Provider>>,
888    ) -> (
889        Result<CatalogDelta, ProviderLifecycleError>,
890        Vec<Arc<dyn Provider>>,
891    ) {
892        let (preparation, reconciled_deferred) = self.prepare_catalog_refresh(
893            provider_id,
894            chrono::Utc::now().timestamp(),
895            true,
896            Some(&attempt.live_provider),
897            Some(&attempt.runtime.catalog_snapshot),
898        );
899        deferred.extend(reconciled_deferred);
900        match preparation {
901            Ok(ProviderRefreshPreparation::Hydrated { delta, .. }) => (Ok(delta), deferred),
902            Ok(ProviderRefreshPreparation::Discover(_)) => self.finish_refresh_failure(
903                provider_id,
904                &attempt.runtime.provider.kind,
905                attempt.generation,
906                &attempt.live_provider,
907                original,
908                deferred,
909            ),
910            Err(error) => self.finish_refresh_failure(
911                provider_id,
912                &attempt.runtime.provider.kind,
913                attempt.generation,
914                &attempt.live_provider,
915                error,
916                deferred,
917            ),
918        }
919    }
920
921    fn finish_refresh_failure(
922        &self,
923        provider_id: &str,
924        expected_kind: &ProviderKind,
925        expected_generation: u64,
926        expected_provider: &Arc<dyn Provider>,
927        original: ProviderLifecycleError,
928        mut deferred: Vec<Arc<dyn Provider>>,
929    ) -> (
930        Result<CatalogDelta, ProviderLifecycleError>,
931        Vec<Arc<dyn Provider>>,
932    ) {
933        let catalog_identity_invalid = matches!(
934            &original,
935            ProviderLifecycleError::Catalog(
936                CatalogError::NamespaceChanged { .. }
937                    | CatalogError::NamespaceInUse { .. }
938                    | CatalogError::NamespaceStore { .. }
939            )
940        );
941        let mut state = self.lock_state();
942        let authoritative = self
943            .hub
944            .load_or_create_auth_provider_runtime_state(provider_id);
945        let error = match authoritative {
946            Err(error) => error.into(),
947            Ok(None) => ProviderLifecycleError::Stale {
948                id: provider_id.to_string(),
949            },
950            Ok(Some(runtime))
951                if !runtime.provider.enabled || runtime.provider.kind != *expected_kind =>
952            {
953                ProviderLifecycleError::Stale {
954                    id: provider_id.to_string(),
955                }
956            }
957            Ok(Some(_)) => {
958                let same_generation =
959                    Self::generation_in(&state, provider_id) == expected_generation;
960                let current = state.live_providers.get(provider_id);
961                let same_provider = current.is_some_and(|current| {
962                    current.kind == *expected_kind
963                        && Arc::ptr_eq(&current.provider, expected_provider)
964                });
965                if same_generation && same_provider && !catalog_identity_invalid {
966                    drop(state);
967                    return (Err(original), deferred);
968                }
969                if catalog_identity_invalid {
970                    original
971                } else if current.is_some_and(|current| current.kind == *expected_kind) {
972                    drop(state);
973                    return (
974                        Err(ProviderLifecycleError::Stale {
975                            id: provider_id.to_string(),
976                        }),
977                        deferred,
978                    );
979                } else {
980                    ProviderLifecycleError::Stale {
981                        id: provider_id.to_string(),
982                    }
983                }
984            }
985        };
986        Self::bump_generation(&mut state, provider_id);
987        let (_, removed) = Self::remove_runtime_provider(&mut state, provider_id, false);
988        deferred.extend(removed);
989        drop(state);
990        (Err(error), deferred)
991    }
992
993    pub async fn restore_provider(
994        &self,
995        provider_id: &str,
996        expected_kind: ProviderKind,
997        live_provider: Arc<dyn Provider>,
998    ) -> Result<ProviderLifecycleOutcome, ProviderLifecycleError> {
999        self.validate_live_name(provider_id, &live_provider)?;
1000        let operation_lock = self.operation_lock(provider_id);
1001        let mut fence = ProviderOperationFence::new(operation_lock.lock_owned().await);
1002        fence.protect(live_provider.clone());
1003        let (result, deferred) =
1004            self.commit_existing_provider_runtime(provider_id, expected_kind, live_provider, false);
1005        fence.extend(deferred);
1006        drop(fence);
1007        result
1008    }
1009
1010    pub async fn enable_provider(
1011        &self,
1012        provider_id: &str,
1013        expected_kind: ProviderKind,
1014        live_provider: Arc<dyn Provider>,
1015    ) -> Result<ProviderLifecycleOutcome, ProviderLifecycleError> {
1016        self.validate_live_name(provider_id, &live_provider)?;
1017        let operation_lock = self.operation_lock(provider_id);
1018        let mut fence = ProviderOperationFence::new(operation_lock.lock_owned().await);
1019        fence.protect(live_provider.clone());
1020        let (result, deferred) =
1021            self.commit_existing_provider_runtime(provider_id, expected_kind, live_provider, true);
1022        fence.extend(deferred);
1023        drop(fence);
1024        result
1025    }
1026
1027    pub fn reconcile_inactive_providers(
1028        &self,
1029    ) -> Result<ProviderReconcileOutcome, ProviderLifecycleError> {
1030        let mut state = self.lock_state();
1031        let removals = self
1032            .hub
1033            .with_auth_provider_runtime_descriptors(|providers| {
1034                let active = providers
1035                    .iter()
1036                    .filter(|provider| provider.enabled)
1037                    .map(|provider| (provider.id.as_str(), &provider.kind))
1038                    .collect::<HashMap<_, _>>();
1039                let mut inactive = state
1040                    .live_providers
1041                    .iter()
1042                    .filter(|(provider_id, live)| {
1043                        active
1044                            .get(provider_id.as_str())
1045                            .is_none_or(|kind| **kind != live.kind)
1046                    })
1047                    .map(|(provider_id, _)| provider_id.clone())
1048                    .collect::<BTreeSet<_>>();
1049                inactive.extend(
1050                    providers
1051                        .iter()
1052                        .filter(|provider| !provider.enabled)
1053                        .map(|provider| provider.id.clone()),
1054                );
1055
1056                inactive
1057                    .into_iter()
1058                    .map(|provider_id| {
1059                        Self::bump_generation(&mut state, &provider_id);
1060                        let removal = Self::remove_live_provider(&mut state, &provider_id);
1061                        (provider_id, removal)
1062                    })
1063                    .collect::<Vec<_>>()
1064            });
1065        let removals = match removals {
1066            Ok(removals) => removals,
1067            Err(error) => {
1068                let provider_ids = state
1069                    .live_providers
1070                    .keys()
1071                    .cloned()
1072                    .collect::<BTreeSet<_>>();
1073                let mut removed_providers = Vec::new();
1074                for provider_id in provider_ids {
1075                    Self::bump_generation(&mut state, &provider_id);
1076                    let removal = Self::remove_live_provider(&mut state, &provider_id);
1077                    removed_providers.extend(removal.removed);
1078                    remove_provider_catalog(&provider_id);
1079                }
1080                drop(state);
1081                drop(removed_providers);
1082                return Err(error.into());
1083            }
1084        };
1085        let mut removed_providers = Vec::new();
1086        let providers = removals
1087            .into_iter()
1088            .map(|(provider_id, removal)| {
1089                removed_providers.extend(removal.removed);
1090                let change = ProviderStateChange {
1091                    auth_changed: false,
1092                    live_changed: removal.changed,
1093                    catalog_changed: remove_provider_catalog(&provider_id),
1094                };
1095                (provider_id, change)
1096            })
1097            .collect();
1098        drop(state);
1099        drop(removed_providers);
1100        Ok(ProviderReconcileOutcome { providers })
1101    }
1102
1103    pub fn disable_provider(
1104        &self,
1105        provider_id: &str,
1106    ) -> Result<ProviderStateChange, ProviderLifecycleError> {
1107        let mut state = self.lock_state();
1108        let auth_changed = match self
1109            .hub
1110            .set_auth_provider_enabled_with_change(provider_id, false)
1111        {
1112            Ok(Some(auth_changed)) => auth_changed,
1113            Ok(None) => {
1114                Self::bump_generation(&mut state, provider_id);
1115                let (_, removed) = Self::remove_runtime_provider(&mut state, provider_id, false);
1116                drop(state);
1117                drop(removed);
1118                return Err(ProviderLifecycleError::ProviderNotFound {
1119                    id: provider_id.to_string(),
1120                });
1121            }
1122            Err(error) => {
1123                Self::bump_generation(&mut state, provider_id);
1124                let (_, removed) = Self::remove_runtime_provider(&mut state, provider_id, false);
1125                drop(state);
1126                drop(removed);
1127                return Err(error.into());
1128            }
1129        };
1130        Self::bump_generation(&mut state, provider_id);
1131        let (change, removed) =
1132            Self::remove_runtime_provider(&mut state, provider_id, auth_changed);
1133        drop(state);
1134        drop(removed);
1135        Ok(change)
1136    }
1137
1138    pub fn remove_provider(
1139        &self,
1140        provider_id: &str,
1141    ) -> Result<ProviderStateChange, ProviderLifecycleError> {
1142        let mut state = self.lock_state();
1143        let auth_changed = match self.hub.remove_auth_provider(provider_id) {
1144            Ok(auth_changed) => auth_changed,
1145            Err(error) => {
1146                Self::bump_generation(&mut state, provider_id);
1147                let (_, removed) = Self::remove_runtime_provider(&mut state, provider_id, false);
1148                drop(state);
1149                drop(removed);
1150                return Err(error.into());
1151            }
1152        };
1153        Self::bump_generation(&mut state, provider_id);
1154        let (change, removed) =
1155            Self::remove_runtime_provider(&mut state, provider_id, auth_changed);
1156        drop(state);
1157        drop(removed);
1158        Ok(change)
1159    }
1160
1161    fn commit_existing_provider_runtime(
1162        &self,
1163        provider_id: &str,
1164        expected_kind: ProviderKind,
1165        live_provider: Arc<dyn Provider>,
1166        enable_if_disabled: bool,
1167    ) -> (
1168        Result<ProviderLifecycleOutcome, ProviderLifecycleError>,
1169        Vec<Arc<dyn Provider>>,
1170    ) {
1171        let mut state = self.lock_state();
1172        let runtime = self
1173            .hub
1174            .load_or_create_auth_provider_runtime_state(provider_id);
1175        let runtime = match runtime {
1176            Ok(Some(runtime)) => runtime,
1177            Ok(None) => {
1178                Self::bump_generation(&mut state, provider_id);
1179                let (_, removed) = Self::remove_runtime_provider(&mut state, provider_id, false);
1180                drop(state);
1181                return (
1182                    Err(ProviderLifecycleError::ProviderNotFound {
1183                        id: provider_id.to_string(),
1184                    }),
1185                    removed,
1186                );
1187            }
1188            Err(error) => {
1189                Self::bump_generation(&mut state, provider_id);
1190                let (_, removed) = Self::remove_runtime_provider(&mut state, provider_id, false);
1191                drop(state);
1192                return (Err(error.into()), removed);
1193            }
1194        };
1195        let provider = runtime.provider;
1196        if provider.kind != expected_kind {
1197            Self::bump_generation(&mut state, provider_id);
1198            let (_, removed) = Self::remove_runtime_provider(&mut state, provider_id, false);
1199            drop(state);
1200            return (
1201                Err(ProviderLifecycleError::Stale {
1202                    id: provider_id.to_string(),
1203                }),
1204                removed,
1205            );
1206        }
1207        if !provider.enabled && !enable_if_disabled {
1208            let validation = self.hub.commit_auth_provider_runtime_if_current_and_then(
1209                &provider,
1210                &runtime.catalog_snapshot,
1211                false,
1212                None,
1213                None,
1214                || (),
1215            );
1216            let error = match validation {
1217                Ok((AuthProviderRuntimeCommit::Missing, _)) => {
1218                    ProviderLifecycleError::ProviderNotFound {
1219                        id: provider_id.to_string(),
1220                    }
1221                }
1222                Ok((AuthProviderRuntimeCommit::Disabled, _)) => {
1223                    ProviderLifecycleError::ProviderDisabled {
1224                        id: provider_id.to_string(),
1225                    }
1226                }
1227                Ok((
1228                    AuthProviderRuntimeCommit::Changed | AuthProviderRuntimeCommit::Applied { .. },
1229                    _,
1230                )) => ProviderLifecycleError::Stale {
1231                    id: provider_id.to_string(),
1232                },
1233                Err(error) => error.into(),
1234            };
1235            Self::bump_generation(&mut state, provider_id);
1236            let (_, removed) = Self::remove_runtime_provider(&mut state, provider_id, false);
1237            drop(state);
1238            return (Err(error), removed);
1239        }
1240        let prepared = match runtime.model_cache.as_deref() {
1241            Some(models) => self
1242                .provider_descriptor_from_state(
1243                    &provider,
1244                    &runtime.provider_ids,
1245                    runtime.model_namespace.as_deref(),
1246                )
1247                .and_then(|descriptor| {
1248                    prepare_provider_catalog(descriptor, models).map_err(Into::into)
1249                })
1250                .map(Some),
1251            None => Ok(None),
1252        };
1253        let prepared = match prepared {
1254            Ok(prepared) => prepared,
1255            Err(error) => {
1256                Self::bump_generation(&mut state, provider_id);
1257                let (_, removed) = Self::remove_runtime_provider(&mut state, provider_id, false);
1258                drop(state);
1259                return (Err(error), removed);
1260            }
1261        };
1262        let model_namespace = prepared
1263            .as_ref()
1264            .map(|prepared| prepared.namespace().to_string());
1265        let selected_provider = state
1266            .live_providers
1267            .get(provider_id)
1268            .filter(|current| current.kind == provider.kind)
1269            .map(|current| current.provider.clone())
1270            .unwrap_or_else(|| live_provider.clone());
1271        let catalog_revision = prepared
1272            .is_some()
1273            .then(crate::model_registry::model_catalog_revision);
1274        let expected_provider_ids = runtime
1275            .model_namespace
1276            .is_none()
1277            .then_some(runtime.provider_ids.as_slice());
1278        let commit = self.hub.commit_auth_provider_runtime_if_current_and_then(
1279            &provider,
1280            &runtime.catalog_snapshot,
1281            enable_if_disabled,
1282            model_namespace.as_deref(),
1283            expected_provider_ids,
1284            || {
1285                let registration = Self::register_live_provider(
1286                    &mut state,
1287                    provider_id,
1288                    provider.kind.clone(),
1289                    selected_provider,
1290                );
1291                let (catalog, catalog_changed) = match prepared {
1292                    Some(prepared) => {
1293                        let catalog = commit_prepared_provider_catalog(prepared);
1294                        let changed = catalog_revision.is_some_and(|revision| {
1295                            crate::model_registry::model_catalog_revision() != revision
1296                        });
1297                        (Some(catalog), changed)
1298                    }
1299                    None => (None, remove_provider_catalog(provider_id)),
1300                };
1301                (catalog, catalog_changed, registration)
1302            },
1303        );
1304        let (commit, applied) = match commit {
1305            Ok(commit) => commit,
1306            Err(error) => {
1307                Self::bump_generation(&mut state, provider_id);
1308                let (_, removed) = Self::remove_runtime_provider(&mut state, provider_id, false);
1309                drop(state);
1310                return (Err(error.into()), removed);
1311            }
1312        };
1313
1314        match commit {
1315            AuthProviderRuntimeCommit::Applied { auth_changed } => {
1316                let (catalog, catalog_changed, registration) =
1317                    applied.expect("applied provider runtime callback was not run");
1318                if auth_changed || catalog_changed || registration.changed {
1319                    Self::bump_generation(&mut state, provider_id);
1320                }
1321                drop(state);
1322                (
1323                    Ok(ProviderLifecycleOutcome {
1324                        state: ProviderStateChange {
1325                            auth_changed,
1326                            live_changed: registration.changed,
1327                            catalog_changed,
1328                        },
1329                        catalog,
1330                    }),
1331                    registration.replaced,
1332                )
1333            }
1334            AuthProviderRuntimeCommit::Missing => {
1335                Self::bump_generation(&mut state, provider_id);
1336                let (_, removed) = Self::remove_runtime_provider(&mut state, provider_id, false);
1337                drop(state);
1338                (
1339                    Err(ProviderLifecycleError::ProviderNotFound {
1340                        id: provider_id.to_string(),
1341                    }),
1342                    removed,
1343                )
1344            }
1345            AuthProviderRuntimeCommit::Disabled => {
1346                Self::bump_generation(&mut state, provider_id);
1347                let (_, removed) = Self::remove_runtime_provider(&mut state, provider_id, false);
1348                drop(state);
1349                (
1350                    Err(ProviderLifecycleError::ProviderDisabled {
1351                        id: provider_id.to_string(),
1352                    }),
1353                    removed,
1354                )
1355            }
1356            AuthProviderRuntimeCommit::Changed => {
1357                Self::bump_generation(&mut state, provider_id);
1358                let (_, removed) = Self::remove_runtime_provider(&mut state, provider_id, false);
1359                drop(state);
1360                (
1361                    Err(ProviderLifecycleError::Stale {
1362                        id: provider_id.to_string(),
1363                    }),
1364                    removed,
1365                )
1366            }
1367        }
1368    }
1369
1370    fn provider_descriptor(
1371        &self,
1372        provider: &StoredProvider,
1373        providers: &[StoredProvider],
1374    ) -> Result<ProviderDescriptor, ProviderLifecycleError> {
1375        let provider_ids = providers
1376            .iter()
1377            .map(|provider| provider.id.clone())
1378            .collect::<Vec<_>>();
1379        let persisted_namespace = self.hub.load_auth_model_namespace(&provider.id)?;
1380        self.provider_descriptor_from_state(provider, &provider_ids, persisted_namespace.as_deref())
1381    }
1382
1383    fn provider_descriptor_from_state(
1384        &self,
1385        provider: &StoredProvider,
1386        provider_ids: &[String],
1387        persisted_namespace: Option<&str>,
1388    ) -> Result<ProviderDescriptor, ProviderLifecycleError> {
1389        let in_memory_namespace = provider_catalog_namespace(&provider.id);
1390        if let (Some(current), Some(persisted)) = (&in_memory_namespace, persisted_namespace)
1391            && current != persisted
1392        {
1393            return Err(CatalogError::NamespaceChanged {
1394                provider_key: provider.id.clone(),
1395                current: current.clone(),
1396                requested: persisted.to_string(),
1397            }
1398            .into());
1399        }
1400        let namespace = in_memory_namespace
1401            .or_else(|| persisted_namespace.map(str::to_string))
1402            .unwrap_or_else(|| {
1403                let short_id = shortest_unique_provider_id(&provider.id, provider_ids);
1404                format!("{short_id}@{}", provider.name)
1405            });
1406        Ok(ProviderDescriptor {
1407            provider_key: provider.id.clone(),
1408            provider_name: provider.name.clone(),
1409            namespace,
1410            wire_profile: wire_profile_for_kind(&provider.kind),
1411        })
1412    }
1413
1414    fn commit_cache_and_catalog_if_current(
1415        &self,
1416        provider: &StoredProvider,
1417        expected_catalog: &crate::auth_store::AuthProviderCatalogSnapshot,
1418        expected_provider_ids: Option<&[String]>,
1419        fetched_at: i64,
1420        models: &[crate::provider::DiscoveredModelDetails],
1421        prepared: PreparedProviderCatalog,
1422    ) -> Result<CatalogDelta, ProviderLifecycleError> {
1423        let namespace = prepared.namespace().to_string();
1424        let (commit, delta) = self
1425            .hub
1426            .update_auth_model_cache_details_if_enabled_and_then(
1427                AuthModelCacheUpdate {
1428                    expected: provider,
1429                    expected_catalog,
1430                    expected_provider_ids,
1431                    model_namespace: &namespace,
1432                    fetched_at,
1433                    models,
1434                },
1435                || commit_prepared_provider_catalog(prepared),
1436            )?;
1437        match (commit, delta) {
1438            (AuthModelCacheCommit::Updated, Some(delta)) => Ok(delta),
1439            (
1440                AuthModelCacheCommit::Missing
1441                | AuthModelCacheCommit::Disabled
1442                | AuthModelCacheCommit::Changed,
1443                _,
1444            ) => Err(ProviderLifecycleError::Stale {
1445                id: provider.id.clone(),
1446            }),
1447            (AuthModelCacheCommit::Updated, None) => unreachable!("updated callback was not run"),
1448        }
1449    }
1450
1451    fn validate_live_name(
1452        &self,
1453        provider_id: &str,
1454        live_provider: &Arc<dyn Provider>,
1455    ) -> Result<(), ProviderLifecycleError> {
1456        if live_provider.name() == provider_id {
1457            return Ok(());
1458        }
1459        Err(ProviderLifecycleError::LiveProviderNameMismatch {
1460            expected: provider_id.to_string(),
1461            actual: live_provider.name().to_string(),
1462        })
1463    }
1464
1465    fn validate_new_provider(
1466        &self,
1467        provider: &StoredProvider,
1468        live_provider: &Arc<dyn Provider>,
1469    ) -> Result<(), ProviderLifecycleError> {
1470        self.validate_live_name(&provider.id, live_provider)?;
1471        if provider.enabled {
1472            return Ok(());
1473        }
1474        Err(ProviderLifecycleError::ProviderDisabled {
1475            id: provider.id.clone(),
1476        })
1477    }
1478
1479    fn operation_lock(&self, provider_id: &str) -> Arc<tokio::sync::Mutex<()>> {
1480        self.lock_state()
1481            .operation_locks
1482            .entry(provider_id.to_string())
1483            .or_default()
1484            .clone()
1485    }
1486
1487    fn refresh_lock(&self, provider_id: &str) -> Arc<tokio::sync::Mutex<()>> {
1488        self.lock_state()
1489            .refresh_locks
1490            .entry(provider_id.to_string())
1491            .or_default()
1492            .clone()
1493    }
1494
1495    fn generation(&self, provider_id: &str) -> u64 {
1496        Self::generation_in(&self.lock_state(), provider_id)
1497    }
1498
1499    fn generation_in(state: &LifecycleState, provider_id: &str) -> u64 {
1500        state.generations.get(provider_id).copied().unwrap_or(0)
1501    }
1502
1503    fn ensure_generation(
1504        &self,
1505        state: &LifecycleState,
1506        provider_id: &str,
1507        expected: u64,
1508    ) -> Result<(), ProviderLifecycleError> {
1509        if Self::generation_in(state, provider_id) == expected {
1510            return Ok(());
1511        }
1512        Err(ProviderLifecycleError::Stale {
1513            id: provider_id.to_string(),
1514        })
1515    }
1516
1517    fn bump_generation(state: &mut LifecycleState, provider_id: &str) {
1518        let generation = state
1519            .generations
1520            .entry(provider_id.to_string())
1521            .or_default();
1522        *generation = generation.wrapping_add(1);
1523    }
1524
1525    fn register_live_provider(
1526        state: &mut LifecycleState,
1527        provider_id: &str,
1528        kind: ProviderKind,
1529        provider: Arc<dyn Provider>,
1530    ) -> LiveProviderRegistration {
1531        let mut changed = state.live_providers.get(provider_id).is_none_or(|current| {
1532            current.kind != kind || !Arc::ptr_eq(&current.provider, &provider)
1533        });
1534        let mut replaced = state
1535            .live_providers
1536            .insert(
1537                provider_id.to_string(),
1538                LiveProviderEntry {
1539                    kind,
1540                    provider: provider.clone(),
1541                },
1542            )
1543            .into_iter()
1544            .map(|live| live.provider)
1545            .collect::<Vec<_>>();
1546        state.registries.retain(|registry| {
1547            let Some(registry) = registry.upgrade() else {
1548                return false;
1549            };
1550            let previous = registry.register_named(provider_id.to_string(), provider.clone());
1551            changed |= previous
1552                .as_ref()
1553                .is_none_or(|previous| !Arc::ptr_eq(previous, &provider));
1554            replaced.extend(previous);
1555            true
1556        });
1557        LiveProviderRegistration { changed, replaced }
1558    }
1559
1560    fn remove_live_provider(state: &mut LifecycleState, provider_id: &str) -> LiveProviderRemoval {
1561        let mut removed = state
1562            .live_providers
1563            .remove(provider_id)
1564            .map(|live| live.provider)
1565            .into_iter()
1566            .collect::<Vec<_>>();
1567        state.registries.retain(|registry| {
1568            let Some(registry) = registry.upgrade() else {
1569                return false;
1570            };
1571            removed.extend(registry.take_named(provider_id));
1572            true
1573        });
1574        LiveProviderRemoval {
1575            changed: !removed.is_empty(),
1576            removed,
1577        }
1578    }
1579
1580    fn remove_runtime_provider(
1581        state: &mut LifecycleState,
1582        provider_id: &str,
1583        auth_changed: bool,
1584    ) -> (ProviderStateChange, Vec<Arc<dyn Provider>>) {
1585        let removal = Self::remove_live_provider(state, provider_id);
1586        (
1587            ProviderStateChange {
1588                auth_changed,
1589                live_changed: removal.changed,
1590                catalog_changed: remove_provider_catalog(provider_id),
1591            },
1592            removal.removed,
1593        )
1594    }
1595
1596    fn lock_state(&self) -> std::sync::MutexGuard<'_, LifecycleState> {
1597        self.state
1598            .lock()
1599            .unwrap_or_else(std::sync::PoisonError::into_inner)
1600    }
1601}
1602
1603pub(crate) fn reload_config_providers_for_hub(hub: &ConfigHub) -> Result<(), ConfigError> {
1604    let state = shared_config_provider_state(hub);
1605    let mut state = state
1606        .lock()
1607        .unwrap_or_else(std::sync::PoisonError::into_inner);
1608    let mut replaced = Vec::new();
1609    hub.reload_and_then(|snapshot| {
1610        let mut providers = snapshot.providers.into_iter().collect::<Vec<_>>();
1611        providers.sort_by(|left, right| left.0.cmp(&right.0));
1612        let current_names = providers
1613            .iter()
1614            .map(|(name, _)| name.clone())
1615            .collect::<BTreeSet<_>>();
1616        let stale_names = state
1617            .config_providers
1618            .keys()
1619            .filter(|name| !current_names.contains(*name))
1620            .cloned()
1621            .collect::<Vec<_>>();
1622        state.registries.retain(|registry| {
1623            let Some(registry) = registry.upgrade() else {
1624                return false;
1625            };
1626            for name in &stale_names {
1627                replaced.extend(registry.take_named(&format!("config:{name}")));
1628            }
1629            for (name, entry) in &providers {
1630                let (_, previous) = crate::config_provider::reconcile_config_provider_deferred(
1631                    &registry, name, entry,
1632                );
1633                replaced.extend(previous);
1634            }
1635            true
1636        });
1637        state.config_providers = providers.into_iter().collect();
1638    })?;
1639    drop(state);
1640    drop(replaced);
1641    Ok(())
1642}
1643
1644pub(crate) fn upsert_config_provider_for_hub(
1645    hub: &ConfigHub,
1646    update: ProviderConfigUpdate<'_>,
1647) -> Result<(), ConfigError> {
1648    mutate_config_provider_for_hub(hub, update, ProviderConfigWriteMode::Upsert)
1649}
1650
1651fn mutate_config_provider_for_hub(
1652    hub: &ConfigHub,
1653    update: ProviderConfigUpdate<'_>,
1654    mode: ProviderConfigWriteMode,
1655) -> Result<(), ConfigError> {
1656    let state = shared_config_provider_state(hub);
1657    let mut state = state
1658        .lock()
1659        .unwrap_or_else(std::sync::PoisonError::into_inner);
1660    let mut replaced = Vec::new();
1661    hub.write_provider_config_and_then(update, mode, |entry| {
1662        state.registries.retain(|registry| {
1663            let Some(registry) = registry.upgrade() else {
1664                return false;
1665            };
1666            let (_, previous) = crate::config_provider::reconcile_config_provider_deferred(
1667                &registry,
1668                &entry.name,
1669                entry,
1670            );
1671            replaced.extend(previous);
1672            true
1673        });
1674        state
1675            .config_providers
1676            .insert(entry.name.clone(), entry.clone());
1677    })?;
1678    drop(state);
1679    drop(replaced);
1680    Ok(())
1681}
1682
1683fn wire_profile_for_kind(kind: &ProviderKind) -> ReasoningWireProfile {
1684    match kind {
1685        ProviderKind::Codex => ReasoningWireProfile::CodexResponses,
1686        ProviderKind::AnthropicOauth => ReasoningWireProfile::AnthropicMessages,
1687        ProviderKind::GitHubCopilot | ProviderKind::Custom => ReasoningWireProfile::Unknown,
1688    }
1689}
1690
1691fn shared_lifecycle_state(hub: &ConfigHub) -> Arc<Mutex<LifecycleState>> {
1692    let key = coordinator_key(hub.auth_path());
1693    let mut coordinators = LIFECYCLE_COORDINATORS
1694        .lock()
1695        .unwrap_or_else(std::sync::PoisonError::into_inner);
1696    coordinators.retain(|_, coordinator| coordinator.strong_count() > 0);
1697    if let Some(state) = coordinators.get(&key).and_then(Weak::upgrade) {
1698        return state;
1699    }
1700    let state = Arc::new(Mutex::new(LifecycleState::default()));
1701    coordinators.insert(key, Arc::downgrade(&state));
1702    state
1703}
1704
1705fn shared_config_provider_state(hub: &ConfigHub) -> Arc<Mutex<ConfigProviderLifecycleState>> {
1706    let key = coordinator_key(&hub.config_toml_path());
1707    let mut coordinators = CONFIG_PROVIDER_COORDINATORS
1708        .lock()
1709        .unwrap_or_else(std::sync::PoisonError::into_inner);
1710    coordinators.retain(|_, coordinator| coordinator.strong_count() > 0);
1711    if let Some(state) = coordinators.get(&key).and_then(Weak::upgrade) {
1712        return state;
1713    }
1714    let state = Arc::new(Mutex::new(ConfigProviderLifecycleState::default()));
1715    coordinators.insert(key, Arc::downgrade(&state));
1716    state
1717}
1718
1719fn coordinator_key(path: &Path) -> PathBuf {
1720    if let Some(parent) = path.parent()
1721        && let Ok(parent) = std::fs::canonicalize(parent)
1722        && let Some(file_name) = path.file_name()
1723    {
1724        return parent.join(file_name);
1725    }
1726    if path.is_absolute() {
1727        return path.to_path_buf();
1728    }
1729    std::env::current_dir()
1730        .map(|current| current.join(path))
1731        .unwrap_or_else(|_| path.to_path_buf())
1732}
1733
1734#[cfg(test)]
1735mod tests {
1736    use std::sync::Mutex as StdMutex;
1737    use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
1738
1739    use tokio::sync::oneshot;
1740
1741    use super::*;
1742    use crate::error::RuntimeError;
1743    use crate::event::Observable;
1744    use crate::provider::{
1745        AssistantMessage, CapabilityKnowledge, DiscoveredModelDetails, LlmRequest,
1746        ModelCapabilities, ReasoningEffort,
1747    };
1748    use crate::tool::BoxFut;
1749
1750    struct TestProvider {
1751        name: String,
1752        models: Arc<StdMutex<Vec<DiscoveredModelDetails>>>,
1753        block: Arc<StdMutex<Option<DiscoveryBlock>>>,
1754        fail: Arc<StdMutex<Option<ModelDiscoveryError>>>,
1755        before_discovery: Arc<StdMutex<Option<DiscoveryHook>>>,
1756        discovery_calls: Arc<AtomicUsize>,
1757        on_drop: StdMutex<Option<DiscoveryHook>>,
1758    }
1759
1760    type DiscoveryHook = Box<dyn FnOnce() + Send>;
1761
1762    struct DiscoveryBlock {
1763        started: oneshot::Sender<()>,
1764        proceed: oneshot::Receiver<()>,
1765    }
1766
1767    impl TestProvider {
1768        fn new(name: &str, models: Vec<DiscoveredModelDetails>) -> Self {
1769            Self {
1770                name: name.into(),
1771                models: Arc::new(StdMutex::new(models)),
1772                block: Arc::new(StdMutex::new(None)),
1773                fail: Arc::new(StdMutex::new(None)),
1774                before_discovery: Arc::new(StdMutex::new(None)),
1775                discovery_calls: Arc::new(AtomicUsize::new(0)),
1776                on_drop: StdMutex::new(None),
1777            }
1778        }
1779
1780        fn set_models(&self, models: Vec<DiscoveredModelDetails>) {
1781            *self.models.lock().unwrap() = models;
1782        }
1783
1784        fn fail_once(&self, error: ModelDiscoveryError) {
1785            *self.fail.lock().unwrap() = Some(error);
1786        }
1787
1788        fn block_once(&self) -> (oneshot::Receiver<()>, oneshot::Sender<()>) {
1789            let (started_tx, started_rx) = oneshot::channel();
1790            let (proceed_tx, proceed_rx) = oneshot::channel();
1791            *self.block.lock().unwrap() = Some(DiscoveryBlock {
1792                started: started_tx,
1793                proceed: proceed_rx,
1794            });
1795            (started_rx, proceed_tx)
1796        }
1797
1798        fn before_discovery_once(&self, hook: impl FnOnce() + Send + 'static) {
1799            *self.before_discovery.lock().unwrap() = Some(Box::new(hook));
1800        }
1801
1802        fn discovery_calls(&self) -> usize {
1803            self.discovery_calls.load(Ordering::SeqCst)
1804        }
1805
1806        fn on_drop(&self, hook: impl FnOnce() + Send + 'static) {
1807            *self.on_drop.lock().unwrap() = Some(Box::new(hook));
1808        }
1809    }
1810
1811    impl Drop for TestProvider {
1812        fn drop(&mut self) {
1813            if let Some(on_drop) = self.on_drop.lock().unwrap().take() {
1814                on_drop();
1815            }
1816        }
1817    }
1818
1819    impl Provider for TestProvider {
1820        fn name(&self) -> &str {
1821            &self.name
1822        }
1823
1824        fn call<'a>(
1825            &'a self,
1826            _req: LlmRequest,
1827        ) -> BoxFut<'a, Result<AssistantMessage, RuntimeError>> {
1828            Box::pin(async { unreachable!("not used by lifecycle tests") })
1829        }
1830
1831        fn call_streaming(&self, _req: LlmRequest) -> Observable<AssistantMessage> {
1832            unreachable!("not used by lifecycle tests")
1833        }
1834
1835        fn try_discover_models(
1836            &self,
1837        ) -> BoxFut<'static, Result<Vec<DiscoveredModelDetails>, ModelDiscoveryError>> {
1838            self.discovery_calls.fetch_add(1, Ordering::SeqCst);
1839            let models = self.models.lock().unwrap().clone();
1840            let block = self.block.lock().unwrap().take();
1841            let error = self.fail.lock().unwrap().take();
1842            let before_discovery = self.before_discovery.lock().unwrap().take();
1843            Box::pin(async move {
1844                if let Some(before_discovery) = before_discovery {
1845                    before_discovery();
1846                }
1847                if let Some(block) = block {
1848                    let _ = block.started.send(());
1849                    let _ = block.proceed.await;
1850                }
1851                if let Some(error) = error {
1852                    return Err(error);
1853                }
1854                Ok(models)
1855            })
1856        }
1857    }
1858
1859    fn provider_record(id: &str) -> StoredProvider {
1860        StoredProvider {
1861            id: id.into(),
1862            name: "Account".into(),
1863            kind: ProviderKind::Codex,
1864            access_token: "access".into(),
1865            refresh_token: Some("refresh".into()),
1866            expires_at: i64::MAX,
1867            account: Some("account@example.com".into()),
1868            enabled: true,
1869            model_cache: None,
1870        }
1871    }
1872
1873    fn model(slug: &str) -> DiscoveredModelDetails {
1874        DiscoveredModelDetails {
1875            slug: slug.into(),
1876            context_budget: Some(128_000),
1877            capability_knowledge: CapabilityKnowledge::Advertised(ModelCapabilities::default()),
1878        }
1879    }
1880
1881    fn model_with_effort(slug: &str, effort: ReasoningEffort) -> DiscoveredModelDetails {
1882        DiscoveredModelDetails {
1883            slug: slug.into(),
1884            context_budget: Some(128_000),
1885            capability_knowledge: CapabilityKnowledge::Advertised(ModelCapabilities {
1886                reasoning_efforts: vec![effort],
1887                ..Default::default()
1888            }),
1889        }
1890    }
1891
1892    fn seed_cached_provider(
1893        lifecycle: &ProviderLifecycle,
1894        id: &str,
1895        enabled: bool,
1896        models: &[DiscoveredModelDetails],
1897    ) -> String {
1898        let mut record = provider_record(id);
1899        record.enabled = enabled;
1900        let namespace = format!("{id}@account");
1901        lifecycle
1902            .hub
1903            .add_auth_provider_with_model_cache_details(record, &namespace, 1, models)
1904            .unwrap();
1905        namespace
1906    }
1907
1908    fn seed_legacy_cached_provider(lifecycle: &ProviderLifecycle, id: &str, enabled: bool) {
1909        let mut record = provider_record(id);
1910        record.enabled = enabled;
1911        record.model_cache = Some(crate::auth_store::ModelCache {
1912            fetched_at: 1,
1913            models: vec![crate::auth_store::CachedModel {
1914                slug: "cached".into(),
1915                context_budget: Some(128_000),
1916                thinking: true,
1917            }],
1918        });
1919        lifecycle.hub.add_auth_provider(record).unwrap();
1920    }
1921
1922    fn fixture(id: &str) -> (tempfile::TempDir, ProviderLifecycle, Arc<TestProvider>) {
1923        let dir = tempfile::tempdir().unwrap();
1924        let lifecycle = ProviderLifecycle::new(
1925            ConfigHub::from_config_dir(dir.path()),
1926            ProviderRegistry::new(),
1927        );
1928        let provider = Arc::new(TestProvider::new(id, vec![model("initial")]));
1929        (dir, lifecycle, provider)
1930    }
1931
1932    fn test_runtime() -> tokio::runtime::Runtime {
1933        tokio::runtime::Builder::new_current_thread()
1934            .enable_all()
1935            .build()
1936            .unwrap()
1937    }
1938
1939    #[test]
1940    fn concurrent_config_provider_updates_leave_config_and_live_registries_in_sync() {
1941        struct ConfigReset;
1942
1943        impl Drop for ConfigReset {
1944            fn drop(&mut self) {
1945                crate::model_registry::set_provider_config(Default::default());
1946            }
1947        }
1948
1949        let _registry_lock = crate::model_registry::MODEL_CONFIG_LOCK
1950            .lock()
1951            .unwrap_or_else(std::sync::PoisonError::into_inner);
1952        let _reset = ConfigReset;
1953        let dir = tempfile::tempdir().unwrap();
1954        let hub = ConfigHub::from_config_dir(dir.path());
1955        let registry = ProviderRegistry::new();
1956        let lifecycle = ProviderLifecycle::new(hub.clone(), registry.clone());
1957        lifecycle
1958            .create_config_provider(ProviderConfigUpdate {
1959                name: "gateway",
1960                kind: "openai-compat",
1961                api_key: Some("initial-key"),
1962                api_key_env: None,
1963                base_url: Some("https://gateway.example/v1"),
1964                max_tokens: Some(8_192),
1965                reasoning_format: None,
1966                prompt_cache_key: None,
1967                enabled: true,
1968            })
1969            .unwrap();
1970
1971        let peer_registry = ProviderRegistry::new();
1972        let _peer = ProviderLifecycle::new(hub.clone(), peer_registry.clone());
1973        let writers = 16;
1974        let barrier = Arc::new(std::sync::Barrier::new(writers));
1975        let handles = (0..writers)
1976            .map(|index| {
1977                let lifecycle = lifecycle.clone();
1978                let barrier = barrier.clone();
1979                std::thread::spawn(move || {
1980                    let api_key = format!("key-{index}");
1981                    barrier.wait();
1982                    lifecycle
1983                        .update_config_provider(ProviderConfigUpdate {
1984                            name: "gateway",
1985                            kind: "openai-compat",
1986                            api_key: Some(&api_key),
1987                            api_key_env: None,
1988                            base_url: Some("https://gateway.example/v1"),
1989                            max_tokens: Some(8_192 + index as u32),
1990                            reasoning_format: None,
1991                            prompt_cache_key: None,
1992                            enabled: index % 2 == 0,
1993                        })
1994                        .unwrap()
1995                })
1996            })
1997            .collect::<Vec<_>>();
1998        for handle in handles {
1999            handle.join().unwrap();
2000        }
2001        let committed = hub.model_config().unwrap().unwrap().providers["gateway"].clone();
2002        let global = crate::model_registry::all_provider_entries()
2003            .into_iter()
2004            .find(|(name, _)| name == "gateway")
2005            .map(|(_, entry)| entry)
2006            .unwrap();
2007        assert_eq!(global.api_key, committed.api_key);
2008        assert_eq!(global.max_tokens, committed.max_tokens);
2009        assert_eq!(global.enabled, committed.enabled);
2010        let available = crate::config_provider::config_provider_availability(&committed)
2011            == crate::config_provider::ConfigProviderAvailability::Available;
2012        assert_eq!(registry.contains("config:gateway"), available);
2013        assert_eq!(peer_registry.contains("config:gateway"), available);
2014    }
2015
2016    #[test]
2017    fn public_config_mutations_sync_existing_and_new_registries() {
2018        struct ConfigReset;
2019
2020        impl Drop for ConfigReset {
2021            fn drop(&mut self) {
2022                crate::model_registry::set_provider_config(Default::default());
2023            }
2024        }
2025
2026        let _registry_lock = crate::model_registry::MODEL_CONFIG_LOCK
2027            .lock()
2028            .unwrap_or_else(std::sync::PoisonError::into_inner);
2029        let _reset = ConfigReset;
2030        let dir = tempfile::tempdir().unwrap();
2031        let hub = ConfigHub::from_config_dir(dir.path());
2032        let active = ProviderConfigUpdate {
2033            name: "gateway",
2034            kind: "openai-compat",
2035            api_key: Some("test-key"),
2036            api_key_env: None,
2037            base_url: Some("https://gateway.example/v1"),
2038            max_tokens: None,
2039            reasoning_format: None,
2040            prompt_cache_key: None,
2041            enabled: true,
2042        };
2043        hub.upsert_provider(active).unwrap();
2044        let registry = ProviderRegistry::new();
2045        let lifecycle = ProviderLifecycle::new(hub.clone(), registry.clone());
2046        lifecycle.reload_config_providers().unwrap();
2047        assert!(registry.contains("config:gateway"));
2048
2049        hub.upsert_provider(ProviderConfigUpdate {
2050            enabled: false,
2051            ..active
2052        })
2053        .unwrap();
2054        assert!(!registry.contains("config:gateway"));
2055        let before_attach_registry = ProviderRegistry::new();
2056        let _before_attach = ProviderLifecycle::new(hub.clone(), before_attach_registry.clone());
2057        assert!(!before_attach_registry.contains("config:gateway"));
2058
2059        hub.upsert_provider(active).unwrap();
2060        assert!(registry.contains("config:gateway"));
2061        assert!(before_attach_registry.contains("config:gateway"));
2062        let after_attach_registry = ProviderRegistry::new();
2063        let _after_attach = ProviderLifecycle::new(hub.clone(), after_attach_registry.clone());
2064        assert!(after_attach_registry.contains("config:gateway"));
2065
2066        std::fs::write(
2067            hub.config_toml_path(),
2068            "[providers.gateway]\nkind = \"openai-compat\"\napi_key = \"reloaded-key\"\nenabled = false\n",
2069        )
2070        .unwrap();
2071        hub.reload().unwrap();
2072        assert!(!registry.contains("config:gateway"));
2073        assert!(!before_attach_registry.contains("config:gateway"));
2074        assert!(!after_attach_registry.contains("config:gateway"));
2075        let reloaded_registry = ProviderRegistry::new();
2076        let _reloaded = ProviderLifecycle::new(hub.clone(), reloaded_registry.clone());
2077        assert!(!reloaded_registry.contains("config:gateway"));
2078    }
2079
2080    #[test]
2081    fn public_config_mutations_fan_out_across_distinct_auth_stores() {
2082        struct ConfigReset;
2083
2084        impl Drop for ConfigReset {
2085            fn drop(&mut self) {
2086                crate::model_registry::set_provider_config(Default::default());
2087            }
2088        }
2089
2090        let _registry_lock = crate::model_registry::MODEL_CONFIG_LOCK
2091            .lock()
2092            .unwrap_or_else(std::sync::PoisonError::into_inner);
2093        let _reset = ConfigReset;
2094        let dir = tempfile::tempdir().unwrap();
2095        let hub_a = ConfigHub::from_auth_path(dir.path().join("auth-a.json"));
2096        let hub_b = ConfigHub::from_auth_path(dir.path().join("auth-b.json"));
2097        let registry_a = ProviderRegistry::new();
2098        let registry_b = ProviderRegistry::new();
2099        let lifecycle_a = ProviderLifecycle::new(hub_a.clone(), registry_a.clone());
2100        let lifecycle_b = ProviderLifecycle::new(hub_b.clone(), registry_b.clone());
2101
2102        assert!(!Arc::ptr_eq(&lifecycle_a.state, &lifecycle_b.state));
2103        assert!(Arc::ptr_eq(
2104            &lifecycle_a.config_state,
2105            &lifecycle_b.config_state
2106        ));
2107
2108        let active = ProviderConfigUpdate {
2109            name: "gateway",
2110            kind: "openai-compat",
2111            api_key: Some("test-key"),
2112            api_key_env: None,
2113            base_url: Some("https://gateway.example/v1"),
2114            max_tokens: None,
2115            reasoning_format: None,
2116            prompt_cache_key: None,
2117            enabled: true,
2118        };
2119        hub_a.upsert_provider(active).unwrap();
2120        assert!(registry_a.contains("config:gateway"));
2121        assert!(registry_b.contains("config:gateway"));
2122
2123        hub_b
2124            .upsert_provider(ProviderConfigUpdate {
2125                enabled: false,
2126                ..active
2127            })
2128            .unwrap();
2129        assert!(!registry_a.contains("config:gateway"));
2130        assert!(!registry_b.contains("config:gateway"));
2131
2132        let registry_c = ProviderRegistry::new();
2133        let lifecycle_c = ProviderLifecycle::new(hub_a.clone(), registry_c.clone());
2134        assert!(Arc::ptr_eq(
2135            &lifecycle_a.config_state,
2136            &lifecycle_c.config_state
2137        ));
2138        assert!(!registry_c.contains("config:gateway"));
2139    }
2140
2141    #[test]
2142    fn public_config_reload_fans_out_across_distinct_auth_stores() {
2143        struct ConfigReset;
2144
2145        impl Drop for ConfigReset {
2146            fn drop(&mut self) {
2147                crate::model_registry::set_provider_config(Default::default());
2148            }
2149        }
2150
2151        let _registry_lock = crate::model_registry::MODEL_CONFIG_LOCK
2152            .lock()
2153            .unwrap_or_else(std::sync::PoisonError::into_inner);
2154        let _reset = ConfigReset;
2155        let dir = tempfile::tempdir().unwrap();
2156        let hub_a = ConfigHub::from_auth_path(dir.path().join("auth-a.json"));
2157        let hub_b = ConfigHub::from_auth_path(dir.path().join("auth-b.json"));
2158        let registry_a = ProviderRegistry::new();
2159        let registry_b = ProviderRegistry::new();
2160        let _lifecycle_a = ProviderLifecycle::new(hub_a.clone(), registry_a.clone());
2161        let _lifecycle_b = ProviderLifecycle::new(hub_b.clone(), registry_b.clone());
2162
2163        std::fs::write(
2164            hub_a.config_toml_path(),
2165            "[providers.gateway]\nkind = \"openai-compat\"\napi_key = \"first-key\"\nenabled = true\n",
2166        )
2167        .unwrap();
2168        hub_a.reload().unwrap();
2169        assert!(registry_a.contains("config:gateway"));
2170        assert!(registry_b.contains("config:gateway"));
2171
2172        std::fs::write(
2173            hub_b.config_toml_path(),
2174            "[providers.gateway]\nkind = \"openai-compat\"\napi_key = \"second-key\"\nenabled = false\n",
2175        )
2176        .unwrap();
2177        hub_b.reload().unwrap();
2178        assert!(!registry_a.contains("config:gateway"));
2179        assert!(!registry_b.contains("config:gateway"));
2180
2181        let projected = crate::model_registry::all_provider_entries()
2182            .into_iter()
2183            .find(|(name, _)| name == "gateway")
2184            .map(|(_, entry)| entry)
2185            .unwrap();
2186        assert_eq!(projected.api_key.as_deref(), Some("second-key"));
2187        assert_eq!(projected.enabled, Some(false));
2188    }
2189
2190    #[test]
2191    fn install_commits_auth_live_provider_and_catalog_together() {
2192        let _registry_lock = crate::model_registry::MODEL_CONFIG_LOCK
2193            .lock()
2194            .unwrap_or_else(std::sync::PoisonError::into_inner);
2195        test_runtime().block_on(async {
2196            let (_dir, lifecycle, provider) = fixture("lifecycle-install");
2197            remove_provider_catalog("lifecycle-install");
2198
2199            let delta = lifecycle
2200                .install_provider(provider_record("lifecycle-install"), provider)
2201                .await
2202                .unwrap();
2203
2204            assert_eq!(delta.total, 1);
2205            assert!(lifecycle.providers.contains("lifecycle-install"));
2206            assert_eq!(
2207                lifecycle.hub.load_auth().unwrap().providers[0].id,
2208                "lifecycle-install"
2209            );
2210            assert!(provider_catalog_namespace("lifecycle-install").is_some());
2211            lifecycle.remove_provider("lifecycle-install").unwrap();
2212        });
2213    }
2214
2215    #[test]
2216    fn pre_discovered_install_does_not_run_discovery() {
2217        let _registry_lock = crate::model_registry::MODEL_CONFIG_LOCK
2218            .lock()
2219            .unwrap_or_else(std::sync::PoisonError::into_inner);
2220        test_runtime().block_on(async {
2221            let (_dir, lifecycle, provider) = fixture("lifecycle-pre-discovered");
2222            remove_provider_catalog("lifecycle-pre-discovered");
2223            provider.fail_once(ModelDiscoveryError::Transport("unused".into()));
2224
2225            lifecycle
2226                .install_pre_discovered_provider(
2227                    provider_record("lifecycle-pre-discovered"),
2228                    provider.clone(),
2229                    vec![model("provided")],
2230                )
2231                .await
2232                .unwrap();
2233
2234            assert_eq!(provider.discovery_calls(), 0);
2235            let namespace = lifecycle
2236                .hub
2237                .load_auth_model_namespace("lifecycle-pre-discovered")
2238                .unwrap()
2239                .unwrap();
2240            assert!(crate::model_registry::model_entry(&format!("{namespace}:provided")).is_some());
2241            assert!(matches!(
2242                lifecycle.refresh_models("lifecycle-pre-discovered").await,
2243                Err(ProviderLifecycleError::Discovery(
2244                    ModelDiscoveryError::Transport(message)
2245                )) if message == "unused"
2246            ));
2247            lifecycle
2248                .remove_provider("lifecycle-pre-discovered")
2249                .unwrap();
2250        });
2251    }
2252
2253    #[test]
2254    fn cached_restore_is_offline_and_reuses_the_shared_provider() {
2255        let _registry_lock = crate::model_registry::MODEL_CONFIG_LOCK
2256            .lock()
2257            .unwrap_or_else(std::sync::PoisonError::into_inner);
2258        test_runtime().block_on(async {
2259            let (_dir, lifecycle, provider) = fixture("lifecycle-restore");
2260            remove_provider_catalog("lifecycle-restore");
2261            seed_legacy_cached_provider(&lifecycle, "lifecycle-restore", true);
2262            provider.fail_once(ModelDiscoveryError::Transport("offline".into()));
2263
2264            let first = lifecycle
2265                .restore_provider("lifecycle-restore", ProviderKind::Codex, provider.clone())
2266                .await
2267                .unwrap();
2268
2269            assert_eq!(provider.discovery_calls(), 0);
2270            assert!(first.state.auth_changed);
2271            assert!(first.state.live_changed);
2272            assert!(first.state.catalog_changed);
2273            let namespace = lifecycle
2274                .hub
2275                .load_auth_model_namespace("lifecycle-restore")
2276                .unwrap()
2277                .unwrap();
2278            assert!(crate::model_registry::model_entry(&format!("{namespace}:cached")).is_some());
2279            let expected: Arc<dyn Provider> = provider.clone();
2280            assert!(Arc::ptr_eq(
2281                &lifecycle.providers.get("lifecycle-restore").unwrap(),
2282                &expected
2283            ));
2284
2285            let peer_registry = ProviderRegistry::new();
2286            let peer = ProviderLifecycle::new(lifecycle.hub.clone(), peer_registry.clone());
2287            let replacement = Arc::new(TestProvider::new(
2288                "lifecycle-restore",
2289                vec![model("replacement")],
2290            ));
2291            let before_revision = crate::model_registry::model_catalog_revision();
2292            let second = peer
2293                .restore_provider(
2294                    "lifecycle-restore",
2295                    ProviderKind::Codex,
2296                    replacement.clone(),
2297                )
2298                .await
2299                .unwrap();
2300
2301            assert_eq!(replacement.discovery_calls(), 0);
2302            assert_eq!(second.state, ProviderStateChange::default());
2303            assert_eq!(
2304                crate::model_registry::model_catalog_revision(),
2305                before_revision
2306            );
2307            assert!(Arc::ptr_eq(
2308                &peer_registry.get("lifecycle-restore").unwrap(),
2309                &expected
2310            ));
2311            assert!(matches!(
2312                lifecycle.refresh_models("lifecycle-restore").await,
2313                Err(ProviderLifecycleError::Discovery(
2314                    ModelDiscoveryError::Transport(message)
2315                )) if message == "offline"
2316            ));
2317            assert_eq!(provider.discovery_calls(), 1);
2318            lifecycle.remove_provider("lifecycle-restore").unwrap();
2319        });
2320    }
2321
2322    #[test]
2323    fn catalog_refresh_plan_contains_restored_live_providers() {
2324        let _registry_lock = crate::model_registry::MODEL_CONFIG_LOCK
2325            .lock()
2326            .unwrap_or_else(std::sync::PoisonError::into_inner);
2327        test_runtime().block_on(async {
2328            let (_dir, lifecycle, provider) = fixture("catalog-refresh-plan");
2329            remove_provider_catalog("catalog-refresh-plan");
2330            seed_legacy_cached_provider(&lifecycle, "catalog-refresh-plan", true);
2331            lifecycle
2332                .hub
2333                .add_auth_provider(provider_record("catalog-refresh-no-live"))
2334                .unwrap();
2335            lifecycle
2336                .restore_provider(
2337                    "catalog-refresh-plan",
2338                    ProviderKind::Codex,
2339                    provider.clone(),
2340                )
2341                .await
2342                .unwrap();
2343            assert_eq!(
2344                lifecycle.catalog_refresh_plan(),
2345                vec!["catalog-refresh-plan"]
2346            );
2347            provider.set_models(vec![model("fresh")]);
2348            lifecycle
2349                .refresh_models("catalog-refresh-plan")
2350                .await
2351                .unwrap();
2352            assert_eq!(
2353                lifecycle.catalog_refresh_plan(),
2354                vec!["catalog-refresh-plan"]
2355            );
2356            assert_eq!(
2357                lifecycle
2358                    .refresh_models_if_stale("catalog-refresh-plan")
2359                    .await
2360                    .unwrap(),
2361                ProviderCatalogRefreshOutcome::NotNeeded
2362            );
2363            assert_eq!(provider.discovery_calls(), 1);
2364
2365            lifecycle.remove_provider("catalog-refresh-plan").unwrap();
2366            lifecycle
2367                .remove_provider("catalog-refresh-no-live")
2368                .unwrap();
2369        });
2370    }
2371
2372    #[test]
2373    fn catalog_refresh_plan_revalidates_external_provider_mutations() {
2374        let _registry_lock = crate::model_registry::MODEL_CONFIG_LOCK
2375            .lock()
2376            .unwrap_or_else(std::sync::PoisonError::into_inner);
2377        test_runtime().block_on(async {
2378            for mutation in ["remove", "disable", "kind"] {
2379                let provider_id = format!("catalog-plan-{mutation}");
2380                let (_dir, lifecycle, provider) = fixture(&provider_id);
2381                remove_provider_catalog(&provider_id);
2382                seed_legacy_cached_provider(&lifecycle, &provider_id, true);
2383                lifecycle
2384                    .restore_provider(&provider_id, ProviderKind::Codex, provider)
2385                    .await
2386                    .unwrap();
2387
2388                let external_hub = ConfigHub::from_config_dir(lifecycle.hub.config_dir());
2389                match mutation {
2390                    "remove" => {
2391                        assert!(external_hub.remove_auth_provider(&provider_id).unwrap());
2392                    }
2393                    "disable" => {
2394                        assert_eq!(
2395                            external_hub
2396                                .set_auth_provider_enabled_with_change(&provider_id, false)
2397                                .unwrap(),
2398                            Some(true)
2399                        );
2400                    }
2401                    "kind" => external_hub
2402                        .update_auth(|store| {
2403                            store.providers[0].kind = ProviderKind::AnthropicOauth;
2404                            Ok(())
2405                        })
2406                        .unwrap(),
2407                    _ => unreachable!(),
2408                }
2409
2410                assert_eq!(lifecycle.catalog_refresh_plan(), vec![provider_id.clone()]);
2411                let result = lifecycle.refresh_models_if_stale(&provider_id).await;
2412                match mutation {
2413                    "remove" => assert!(matches!(
2414                        result,
2415                        Err(ProviderLifecycleError::ProviderNotFound { .. })
2416                    )),
2417                    "disable" => assert!(matches!(
2418                        result,
2419                        Err(ProviderLifecycleError::ProviderDisabled { .. })
2420                    )),
2421                    "kind" => assert!(matches!(result, Err(ProviderLifecycleError::Stale { .. }))),
2422                    _ => unreachable!(),
2423                }
2424                assert!(!lifecycle.providers.contains(&provider_id));
2425                assert!(provider_catalog_namespace(&provider_id).is_none());
2426                if mutation != "remove" {
2427                    lifecycle.remove_provider(&provider_id).unwrap();
2428                }
2429            }
2430        });
2431    }
2432
2433    #[test]
2434    fn fresh_external_cache_is_hydrated_when_plan_observes_fresh_storage() {
2435        let _registry_lock = crate::model_registry::MODEL_CONFIG_LOCK
2436            .lock()
2437            .unwrap_or_else(std::sync::PoisonError::into_inner);
2438        test_runtime().block_on(async {
2439            let (_dir, lifecycle, provider) = fixture("catalog-refresh-external-cache");
2440            remove_provider_catalog("catalog-refresh-external-cache");
2441            seed_legacy_cached_provider(&lifecycle, "catalog-refresh-external-cache", true);
2442            lifecycle
2443                .restore_provider(
2444                    "catalog-refresh-external-cache",
2445                    ProviderKind::Codex,
2446                    provider.clone(),
2447                )
2448                .await
2449                .unwrap();
2450            let namespace = lifecycle
2451                .hub
2452                .load_auth_model_namespace("catalog-refresh-external-cache")
2453                .unwrap()
2454                .unwrap();
2455            let external_hub = ConfigHub::from_config_dir(lifecycle.hub.config_dir());
2456            external_hub
2457                .update_auth_model_cache_details(
2458                    "catalog-refresh-external-cache",
2459                    &namespace,
2460                    chrono::Utc::now().timestamp(),
2461                    &[model_with_effort("cached", ReasoningEffort::High)],
2462                )
2463                .unwrap();
2464            assert_eq!(
2465                lifecycle.catalog_refresh_plan(),
2466                vec!["catalog-refresh-external-cache"]
2467            );
2468            let registry_key = format!("{namespace}:cached");
2469            assert!(
2470                crate::model_registry::model_entry(&registry_key)
2471                    .unwrap()
2472                    .reasoning_efforts
2473                    .is_empty()
2474            );
2475
2476            assert!(matches!(
2477                lifecycle
2478                    .refresh_models_if_stale("catalog-refresh-external-cache")
2479                    .await
2480                    .unwrap(),
2481                ProviderCatalogRefreshOutcome::CatalogUpdated(delta)
2482                    if delta.updated == 1 && delta.total == 1
2483            ));
2484            assert_eq!(provider.discovery_calls(), 0);
2485            assert_eq!(
2486                crate::model_registry::model_entry(&registry_key)
2487                    .unwrap()
2488                    .reasoning_efforts,
2489                vec![ReasoningEffort::High]
2490            );
2491
2492            lifecycle
2493                .remove_provider("catalog-refresh-external-cache")
2494                .unwrap();
2495        });
2496    }
2497
2498    #[test]
2499    fn stale_network_commit_hydrates_the_external_cache_winner() {
2500        let _registry_lock = crate::model_registry::MODEL_CONFIG_LOCK
2501            .lock()
2502            .unwrap_or_else(std::sync::PoisonError::into_inner);
2503        test_runtime().block_on(async {
2504            let (_dir, lifecycle, provider) = fixture("catalog-refresh-external-winner");
2505            remove_provider_catalog("catalog-refresh-external-winner");
2506            seed_legacy_cached_provider(&lifecycle, "catalog-refresh-external-winner", true);
2507            lifecycle
2508                .restore_provider(
2509                    "catalog-refresh-external-winner",
2510                    ProviderKind::Codex,
2511                    provider.clone(),
2512                )
2513                .await
2514                .unwrap();
2515            let namespace = lifecycle
2516                .hub
2517                .load_auth_model_namespace("catalog-refresh-external-winner")
2518                .unwrap()
2519                .unwrap();
2520            provider.set_models(vec![model_with_effort("loser", ReasoningEffort::Low)]);
2521            let (started, proceed) = provider.block_once();
2522            let background_lifecycle = lifecycle.clone();
2523            let background = tokio::spawn(async move {
2524                background_lifecycle
2525                    .refresh_models_if_stale("catalog-refresh-external-winner")
2526                    .await
2527            });
2528            started.await.unwrap();
2529
2530            let external_hub = ConfigHub::from_config_dir(lifecycle.hub.config_dir());
2531            external_hub
2532                .update_auth_model_cache_details(
2533                    "catalog-refresh-external-winner",
2534                    &namespace,
2535                    chrono::Utc::now().timestamp(),
2536                    &[model_with_effort("winner", ReasoningEffort::High)],
2537                )
2538                .unwrap();
2539            proceed.send(()).unwrap();
2540
2541            assert!(matches!(
2542                background.await.unwrap().unwrap(),
2543                ProviderCatalogRefreshOutcome::CatalogUpdated(delta)
2544                    if delta.added == 1 && delta.removed == 1 && delta.total == 1
2545            ));
2546            assert_eq!(provider.discovery_calls(), 1);
2547            assert!(crate::model_registry::model_entry(&format!("{namespace}:loser")).is_none());
2548            assert_eq!(
2549                crate::model_registry::model_entry(&format!("{namespace}:winner"))
2550                    .unwrap()
2551                    .reasoning_efforts,
2552                vec![ReasoningEffort::High]
2553            );
2554            assert_eq!(
2555                lifecycle.hub.load_auth().unwrap().providers[0]
2556                    .model_cache
2557                    .as_ref()
2558                    .unwrap()
2559                    .models[0]
2560                    .slug,
2561                "winner"
2562            );
2563
2564            lifecycle
2565                .remove_provider("catalog-refresh-external-winner")
2566                .unwrap();
2567        });
2568    }
2569
2570    #[test]
2571    fn failed_network_discovery_hydrates_the_external_cache_winner() {
2572        let _registry_lock = crate::model_registry::MODEL_CONFIG_LOCK
2573            .lock()
2574            .unwrap_or_else(std::sync::PoisonError::into_inner);
2575        test_runtime().block_on(async {
2576            let (_dir, lifecycle, provider) = fixture("catalog-refresh-error-winner");
2577            remove_provider_catalog("catalog-refresh-error-winner");
2578            seed_legacy_cached_provider(&lifecycle, "catalog-refresh-error-winner", true);
2579            lifecycle
2580                .restore_provider(
2581                    "catalog-refresh-error-winner",
2582                    ProviderKind::Codex,
2583                    provider.clone(),
2584                )
2585                .await
2586                .unwrap();
2587            let namespace = lifecycle
2588                .hub
2589                .load_auth_model_namespace("catalog-refresh-error-winner")
2590                .unwrap()
2591                .unwrap();
2592            provider.fail_once(ModelDiscoveryError::Transport("offline".into()));
2593            let (started, proceed) = provider.block_once();
2594            let background_lifecycle = lifecycle.clone();
2595            let background = tokio::spawn(async move {
2596                background_lifecycle
2597                    .refresh_models_if_stale("catalog-refresh-error-winner")
2598                    .await
2599            });
2600            started.await.unwrap();
2601
2602            ConfigHub::from_config_dir(lifecycle.hub.config_dir())
2603                .update_auth_model_cache_details(
2604                    "catalog-refresh-error-winner",
2605                    &namespace,
2606                    chrono::Utc::now().timestamp(),
2607                    &[model_with_effort("winner", ReasoningEffort::High)],
2608                )
2609                .unwrap();
2610            proceed.send(()).unwrap();
2611
2612            assert!(matches!(
2613                background.await.unwrap().unwrap(),
2614                ProviderCatalogRefreshOutcome::CatalogUpdated(delta)
2615                    if delta.added == 1 && delta.removed == 1 && delta.total == 1
2616            ));
2617            assert_eq!(provider.discovery_calls(), 1);
2618            assert_eq!(
2619                crate::model_registry::model_entry(&format!("{namespace}:winner"))
2620                    .unwrap()
2621                    .reasoning_efforts,
2622                vec![ReasoningEffort::High]
2623            );
2624
2625            lifecycle
2626                .remove_provider("catalog-refresh-error-winner")
2627                .unwrap();
2628        });
2629    }
2630
2631    #[test]
2632    fn cache_hydration_cas_remove_and_disable_prune_local_runtime() {
2633        let _registry_lock = crate::model_registry::MODEL_CONFIG_LOCK
2634            .lock()
2635            .unwrap_or_else(std::sync::PoisonError::into_inner);
2636        test_runtime().block_on(async {
2637            for (provider_id, remove) in [
2638                ("catalog-hydrate-removed", true),
2639                ("catalog-hydrate-disabled", false),
2640            ] {
2641                let (_dir, lifecycle, provider) = fixture(provider_id);
2642                remove_provider_catalog(provider_id);
2643                seed_cached_provider(&lifecycle, provider_id, true, &[model("cached")]);
2644                let expected_live: Arc<dyn Provider> = provider.clone();
2645                lifecycle
2646                    .restore_provider(provider_id, ProviderKind::Codex, provider)
2647                    .await
2648                    .unwrap();
2649                let (attempt, deferred) = lifecycle.load_catalog_refresh_attempt(provider_id, 1);
2650                assert!(deferred.is_empty());
2651                let attempt = attempt.unwrap();
2652                let external_hub = ConfigHub::from_config_dir(lifecycle.hub.config_dir());
2653                if remove {
2654                    external_hub.remove_auth_provider(provider_id).unwrap();
2655                } else {
2656                    external_hub
2657                        .set_auth_provider_enabled(provider_id, false)
2658                        .unwrap();
2659                }
2660
2661                let (result, deferred) =
2662                    lifecycle.hydrate_cached_catalog_if_current(provider_id, &attempt);
2663                drop(deferred);
2664                if remove {
2665                    assert!(matches!(
2666                        result,
2667                        Err(ProviderLifecycleError::ProviderNotFound { .. })
2668                    ));
2669                } else {
2670                    assert!(matches!(result, Err(ProviderLifecycleError::Stale { .. })));
2671                    let (result, deferred) = lifecycle.prepare_catalog_refresh(
2672                        provider_id,
2673                        1,
2674                        true,
2675                        Some(&expected_live),
2676                        Some(&attempt.runtime.catalog_snapshot),
2677                    );
2678                    drop(deferred);
2679                    assert!(matches!(
2680                        result,
2681                        Err(ProviderLifecycleError::ProviderDisabled { .. })
2682                    ));
2683                }
2684                assert!(!lifecycle.providers.contains(provider_id));
2685                assert!(provider_catalog_namespace(provider_id).is_none());
2686            }
2687        });
2688    }
2689
2690    #[test]
2691    fn cache_hydration_cas_kind_change_is_reconciled_fail_closed() {
2692        let _registry_lock = crate::model_registry::MODEL_CONFIG_LOCK
2693            .lock()
2694            .unwrap_or_else(std::sync::PoisonError::into_inner);
2695        test_runtime().block_on(async {
2696            let provider_id = "catalog-hydrate-kind-change";
2697            let (_dir, lifecycle, provider) = fixture(provider_id);
2698            remove_provider_catalog(provider_id);
2699            seed_cached_provider(&lifecycle, provider_id, true, &[model("cached")]);
2700            lifecycle
2701                .restore_provider(provider_id, ProviderKind::Codex, provider.clone())
2702                .await
2703                .unwrap();
2704            let expected_live: Arc<dyn Provider> = provider;
2705            let (attempt, deferred) = lifecycle.load_catalog_refresh_attempt(provider_id, 1);
2706            assert!(deferred.is_empty());
2707            let attempt = attempt.unwrap();
2708            lifecycle
2709                .hub
2710                .update_auth(|store| {
2711                    store.providers[0].kind = ProviderKind::AnthropicOauth;
2712                    Ok(())
2713                })
2714                .unwrap();
2715
2716            let (result, deferred) =
2717                lifecycle.hydrate_cached_catalog_if_current(provider_id, &attempt);
2718            assert!(deferred.is_empty());
2719            assert!(matches!(result, Err(ProviderLifecycleError::Stale { .. })));
2720            let (result, deferred) = lifecycle.prepare_catalog_refresh(
2721                provider_id,
2722                1,
2723                true,
2724                Some(&expected_live),
2725                Some(&attempt.runtime.catalog_snapshot),
2726            );
2727            drop(deferred);
2728            assert!(matches!(result, Err(ProviderLifecycleError::Stale { .. })));
2729            assert!(!lifecycle.providers.contains(provider_id));
2730            assert!(provider_catalog_namespace(provider_id).is_none());
2731            lifecycle.remove_provider(provider_id).unwrap();
2732        });
2733    }
2734
2735    #[test]
2736    fn background_refresh_is_deduplicated_without_blocking_runtime_restore() {
2737        let _registry_lock = crate::model_registry::MODEL_CONFIG_LOCK
2738            .lock()
2739            .unwrap_or_else(std::sync::PoisonError::into_inner);
2740        test_runtime().block_on(async {
2741            let (_dir, lifecycle, provider) = fixture("catalog-refresh-background");
2742            remove_provider_catalog("catalog-refresh-background");
2743            seed_legacy_cached_provider(&lifecycle, "catalog-refresh-background", true);
2744            lifecycle
2745                .restore_provider(
2746                    "catalog-refresh-background",
2747                    ProviderKind::Codex,
2748                    provider.clone(),
2749                )
2750                .await
2751                .unwrap();
2752            provider.set_models(vec![model("fresh")]);
2753            let (started, proceed) = provider.block_once();
2754            let background_lifecycle = lifecycle.clone();
2755            let background = tokio::spawn(async move {
2756                background_lifecycle
2757                    .refresh_models_if_stale("catalog-refresh-background")
2758                    .await
2759            });
2760            started.await.unwrap();
2761
2762            assert!(
2763                lifecycle
2764                    .operation_lock("catalog-refresh-background")
2765                    .try_lock()
2766                    .is_ok()
2767            );
2768            assert_eq!(
2769                lifecycle
2770                    .refresh_models_if_stale("catalog-refresh-background")
2771                    .await
2772                    .unwrap(),
2773                ProviderCatalogRefreshOutcome::AlreadyInFlight
2774            );
2775            let restored = lifecycle
2776                .restore_provider(
2777                    "catalog-refresh-background",
2778                    ProviderKind::Codex,
2779                    provider.clone(),
2780                )
2781                .await
2782                .unwrap();
2783            assert_eq!(restored.state, ProviderStateChange::default());
2784
2785            proceed.send(()).unwrap();
2786            assert!(matches!(
2787                background.await.unwrap().unwrap(),
2788                ProviderCatalogRefreshOutcome::CatalogUpdated(delta) if delta.total == 1
2789            ));
2790            assert_eq!(provider.discovery_calls(), 1);
2791            assert_eq!(
2792                lifecycle
2793                    .refresh_models_if_stale("catalog-refresh-background")
2794                    .await
2795                    .unwrap(),
2796                ProviderCatalogRefreshOutcome::NotNeeded
2797            );
2798            assert_eq!(provider.discovery_calls(), 1);
2799
2800            lifecycle
2801                .remove_provider("catalog-refresh-background")
2802                .unwrap();
2803        });
2804    }
2805
2806    #[test]
2807    fn concurrent_background_refresh_returns_already_in_flight() {
2808        let _registry_lock = crate::model_registry::MODEL_CONFIG_LOCK
2809            .lock()
2810            .unwrap_or_else(std::sync::PoisonError::into_inner);
2811        test_runtime().block_on(async {
2812            let (_dir, lifecycle, provider) = fixture("catalog-refresh-failure");
2813            remove_provider_catalog("catalog-refresh-failure");
2814            seed_legacy_cached_provider(&lifecycle, "catalog-refresh-failure", true);
2815            lifecycle
2816                .restore_provider(
2817                    "catalog-refresh-failure",
2818                    ProviderKind::Codex,
2819                    provider.clone(),
2820                )
2821                .await
2822                .unwrap();
2823            provider.fail_once(ModelDiscoveryError::Transport("offline".into()));
2824            let (started, proceed) = provider.block_once();
2825            let background_lifecycle = lifecycle.clone();
2826            let background = tokio::spawn(async move {
2827                background_lifecycle
2828                    .refresh_models_if_stale("catalog-refresh-failure")
2829                    .await
2830            });
2831            started.await.unwrap();
2832
2833            assert_eq!(
2834                lifecycle
2835                    .refresh_models_if_stale("catalog-refresh-failure")
2836                    .await
2837                    .unwrap(),
2838                ProviderCatalogRefreshOutcome::AlreadyInFlight
2839            );
2840            assert_eq!(provider.discovery_calls(), 1);
2841            proceed.send(()).unwrap();
2842            assert!(matches!(
2843                background.await.unwrap(),
2844                Err(ProviderLifecycleError::Discovery(
2845                    ModelDiscoveryError::Transport(message)
2846                )) if message == "offline"
2847            ));
2848            assert_eq!(provider.discovery_calls(), 1);
2849
2850            lifecycle
2851                .remove_provider("catalog-refresh-failure")
2852                .unwrap();
2853        });
2854    }
2855
2856    #[test]
2857    fn enabling_a_cached_provider_is_atomic_and_offline() {
2858        let _registry_lock = crate::model_registry::MODEL_CONFIG_LOCK
2859            .lock()
2860            .unwrap_or_else(std::sync::PoisonError::into_inner);
2861        test_runtime().block_on(async {
2862            let (_dir, lifecycle, provider) = fixture("lifecycle-enable");
2863            remove_provider_catalog("lifecycle-enable");
2864            let namespace =
2865                seed_cached_provider(&lifecycle, "lifecycle-enable", false, &[model("cached")]);
2866            provider.fail_once(ModelDiscoveryError::Transport("unused".into()));
2867
2868            let outcome = lifecycle
2869                .enable_provider("lifecycle-enable", ProviderKind::Codex, provider.clone())
2870                .await
2871                .unwrap();
2872
2873            assert_eq!(provider.discovery_calls(), 0);
2874            assert_eq!(
2875                outcome.state,
2876                ProviderStateChange {
2877                    auth_changed: true,
2878                    live_changed: true,
2879                    catalog_changed: true,
2880                }
2881            );
2882            assert!(lifecycle.hub.load_auth().unwrap().providers[0].enabled);
2883            assert!(lifecycle.providers.contains("lifecycle-enable"));
2884            assert!(crate::model_registry::model_entry(&format!("{namespace}:cached")).is_some());
2885            lifecycle.remove_provider("lifecycle-enable").unwrap();
2886        });
2887    }
2888
2889    #[test]
2890    fn disabled_restore_ignores_an_invalid_cached_catalog() {
2891        let _registry_lock = crate::model_registry::MODEL_CONFIG_LOCK
2892            .lock()
2893            .unwrap_or_else(std::sync::PoisonError::into_inner);
2894        test_runtime().block_on(async {
2895            let (_dir, lifecycle, provider) = fixture("lifecycle-disabled-cache");
2896            remove_provider_catalog("lifecycle-disabled-cache");
2897            let mut record = provider_record("lifecycle-disabled-cache");
2898            record.enabled = false;
2899            record.model_cache = Some(crate::auth_store::ModelCache {
2900                fetched_at: 1,
2901                models: vec![
2902                    crate::auth_store::CachedModel {
2903                        slug: "duplicate".into(),
2904                        context_budget: None,
2905                        thinking: false,
2906                    },
2907                    crate::auth_store::CachedModel {
2908                        slug: "duplicate".into(),
2909                        context_budget: None,
2910                        thinking: false,
2911                    },
2912                ],
2913            });
2914            lifecycle.hub.add_auth_provider(record).unwrap();
2915
2916            assert!(matches!(
2917                lifecycle
2918                    .restore_provider(
2919                        "lifecycle-disabled-cache",
2920                        ProviderKind::Codex,
2921                        provider.clone(),
2922                    )
2923                    .await,
2924                Err(ProviderLifecycleError::ProviderDisabled { id })
2925                    if id == "lifecycle-disabled-cache"
2926            ));
2927            assert_eq!(provider.discovery_calls(), 0);
2928            assert!(!lifecycle.providers.contains("lifecycle-disabled-cache"));
2929            assert!(provider_catalog_namespace("lifecycle-disabled-cache").is_none());
2930            lifecycle
2931                .remove_provider("lifecycle-disabled-cache")
2932                .unwrap();
2933        });
2934    }
2935
2936    #[test]
2937    fn restoring_without_cache_removes_the_stale_catalog() {
2938        let _registry_lock = crate::model_registry::MODEL_CONFIG_LOCK
2939            .lock()
2940            .unwrap_or_else(std::sync::PoisonError::into_inner);
2941        test_runtime().block_on(async {
2942            let (_dir, lifecycle, provider) = fixture("lifecycle-empty-cache");
2943            remove_provider_catalog("lifecycle-empty-cache");
2944            lifecycle
2945                .install_provider(provider_record("lifecycle-empty-cache"), provider.clone())
2946                .await
2947                .unwrap();
2948            lifecycle
2949                .hub
2950                .update_auth(|store| {
2951                    store.providers[0].model_cache = None;
2952                    Ok(())
2953                })
2954                .unwrap();
2955            let replacement = Arc::new(TestProvider::new(
2956                "lifecycle-empty-cache",
2957                vec![model("unused")],
2958            ));
2959
2960            let outcome = lifecycle
2961                .restore_provider(
2962                    "lifecycle-empty-cache",
2963                    ProviderKind::Codex,
2964                    replacement.clone(),
2965                )
2966                .await
2967                .unwrap();
2968
2969            assert_eq!(provider.discovery_calls(), 1);
2970            assert_eq!(replacement.discovery_calls(), 0);
2971            assert!(!outcome.state.live_changed);
2972            assert!(outcome.state.catalog_changed);
2973            assert!(outcome.catalog.is_none());
2974            assert!(lifecycle.providers.contains("lifecycle-empty-cache"));
2975            assert!(provider_catalog_namespace("lifecycle-empty-cache").is_none());
2976            lifecycle.remove_provider("lifecycle-empty-cache").unwrap();
2977        });
2978    }
2979
2980    #[test]
2981    fn reconcile_prunes_external_disable_remove_and_kind_change() {
2982        let _registry_lock = crate::model_registry::MODEL_CONFIG_LOCK
2983            .lock()
2984            .unwrap_or_else(std::sync::PoisonError::into_inner);
2985        test_runtime().block_on(async {
2986            let dir = tempfile::tempdir().unwrap();
2987            let lifecycle = ProviderLifecycle::new(
2988                ConfigHub::from_config_dir(dir.path()),
2989                ProviderRegistry::new(),
2990            );
2991            let ids = ["lifecycle-disabled", "lifecycle-removed", "lifecycle-kind"];
2992            for id in ids {
2993                remove_provider_catalog(id);
2994                lifecycle
2995                    .install_provider(
2996                        provider_record(id),
2997                        Arc::new(TestProvider::new(id, vec![model("cached")])),
2998                    )
2999                    .await
3000                    .unwrap();
3001            }
3002            let peer_registry = ProviderRegistry::new();
3003            let _peer = ProviderLifecycle::new(lifecycle.hub.clone(), peer_registry.clone());
3004
3005            lifecycle
3006                .hub
3007                .set_auth_provider_enabled("lifecycle-disabled", false)
3008                .unwrap();
3009            lifecycle
3010                .hub
3011                .remove_auth_provider("lifecycle-removed")
3012                .unwrap();
3013            lifecycle
3014                .hub
3015                .update_auth(|store| {
3016                    store
3017                        .providers
3018                        .iter_mut()
3019                        .find(|provider| provider.id == "lifecycle-kind")
3020                        .unwrap()
3021                        .kind = ProviderKind::AnthropicOauth;
3022                    Ok(())
3023                })
3024                .unwrap();
3025
3026            let outcome = lifecycle.reconcile_inactive_providers().unwrap();
3027
3028            assert_eq!(outcome.providers.len(), 3);
3029            for id in ids {
3030                assert!(!lifecycle.providers.contains(id));
3031                assert!(!peer_registry.contains(id));
3032                assert!(provider_catalog_namespace(id).is_none());
3033            }
3034            let store = lifecycle.hub.load_auth().unwrap();
3035            assert_eq!(store.providers.len(), 2);
3036            assert!(
3037                !store
3038                    .providers
3039                    .iter()
3040                    .find(|provider| provider.id == "lifecycle-disabled")
3041                    .unwrap()
3042                    .enabled
3043            );
3044            assert_eq!(
3045                store
3046                    .providers
3047                    .iter()
3048                    .find(|provider| provider.id == "lifecycle-kind")
3049                    .unwrap()
3050                    .kind,
3051                ProviderKind::AnthropicOauth
3052            );
3053            lifecycle.remove_provider("lifecycle-disabled").unwrap();
3054            lifecycle.remove_provider("lifecycle-kind").unwrap();
3055        });
3056    }
3057
3058    #[test]
3059    fn reconcile_auth_parse_error_prunes_live_runtime() {
3060        let _registry_lock = crate::model_registry::MODEL_CONFIG_LOCK
3061            .lock()
3062            .unwrap_or_else(std::sync::PoisonError::into_inner);
3063        test_runtime().block_on(async {
3064            let (_dir, lifecycle, provider) = fixture("lifecycle-reconcile-invalid");
3065            remove_provider_catalog("lifecycle-reconcile-invalid");
3066            lifecycle
3067                .install_provider(provider_record("lifecycle-reconcile-invalid"), provider)
3068                .await
3069                .unwrap();
3070            std::fs::write(lifecycle.hub.auth_path(), b"{").unwrap();
3071
3072            assert!(matches!(
3073                lifecycle.reconcile_inactive_providers(),
3074                Err(ProviderLifecycleError::Config(_))
3075            ));
3076            assert!(!lifecycle.providers.contains("lifecycle-reconcile-invalid"));
3077            assert!(provider_catalog_namespace("lifecycle-reconcile-invalid").is_none());
3078        });
3079    }
3080
3081    #[test]
3082    fn provider_mutation_auth_errors_fail_closed() {
3083        let _registry_lock = crate::model_registry::MODEL_CONFIG_LOCK
3084            .lock()
3085            .unwrap_or_else(std::sync::PoisonError::into_inner);
3086        test_runtime().block_on(async {
3087            for (id, remove) in [
3088                ("lifecycle-disable-invalid", false),
3089                ("lifecycle-remove-invalid", true),
3090            ] {
3091                let (_dir, lifecycle, provider) = fixture(id);
3092                remove_provider_catalog(id);
3093                lifecycle
3094                    .install_provider(provider_record(id), provider)
3095                    .await
3096                    .unwrap();
3097                std::fs::write(lifecycle.hub.auth_path(), b"{").unwrap();
3098
3099                let result = if remove {
3100                    lifecycle.remove_provider(id)
3101                } else {
3102                    lifecycle.disable_provider(id)
3103                };
3104                assert!(matches!(result, Err(ProviderLifecycleError::Config(_))));
3105                assert!(!lifecycle.providers.contains(id));
3106                assert!(provider_catalog_namespace(id).is_none());
3107            }
3108        });
3109    }
3110
3111    #[test]
3112    fn kind_change_after_construction_fails_closed() {
3113        let _registry_lock = crate::model_registry::MODEL_CONFIG_LOCK
3114            .lock()
3115            .unwrap_or_else(std::sync::PoisonError::into_inner);
3116        test_runtime().block_on(async {
3117            let (_dir, lifecycle, provider) = fixture("lifecycle-kind-race");
3118            remove_provider_catalog("lifecycle-kind-race");
3119            seed_cached_provider(&lifecycle, "lifecycle-kind-race", true, &[model("cached")]);
3120            lifecycle
3121                .restore_provider("lifecycle-kind-race", ProviderKind::Codex, provider)
3122                .await
3123                .unwrap();
3124            lifecycle
3125                .hub
3126                .update_auth(|store| {
3127                    store.providers[0].kind = ProviderKind::AnthropicOauth;
3128                    Ok(())
3129                })
3130                .unwrap();
3131            let stale_candidate = Arc::new(TestProvider::new(
3132                "lifecycle-kind-race",
3133                vec![model("unused")],
3134            ));
3135
3136            assert!(matches!(
3137                lifecycle
3138                    .restore_provider(
3139                        "lifecycle-kind-race",
3140                        ProviderKind::Codex,
3141                        stale_candidate,
3142                    )
3143                    .await,
3144                Err(ProviderLifecycleError::Stale { id }) if id == "lifecycle-kind-race"
3145            ));
3146            assert!(!lifecycle.providers.contains("lifecycle-kind-race"));
3147            assert!(provider_catalog_namespace("lifecycle-kind-race").is_none());
3148            lifecycle.remove_provider("lifecycle-kind-race").unwrap();
3149        });
3150    }
3151
3152    #[test]
3153    fn invalid_cached_catalog_fails_closed() {
3154        let _registry_lock = crate::model_registry::MODEL_CONFIG_LOCK
3155            .lock()
3156            .unwrap_or_else(std::sync::PoisonError::into_inner);
3157        test_runtime().block_on(async {
3158            let (_dir, lifecycle, provider) = fixture("lifecycle-invalid-cache");
3159            remove_provider_catalog("lifecycle-invalid-cache");
3160            seed_cached_provider(
3161                &lifecycle,
3162                "lifecycle-invalid-cache",
3163                true,
3164                &[model("cached")],
3165            );
3166            let operation_lock = lifecycle.operation_lock("lifecycle-invalid-cache");
3167            let dropped = Arc::new(AtomicBool::new(false));
3168            let dropped_from_callback = dropped.clone();
3169            provider.on_drop(move || {
3170                assert!(operation_lock.try_lock().is_ok());
3171                dropped_from_callback.store(true, Ordering::SeqCst);
3172            });
3173            lifecycle
3174                .restore_provider("lifecycle-invalid-cache", ProviderKind::Codex, provider)
3175                .await
3176                .unwrap();
3177            lifecycle
3178                .hub
3179                .update_auth(|store| {
3180                    store.providers[0].model_cache = Some(crate::auth_store::ModelCache {
3181                        fetched_at: 2,
3182                        models: vec![
3183                            crate::auth_store::CachedModel {
3184                                slug: "duplicate".into(),
3185                                context_budget: None,
3186                                thinking: false,
3187                            },
3188                            crate::auth_store::CachedModel {
3189                                slug: "duplicate".into(),
3190                                context_budget: None,
3191                                thinking: false,
3192                            },
3193                        ],
3194                    });
3195                    Ok(())
3196                })
3197                .unwrap();
3198
3199            assert!(matches!(
3200                lifecycle
3201                    .restore_provider(
3202                        "lifecycle-invalid-cache",
3203                        ProviderKind::Codex,
3204                        Arc::new(TestProvider::new(
3205                            "lifecycle-invalid-cache",
3206                            vec![model("unused")],
3207                        )),
3208                    )
3209                    .await,
3210                Err(ProviderLifecycleError::Catalog(CatalogError::DuplicateModel { model }))
3211                    if model == "duplicate"
3212            ));
3213            assert!(!lifecycle.providers.contains("lifecycle-invalid-cache"));
3214            assert!(provider_catalog_namespace("lifecycle-invalid-cache").is_none());
3215            assert!(dropped.load(Ordering::SeqCst));
3216            lifecycle
3217                .remove_provider("lifecycle-invalid-cache")
3218                .unwrap();
3219        });
3220    }
3221
3222    #[test]
3223    fn refresh_cancellation_releases_the_operation_before_provider_drop() {
3224        let _registry_lock = crate::model_registry::MODEL_CONFIG_LOCK
3225            .lock()
3226            .unwrap_or_else(std::sync::PoisonError::into_inner);
3227        test_runtime().block_on(async {
3228            let (_dir, lifecycle, provider) = fixture("lifecycle-refresh-cancel");
3229            remove_provider_catalog("lifecycle-refresh-cancel");
3230            lifecycle
3231                .install_provider(
3232                    provider_record("lifecycle-refresh-cancel"),
3233                    provider.clone(),
3234                )
3235                .await
3236                .unwrap();
3237            let operation_lock = lifecycle.operation_lock("lifecycle-refresh-cancel");
3238            let dropped = Arc::new(AtomicBool::new(false));
3239            let dropped_from_callback = dropped.clone();
3240            provider.on_drop(move || {
3241                assert!(operation_lock.try_lock().is_ok());
3242                dropped_from_callback.store(true, Ordering::SeqCst);
3243            });
3244            let (started, _proceed) = provider.block_once();
3245            drop(provider);
3246            let refresh_lifecycle = lifecycle.clone();
3247            let refresh = tokio::spawn(async move {
3248                refresh_lifecycle
3249                    .refresh_models("lifecycle-refresh-cancel")
3250                    .await
3251            });
3252            started.await.unwrap();
3253            lifecycle
3254                .disable_provider("lifecycle-refresh-cancel")
3255                .unwrap();
3256
3257            refresh.abort();
3258            assert!(refresh.await.unwrap_err().is_cancelled());
3259            assert!(dropped.load(Ordering::SeqCst));
3260            lifecycle
3261                .remove_provider("lifecycle-refresh-cancel")
3262                .unwrap();
3263        });
3264    }
3265
3266    #[test]
3267    fn provider_id_compare_and_swap_rejects_stale_namespaces() {
3268        let dir = tempfile::tempdir().unwrap();
3269        let hub = ConfigHub::from_config_dir(dir.path());
3270        let mut target = provider_record("namespace-target");
3271        target.enabled = false;
3272        hub.add_auth_provider(target).unwrap();
3273        let runtime = hub
3274            .load_or_create_auth_provider_runtime_state("namespace-target")
3275            .unwrap()
3276            .unwrap();
3277        hub.add_auth_provider(provider_record("namespace-peer"))
3278            .unwrap();
3279        let callback_ran = AtomicBool::new(false);
3280
3281        let (commit, callback) = hub
3282            .commit_auth_provider_runtime_if_current_and_then(
3283                &runtime.provider,
3284                &runtime.catalog_snapshot,
3285                true,
3286                Some("namespace-target@account"),
3287                Some(&runtime.provider_ids),
3288                || callback_ran.store(true, Ordering::SeqCst),
3289            )
3290            .unwrap();
3291
3292        assert_eq!(commit, AuthProviderRuntimeCommit::Changed);
3293        assert!(callback.is_none());
3294        assert!(!callback_ran.load(Ordering::SeqCst));
3295        let stored = hub
3296            .load_auth()
3297            .unwrap()
3298            .providers
3299            .into_iter()
3300            .find(|provider| provider.id == "namespace-target")
3301            .unwrap();
3302        assert!(!stored.enabled);
3303        assert_eq!(
3304            hub.load_auth_model_namespace("namespace-target").unwrap(),
3305            None
3306        );
3307    }
3308
3309    #[test]
3310    fn provider_insert_rejects_a_changed_provider_id_set() {
3311        let dir = tempfile::tempdir().unwrap();
3312        let hub = ConfigHub::from_config_dir(dir.path());
3313        let expected_provider_ids = Vec::new();
3314        hub.add_auth_provider(provider_record("insert-peer"))
3315            .unwrap();
3316        let callback_ran = AtomicBool::new(false);
3317
3318        let (commit, callback) = hub
3319            .add_auth_provider_with_model_cache_details_if_provider_ids_and_then(
3320                provider_record("insert-target"),
3321                Some(&expected_provider_ids),
3322                "insert-target@account",
3323                1,
3324                &[model("cached")],
3325                || callback_ran.store(true, Ordering::SeqCst),
3326            )
3327            .unwrap();
3328
3329        assert_eq!(commit, AuthProviderInsertCommit::Changed);
3330        assert!(callback.is_none());
3331        assert!(!callback_ran.load(Ordering::SeqCst));
3332        assert!(
3333            hub.load_auth()
3334                .unwrap()
3335                .providers
3336                .iter()
3337                .all(|provider| provider.id != "insert-target")
3338        );
3339    }
3340
3341    #[test]
3342    fn provider_drop_runs_after_lifecycle_locks_are_released() {
3343        let _registry_lock = crate::model_registry::MODEL_CONFIG_LOCK
3344            .lock()
3345            .unwrap_or_else(std::sync::PoisonError::into_inner);
3346        test_runtime().block_on(async {
3347            let (_dir, lifecycle, provider) = fixture("lifecycle-drop");
3348            remove_provider_catalog("lifecycle-drop");
3349            lifecycle
3350                .install_provider(provider_record("lifecycle-drop"), provider.clone())
3351                .await
3352                .unwrap();
3353            let state = Arc::downgrade(&lifecycle.state);
3354            let registry = lifecycle.providers.clone();
3355            let hub = lifecycle.hub.clone();
3356            let dropped = Arc::new(AtomicBool::new(false));
3357            let dropped_from_callback = dropped.clone();
3358            provider.on_drop(move || {
3359                assert!(state.upgrade().unwrap().try_lock().is_ok());
3360                assert!(registry.get("lifecycle-drop").is_none());
3361                assert!(
3362                    !hub.set_auth_provider_enabled("lifecycle-drop", false)
3363                        .unwrap()
3364                );
3365                dropped_from_callback.store(true, Ordering::SeqCst);
3366            });
3367            drop(provider);
3368
3369            lifecycle.remove_provider("lifecycle-drop").unwrap();
3370
3371            assert!(dropped.load(Ordering::SeqCst));
3372        });
3373    }
3374
3375    #[test]
3376    fn failed_refresh_preserves_cache_and_catalog_revision() {
3377        let _registry_lock = crate::model_registry::MODEL_CONFIG_LOCK
3378            .lock()
3379            .unwrap_or_else(std::sync::PoisonError::into_inner);
3380        test_runtime().block_on(async {
3381            let (_dir, lifecycle, provider) = fixture("lifecycle-failure");
3382            remove_provider_catalog("lifecycle-failure");
3383            lifecycle
3384                .install_provider(provider_record("lifecycle-failure"), provider.clone())
3385                .await
3386                .unwrap();
3387            let auth_path = lifecycle.hub.config_dir().join("auth.json");
3388            let before_auth = std::fs::read(&auth_path).unwrap();
3389            let before_revision = crate::model_registry::model_catalog_revision();
3390            provider.fail_once(ModelDiscoveryError::Transport("offline".into()));
3391
3392            assert!(matches!(
3393                lifecycle.refresh_models("lifecycle-failure").await,
3394                Err(ProviderLifecycleError::Discovery(
3395                    ModelDiscoveryError::Transport(message)
3396                )) if message == "offline"
3397            ));
3398            assert_eq!(std::fs::read(auth_path).unwrap(), before_auth);
3399            assert_eq!(
3400                crate::model_registry::model_catalog_revision(),
3401                before_revision
3402            );
3403            lifecycle.remove_provider("lifecycle-failure").unwrap();
3404        });
3405    }
3406
3407    #[test]
3408    fn refresh_auth_parse_error_prunes_live_runtime() {
3409        let _registry_lock = crate::model_registry::MODEL_CONFIG_LOCK
3410            .lock()
3411            .unwrap_or_else(std::sync::PoisonError::into_inner);
3412        test_runtime().block_on(async {
3413            let (_dir, lifecycle, provider) = fixture("lifecycle-refresh-invalid-auth");
3414            remove_provider_catalog("lifecycle-refresh-invalid-auth");
3415            lifecycle
3416                .install_provider(
3417                    provider_record("lifecycle-refresh-invalid-auth"),
3418                    provider.clone(),
3419                )
3420                .await
3421                .unwrap();
3422            let auth_path = lifecycle.hub.auth_path().to_path_buf();
3423            provider.before_discovery_once(move || {
3424                std::fs::write(auth_path, b"{").unwrap();
3425            });
3426
3427            assert!(matches!(
3428                lifecycle
3429                    .refresh_models("lifecycle-refresh-invalid-auth")
3430                    .await,
3431                Err(ProviderLifecycleError::Config(_))
3432            ));
3433            assert!(
3434                !lifecycle
3435                    .providers
3436                    .contains("lifecycle-refresh-invalid-auth")
3437            );
3438            assert!(provider_catalog_namespace("lifecycle-refresh-invalid-auth").is_none());
3439        });
3440    }
3441
3442    #[test]
3443    fn disable_invalidates_in_flight_refresh_without_resurrection() {
3444        let _registry_lock = crate::model_registry::MODEL_CONFIG_LOCK
3445            .lock()
3446            .unwrap_or_else(std::sync::PoisonError::into_inner);
3447        test_runtime().block_on(async {
3448            let (_dir, lifecycle, provider) = fixture("lifecycle-race");
3449            remove_provider_catalog("lifecycle-race");
3450            lifecycle
3451                .install_provider(provider_record("lifecycle-race"), provider.clone())
3452                .await
3453                .unwrap();
3454            let peer_registry = ProviderRegistry::new();
3455            let peer = ProviderLifecycle::new(lifecycle.hub.clone(), peer_registry.clone());
3456            assert!(Arc::ptr_eq(&lifecycle.state, &peer.state));
3457            assert!(peer_registry.contains("lifecycle-race"));
3458            provider.set_models(vec![model("replacement")]);
3459            let (started, proceed) = provider.block_once();
3460            let refresh_lifecycle = lifecycle.clone();
3461            let refresh =
3462                tokio::spawn(
3463                    async move { refresh_lifecycle.refresh_models("lifecycle-race").await },
3464                );
3465            started.await.unwrap();
3466
3467            peer.disable_provider("lifecycle-race").unwrap();
3468            proceed.send(()).unwrap();
3469            assert!(matches!(
3470                refresh.await.unwrap(),
3471                Err(ProviderLifecycleError::Stale { id }) if id == "lifecycle-race"
3472            ));
3473            assert!(!lifecycle.providers.contains("lifecycle-race"));
3474            assert!(!peer_registry.contains("lifecycle-race"));
3475            assert!(provider_catalog_namespace("lifecycle-race").is_none());
3476            let stored = lifecycle.hub.load_auth().unwrap().providers.remove(0);
3477            assert!(!stored.enabled);
3478            assert_eq!(stored.model_cache.unwrap().models[0].slug, "initial");
3479            lifecycle.remove_provider("lifecycle-race").unwrap();
3480        });
3481    }
3482
3483    #[test]
3484    fn independent_lifecycles_serialize_refreshes_for_the_same_auth_store() {
3485        let _registry_lock = crate::model_registry::MODEL_CONFIG_LOCK
3486            .lock()
3487            .unwrap_or_else(std::sync::PoisonError::into_inner);
3488        test_runtime().block_on(async {
3489            let (_dir, lifecycle, provider) = fixture("lifecycle-shared-gate");
3490            remove_provider_catalog("lifecycle-shared-gate");
3491            lifecycle
3492                .install_provider(provider_record("lifecycle-shared-gate"), provider.clone())
3493                .await
3494                .unwrap();
3495            let peer_registry = ProviderRegistry::new();
3496            let peer = ProviderLifecycle::new(lifecycle.hub.clone(), peer_registry.clone());
3497            assert!(Arc::ptr_eq(&lifecycle.state, &peer.state));
3498            assert!(peer_registry.contains("lifecycle-shared-gate"));
3499
3500            provider.set_models(vec![model("slow")]);
3501            let (started, proceed) = provider.block_once();
3502            let slow_lifecycle = lifecycle.clone();
3503            let slow = tokio::spawn(async move {
3504                slow_lifecycle.refresh_models("lifecycle-shared-gate").await
3505            });
3506            started.await.unwrap();
3507            provider.set_models(vec![model("fresh")]);
3508            let fast =
3509                tokio::spawn(async move { peer.refresh_models("lifecycle-shared-gate").await });
3510            tokio::task::yield_now().await;
3511            assert!(!fast.is_finished());
3512
3513            proceed.send(()).unwrap();
3514            slow.await.unwrap().unwrap();
3515            fast.await.unwrap().unwrap();
3516
3517            let stored = lifecycle.hub.load_auth().unwrap().providers.remove(0);
3518            assert_eq!(stored.model_cache.unwrap().models[0].slug, "fresh");
3519            let namespace = lifecycle
3520                .hub
3521                .load_auth_model_namespace("lifecycle-shared-gate")
3522                .unwrap()
3523                .unwrap();
3524            assert!(crate::model_registry::model_entry(&format!("{namespace}:fresh")).is_some());
3525            assert!(crate::model_registry::model_entry(&format!("{namespace}:slow")).is_none());
3526            lifecycle.remove_provider("lifecycle-shared-gate").unwrap();
3527        });
3528    }
3529
3530    #[test]
3531    fn external_cache_aba_hydrates_the_authoritative_catalog() {
3532        let _registry_lock = crate::model_registry::MODEL_CONFIG_LOCK
3533            .lock()
3534            .unwrap_or_else(std::sync::PoisonError::into_inner);
3535        test_runtime().block_on(async {
3536            let (_dir, lifecycle, provider) = fixture("lifecycle-cache-cas");
3537            remove_provider_catalog("lifecycle-cache-cas");
3538            lifecycle
3539                .install_provider(provider_record("lifecycle-cache-cas"), provider.clone())
3540                .await
3541                .unwrap();
3542            let namespace = lifecycle
3543                .hub
3544                .load_auth_model_namespace("lifecycle-cache-cas")
3545                .unwrap()
3546                .unwrap();
3547            let initial_fetched_at = lifecycle.hub.load_auth().unwrap().providers[0]
3548                .model_cache
3549                .as_ref()
3550                .unwrap()
3551                .fetched_at;
3552            let initial = lifecycle
3553                .hub
3554                .load_auth_model_cache_details("lifecycle-cache-cas")
3555                .unwrap()
3556                .unwrap();
3557            provider.set_models(vec![model("slow")]);
3558            let (started, proceed) = provider.block_once();
3559            let refresh_lifecycle = lifecycle.clone();
3560            let refresh = tokio::spawn(async move {
3561                refresh_lifecycle
3562                    .refresh_models("lifecycle-cache-cas")
3563                    .await
3564            });
3565            started.await.unwrap();
3566
3567            assert!(
3568                lifecycle
3569                    .hub
3570                    .update_auth_model_cache_details(
3571                        "lifecycle-cache-cas",
3572                        &namespace,
3573                        99,
3574                        &[model("external")],
3575                    )
3576                    .unwrap()
3577            );
3578            assert!(
3579                lifecycle
3580                    .hub
3581                    .update_auth_model_cache_details(
3582                        "lifecycle-cache-cas",
3583                        &namespace,
3584                        initial_fetched_at,
3585                        &initial,
3586                    )
3587                    .unwrap()
3588            );
3589            proceed.send(()).unwrap();
3590            assert_eq!(
3591                refresh.await.unwrap().unwrap(),
3592                CatalogDelta {
3593                    total: 1,
3594                    ..Default::default()
3595                }
3596            );
3597
3598            let stored = lifecycle.hub.load_auth().unwrap().providers.remove(0);
3599            assert_eq!(stored.model_cache.unwrap().models[0].slug, "initial");
3600            assert!(crate::model_registry::model_entry(&format!("{namespace}:initial")).is_some());
3601            assert!(crate::model_registry::model_entry(&format!("{namespace}:slow")).is_none());
3602            lifecycle.remove_provider("lifecycle-cache-cas").unwrap();
3603        });
3604    }
3605
3606    #[test]
3607    fn external_enable_aba_preserves_the_authoritative_cache() {
3608        let _registry_lock = crate::model_registry::MODEL_CONFIG_LOCK
3609            .lock()
3610            .unwrap_or_else(std::sync::PoisonError::into_inner);
3611        test_runtime().block_on(async {
3612            let (_dir, lifecycle, provider) = fixture("lifecycle-enable-aba");
3613            remove_provider_catalog("lifecycle-enable-aba");
3614            lifecycle
3615                .install_provider(provider_record("lifecycle-enable-aba"), provider.clone())
3616                .await
3617                .unwrap();
3618            provider.set_models(vec![model("slow")]);
3619            let (started, proceed) = provider.block_once();
3620            let refresh_lifecycle = lifecycle.clone();
3621            let refresh = tokio::spawn(async move {
3622                refresh_lifecycle
3623                    .refresh_models("lifecycle-enable-aba")
3624                    .await
3625            });
3626            started.await.unwrap();
3627
3628            lifecycle
3629                .hub
3630                .set_auth_provider_enabled("lifecycle-enable-aba", false)
3631                .unwrap();
3632            lifecycle
3633                .hub
3634                .set_auth_provider_enabled("lifecycle-enable-aba", true)
3635                .unwrap();
3636            proceed.send(()).unwrap();
3637
3638            assert_eq!(
3639                refresh.await.unwrap().unwrap(),
3640                CatalogDelta {
3641                    total: 1,
3642                    ..Default::default()
3643                }
3644            );
3645            let stored = lifecycle.hub.load_auth().unwrap().providers.remove(0);
3646            assert!(stored.enabled);
3647            assert_eq!(stored.model_cache.unwrap().models[0].slug, "initial");
3648            lifecycle.remove_provider("lifecycle-enable-aba").unwrap();
3649        });
3650    }
3651
3652    #[test]
3653    fn external_kind_change_during_refresh_prunes_the_old_runtime() {
3654        let _registry_lock = crate::model_registry::MODEL_CONFIG_LOCK
3655            .lock()
3656            .unwrap_or_else(std::sync::PoisonError::into_inner);
3657        test_runtime().block_on(async {
3658            let (_dir, lifecycle, provider) = fixture("lifecycle-refresh-kind");
3659            remove_provider_catalog("lifecycle-refresh-kind");
3660            lifecycle
3661                .install_provider(provider_record("lifecycle-refresh-kind"), provider.clone())
3662                .await
3663                .unwrap();
3664            provider.set_models(vec![model("replacement")]);
3665            let (started, proceed) = provider.block_once();
3666            let refresh_lifecycle = lifecycle.clone();
3667            let refresh = tokio::spawn(async move {
3668                refresh_lifecycle
3669                    .refresh_models("lifecycle-refresh-kind")
3670                    .await
3671            });
3672            started.await.unwrap();
3673            lifecycle
3674                .hub
3675                .update_auth(|store| {
3676                    store.providers[0].kind = ProviderKind::AnthropicOauth;
3677                    Ok(())
3678                })
3679                .unwrap();
3680            proceed.send(()).unwrap();
3681
3682            assert!(matches!(
3683                refresh.await.unwrap(),
3684                Err(ProviderLifecycleError::Stale { id }) if id == "lifecycle-refresh-kind"
3685            ));
3686            assert!(!lifecycle.providers.contains("lifecycle-refresh-kind"));
3687            assert!(provider_catalog_namespace("lifecycle-refresh-kind").is_none());
3688            lifecycle.remove_provider("lifecycle-refresh-kind").unwrap();
3689        });
3690    }
3691
3692    #[test]
3693    fn discovery_failure_observes_an_external_disable() {
3694        let _registry_lock = crate::model_registry::MODEL_CONFIG_LOCK
3695            .lock()
3696            .unwrap_or_else(std::sync::PoisonError::into_inner);
3697        test_runtime().block_on(async {
3698            let (_dir, lifecycle, provider) = fixture("lifecycle-refresh-disabled");
3699            remove_provider_catalog("lifecycle-refresh-disabled");
3700            lifecycle
3701                .install_provider(
3702                    provider_record("lifecycle-refresh-disabled"),
3703                    provider.clone(),
3704                )
3705                .await
3706                .unwrap();
3707            let hub = lifecycle.hub.clone();
3708            provider.before_discovery_once(move || {
3709                assert!(
3710                    hub.set_auth_provider_enabled("lifecycle-refresh-disabled", false)
3711                        .unwrap()
3712                );
3713            });
3714            provider.fail_once(ModelDiscoveryError::Transport("offline".into()));
3715
3716            assert!(matches!(
3717                lifecycle.refresh_models("lifecycle-refresh-disabled").await,
3718                Err(ProviderLifecycleError::Stale { id }) if id == "lifecycle-refresh-disabled"
3719            ));
3720            assert!(!lifecycle.providers.contains("lifecycle-refresh-disabled"));
3721            assert!(provider_catalog_namespace("lifecycle-refresh-disabled").is_none());
3722            lifecycle
3723                .remove_provider("lifecycle-refresh-disabled")
3724                .unwrap();
3725        });
3726    }
3727
3728    #[test]
3729    fn namespace_mismatch_during_refresh_fails_closed() {
3730        let _registry_lock = crate::model_registry::MODEL_CONFIG_LOCK
3731            .lock()
3732            .unwrap_or_else(std::sync::PoisonError::into_inner);
3733        test_runtime().block_on(async {
3734            let (_dir, lifecycle, provider) = fixture("lifecycle-refresh-namespace");
3735            remove_provider_catalog("lifecycle-refresh-namespace");
3736            lifecycle
3737                .install_provider(
3738                    provider_record("lifecycle-refresh-namespace"),
3739                    provider.clone(),
3740                )
3741                .await
3742                .unwrap();
3743            remove_provider_catalog("lifecycle-refresh-namespace");
3744            crate::model_registry::replace_provider_catalog(
3745                ProviderDescriptor {
3746                    provider_key: "lifecycle-refresh-namespace".into(),
3747                    provider_name: "Account".into(),
3748                    namespace: "unexpected@account".into(),
3749                    wire_profile: ReasoningWireProfile::CodexResponses,
3750                },
3751                &[model("replacement")],
3752            )
3753            .unwrap();
3754
3755            assert!(matches!(
3756                lifecycle
3757                    .refresh_models("lifecycle-refresh-namespace")
3758                    .await,
3759                Err(ProviderLifecycleError::Catalog(
3760                    CatalogError::NamespaceChanged { .. }
3761                ))
3762            ));
3763            assert!(!lifecycle.providers.contains("lifecycle-refresh-namespace"));
3764            assert!(provider_catalog_namespace("lifecycle-refresh-namespace").is_none());
3765            lifecycle
3766                .remove_provider("lifecycle-refresh-namespace")
3767                .unwrap();
3768        });
3769    }
3770
3771    #[test]
3772    fn registry_key_conflict_during_refresh_preserves_last_good_catalog() {
3773        struct ConfigReset;
3774
3775        impl Drop for ConfigReset {
3776            fn drop(&mut self) {
3777                crate::model_registry::set_provider_config(
3778                    crate::model_registry::ProviderConfig::default(),
3779                );
3780            }
3781        }
3782
3783        let _registry_lock = crate::model_registry::MODEL_CONFIG_LOCK
3784            .lock()
3785            .unwrap_or_else(std::sync::PoisonError::into_inner);
3786        let _config_reset = ConfigReset;
3787        test_runtime().block_on(async {
3788            let (_dir, lifecycle, provider) = fixture("lifecycle-refresh-registry-key");
3789            remove_provider_catalog("lifecycle-refresh-registry-key");
3790            lifecycle
3791                .hub
3792                .add_auth_provider_with_model_cache_details(
3793                    provider_record("lifecycle-refresh-registry-key"),
3794                    "stable",
3795                    1,
3796                    &[model("initial")],
3797                )
3798                .unwrap();
3799            lifecycle
3800                .restore_provider(
3801                    "lifecycle-refresh-registry-key",
3802                    ProviderKind::Codex,
3803                    provider.clone(),
3804                )
3805                .await
3806                .unwrap();
3807            let mut config = crate::model_registry::ProviderConfig::default();
3808            config.providers.insert(
3809                "stable".into(),
3810                crate::model_registry::ProviderEntry {
3811                    kind: "openai".into(),
3812                    base_url: Some("https://api.openai.com/v1".into()),
3813                    ..Default::default()
3814                },
3815            );
3816            crate::model_registry::set_provider_config(config);
3817            let before_revision = crate::model_registry::model_catalog_revision();
3818            provider.set_models(vec![model("gpt-4o")]);
3819
3820            assert!(matches!(
3821                lifecycle
3822                    .refresh_models("lifecycle-refresh-registry-key")
3823                    .await,
3824                Err(ProviderLifecycleError::Catalog(
3825                    CatalogError::RegistryKeyInUse { .. }
3826                ))
3827            ));
3828            assert!(
3829                lifecycle
3830                    .providers
3831                    .contains("lifecycle-refresh-registry-key")
3832            );
3833            assert!(crate::model_registry::model_entry("stable:initial").is_some());
3834            assert_eq!(
3835                crate::model_registry::model_catalog_revision(),
3836                before_revision
3837            );
3838            lifecycle
3839                .remove_provider("lifecycle-refresh-registry-key")
3840                .unwrap();
3841        });
3842    }
3843
3844    #[test]
3845    fn credential_rotation_during_discovery_does_not_invalidate_catalog_refresh() {
3846        let _registry_lock = crate::model_registry::MODEL_CONFIG_LOCK
3847            .lock()
3848            .unwrap_or_else(std::sync::PoisonError::into_inner);
3849        test_runtime().block_on(async {
3850            let (_dir, lifecycle, provider) = fixture("lifecycle-token-refresh");
3851            remove_provider_catalog("lifecycle-token-refresh");
3852            lifecycle
3853                .install_provider(provider_record("lifecycle-token-refresh"), provider.clone())
3854                .await
3855                .unwrap();
3856            provider.set_models(vec![model("replacement")]);
3857            let hub = lifecycle.hub.clone();
3858            provider.before_discovery_once(move || {
3859                assert!(
3860                    hub.update_auth_tokens(
3861                        "lifecycle-token-refresh",
3862                        crate::config_hub::AuthTokenUpdate {
3863                            access_token: "rotated-access".into(),
3864                            refresh_token: Some("rotated-refresh".into()),
3865                            expires_at: 123,
3866                            account: Some("rotated@example.com".into()),
3867                        },
3868                    )
3869                    .unwrap()
3870                );
3871            });
3872
3873            let delta = lifecycle
3874                .refresh_models("lifecycle-token-refresh")
3875                .await
3876                .unwrap();
3877
3878            assert_eq!(delta.total, 1);
3879            let stored = lifecycle.hub.load_auth().unwrap().providers.remove(0);
3880            assert_eq!(stored.access_token, "rotated-access");
3881            assert_eq!(stored.refresh_token.as_deref(), Some("rotated-refresh"));
3882            assert_eq!(stored.model_cache.unwrap().models[0].slug, "replacement");
3883            lifecycle
3884                .remove_provider("lifecycle-token-refresh")
3885                .unwrap();
3886        });
3887    }
3888}