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#[derive(Debug, Clone, Copy, PartialEq, Eq)]
61#[non_exhaustive]
62pub enum ProviderCatalogRefreshOutcome {
63 NotNeeded,
65 AlreadyInFlight,
67 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 pub fn reload_config_providers(&self) -> Result<(), ProviderLifecycleError> {
279 reload_config_providers_for_hub(&self.hub)?;
280 Ok(())
281 }
282
283 pub fn create_config_provider(
285 &self,
286 update: ProviderConfigUpdate<'_>,
287 ) -> Result<(), ProviderLifecycleError> {
288 self.commit_config_provider(update, ProviderConfigWriteMode::Create)
289 }
290
291 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 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 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, ¤t_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, ¤t_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(¤t.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(¤t.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 ®istry, 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 ®istry,
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(®istry_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(®istry_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}