From 466c7918a19f9e98a5b65e00c9c4b02d34c717a6 Mon Sep 17 00:00:00 2001 From: elky Date: Wed, 7 Oct 2026 00:31:53 +0800 Subject: [PATCH] feat: unify provider scheduling workspace --- .../ai_serving/planner/candidate_source.rs | 78 ++ .../src/ai_serving/planner/decision_input.rs | 2 +- apps/aether-gateway/src/data/state/catalog.rs | 24 + .../handlers/admin/provider/crud/writes.rs | 31 +- .../admin/provider/shared/payloads.rs | 2 + .../admin/request/provider/catalog.rs | 18 + .../src/handlers/admin/routing/mod.rs | 4 + apps/aether-gateway/src/handlers/proxy/mod.rs | 8 + apps/aether-gateway/src/routing/resolver.rs | 93 +- apps/aether-gateway/src/scheduler/config.rs | 39 +- apps/aether-gateway/src/state/catalog.rs | 38 + .../src/state/routing_profiles.rs | 12 +- .../src/tests/control/admin/providers.rs | 84 ++ .../adapters/postgres/src/provider_catalog.rs | 58 + .../src/repository/provider_catalog/types.rs | 13 + .../src/repository/routing_profiles/types.rs | 19 +- .../runtime/src/lifecycle/migrate/tests.rs | 1 + .../migrate/tests/scoped_provider_creation.rs | 131 ++ .../src/repository/provider_catalog/memory.rs | 149 ++- .../src/repository/routing_profiles/memory.rs | 52 + crates/aether-routing-core/src/model.rs | 4 + crates/aether-routing-core/src/policy.rs | 185 ++- crates/aether-routing-core/src/ranking.rs | 52 +- frontend/src/api/endpoints/providers.ts | 1 + frontend/src/api/routing-profiles.ts | 1 + frontend/src/components/common/HelpHint.vue | 28 +- .../components/ProviderBatchActionDialog.vue | 6 +- .../providers/components/ProviderCard.vue | 36 +- .../components/ProviderDetailDrawer.vue | 9 + .../components/ProviderDragHandle.vue | 4 +- .../components/ProviderEmptyState.vue | 10 +- .../components/ProviderFormDialog.vue | 12 +- .../components/ProviderGroupControls.vue | 39 + .../components/ProviderGroupToggleButton.vue | 31 + .../components/ProviderMobileCard.vue | 8 +- .../components/ProviderPriorityInput.vue | 78 ++ .../components/ProviderSchedulingStatus.vue | 30 + .../components/ProviderSchedulingView.vue | 808 ++++++++++++ .../components/ProviderTableHeader.vue | 34 +- .../providers/components/ProviderTableRow.vue | 29 +- .../ProviderFormDialog.routing-group.spec.ts | 78 ++ .../__tests__/ProviderGroupControls.spec.ts | 105 ++ .../__tests__/ProviderPriorityInput.spec.ts | 123 ++ .../ProviderSchedulingView.navigation.spec.ts | 354 +++++ .../useSchedulingProviderBalance.spec.ts | 129 ++ ...ayOrder.ts => useProviderPriorityOrder.ts} | 45 +- .../useSchedulingProviderBalance.ts | 74 ++ .../utils/__tests__/groupPriority.spec.ts | 38 + .../features/providers/utils/groupPriority.ts | 42 + .../RoutingPriorityPolicyEditor.spec.ts | 202 +++ ...ngSchedulingPolicyEditor.providers.spec.ts | 16 +- .../RoutingSchedulingPolicyEditor.spec.ts | 317 ++++- .../__tests__/schedulingPolicies.spec.ts | 45 +- .../RoutingFailoverPolicyEditor.vue | 43 +- .../components/RoutingModelSelector.vue | 60 +- .../RoutingPriorityPolicyEditor.vue | 739 +++-------- .../RoutingSchedulingPolicyEditor.vue | 456 ++++++- .../features/routing/utils/routingPolicy.ts | 5 + .../routing/utils/schedulingPolicies.ts | 31 +- frontend/src/i18n/messages.ts | 8 + .../main-layout/__tests__/navigation.spec.ts | 61 +- .../src/layouts/main-layout/navigation.ts | 13 - .../routes/__tests__/admin.scheduling.spec.ts | 51 + frontend/src/router/routes/admin.ts | 18 +- .../src/views/admin/ProviderManagement.vue | 822 +++++++----- frontend/src/views/admin/RoutingProfiles.vue | 1166 ----------------- .../ProviderManagement.card-view.spec.ts | 561 +++++++- .../ProviderSchedulingView.failover.spec.ts | 366 ++++++ .../RoutingProfiles.failover.spec.ts | 247 ---- 69 files changed, 5810 insertions(+), 2666 deletions(-) create mode 100644 crates/aether-data/runtime/src/lifecycle/migrate/tests/scoped_provider_creation.rs create mode 100644 frontend/src/features/providers/components/ProviderGroupControls.vue create mode 100644 frontend/src/features/providers/components/ProviderGroupToggleButton.vue create mode 100644 frontend/src/features/providers/components/ProviderPriorityInput.vue create mode 100644 frontend/src/features/providers/components/ProviderSchedulingStatus.vue create mode 100644 frontend/src/features/providers/components/ProviderSchedulingView.vue create mode 100644 frontend/src/features/providers/components/__tests__/ProviderFormDialog.routing-group.spec.ts create mode 100644 frontend/src/features/providers/components/__tests__/ProviderGroupControls.spec.ts create mode 100644 frontend/src/features/providers/components/__tests__/ProviderPriorityInput.spec.ts create mode 100644 frontend/src/features/providers/components/__tests__/ProviderSchedulingView.navigation.spec.ts create mode 100644 frontend/src/features/providers/composables/__tests__/useSchedulingProviderBalance.spec.ts rename frontend/src/features/providers/composables/{useProviderDisplayOrder.ts => useProviderPriorityOrder.ts} (79%) create mode 100644 frontend/src/features/providers/composables/useSchedulingProviderBalance.ts create mode 100644 frontend/src/features/providers/utils/__tests__/groupPriority.spec.ts create mode 100644 frontend/src/features/providers/utils/groupPriority.ts create mode 100644 frontend/src/features/routing/__tests__/RoutingPriorityPolicyEditor.spec.ts create mode 100644 frontend/src/router/routes/__tests__/admin.scheduling.spec.ts delete mode 100644 frontend/src/views/admin/RoutingProfiles.vue create mode 100644 frontend/src/views/admin/__tests__/ProviderSchedulingView.failover.spec.ts delete mode 100644 frontend/src/views/admin/__tests__/RoutingProfiles.failover.spec.ts diff --git a/apps/aether-gateway/src/ai_serving/planner/candidate_source.rs b/apps/aether-gateway/src/ai_serving/planner/candidate_source.rs index aeb5792bf..50479935a 100644 --- a/apps/aether-gateway/src/ai_serving/planner/candidate_source.rs +++ b/apps/aether-gateway/src/ai_serving/planner/candidate_source.rs @@ -1892,6 +1892,84 @@ mod tests { .is_none()); } + #[tokio::test] + async fn routing_policy_excludes_group_disabled_providers_from_candidate_pages() { + let repository: Arc = + Arc::new(InMemoryMinimalCandidateSelectionReadRepository::seed([ + standard_candidate_row("provider-disabled", "openai:chat", 0), + standard_candidate_row("provider-enabled", "openai:chat", 1), + ])); + let app = AppState::new() + .expect("gateway state should build") + .with_data_state_for_tests( + GatewayDataState::with_minimal_candidate_selection_reader_for_tests(repository), + ); + let auth_snapshot = unrestricted_auth_snapshot(); + let model_directive_policy = + crate::system_features::ModelDirectivePolicySnapshot::load(&app).await; + let config = serde_json::from_value(serde_json::json!({ + "disabled_providers": ["provider-disabled"], + "model_policies": [{ + "model": "*", + "allowed_providers": ["provider-disabled", "provider-enabled"] + }] + })) + .expect("routing config should parse"); + let routing_policy = aether_routing_core::resolve_routing_policy( + &config, + aether_routing_core::RoutingPolicyInput { + group_id: Some("routing-group-1"), + group_version: Some(1), + selection_source: "test", + requested_model: "gpt-5", + resolved_model: "gpt-5", + api_format: "openai:chat", + user_id: None, + api_key_id: None, + headers: &serde_json::json!({}), + body: &serde_json::json!({}), + phase: aether_routing_core::RoutingRulePhase::ClientRequest, + }, + ) + .expect("routing policy should resolve"); + let mut cursor = LocalCandidatePreselectionPageCursor::new( + PlannerAppState::new(&app), + &model_directive_policy, + "openai:chat", + "gpt-5", + None, + false, + None, + &auth_snapshot, + Some(&routing_policy), + None, + None, + true, + LocalCandidatePreselectionKeyMode::ProviderEndpointKeyModelAndApiFormat, + false, + None, + ) + .await; + + let page = cursor + .next_page() + .await + .expect("routing candidate scan should succeed") + .expect("the enabled provider should remain"); + assert_eq!( + page.candidates + .iter() + .map(|candidate| candidate.provider_id.as_str()) + .collect::>(), + vec!["provider-enabled"] + ); + assert!(cursor + .next_page() + .await + .expect("routing scan should finish") + .is_none()); + } + #[tokio::test] async fn routing_policy_collects_candidate_pages_before_final_ranking() { let rows = (0..300) diff --git a/apps/aether-gateway/src/ai_serving/planner/decision_input.rs b/apps/aether-gateway/src/ai_serving/planner/decision_input.rs index 2bcf078ba..e29a98f08 100644 --- a/apps/aether-gateway/src/ai_serving/planner/decision_input.rs +++ b/apps/aether-gateway/src/ai_serving/planner/decision_input.rs @@ -1735,7 +1735,7 @@ mod tests { assert_eq!(policy.group_version, Some(4)); assert_eq!( policy.priority_mode, - aether_routing_core::RoutingSetPriorityMode::GlobalKey + aether_routing_core::RoutingSetPriorityMode::Provider ); assert_eq!( policy.scheduling_mode, diff --git a/apps/aether-gateway/src/data/state/catalog.rs b/apps/aether-gateway/src/data/state/catalog.rs index ddedda015..bf91aa938 100644 --- a/apps/aether-gateway/src/data/state/catalog.rs +++ b/apps/aether-gateway/src/data/state/catalog.rs @@ -562,6 +562,30 @@ impl GatewayDataState { Ok(created) } + pub(crate) async fn create_provider_catalog_provider_in_routing_group( + &self, + provider: &StoredProviderCatalogProvider, + shift_existing_priorities_from: Option, + routing_group_id: &str, + ) -> Result, DataLayerError> { + let created = match &self.provider_catalog_writer { + Some(repository) => repository + .create_provider_in_routing_group( + provider, + shift_existing_priorities_from, + routing_group_id, + ) + .await + .map(Some), + None => Ok(None), + }?; + if created.is_some() { + self.clear_provider_catalog_cache(); + self.clear_routing_group_cache(); + } + Ok(created) + } + pub(crate) async fn update_provider_catalog_provider( &self, provider: &StoredProviderCatalogProvider, diff --git a/apps/aether-gateway/src/handlers/admin/provider/crud/writes.rs b/apps/aether-gateway/src/handlers/admin/provider/crud/writes.rs index 24ec9591a..9ac782fe1 100644 --- a/apps/aether-gateway/src/handlers/admin/provider/crud/writes.rs +++ b/apps/aether-gateway/src/handlers/admin/provider/crud/writes.rs @@ -62,6 +62,16 @@ pub(crate) async fn maybe_build_local_admin_provider_writes_response( ))); } }; + let routing_group_id = payload + .routing_group_id + .as_deref() + .map(str::trim) + .map(str::to_string); + if routing_group_id.as_deref() == Some("") { + return Ok(Some(build_admin_provider_bad_request_response( + "routing_group_id 不能为空", + ))); + } let (record, shift_existing_priorities_from) = match state.build_admin_create_provider_record(payload).await { Ok(record) => record, @@ -69,10 +79,23 @@ pub(crate) async fn maybe_build_local_admin_provider_writes_response( return Ok(Some(build_admin_provider_bad_request_response(message))); } }; - let Some(created_provider) = state - .create_provider_catalog_provider(&record, shift_existing_priorities_from) - .await? - else { + let created = match routing_group_id.as_deref() { + Some(group_id) => { + state + .create_provider_catalog_provider_in_routing_group( + &record, + shift_existing_priorities_from, + group_id, + ) + .await? + } + None => { + state + .create_provider_catalog_provider(&record, shift_existing_priorities_from) + .await? + } + }; + let Some(created_provider) = created else { return Ok(Some(build_admin_providers_data_unavailable_response())); }; diff --git a/apps/aether-gateway/src/handlers/admin/provider/shared/payloads.rs b/apps/aether-gateway/src/handlers/admin/provider/shared/payloads.rs index 0b74b8778..7bc637609 100644 --- a/apps/aether-gateway/src/handlers/admin/provider/shared/payloads.rs +++ b/apps/aether-gateway/src/handlers/admin/provider/shared/payloads.rs @@ -179,6 +179,8 @@ pub(crate) struct AdminCodexResetCreditConsumeRequest { pub(crate) struct AdminProviderCreateRequest { pub(crate) name: String, #[serde(default)] + pub(crate) routing_group_id: Option, + #[serde(default)] pub(crate) provider_type: Option, #[serde(default)] pub(crate) description: Option, diff --git a/apps/aether-gateway/src/handlers/admin/request/provider/catalog.rs b/apps/aether-gateway/src/handlers/admin/request/provider/catalog.rs index 0b73678a5..2bf8da95f 100644 --- a/apps/aether-gateway/src/handlers/admin/request/provider/catalog.rs +++ b/apps/aether-gateway/src/handlers/admin/request/provider/catalog.rs @@ -420,6 +420,24 @@ impl<'a> AdminAppState<'a> { .await } + pub(crate) async fn create_provider_catalog_provider_in_routing_group( + &self, + provider: &aether_data_contracts::repository::provider_catalog::StoredProviderCatalogProvider, + shift_existing_priorities_from: Option, + routing_group_id: &str, + ) -> Result< + Option, + GatewayError, + > { + self.app + .create_provider_catalog_provider_in_routing_group( + provider, + shift_existing_priorities_from, + routing_group_id, + ) + .await + } + pub(crate) async fn update_provider_catalog_provider( &self, provider: &aether_data_contracts::repository::provider_catalog::StoredProviderCatalogProvider, diff --git a/apps/aether-gateway/src/handlers/admin/routing/mod.rs b/apps/aether-gateway/src/handlers/admin/routing/mod.rs index 49d5e904a..44a9e53e7 100644 --- a/apps/aether-gateway/src/handlers/admin/routing/mod.rs +++ b/apps/aether-gateway/src/handlers/admin/routing/mod.rs @@ -350,6 +350,7 @@ async fn publish_routing_group( .update_routing_group( group_id, UpdateRoutingGroupRecord { + expected_version: Some(group.version), version: Some(next_version), updated_at: now, published_at: Some(Some(now)), @@ -455,6 +456,9 @@ fn build_routing_group_update_patch( updated_at: current_unix_secs() as i64, ..UpdateRoutingGroupRecord::default() }; + if let Some(value) = object.get("expected_version") { + patch.expected_version = Some(required_i64(value, "expected_version")?); + } if let Some(value) = object.get("name") { patch.name = Some(required_string(value, "name")?); } diff --git a/apps/aether-gateway/src/handlers/proxy/mod.rs b/apps/aether-gateway/src/handlers/proxy/mod.rs index f678e6d15..a30a65512 100644 --- a/apps/aether-gateway/src/handlers/proxy/mod.rs +++ b/apps/aether-gateway/src/handlers/proxy/mod.rs @@ -2909,6 +2909,10 @@ mod tests { allowed_keys: vec!["key-other".to_string()], ..matching.clone() }; + let disabled_provider = aether_routing_core::RankingOverlay { + disabled_providers: vec!["provider-allowed".to_string()], + ..matching.clone() + }; assert!(routing_overlay_allows_affinity_target(None, &target)); assert!(routing_overlay_allows_affinity_target( @@ -2923,6 +2927,10 @@ mod tests { Some(&wrong_key), &target )); + assert!(!routing_overlay_allows_affinity_target( + Some(&disabled_provider), + &target + )); } #[test] diff --git a/apps/aether-gateway/src/routing/resolver.rs b/apps/aether-gateway/src/routing/resolver.rs index 49e28bbf9..547df881c 100644 --- a/apps/aether-gateway/src/routing/resolver.rs +++ b/apps/aether-gateway/src/routing/resolver.rs @@ -92,7 +92,8 @@ pub(crate) fn resolve_gateway_static_default_routing_policy( selection_source: input.selection_source.to_string(), requested_model: input.requested_model.to_string(), resolved_model: input.resolved_model.to_string(), - priority_mode: default_policy.priority_mode, + // Match the full resolver even when a stored default still says global_key. + priority_mode: RoutingSetPriorityMode::Provider, scheduling_mode: default_policy.scheduling_mode, keep_priority_on_conversion: default_policy.keep_priority_on_conversion, sticky_key_attempts: default_policy.sticky_key_attempts, @@ -110,10 +111,11 @@ fn static_default_policy_fields( let Some(object) = config_json.as_object() else { return Ok(None); }; - // A strategy's default policy applies to every model. Only model policies - // and rules require the request-context-aware resolver; unknown legacy - // fields (including the removed group allowlist) are intentionally ignored. - if !routing_array_field_is_missing_or_empty(object, "model_policies") + // A strategy's default policy applies to every model. Provider exclusions, + // model policies and rules require the full resolver to build the overlay; + // unknown legacy fields (including the removed group allowlist) are ignored. + if !routing_array_field_is_missing_or_empty(object, "disabled_providers") + || !routing_array_field_is_missing_or_empty(object, "model_policies") || !routing_array_field_is_missing_or_empty(object, "rules") { return Ok(None); @@ -275,7 +277,7 @@ mod tests { ); assert_eq!( static_policy.priority_mode, - RoutingSetPriorityMode::GlobalKey + RoutingSetPriorityMode::Provider ); assert_eq!( static_policy.scheduling_mode, @@ -313,6 +315,85 @@ mod tests { assert!(policy.is_none()); } + #[test] + fn provider_exclusions_without_model_policies_or_rules_use_full_resolver() { + for model in ["model-a", "future-model"] { + let config = json!({ + "disabled_providers": ["provider-disabled"], + "model_policies": [], + "rules": [] + }); + let static_policy = resolve_gateway_static_default_routing_policy( + GatewayStaticRoutingPolicyInput { + group_id: Some("group-1"), + group_version: Some(1), + group_config_json: &config, + selection_source: "system_default", + requested_model: model, + resolved_model: model, + }, + ) + .expect("provider exclusions should defer to the full resolver"); + assert!(static_policy.is_none()); + + let policy = resolve_gateway_routing_policy(GatewayRoutingPolicyInput { + group_id: Some("group-1"), + group_version: Some(1), + group_config_json: &config, + selection_source: "system_default", + requested_model: model, + resolved_model: model, + api_format: "openai:chat", + user_id: None, + api_key_id: None, + headers: &json!({}), + body: &json!({}), + phase: RoutingRulePhase::ClientRequest, + }) + .expect("group exclusions should resolve"); + assert!(!policy.ranking_overlay.provider_allowed("provider-disabled")); + assert!(policy.ranking_overlay.provider_allowed("provider-enabled")); + } + } + + #[test] + fn empty_provider_exclusions_preserve_static_default_fast_path() { + for config in [json!({}), json!({"disabled_providers": []})] { + assert!(static_default_policy_fields(&config) + .expect("empty provider exclusions should be valid") + .is_some()); + } + } + + #[test] + fn malformed_provider_exclusions_cannot_bypass_full_config_validation() { + for disabled in [json!("provider-disabled"), json!([42]), Value::Null] { + let config = json!({"disabled_providers": disabled}); + let error = resolve_gateway_routing_policy(GatewayRoutingPolicyInput { + group_id: Some("group-1"), + group_version: Some(1), + group_config_json: &config, + selection_source: "system_default", + requested_model: "model-a", + resolved_model: "model-a", + api_format: "openai:chat", + user_id: None, + api_key_id: None, + headers: &json!({}), + body: &json!({}), + phase: RoutingRulePhase::ClientRequest, + }) + .expect_err("malformed provider exclusions must not be silently ignored"); + assert!(matches!( + error, + GatewayError::Client { + status: StatusCode::BAD_REQUEST, + .. + } + )); + } + } + #[test] fn routing_config_errors_do_not_echo_config_values() { let secret = "https://internal.example/?token=Bearer-secret"; diff --git a/apps/aether-gateway/src/scheduler/config.rs b/apps/aether-gateway/src/scheduler/config.rs index 29ff4f6de..1f1caf5a8 100644 --- a/apps/aether-gateway/src/scheduler/config.rs +++ b/apps/aether-gateway/src/scheduler/config.rs @@ -102,8 +102,12 @@ impl SchedulerOrderingConfig { fn scheduler_priority_mode_from_routing(mode: RoutingSetPriorityMode) -> SchedulerPriorityMode { match mode { - RoutingSetPriorityMode::Provider => SchedulerPriorityMode::Provider, - RoutingSetPriorityMode::GlobalKey => SchedulerPriorityMode::GlobalKey, + // Defaults and previously resolved snapshots may still carry the + // removed global_key mode. Keep that compatibility at this boundary, + // without changing the scheduler's independent GlobalKey capability. + RoutingSetPriorityMode::Provider | RoutingSetPriorityMode::GlobalKey => { + SchedulerPriorityMode::Provider + } } } @@ -161,6 +165,33 @@ mod tests { use super::*; use crate::data::GatewayDataState; + #[test] + fn legacy_resolved_key_snapshot_uses_provider_ordering() { + let snapshot: ResolvedRoutingPolicy = serde_json::from_value(json!({ + "group_id": "legacy-group", + "selection_source": "system_default", + "requested_model": "model-a", + "resolved_model": "model-a", + "priority_mode": "global_key", + "scheduling_mode": "fixed_order", + "keep_priority_on_conversion": true, + "sticky_key_attempts": 4, + "ranking_overlay": { "key_priority_overrides": { "key-a": 2 } }, + "mutation_plan": { "body_patch": [], "header_patch": [] } + })) + .expect("legacy snapshots must remain readable"); + + let ordering = SchedulerOrderingConfig::from_routing_policy(&snapshot); + assert_eq!(ordering.priority_mode, SchedulerPriorityMode::Provider); + assert_eq!( + ordering.scheduling_mode, + SchedulerSchedulingMode::FixedOrder + ); + assert!(ordering.keep_priority_on_conversion); + assert_eq!(ordering.sticky_key_attempts, 4); + assert_eq!(snapshot.ranking_overlay.key_priority_overrides["key-a"], 2); + } + async fn create_system_default( repository: &InMemoryRoutingGroupRepository, enabled: bool, @@ -185,14 +216,14 @@ mod tests { } #[tokio::test] - async fn system_default_routing_group_exposes_strategy_ordering() { + async fn legacy_system_default_routing_group_uses_provider_ordering() { let repository = Arc::new(InMemoryRoutingGroupRepository::default()); create_system_default( &repository, true, json!({ "default_policy": { - "priority_mode": "provider", + "priority_mode": "global_key", "scheduling_mode": "fixed_order", "keep_priority_on_conversion": false } diff --git a/apps/aether-gateway/src/state/catalog.rs b/apps/aether-gateway/src/state/catalog.rs index baa8fb6ec..bca28244c 100644 --- a/apps/aether-gateway/src/state/catalog.rs +++ b/apps/aether-gateway/src/state/catalog.rs @@ -639,6 +639,44 @@ impl AppState { } } + pub(crate) async fn create_provider_catalog_provider_in_routing_group( + &self, + provider: &provider_catalog::StoredProviderCatalogProvider, + shift_existing_priorities_from: Option, + routing_group_id: &str, + ) -> Result, GatewayError> { + let protected = self.protect_provider_catalog_provider(provider)?; + let created = self + .data + .create_provider_catalog_provider_in_routing_group( + &protected, + shift_existing_priorities_from, + routing_group_id, + ) + .await + .map_err(|err| match err { + aether_data_contracts::DataLayerError::InvalidInput(ref message) + if message == "routing_group_not_found" => + { + GatewayError::Client { + status: axum::http::StatusCode::NOT_FOUND, + message: "策略分组不存在，请刷新后重试".to_string(), + } + } + _ => GatewayError::Internal(err.to_string()), + })?; + if created.is_some() { + self.invalidate_provider_routing_caches(); + } + match created { + Some(provider) => self + .open_provider_catalog_provider(provider) + .await + .map(Some), + None => Ok(None), + } + } + pub(crate) async fn update_provider_catalog_provider( &self, provider: &provider_catalog::StoredProviderCatalogProvider, diff --git a/apps/aether-gateway/src/state/routing_profiles.rs b/apps/aether-gateway/src/state/routing_profiles.rs index 8bbb4ff9b..d6263fb11 100644 --- a/apps/aether-gateway/src/state/routing_profiles.rs +++ b/apps/aether-gateway/src/state/routing_profiles.rs @@ -160,7 +160,17 @@ impl AppState { .data .update_routing_group(id, patch) .await - .map_err(|err| GatewayError::Internal(err.to_string()))?; + .map_err(|err| match err { + aether_data_contracts::DataLayerError::InvalidInput(ref message) + if message == "routing_group_version_conflict" => + { + GatewayError::Client { + status: axum::http::StatusCode::CONFLICT, + message: "策略分组已被修改，请刷新后重试".to_string(), + } + } + _ => GatewayError::Internal(err.to_string()), + })?; if updated.is_some() { self.invalidate_provider_routing_caches(); } diff --git a/apps/aether-gateway/src/tests/control/admin/providers.rs b/apps/aether-gateway/src/tests/control/admin/providers.rs index eb9467ed5..4d52d0678 100644 --- a/apps/aether-gateway/src/tests/control/admin/providers.rs +++ b/apps/aether-gateway/src/tests/control/admin/providers.rs @@ -42,6 +42,90 @@ const ADMIN_PROVIDERS_DATA_UNAVAILABLE_DETAIL: &str = "Admin provider catalog da mod health; +#[tokio::test] +async fn gateway_creates_provider_in_selected_routing_group_and_rejects_stale_saves() { + use aether_data::repository::routing_profiles::InMemoryRoutingGroupRepository; + use aether_data_contracts::repository::routing_profiles::{ + CreateRoutingGroupRecord, RoutingGroupReadRepository, RoutingGroupWriteRepository, + }; + let groups = Arc::new(InMemoryRoutingGroupRepository::default()); + for id in ["selected", "other"] { + groups + .create_routing_group(CreateRoutingGroupRecord { + id: id.into(), + name: id.into(), + description: None, + enabled: true, + is_system_default: id == "selected", + sort_order: 0, + config_json: json!({}), + version: 1, + created_at: 1, + updated_at: 1, + published_at: None, + }) + .await + .unwrap(); + } + let providers = Arc::new( + InMemoryProviderCatalogReadRepository::default().with_routing_groups(groups.clone()), + ); + let gateway = build_router_with_state( + AppState::new().unwrap().with_data_state_for_tests( + GatewayDataState::with_provider_catalog_repository_for_tests(providers.clone()) + .with_routing_group_repository_for_tests(groups.clone()), + ), + ); + let (gateway_url, gateway_handle) = start_server(gateway).await; + let client = reqwest::Client::new(); + let request = |name: &str, group: &str| { + client + .post(format!("{gateway_url}/api/admin/providers/")) + .header(GATEWAY_HEADER, "rust-phase3b") + .header(TRUSTED_ADMIN_USER_ID_HEADER, "admin-user-123") + .header(TRUSTED_ADMIN_USER_ROLE_HEADER, "admin") + .header(TRUSTED_ADMIN_SESSION_ID_HEADER, "session-123") + .json(&json!({"name": name, "routing_group_id": group})) + }; + let missing = request("missing", "gone").send().await.unwrap(); + assert_eq!(missing.status(), StatusCode::NOT_FOUND); + assert!(providers.list_providers(false).await.unwrap().is_empty()); + + let response = request("scoped", "selected").send().await.unwrap(); + let status = response.status(); + let payload: serde_json::Value = response.json().await.unwrap(); + assert_eq!(status, StatusCode::OK, "{payload}"); + let provider_id = payload["id"].as_str().unwrap(); + for group in groups.list_routing_groups().await.unwrap() { + assert_eq!(group.version, if group.id == "selected" { 1 } else { 2 }); + assert_eq!( + group.config_json["disabled_providers"], + if group.id == "selected" { + json!(null) + } else { + json!([provider_id]) + } + ); + } + let response = client + .patch(format!("{gateway_url}/api/admin/routing/groups/other")) + .header(GATEWAY_HEADER, "rust-phase3b") + .header(TRUSTED_ADMIN_USER_ID_HEADER, "admin-user-123") + .header(TRUSTED_ADMIN_USER_ROLE_HEADER, "admin") + .header(TRUSTED_ADMIN_SESSION_ID_HEADER, "session-123") + .json(&json!({"expected_version": 1, "config_json": {"disabled_providers": []}})) + .send() + .await + .unwrap(); + assert_eq!( + response.status(), + StatusCode::CONFLICT, + "{}", + response.text().await.unwrap() + ); + gateway_handle.abort(); +} + async fn provider_health_summary( endpoints: &[StoredProviderCatalogEndpoint], keys: &[StoredProviderCatalogKey], diff --git a/crates/aether-data/adapters/postgres/src/provider_catalog.rs b/crates/aether-data/adapters/postgres/src/provider_catalog.rs index fa19b83ad..4a4d9aeef 100644 --- a/crates/aether-data/adapters/postgres/src/provider_catalog.rs +++ b/crates/aether-data/adapters/postgres/src/provider_catalog.rs @@ -1179,6 +1179,16 @@ WHERE id = $1 &self, provider: &StoredProviderCatalogProvider, shift_existing_priorities_from: Option, + ) -> Result { + self.create_provider_with_routing_group(provider, shift_existing_priorities_from, None) + .await + } + + async fn create_provider_with_routing_group( + &self, + provider: &StoredProviderCatalogProvider, + shift_existing_priorities_from: Option, + routing_group_id: Option<&str>, ) -> Result { if provider.id.trim().is_empty() { return Err(DataLayerError::InvalidInput( @@ -1208,6 +1218,40 @@ WHERE id = $1 let mut tx = self.pool.begin().await.map_postgres_err()?; + if let Some(group_id) = routing_group_id { + // Group edits use the same lock: validation, exclusions, and provider + // creation are committed together, including concurrent deletions. + sqlx::query("LOCK TABLE routing_groups IN SHARE ROW EXCLUSIVE MODE") + .execute(&mut *tx) + .await + .map_postgres_err()?; + let exists: bool = + sqlx::query_scalar("SELECT EXISTS(SELECT 1 FROM routing_groups WHERE id = $1)") + .bind(group_id) + .fetch_one(&mut *tx) + .await + .map_postgres_err()?; + if !exists { + return Err(DataLayerError::InvalidInput( + "routing_group_not_found".to_string(), + )); + } + sqlx::query(r#" +UPDATE routing_groups +SET config_json = jsonb_set(config_json::jsonb, '{disabled_providers}', + CASE WHEN id = $1 + THEN COALESCE(config_json::jsonb -> 'disabled_providers', '[]'::jsonb) - $2::text + ELSE (COALESCE(config_json::jsonb -> 'disabled_providers', '[]'::jsonb) - $2::text) || jsonb_build_array($2::text) + END), + version = version + 1, + updated_at = EXTRACT(EPOCH FROM NOW())::bigint +WHERE (id <> $1 AND NOT (COALESCE(config_json::jsonb -> 'disabled_providers', '[]'::jsonb) ? $2::text)) + OR (id = $1 AND (COALESCE(config_json::jsonb -> 'disabled_providers', '[]'::jsonb) ? $2::text)) +"#) + .bind(group_id).bind(&provider.id) + .execute(&mut *tx).await.map_postgres_err()?; + } + if let Some(target_priority) = shift_existing_priorities_from { sqlx::query( r#" @@ -3015,6 +3059,20 @@ impl ProviderCatalogReadRepository for SqlxProviderCatalogReadRepository { #[async_trait] impl ProviderCatalogWriteRepository for SqlxProviderCatalogReadRepository { + async fn create_provider_in_routing_group( + &self, + provider: &StoredProviderCatalogProvider, + shift_existing_priorities_from: Option, + routing_group_id: &str, + ) -> Result { + self.create_provider_with_routing_group( + provider, + shift_existing_priorities_from, + Some(routing_group_id), + ) + .await + } + async fn create_provider( &self, provider: &StoredProviderCatalogProvider, diff --git a/crates/aether-data/contracts/src/repository/provider_catalog/types.rs b/crates/aether-data/contracts/src/repository/provider_catalog/types.rs index 37f65824d..384d95ca2 100644 --- a/crates/aether-data/contracts/src/repository/provider_catalog/types.rs +++ b/crates/aether-data/contracts/src/repository/provider_catalog/types.rs @@ -1463,6 +1463,19 @@ pub trait ProviderCatalogReadRepository: Send + Sync { #[async_trait] pub trait ProviderCatalogWriteRepository: Send + Sync { + /// Create a provider and exclude it from every other existing routing group + /// in one transaction. Implementations must fail closed if unsupported. + async fn create_provider_in_routing_group( + &self, + _provider: &StoredProviderCatalogProvider, + _shift_existing_priorities_from: Option, + _routing_group_id: &str, + ) -> Result { + Err(crate::DataLayerError::InvalidConfiguration( + "atomic provider creation in a routing group is not supported".to_string(), + )) + } + async fn create_provider( &self, provider: &StoredProviderCatalogProvider, diff --git a/crates/aether-data/contracts/src/repository/routing_profiles/types.rs b/crates/aether-data/contracts/src/repository/routing_profiles/types.rs index d949d7836..ccb311f9d 100644 --- a/crates/aether-data/contracts/src/repository/routing_profiles/types.rs +++ b/crates/aether-data/contracts/src/repository/routing_profiles/types.rs @@ -58,6 +58,7 @@ pub struct CreateRoutingGroupRecord { #[derive(Debug, Clone, PartialEq, Eq, Default)] pub struct UpdateRoutingGroupRecord { + pub expected_version: Option, pub name: Option, pub description: Option>, pub enabled: Option, @@ -244,6 +245,16 @@ pub fn apply_group_patch( group: &mut StoredRoutingGroup, patch: UpdateRoutingGroupRecord, ) -> Result<(), crate::DataLayerError> { + if patch + .expected_version + .is_some_and(|version| version != group.version) + { + return Err(crate::DataLayerError::InvalidInput( + "routing_group_version_conflict".to_string(), + )); + } + let previous_version = group.version; + let config_changed = patch.config_json.is_some(); if let Some(name) = patch.name { if name.trim().is_empty() { return Err(crate::DataLayerError::InvalidInput( @@ -273,7 +284,13 @@ pub fn apply_group_patch( group.config_json = config_json; } if let Some(version) = patch.version { - group.version = version.max(1); + group.version = if config_changed { + version.max(previous_version.saturating_add(1)) + } else { + version.max(previous_version) + }; + } else if config_changed { + group.version = previous_version.saturating_add(1); } if let Some(published_at) = patch.published_at { group.published_at = published_at; diff --git a/crates/aether-data/runtime/src/lifecycle/migrate/tests.rs b/crates/aether-data/runtime/src/lifecycle/migrate/tests.rs index 67baa12cf..380ec6481 100644 --- a/crates/aether-data/runtime/src/lifecycle/migrate/tests.rs +++ b/crates/aether-data/runtime/src/lifecycle/migrate/tests.rs @@ -35,6 +35,7 @@ mod overview_fact_metadata; mod overview_migration_safety; mod policy_nulls; mod provider_expenses; +mod scoped_provider_creation; /// A clean PostgreSQL database is bootstrapped from the schema snapshot first; /// migrations after the privacy/security frontier are intentionally left diff --git a/crates/aether-data/runtime/src/lifecycle/migrate/tests/scoped_provider_creation.rs b/crates/aether-data/runtime/src/lifecycle/migrate/tests/scoped_provider_creation.rs new file mode 100644 index 000000000..7050998a7 --- /dev/null +++ b/crates/aether-data/runtime/src/lifecycle/migrate/tests/scoped_provider_creation.rs @@ -0,0 +1,131 @@ +use super::*; +use aether_data_contracts::repository::{ + provider_catalog::{ProviderCatalogWriteRepository, StoredProviderCatalogProvider}, + routing_profiles::{ + CreateRoutingGroupRecord, RoutingGroupReadRepository, RoutingGroupWriteRepository, + UpdateRoutingGroupRecord, + }, +}; +use serde_json::json; + +#[tokio::test] +async fn postgres_scoped_provider_creation_rolls_back_and_serializes_group_saves() { + let Some(server) = ManagedPostgresServer::try_start() + .await + .expect("local postgres should start or skip") + else { + return; + }; + let pool = PgPool::connect(server.database_url()).await.unwrap(); + prepare_and_apply_clean_postgres_database(&pool).await; + let groups = + crate::repository::routing_profiles::PostgresRoutingGroupRepository::new(pool.clone()); + let providers = + crate::repository::provider_catalog::SqlxProviderCatalogReadRepository::new(pool.clone()); + for id in ["selected", "other"] { + groups + .create_routing_group(CreateRoutingGroupRecord { + id: id.into(), + name: id.into(), + description: None, + enabled: true, + is_system_default: id == "selected", + sort_order: 0, + config_json: json!({"disabled_providers": ["existing-disabled"]}), + version: 1, + created_at: 1, + updated_at: 1, + published_at: None, + }) + .await + .unwrap(); + } + let provider = + StoredProviderCatalogProvider::new("new".into(), "new".into(), None, "custom".into()) + .unwrap(); + assert!(providers + .create_provider_in_routing_group(&provider, None, "missing") + .await + .is_err()); + assert_eq!( + sqlx::query_scalar::<_, i64>("SELECT COUNT(*) FROM providers") + .fetch_one(&pool) + .await + .unwrap(), + 0 + ); + assert!(groups + .list_routing_groups() + .await + .unwrap() + .iter() + .all(|group| group.version == 1)); + + providers + .create_provider_in_routing_group(&provider, None, "selected") + .await + .unwrap(); + let before = groups.list_routing_groups().await.unwrap(); + for group in &before { + assert_eq!(group.version, if group.id == "selected" { 1 } else { 2 }); + assert_eq!( + group.config_json["disabled_providers"], + if group.id == "selected" { + json!(["existing-disabled"]) + } else { + json!(["existing-disabled", "new"]) + } + ); + } + // The INSERT fails after group updates execute. Its transaction must undo + // every exclusion and version change along with any priority shifts. + assert!(providers + .create_provider_in_routing_group(&provider, Some(0), "other") + .await + .is_err()); + assert_eq!(groups.list_routing_groups().await.unwrap(), before); + assert!(groups + .update_routing_group( + "other", + UpdateRoutingGroupRecord { + expected_version: Some(1), + config_json: Some(json!({"disabled_providers": []})), + ..Default::default() + } + ) + .await + .is_err()); + assert_eq!(groups.list_routing_groups().await.unwrap(), before); + + let concurrent_provider = StoredProviderCatalogProvider::new( + "concurrent".into(), + "concurrent".into(), + None, + "custom".into(), + ) + .unwrap(); + let (created, edited) = tokio::join!( + providers.create_provider_in_routing_group(&concurrent_provider, None, "selected"), + groups.update_routing_group( + "other", + UpdateRoutingGroupRecord { + expected_version: Some(2), + config_json: Some(json!({"disabled_providers": ["new"]})), + ..Default::default() + } + ) + ); + created.unwrap(); + let other = groups + .list_routing_groups() + .await + .unwrap() + .into_iter() + .find(|group| group.id == "other") + .unwrap(); + assert!(other.config_json["disabled_providers"] + .as_array() + .unwrap() + .contains(&json!("concurrent"))); + assert_eq!(other.version, if edited.is_ok() { 4 } else { 3 }); +} diff --git a/crates/aether-data/runtime/src/repository/provider_catalog/memory.rs b/crates/aether-data/runtime/src/repository/provider_catalog/memory.rs index 988b7bf02..a127476e5 100644 --- a/crates/aether-data/runtime/src/repository/provider_catalog/memory.rs +++ b/crates/aether-data/runtime/src/repository/provider_catalog/memory.rs @@ -1,5 +1,5 @@ use std::collections::BTreeMap; -use std::sync::RwLock; +use std::sync::{Arc, RwLock}; use std::time::{SystemTime, UNIX_EPOCH}; use async_trait::async_trait; @@ -31,6 +31,8 @@ struct MemoryProviderCatalogIndex { #[derive(Debug, Default)] pub struct InMemoryProviderCatalogReadRepository { index: RwLock, + routing_groups: + Option>, } impl InMemoryProviderCatalogReadRepository { @@ -40,6 +42,7 @@ impl InMemoryProviderCatalogReadRepository { keys: Vec, ) -> Self { Self { + routing_groups: None, index: RwLock::new(MemoryProviderCatalogIndex { providers: providers .into_iter() @@ -54,6 +57,14 @@ impl InMemoryProviderCatalogReadRepository { } } + pub fn with_routing_groups( + mut self, + repository: Arc, + ) -> Self { + self.routing_groups = Some(repository); + self + } + fn snapshot(&self) -> ProviderCatalogSnapshot { let index = self.index.read().expect("provider catalog repository lock"); ProviderCatalogSnapshot::new( @@ -416,6 +427,46 @@ impl ProviderCatalogReadRepository for InMemoryProviderCatalogReadRepository { #[async_trait] impl ProviderCatalogWriteRepository for InMemoryProviderCatalogReadRepository { + async fn create_provider_in_routing_group( + &self, + provider: &StoredProviderCatalogProvider, + shift_existing_priorities_from: Option, + routing_group_id: &str, + ) -> Result { + let groups = self.routing_groups.as_ref().ok_or_else(|| { + DataLayerError::InvalidConfiguration( + "atomic provider creation requires a shared routing group repository".to_string(), + ) + })?; + groups.create_scoped_provider(routing_group_id, &provider.id, || { + let mut index = self + .index + .write() + .expect("provider catalog repository lock"); + if index.providers.contains_key(&provider.id) + || index + .providers + .values() + .any(|existing| existing.name == provider.name) + { + return Err(DataLayerError::InvalidInput( + "provider already exists".to_string(), + )); + } + if let Some(target_priority) = shift_existing_priorities_from { + for existing in index.providers.values_mut() { + if existing.provider_priority >= target_priority { + existing.provider_priority += 1; + } + } + } + index + .providers + .insert(provider.id.clone(), provider.clone()); + Ok(provider.clone()) + }) + } + async fn create_provider( &self, provider: &StoredProviderCatalogProvider, @@ -1587,6 +1638,102 @@ mod tests { .expect("key should build") } + #[tokio::test] + async fn scoped_provider_creation_is_atomic_and_rejects_stale_group_updates() { + use crate::repository::routing_profiles::InMemoryRoutingGroupRepository; + use aether_data_contracts::repository::routing_profiles::{ + CreateRoutingGroupRecord, RoutingGroupReadRepository, RoutingGroupWriteRepository, + UpdateRoutingGroupRecord, + }; + let groups = Arc::new(InMemoryRoutingGroupRepository::default()); + for id in ["selected", "other", "disabled"] { + groups + .create_routing_group(CreateRoutingGroupRecord { + id: id.into(), + name: id.into(), + description: None, + enabled: id != "disabled", + is_system_default: id == "selected", + sort_order: 0, + config_json: json!({"disabled_providers": ["already-disabled"], "rules": []}), + version: 1, + created_at: 1, + updated_at: 1, + published_at: None, + }) + .await + .unwrap(); + } + let repository = + InMemoryProviderCatalogReadRepository::default().with_routing_groups(groups.clone()); + let provider = sample_provider("new"); + assert!(repository + .create_provider_in_routing_group(&provider, Some(0), "missing") + .await + .is_err()); + assert!(repository.list_providers(false).await.unwrap().is_empty()); + assert!(groups + .list_routing_groups() + .await + .unwrap() + .iter() + .all(|group| group.version == 1)); + + repository + .create_provider_in_routing_group(&provider, None, "selected") + .await + .unwrap(); + for group in groups.list_routing_groups().await.unwrap() { + assert_eq!(group.version, if group.id == "selected" { 1 } else { 2 }); + assert_eq!( + group.config_json["disabled_providers"], + if group.id == "selected" { + json!(["already-disabled"]) + } else { + json!(["already-disabled", "new"]) + } + ); + } + let before = groups.list_routing_groups().await.unwrap(); + assert!(repository + .create_provider_in_routing_group(&provider, Some(0), "other") + .await + .is_err()); + assert_eq!(groups.list_routing_groups().await.unwrap(), before); + let stale = groups + .update_routing_group( + "other", + UpdateRoutingGroupRecord { + expected_version: Some(1), + config_json: Some(json!({"disabled_providers": []})), + ..Default::default() + }, + ) + .await; + assert!( + matches!(stale, Err(DataLayerError::InvalidInput(message)) if message == "routing_group_version_conflict") + ); + assert_eq!(groups.list_routing_groups().await.unwrap(), before); + let updated = groups + .update_routing_group( + "other", + UpdateRoutingGroupRecord { + expected_version: Some(2), + config_json: Some(json!({"disabled_providers": ["already-disabled"]})), + version: Some(2), + ..Default::default() + }, + ) + .await + .unwrap() + .unwrap(); + assert_eq!(updated.version, 3); + assert_eq!( + updated.config_json["disabled_providers"], + json!(["already-disabled"]) + ); + } + #[tokio::test] async fn reads_provider_catalog_items_by_id() { let repository = InMemoryProviderCatalogReadRepository::seed( diff --git a/crates/aether-data/runtime/src/repository/routing_profiles/memory.rs b/crates/aether-data/runtime/src/repository/routing_profiles/memory.rs index c0282763d..95a9e96b7 100644 --- a/crates/aether-data/runtime/src/repository/routing_profiles/memory.rs +++ b/crates/aether-data/runtime/src/repository/routing_profiles/memory.rs @@ -20,6 +20,58 @@ pub struct InMemoryRoutingGroupRepository { } impl InMemoryRoutingGroupRepository { + pub(crate) fn create_scoped_provider( + &self, + selected_group_id: &str, + provider_id: &str, + create: impl FnOnce() -> Result, + ) -> Result { + let mut groups = self.groups.write().expect("routing group repository lock"); + if !groups.contains_key(selected_group_id) { + return Err(DataLayerError::InvalidInput( + "routing_group_not_found".to_string(), + )); + } + let mut updated = groups.clone(); + for group in updated.values_mut() { + let had_provider = group + .config_json + .get("disabled_providers") + .and_then(serde_json::Value::as_array) + .is_some_and(|disabled| { + disabled + .iter() + .any(|value| value.as_str() == Some(provider_id)) + }); + if (group.id == selected_group_id && !had_provider) + || (group.id != selected_group_id && had_provider) + { + continue; + } + let object = group.config_json.as_object_mut().ok_or_else(|| { + DataLayerError::InvalidInput("routing group config must be an object".to_string()) + })?; + let disabled = object + .entry("disabled_providers") + .or_insert_with(|| serde_json::json!([])); + let disabled = disabled.as_array_mut().ok_or_else(|| { + DataLayerError::InvalidInput("disabled_providers must be an array".to_string()) + })?; + disabled.retain(|value| value.as_str() != Some(provider_id)); + if group.id != selected_group_id { + disabled.push(serde_json::json!(provider_id)); + } + group.version = group.version.saturating_add(1); + group.updated_at = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap_or_default() + .as_secs() as i64; + } + let created = create()?; + *groups = updated; + Ok(created) + } + pub fn seed(groups: I, bindings: B, versions: V) -> Self where I: IntoIterator, diff --git a/crates/aether-routing-core/src/model.rs b/crates/aether-routing-core/src/model.rs index fc48854e9..88e866765 100644 --- a/crates/aether-routing-core/src/model.rs +++ b/crates/aether-routing-core/src/model.rs @@ -224,6 +224,10 @@ pub struct RoutingRule { #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, Default)] pub struct RoutingGroupConfig { + /// Providers excluded from every model in this group, including providers + /// otherwise selected by model policies or routing rules. + #[serde(default)] + pub disabled_providers: Vec, /// The default policy is global for the selected strategy group. Model /// differences are expressed through `model_policies` and `rules`. #[serde(default)] diff --git a/crates/aether-routing-core/src/policy.rs b/crates/aether-routing-core/src/policy.rs index 1b2972da1..1cec1bff6 100644 --- a/crates/aether-routing-core/src/policy.rs +++ b/crates/aether-routing-core/src/policy.rs @@ -85,12 +85,17 @@ pub fn resolve_routing_policy( selection_source: input.selection_source.to_string(), requested_model: input.requested_model.to_string(), resolved_model: input.resolved_model.to_string(), - priority_mode: config.default_policy.priority_mode, + // Legacy global_key values remain readable, but routing groups now + // always rank providers before their keys. + priority_mode: RoutingSetPriorityMode::Provider, scheduling_mode: config.default_policy.scheduling_mode, keep_priority_on_conversion: config.default_policy.keep_priority_on_conversion, sticky_key_attempts: config.default_policy.sticky_key_attempts, execution_policy: config.default_policy.execution_policy.clone(), - ranking_overlay: RankingOverlay::default(), + ranking_overlay: RankingOverlay { + disabled_providers: config.disabled_providers.clone(), + ..RankingOverlay::default() + }, mutation_plan: MutationPlan::default(), pool_policy_overrides: BTreeMap::new(), matched_rules: Vec::new(), @@ -203,14 +208,13 @@ fn apply_action( policy.ranking_overlay.allowed_keys = key_ids.clone(); } RoutingAction::SetScheduling { - priority_mode, + // Keep accepting the legacy field without re-enabling key-first + // scheduling through a model rule. + priority_mode: _, scheduling_mode, keep_priority_on_conversion, sticky_key_attempts, } => { - if let Some(priority_mode) = priority_mode { - policy.priority_mode = *priority_mode; - } if let Some(scheduling_mode) = scheduling_mode { policy.scheduling_mode = *scheduling_mode; } @@ -316,6 +320,164 @@ mod tests { use super::*; + #[test] + fn legacy_key_scheduling_keeps_overrides_but_resolves_to_provider_ordering() { + let config: RoutingGroupConfig = serde_json::from_value(json!({ + "default_policy": { "priority_mode": "global_key" }, + "model_policies": [{ + "model": "*", + "provider_priority_overrides": { "provider-a": 7 }, + "key_priority_overrides": { "key-a": 2 }, + "key_priority_overrides_by_format": { "openai:chat": { "key-a": 3 } }, + "pool_priority_overrides": { "provider-pool": 4 } + }], + "rules": [{ + "id": "legacy-key-client", "phase": "client_request", + "actions": [{ "type": "set_scheduling", "priority_mode": "global_key", "scheduling_mode": "fixed_order" }] + }] + })) + .expect("legacy key scheduling must stay readable"); + let stored = serde_json::to_value(&config).unwrap(); + assert_eq!(stored["default_policy"]["priority_mode"], "global_key"); + assert_eq!( + stored["rules"][0]["actions"][0]["priority_mode"], + "global_key" + ); + assert_eq!( + serde_json::from_value::(stored).unwrap(), + config + ); + + for phase in [ + RoutingRulePhase::ClientRequest, + RoutingRulePhase::ProviderRequest, + ] { + let policy = resolve_routing_policy( + &config, + RoutingPolicyInput { + group_id: Some("legacy-group"), + group_version: Some(1), + selection_source: "explicit", + requested_model: "model-a", + resolved_model: "model-a", + api_format: "openai:chat", + user_id: None, + api_key_id: None, + headers: &json!({}), + body: &json!({}), + phase, + }, + ) + .unwrap(); + assert_eq!(policy.priority_mode, RoutingSetPriorityMode::Provider); + assert_eq!( + policy.scheduling_mode, + if phase == RoutingRulePhase::ClientRequest { + RoutingSchedulingMode::FixedOrder + } else { + RoutingSchedulingMode::CacheAffinity + } + ); + assert_eq!( + policy.matched_rules.len(), + usize::from(phase == RoutingRulePhase::ClientRequest) + ); + assert_eq!( + policy.ranking_overlay.provider_priority_overrides["provider-a"], + 7 + ); + assert_eq!(policy.ranking_overlay.key_priority_overrides["key-a"], 2); + assert_eq!( + policy.ranking_overlay.pool_priority_overrides["provider-pool"], + 4 + ); + assert_eq!( + policy + .ranking_overlay + .key_priority_for_format("key-a", "openai:chat", 99), + 3 + ); + } + assert_eq!( + config.default_policy.priority_mode, + RoutingSetPriorityMode::GlobalKey + ); + } + + #[test] + fn group_disabled_providers_apply_to_every_model_and_cannot_be_reenabled() { + let config: RoutingGroupConfig = serde_json::from_value(json!({ + "disabled_providers": ["provider-disabled"], + "model_policies": [{ + "model": "model-allowlist", + "allowed_providers": ["provider-disabled", "provider-enabled"] + }], + "rules": [{ + "id": "replace-provider-allowlist", + "conditions": { "field": "model", "op": "eq", "value": "rule-allowlist" }, + "actions": [{ + "type": "restrict_providers", + "provider_ids": ["provider-disabled", "provider-enabled"] + }, { + "type": "set_provider_priority", + "provider_id": "provider-disabled", + "priority": 0 + }] + }, { + "id": "clear-provider-allowlist", + "conditions": { "field": "model", "op": "eq", "value": "rule-unrestricted" }, + "actions": [{ "type": "restrict_providers", "provider_ids": [] }] + }] + })) + .expect("group provider exclusions should deserialize"); + + // The field survives the same round trip used when persisting or + // publishing strategy configuration. + let stored_config = serde_json::to_value(&config).unwrap(); + assert_eq!( + stored_config["disabled_providers"], + json!(["provider-disabled"]) + ); + let config: RoutingGroupConfig = serde_json::from_value(stored_config).unwrap(); + + for model in [ + "future-model", + "model-allowlist", + "rule-allowlist", + "rule-unrestricted", + ] { + let policy = resolve_routing_policy( + &config, + RoutingPolicyInput { + group_id: Some("group-1"), + group_version: Some(1), + selection_source: "explicit", + requested_model: model, + resolved_model: model, + api_format: "openai:chat", + user_id: None, + api_key_id: None, + headers: &json!({}), + body: &json!({}), + phase: RoutingRulePhase::ClientRequest, + }, + ) + .expect("policy with group provider exclusions should resolve"); + + assert!( + !policy.ranking_overlay.provider_allowed("provider-disabled"), + "{model} must retain the group exclusion" + ); + assert!(policy.ranking_overlay.provider_allowed("provider-enabled")); + let has_allowlist = matches!(model, "model-allowlist" | "rule-allowlist"); + assert_eq!( + policy.ranking_overlay.provider_allowed("provider-unlisted"), + !has_allowlist, + "{model} should preserve its normal allowlist behavior" + ); + } + } + #[test] fn all_model_scheduling_and_rankings_apply_to_future_models() { let config: RoutingGroupConfig = serde_json::from_value(json!({ @@ -330,6 +492,7 @@ mod tests { "rules": [] })) .expect("all-model scheduling config should deserialize"); + assert!(config.disabled_providers.is_empty()); for model in ["existing-model", "future-model"] { let policy = resolve_routing_policy( @@ -349,7 +512,7 @@ mod tests { }, ) .expect("all-model scheduling policy should resolve"); - assert_eq!(policy.priority_mode, RoutingSetPriorityMode::GlobalKey); + assert_eq!(policy.priority_mode, RoutingSetPriorityMode::Provider); assert_eq!(policy.scheduling_mode, RoutingSchedulingMode::LoadBalance); assert_eq!( policy @@ -419,7 +582,7 @@ mod tests { .is_empty()); assert!(policy.matched_rules.is_empty()); } else { - assert_eq!(policy.priority_mode, RoutingSetPriorityMode::GlobalKey); + assert_eq!(policy.priority_mode, RoutingSetPriorityMode::Provider); assert_eq!(policy.scheduling_mode, RoutingSchedulingMode::FixedOrder); assert_eq!( policy @@ -436,6 +599,7 @@ mod tests { #[test] fn resolves_model_policy_and_matching_rule() { let config = RoutingGroupConfig { + disabled_providers: vec![], default_policy: RoutingDefaultPolicy::default(), model_policies: vec![RoutingModelPolicy { model: "gpt-5".to_string(), @@ -504,6 +668,7 @@ mod tests { #[test] fn default_policy_applies_to_models_without_an_override() { let config = RoutingGroupConfig { + disabled_providers: vec![], default_policy: RoutingDefaultPolicy { priority_mode: RoutingSetPriorityMode::GlobalKey, scheduling_mode: RoutingSchedulingMode::LoadBalance, @@ -538,7 +703,7 @@ mod tests { ) .expect("the specially configured model should resolve"); - assert_eq!(special.priority_mode, RoutingSetPriorityMode::GlobalKey); + assert_eq!(special.priority_mode, RoutingSetPriorityMode::Provider); assert_eq!(special.scheduling_mode, RoutingSchedulingMode::LoadBalance); assert!(special.keep_priority_on_conversion); assert_eq!(special.sticky_key_attempts, 3); @@ -572,7 +737,7 @@ mod tests { ) .expect("an unconfigured model should keep using the default policy"); - assert_eq!(ordinary.priority_mode, RoutingSetPriorityMode::GlobalKey); + assert_eq!(ordinary.priority_mode, RoutingSetPriorityMode::Provider); assert_eq!(ordinary.scheduling_mode, RoutingSchedulingMode::LoadBalance); assert!(ordinary.keep_priority_on_conversion); assert_eq!(ordinary.sticky_key_attempts, 3); diff --git a/crates/aether-routing-core/src/ranking.rs b/crates/aether-routing-core/src/ranking.rs index c12f52e07..a86190dcc 100644 --- a/crates/aether-routing-core/src/ranking.rs +++ b/crates/aether-routing-core/src/ranking.rs @@ -13,6 +13,9 @@ pub enum CandidateKind { #[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)] pub struct RankingOverlay { + /// Group-wide exclusions take precedence over every provider allowlist. + #[serde(default)] + pub disabled_providers: Vec, #[serde(default)] pub allowed_providers: Vec, #[serde(default)] @@ -112,11 +115,15 @@ impl RankingOverlay { } pub fn provider_allowed(&self, provider_id: &str) -> bool { - self.allowed_providers.is_empty() - || self - .allowed_providers - .iter() - .any(|item| item == provider_id) + !self + .disabled_providers + .iter() + .any(|item| item == provider_id) + && (self.allowed_providers.is_empty() + || self + .allowed_providers + .iter() + .any(|item| item == provider_id)) } pub fn key_allowed(&self, key_id: &str) -> bool { @@ -180,6 +187,41 @@ mod tests { use super::*; + #[test] + fn disabled_providers_take_precedence_over_allowlists() { + let mut overlay = RankingOverlay { + disabled_providers: vec!["provider-disabled".to_string()], + ..RankingOverlay::default() + }; + assert!(!overlay.provider_allowed("provider-disabled")); + assert!(overlay.provider_allowed("provider-enabled")); + + overlay.allowed_providers = vec![ + "provider-disabled".to_string(), + "provider-enabled".to_string(), + ]; + assert!(!overlay.provider_allowed("provider-disabled")); + assert!(overlay.provider_allowed("provider-enabled")); + assert!(!overlay.provider_allowed("provider-unlisted")); + + // An allowlist containing only disabled providers must not become an + // empty allowlist, which would otherwise allow unrelated providers. + overlay.allowed_providers = vec!["provider-disabled".to_string()]; + assert!(!overlay.provider_allowed("provider-disabled")); + assert!(!overlay.provider_allowed("provider-enabled")); + } + + #[test] + fn legacy_overlay_without_disabled_providers_preserves_provider_selection() { + let overlay: RankingOverlay = serde_json::from_value(serde_json::json!({ + "allowed_providers": ["provider-enabled"] + })) + .expect("legacy overlays should remain readable"); + assert!(overlay.disabled_providers.is_empty()); + assert!(overlay.provider_allowed("provider-enabled")); + assert!(!overlay.provider_allowed("provider-unlisted")); + } + #[test] fn overlay_applies_provider_and_key_priority() { let overlay = RankingOverlay { diff --git a/frontend/src/api/endpoints/providers.ts b/frontend/src/api/endpoints/providers.ts index d73206ebf..83f2c559c 100644 --- a/frontend/src/api/endpoints/providers.ts +++ b/frontend/src/api/endpoints/providers.ts @@ -159,6 +159,7 @@ export async function updateProvider( */ export async function createProvider( data: { + routing_group_id?: string name: string provider_type?: ProviderType description?: string diff --git a/frontend/src/api/routing-profiles.ts b/frontend/src/api/routing-profiles.ts index a548c14da..5e024e00a 100644 --- a/frontend/src/api/routing-profiles.ts +++ b/frontend/src/api/routing-profiles.ts @@ -67,6 +67,7 @@ export interface RoutingGroupCreateRequest { } export interface RoutingGroupUpdateRequest { + expected_version?: number name?: string description?: string | null enabled?: boolean diff --git a/frontend/src/components/common/HelpHint.vue b/frontend/src/components/common/HelpHint.vue index 48c7050d1..1db2b36e4 100644 --- a/frontend/src/components/common/HelpHint.vue +++ b/frontend/src/components/common/HelpHint.vue @@ -1,17 +1,43 @@