i/Aether
1
0
Fork 0
mirror of https://github.com/fawney19/Aether.git synced 2026-09-03 09:50:21 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat: provider api_formats 可空继承、OpenAI 图片 edit/variation 与用量配额多项补强

Browse Source 
- 鉴权: provider_api_keys.api_formats 改为可空，OAuth 托管 key 自动继承 provider endpoints 激活格式，相关 handler/测试同步更新
- 图片 planner: OpenAI 图片路由新增 edit/variation 操作并完善参数校验、响应合并与流式处理
- 用量: user me usage 返回区分 client_requested_stream/upstream_is_stream，前端 usage 列表筛选与展示增强
- 统计: stats_daily_model 新增 cache_creation_ephemeral_5m/1h tokens 字段与回填链路
- 配额/observability: quota repository 新增内存与 SQL 扩展，admin observability usage 字段扩充
- 其它: OAuth 导入/轮询收敛、provider 汇总与 pool admin 读写链路小修、新增 system_config 缓存与 provider template handler

Closes #318

Co-authored-by: Entropy.Xu <53283266+Entropy-Xu@users.noreply.github.com>
This commit is contained in:
fawney19
2026-04-23 14:42:51 +08:00
parent f55f22d2e8
commit fa328e18a1
128 changed files with 5583 additions and 690 deletions
Download Patch File Download Diff File Expand all files Collapse all files

34
apps/aether-gateway/src/handlers/admin/endpoint/health_builders/keys.rs
View File

@@ -1,5 +1,5 @@
use crate::handlers::admin::request::AdminAppState;
use crate::handlers::public::provider_key_api_formats;
use crate::provider_key_auth::provider_key_effective_api_formats;
use aether_scheduler_core::count_recent_rpm_requests_for_provider_key_since;
use serde_json::json;
use std::time::{SystemTime, UNIX_EPOCH};
@@ -18,6 +18,16 @@ pub(crate) async fn build_admin_key_health_payload(
.await
.ok()
.and_then(|mut keys| keys.drain(..).next())?;
let provider = state
.read_provider_catalog_providers_by_ids(std::slice::from_ref(&key.provider_id))
.await
.ok()
.and_then(|mut providers| providers.drain(..).next())?;
let endpoints = state
.list_provider_catalog_endpoints_by_provider_ids(std::slice::from_ref(&key.provider_id))
.await
.ok()
.unwrap_or_default();
let request_count = key.request_count.unwrap_or(0);
let success_count = key.success_count.unwrap_or(0);
@@ -97,7 +107,9 @@ pub(crate) async fn build_admin_key_health_payload(
.unwrap_or(0));
} else {
let mut formats_payload = serde_json::Map::new();
for format_name in provider_key_api_formats(&key) {
for format_name in
provider_key_effective_api_formats(&key, &provider.provider_type, &endpoints)
{
let health_data = health_by_format.and_then(|formats| formats.get(&format_name));
let circuit_data = circuit_by_format.and_then(|formats| formats.get(&format_name));
let is_open = circuit_data
@@ -363,11 +375,27 @@ pub(crate) async fn recover_all_admin_key_health(
if !updated {
continue;
}
let provider = state
.read_provider_catalog_providers_by_ids(std::slice::from_ref(&key.provider_id))
.await
.ok()
.and_then(|mut providers| providers.drain(..).next());
let endpoints = state
.list_provider_catalog_endpoints_by_provider_ids(std::slice::from_ref(&key.provider_id))
.await
.ok()
.unwrap_or_default();
let api_formats = provider
.as_ref()
.map(|provider| {
provider_key_effective_api_formats(&key, &provider.provider_type, &endpoints)
})
.unwrap_or_default();
payload_items.push(json!({
"key_id": key.id,
"key_name": key.name,
"provider_id": key.provider_id,
"api_formats": key.api_formats.unwrap_or_else(|| json!([])),
"api_formats": api_formats,
}));
}

28
apps/aether-gateway/src/handlers/admin/endpoint/health_builders/status.rs
View File

@@ -1,9 +1,8 @@
use crate::handlers::admin::request::AdminAppState;
use crate::handlers::admin::shared::unix_secs_to_rfc3339;
use crate::handlers::public::{
api_format_display_name, build_public_health_timeline, provider_key_api_formats,
};
use crate::handlers::public::{api_format_display_name, build_public_health_timeline};
use crate::handlers::shared::unix_ms_to_rfc3339;
use crate::provider_key_auth::provider_key_effective_api_formats;
use aether_data_contracts::repository::candidates::PublicHealthTimelineBucket;
use aether_scheduler_core::{is_provider_key_circuit_open, provider_key_health_score};
use serde_json::json;
@@ -50,17 +49,26 @@ pub(crate) async fn build_admin_endpoint_health_status_payload(
let mut endpoint_to_format = BTreeMap::<String, String>::new();
let mut provider_ids_by_format = BTreeMap::<String, BTreeSet<String>>::new();
let mut active_provider_formats = BTreeSet::<(String, String)>::new();
let provider_type_by_id = providers
.iter()
.map(|provider| (provider.id.clone(), provider.provider_type.clone()))
.collect::<BTreeMap<_, _>>();
let mut active_endpoints_by_provider = BTreeMap::<String, Vec<_>>::new();
for endpoint in active_endpoints {
endpoint_to_format.insert(endpoint.id.clone(), endpoint.api_format.clone());
endpoint_ids_by_format
.entry(endpoint.api_format.clone())
.or_default()
.push(endpoint.id);
.push(endpoint.id.clone());
provider_ids_by_format
.entry(endpoint.api_format.clone())
.or_default()
.insert(endpoint.provider_id.clone());
active_provider_formats.insert((endpoint.provider_id, endpoint.api_format));
active_provider_formats.insert((endpoint.provider_id.clone(), endpoint.api_format.clone()));
active_endpoints_by_provider
.entry(endpoint.provider_id.clone())
.or_default()
.push(endpoint);
}
let all_endpoint_ids = endpoint_to_format.keys().cloned().collect::<Vec<_>>();
@@ -74,7 +82,15 @@ pub(crate) async fn build_admin_endpoint_health_status_payload(
.ok()
.unwrap_or_default();
for key in keys {
for api_format in provider_key_api_formats(&key) {
let provider_type = provider_type_by_id
.get(&key.provider_id)
.map(String::as_str)
.unwrap_or("");
let endpoints = active_endpoints_by_provider
.get(&key.provider_id)
.map(Vec::as_slice)
.unwrap_or(&[]);
for api_format in provider_key_effective_api_formats(&key, provider_type, endpoints) {
if !active_provider_formats.contains(&(key.provider_id.clone(), api_format.clone()))
{
continue;

8
apps/aether-gateway/src/handlers/admin/model/routing.rs
View File

@@ -122,7 +122,13 @@ pub(crate) async fn build_admin_global_model_routing_payload(
.cloned()
.unwrap_or_default()
.into_iter()
.filter(|key| provider_catalog_key_supports_format(key, &endpoint.api_format))
.filter(|key| {
provider_catalog_key_supports_format(
key,
provider.provider_type.as_str(),
&endpoint.api_format,
)
})
.filter(|key| {
key_allowed_models_match_global_model_for_routing(
key.allowed_models.as_ref(),

33
apps/aether-gateway/src/handlers/admin/provider/crud/writes.rs
View File

@@ -5,7 +5,9 @@ use crate::handlers::admin::provider::shared::paths::{
use crate::handlers::admin::provider::shared::payloads::{
AdminProviderCreateRequest, AdminProviderUpdatePatch,
};
use crate::handlers::admin::provider::write::provider::build_admin_fixed_provider_endpoint_record;
use crate::handlers::admin::provider::write::provider::{
reconcile_admin_fixed_provider_template_endpoints, reconcile_admin_fixed_provider_template_keys,
};
use crate::handlers::admin::request::{AdminAppState, AdminRequestContext};
use crate::handlers::admin::shared::attach_admin_audit_response;
use crate::GatewayError;
@@ -73,24 +75,12 @@ pub(crate) async fn maybe_build_local_admin_provider_writes_response(
return Ok(Some(build_admin_providers_data_unavailable_response()));
};
if let Some((base_url, endpoint_signatures)) =
state.fixed_provider_template(&created_provider.provider_type)
if state
.fixed_provider_template(&created_provider.provider_type)
.is_some()
{
for endpoint_signature in endpoint_signatures {
let endpoint = match build_admin_fixed_provider_endpoint_record(
&created_provider,
endpoint_signature,
base_url,
) {
Ok(endpoint) => endpoint,
Err(message) => {
return Ok(Some(build_admin_provider_bad_request_response(message)));
}
};
let Some(_) = state.create_provider_catalog_endpoint(&endpoint).await? else {
return Ok(Some(build_admin_providers_data_unavailable_response()));
};
}
reconcile_admin_fixed_provider_template_endpoints(state, &created_provider).await?;
reconcile_admin_fixed_provider_template_keys(state, &created_provider).await?;
}
return Ok(Some(attach_admin_audit_response(
Json(json!({
@@ -167,6 +157,13 @@ pub(crate) async fn maybe_build_local_admin_provider_writes_response(
else {
return Ok(Some(build_admin_providers_data_unavailable_response()));
};
if state
.fixed_provider_template(&updated_record.provider_type)
.is_some()
{
reconcile_admin_fixed_provider_template_endpoints(state, &updated_record).await?;
reconcile_admin_fixed_provider_template_keys(state, &updated_record).await?;
}
return Ok(Some(
match state
.build_admin_provider_summary_payload(&provider_id)

7
apps/aether-gateway/src/handlers/admin/provider/endpoint_keys/mutations/create.rs
View File

@@ -1,6 +1,7 @@
use crate::handlers::admin::provider::shared::paths::admin_provider_id_for_keys;
use crate::handlers::admin::provider::shared::payloads::AdminProviderKeyCreateRequest;
use crate::handlers::admin::request::{AdminAppState, AdminRequestContext};
use crate::provider_key_auth::provider_key_effective_api_formats;
use crate::{model_fetch::perform_model_fetch_for_key, GatewayError};
use axum::{
body::{Body, Bytes},
@@ -94,11 +95,17 @@ pub(super) async fn maybe_handle(
.ok()
.map(|duration| duration.as_secs())
.unwrap_or(0);
let endpoints = state
.list_provider_catalog_endpoints_by_provider_ids(std::slice::from_ref(&provider.id))
.await?;
let api_formats =
provider_key_effective_api_formats(&created, &provider.provider_type, &endpoints);
Ok(Some(
Json(state.build_admin_provider_key_response(
&created,
&provider.provider_type,
&api_formats,
now_unix_secs,
))
.into_response(),

7
apps/aether-gateway/src/handlers/admin/provider/endpoint_keys/mutations/update.rs
View File

@@ -1,6 +1,7 @@
use crate::handlers::admin::provider::shared::paths::admin_update_key_id;
use crate::handlers::admin::provider::shared::payloads::AdminProviderKeyUpdatePatch;
use crate::handlers::admin::request::{AdminAppState, AdminRequestContext};
use crate::provider_key_auth::provider_key_effective_api_formats;
use crate::{model_fetch::perform_model_fetch_for_key, GatewayError};
use axum::{
body::{Body, Bytes},
@@ -117,11 +118,17 @@ pub(super) async fn maybe_handle(
.ok()
.map(|duration| duration.as_secs())
.unwrap_or(0);
let endpoints = state
.list_provider_catalog_endpoints_by_provider_ids(std::slice::from_ref(&provider.id))
.await?;
let api_formats =
provider_key_effective_api_formats(&updated, &provider.provider_type, &endpoints);
Ok(Some(
Json(state.build_admin_provider_key_response(
&updated,
&provider.provider_type,
&api_formats,
now_unix_secs,
))
.into_response(),

3
apps/aether-gateway/src/handlers/admin/provider/oauth/dispatch/batch/execution.rs
View File

@@ -201,8 +201,10 @@ pub(super) async fn execute_admin_provider_oauth_batch_import(
match update_existing_provider_oauth_catalog_key(
state,
&existing_key,
provider_type,
&access_token,
&auth_config,
&api_formats,
None,
expires_at,
)
@@ -242,6 +244,7 @@ pub(super) async fn execute_admin_provider_oauth_batch_import(
match create_provider_oauth_catalog_key(
state,
provider_id,
provider_type,
key_name.as_str(),
&access_token,
&auth_config,

6
apps/aether-gateway/src/handlers/admin/provider/oauth/dispatch/batch/kiro_import.rs
View File

@@ -57,6 +57,7 @@ pub(super) async fn execute_admin_provider_oauth_kiro_batch_import(
let endpoints = state
.list_provider_catalog_endpoints_by_provider_ids(&[provider_id.to_string()])
.await?;
let api_formats = provider_oauth_active_api_formats(&endpoints);
let runtime_endpoint = provider_oauth_runtime_endpoint_for_provider("kiro", &endpoints);
let request_proxy = state
.resolve_admin_provider_oauth_operation_proxy_snapshot(
@@ -191,8 +192,10 @@ pub(super) async fn execute_admin_provider_oauth_kiro_batch_import(
match update_existing_provider_oauth_catalog_key(
state,
&existing_key,
provider.provider_type.as_str(),
&access_token,
&auth_config,
&api_formats,
None,
refreshed_auth_config.expires_at,
)
@@ -223,10 +226,11 @@ pub(super) async fn execute_admin_provider_oauth_kiro_batch_import(
match create_provider_oauth_catalog_key(
state,
provider_id,
provider.provider_type.as_str(),
&key_name,
&access_token,
&auth_config,
&provider_oauth_active_api_formats(&endpoints),
&api_formats,
None,
refreshed_auth_config.expires_at,
)

3
apps/aether-gateway/src/handlers/admin/provider/oauth/dispatch/complete/provider.rs
View File

@@ -174,8 +174,10 @@ pub(super) async fn handle_admin_provider_oauth_complete_provider(
match state
.update_existing_provider_oauth_catalog_key(
&existing_key,
&provider_type,
&access_token,
&auth_config,
&api_formats,
None,
expires_at,
)
@@ -213,6 +215,7 @@ pub(super) async fn handle_admin_provider_oauth_complete_provider(
match state
.create_provider_oauth_catalog_key(
&provider_id,
&provider_type,
&name,
&access_token,
&auth_config,

3
apps/aether-gateway/src/handlers/admin/provider/oauth/dispatch/device/poll.rs
View File

@@ -351,8 +351,10 @@ pub(super) async fn handle_admin_provider_oauth_device_poll(
match state
.update_existing_provider_oauth_catalog_key(
&existing_key,
&provider.provider_type,
&access_token,
&auth_config,
&api_formats,
key_proxy.clone(),
Some(expires_at),
)
@@ -374,6 +376,7 @@ pub(super) async fn handle_admin_provider_oauth_device_poll(
match state
.create_provider_oauth_catalog_key(
&provider_id,
&provider.provider_type,
&key_name,
&access_token,
&auth_config,

3
apps/aether-gateway/src/handlers/admin/provider/oauth/dispatch/import.rs
View File

@@ -166,8 +166,10 @@ pub(super) async fn handle_admin_provider_oauth_import_refresh_token(
match state
.update_existing_provider_oauth_catalog_key(
&existing_key,
&provider_type,
&access_token,
&auth_config,
&api_formats,
None,
expires_at,
)
@@ -204,6 +206,7 @@ pub(super) async fn handle_admin_provider_oauth_import_refresh_token(
match state
.create_provider_oauth_catalog_key(
&provider_id,
&provider_type,
&name,
&access_token,
&auth_config,

31
apps/aether-gateway/src/handlers/admin/provider/oauth/provisioning.rs
View File

@@ -2,12 +2,13 @@ use super::state::{
enrich_admin_provider_oauth_auth_config, json_non_empty_string, json_u64_value,
};
use crate::handlers::admin::request::AdminAppState;
use crate::provider_key_auth::provider_active_api_formats;
use crate::GatewayError;
use aether_data_contracts::repository::provider_catalog::{
StoredProviderCatalogEndpoint, StoredProviderCatalogKey,
};
use aether_provider_transport::provider_types::provider_type_is_fixed;
use serde_json::json;
use std::collections::BTreeSet;
use std::time::{SystemTime, UNIX_EPOCH};
use uuid::Uuid;
@@ -23,16 +24,7 @@ pub(crate) fn provider_oauth_key_proxy_value(
pub(crate) fn provider_oauth_active_api_formats(
endpoints: &[StoredProviderCatalogEndpoint],
) -> Vec<String> {
let mut formats = Vec::new();
let mut seen = BTreeSet::new();
for endpoint in endpoints.iter().filter(|endpoint| endpoint.is_active) {
let api_format = endpoint.api_format.trim();
if api_format.is_empty() || !seen.insert(api_format.to_string()) {
continue;
}
formats.push(api_format.to_string());
}
formats
provider_active_api_formats(endpoints)
}
pub(crate) fn build_provider_oauth_auth_config_from_token_payload(
@@ -76,6 +68,7 @@ pub(crate) fn build_provider_oauth_auth_config_from_token_payload(
pub(crate) async fn create_provider_oauth_catalog_key(
state: &AdminAppState<'_>,
provider_id: &str,
provider_type: &str,
name: &str,
access_token: &str,
auth_config: &serde_json::Map<String, serde_json::Value>,
@@ -108,7 +101,7 @@ pub(crate) async fn create_provider_oauth_catalog_key(
)
.map_err(|err| GatewayError::Internal(err.to_string()))?
.with_transport_fields(
Some(json!(api_formats)),
provider_oauth_catalog_key_api_formats(provider_type, api_formats),
encrypted_api_key,
Some(encrypted_auth_config),
None,
@@ -136,8 +129,10 @@ pub(crate) async fn create_provider_oauth_catalog_key(
pub(crate) async fn update_existing_provider_oauth_catalog_key(
state: &AdminAppState<'_>,
existing_key: &StoredProviderCatalogKey,
provider_type: &str,
access_token: &str,
auth_config: &serde_json::Map<String, serde_json::Value>,
api_formats: &[String],
proxy: Option<serde_json::Value>,
expires_at_unix_secs: Option<u64>,
) -> Result<Option<StoredProviderCatalogKey>, GatewayError> {
@@ -159,6 +154,7 @@ pub(crate) async fn update_existing_provider_oauth_catalog_key(
let mut updated = existing_key.clone();
updated.encrypted_api_key = encrypted_api_key;
updated.encrypted_auth_config = Some(encrypted_auth_config);
updated.api_formats = provider_oauth_catalog_key_api_formats(provider_type, api_formats);
updated.is_active = true;
updated.expires_at_unix_secs = expires_at_unix_secs;
updated.oauth_invalid_at_unix_secs = None;
@@ -172,3 +168,14 @@ pub(crate) async fn update_existing_provider_oauth_catalog_key(
updated.updated_at_unix_secs = Some(now_unix_secs);
state.update_provider_catalog_key(&updated).await
}
fn provider_oauth_catalog_key_api_formats(
provider_type: &str,
api_formats: &[String],
) -> Option<serde_json::Value> {
if provider_type_is_fixed(provider_type) {
None
} else {
Some(json!(api_formats))
}
}

27
apps/aether-gateway/src/handlers/admin/provider/pool_admin/payloads.rs
View File

@@ -3,26 +3,14 @@ use crate::handlers::admin::provider::shared::support::{
};
use crate::handlers::admin::request::AdminAppState;
use crate::handlers::admin::shared::{provider_key_status_snapshot_payload, unix_secs_to_rfc3339};
use crate::provider_key_auth::provider_key_auth_semantics;
use crate::provider_key_auth::{provider_key_auth_semantics, provider_key_effective_api_formats};
use aether_admin::provider::pool as admin_provider_pool_pure;
use aether_admin::provider::quota as admin_provider_quota_pure;
use aether_data_contracts::repository::provider_catalog::StoredProviderCatalogKey;
use aether_data_contracts::repository::provider_catalog::{
StoredProviderCatalogEndpoint, StoredProviderCatalogKey,
};
use serde_json::json;
pub(super) fn admin_pool_api_formats(key: &StoredProviderCatalogKey) -> Vec<String> {
key.api_formats
.as_ref()
.and_then(serde_json::Value::as_array)
.map(|values| {
values
.iter()
.filter_map(serde_json::Value::as_str)
.map(ToOwned::to_owned)
.collect::<Vec<_>>()
})
.unwrap_or_default()
}
fn admin_pool_string_list(value: Option<&serde_json::Value>) -> Option<Vec<String>> {
let values = value
.and_then(serde_json::Value::as_array)
@@ -749,6 +737,7 @@ fn admin_pool_scheduling_payload(
pub(super) fn build_admin_pool_key_payload(
state: &AdminAppState<'_>,
provider_type: &str,
endpoints: &[StoredProviderCatalogEndpoint],
key: &StoredProviderCatalogKey,
runtime: &AdminProviderPoolRuntimeState,
pool_config: Option<AdminProviderPoolConfig>,
@@ -915,7 +904,11 @@ pub(super) fn build_admin_pool_key_payload(
);
payload.insert(
"api_formats".to_string(),
json!(admin_pool_api_formats(key)),
json!(provider_key_effective_api_formats(
key,
provider_type,
endpoints,
)),
);
payload.insert(
"rate_multipliers".to_string(),

4
apps/aether-gateway/src/handlers/admin/provider/pool_admin/read_routes/keys.rs
View File

@@ -124,6 +124,9 @@ pub(super) async fn build_admin_pool_list_keys_response(
};
let key_ids = keys.iter().map(|key| key.id.clone()).collect::<Vec<_>>();
let endpoints = state
.list_provider_catalog_endpoints_by_provider_ids(std::slice::from_ref(&provider.id))
.await?;
let runtime = match (state.redis_kv_runner(), pool_config.as_ref()) {
(Some(runner), Some(pool_config)) if !key_ids.is_empty() => {
read_admin_provider_pool_runtime_state(
@@ -143,6 +146,7 @@ pub(super) async fn build_admin_pool_list_keys_response(
pool_payloads::build_admin_pool_key_payload(
state,
&provider.provider_type,
&endpoints,
&key,
&runtime,
pool_config.clone(),

39
apps/aether-gateway/src/handlers/admin/provider/query/models.rs
View File

@@ -14,6 +14,9 @@ use crate::ai_pipeline::{maybe_build_sync_finalize_outcome, GatewayControlDecisi
use crate::execution_runtime;
use crate::handlers::admin::request::{AdminAppState, AdminGatewayProviderTransportSnapshot};
use crate::model_fetch::ModelFetchRuntimeState;
use crate::provider_key_auth::{
provider_key_configured_api_formats, provider_key_inherits_provider_api_formats,
};
use crate::provider_transport::kiro::{
build_kiro_generate_assistant_response_url, build_kiro_provider_headers,
build_kiro_provider_request_body, supports_local_kiro_request_transport_with_network,
@@ -272,9 +275,13 @@ fn provider_query_select_kiro_endpoint<'a>(
fn provider_query_key_supports_endpoint(
key: &StoredProviderCatalogKey,
provider_type: &str,
endpoint_api_format: &str,
) -> bool {
let formats = json_string_list(key.api_formats.as_ref());
if provider_key_inherits_provider_api_formats(key, provider_type) {
return true;
}
let formats = provider_key_configured_api_formats(key);
formats.is_empty()
|| formats
.iter()
@@ -321,7 +328,11 @@ async fn provider_query_select_preferred_non_kiro_endpoint(
for key in keys {
if !key.is_active
|| selected_key_id.is_some_and(|value| value != key.id.as_str())
|| !provider_query_key_supports_endpoint(key, &endpoint.api_format)
|| !provider_query_key_supports_endpoint(
key,
&provider.provider_type,
&endpoint.api_format,
)
{
continue;
}
@@ -348,7 +359,11 @@ async fn provider_query_select_preferred_non_kiro_endpoint(
for key in keys {
if !key.is_active
|| selected_key_id.is_some_and(|value| value != key.id.as_str())
|| !provider_query_key_supports_endpoint(key, &endpoint.api_format)
|| !provider_query_key_supports_endpoint(
key,
&provider.provider_type,
&endpoint.api_format,
)
{
continue;
}
@@ -375,7 +390,11 @@ async fn provider_query_select_preferred_non_kiro_endpoint(
&& keys.iter().any(|key| {
key.is_active
&& selected_key_id.is_none_or(|value| value == key.id.as_str())
&& provider_query_key_supports_endpoint(key, &endpoint.api_format)
&& provider_query_key_supports_endpoint(
key,
&provider.provider_type,
&endpoint.api_format,
)
})
})
.or_else(|| endpoints.iter().find(|endpoint| endpoint.is_active))
@@ -531,7 +550,13 @@ async fn provider_query_build_kiro_test_candidates(
ADMIN_PROVIDER_QUERY_API_KEY_NOT_FOUND_DETAIL,
));
};
if !key.is_active || !provider_query_key_supports_endpoint(key, &endpoint.api_format) {
if !key.is_active
|| !provider_query_key_supports_endpoint(
key,
&provider.provider_type,
&endpoint.api_format,
)
{
return Err(build_admin_provider_query_not_found_response(
ADMIN_PROVIDER_QUERY_NO_ACTIVE_TEST_CANDIDATE_DETAIL,
));
@@ -567,7 +592,9 @@ async fn provider_query_build_kiro_test_candidates(
.as_deref()
.is_none_or(|value| value == key.id.as_str())
})
.filter(|key| provider_query_key_supports_endpoint(key, &endpoint.api_format))
.filter(|key| {
provider_query_key_supports_endpoint(key, &provider.provider_type, &endpoint.api_format)
})
.collect::<Vec<_>>();
keys.sort_by_key(|key| {
provider_query_test_key_sort_key(provider.provider_type.as_str(), key, &endpoint.api_format)

17
apps/aether-gateway/src/handlers/admin/provider/summary/aggregates.rs
View File

@@ -3,8 +3,6 @@ use crate::handlers::admin::request::AdminAppState;
use aether_data_contracts::repository::provider_catalog::{
StoredProviderCatalogEndpoint, StoredProviderCatalogKey,
};
use aether_data_contracts::repository::quota::StoredProviderQuotaSnapshot;
use futures_util::future::join_all;
use serde_json::json;
use std::collections::{BTreeMap, BTreeSet};
use std::time::{SystemTime, UNIX_EPOCH};
@@ -235,17 +233,6 @@ pub(crate) async fn build_admin_providers_summary_payload(
.or_default()
.insert(row.global_model_id);
}
let quota_snapshots_by_provider = join_all(provider_ids.iter().map(|provider_id| async {
let quota_snapshot = state
.read_provider_quota_snapshot(provider_id)
.await
.ok()
.flatten();
(provider_id.clone(), quota_snapshot)
}))
.await
.into_iter()
.collect::<BTreeMap<String, Option<StoredProviderQuotaSnapshot>>>();
let now_unix_secs = SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap_or_default()
@@ -268,9 +255,7 @@ pub(crate) async fn build_admin_providers_summary_payload(
.get(&provider.id)
.map(Vec::as_slice)
.unwrap_or(&[]),
quota_snapshots_by_provider
.get(&provider.id)
.and_then(Option::as_ref),
None,
model_stats_by_provider.get(&provider.id),
active_global_model_ids,
now_unix_secs,

45
apps/aether-gateway/src/handlers/admin/provider/summary/value.rs
View File

@@ -1,7 +1,6 @@
use crate::handlers::admin::shared::unix_secs_to_rfc3339;
use crate::handlers::public::{
provider_key_api_formats, request_candidate_event_unix_ms, request_candidate_status_label,
};
use crate::handlers::public::{request_candidate_event_unix_ms, request_candidate_status_label};
use crate::provider_key_auth::provider_key_effective_api_formats;
use aether_data_contracts::repository::candidates::{
RequestCandidateStatus, StoredRequestCandidate,
};
@@ -66,7 +65,9 @@ pub(crate) fn build_admin_provider_summary_value(
keys_by_endpoint.entry(endpoint.id.clone()).or_default();
}
for key in keys {
for api_format in provider_key_api_formats(key) {
for api_format in
provider_key_effective_api_formats(key, &provider.provider_type, endpoints)
{
if let Some(endpoint_id) = format_to_endpoint_id.get(&api_format) {
keys_by_endpoint
.entry(endpoint_id.clone())
@@ -131,6 +132,26 @@ pub(crate) fn build_admin_provider_summary_value(
.and_then(|cfg| cfg.get("architecture_id"))
.and_then(serde_json::Value::as_str)
.map(ToOwned::to_owned);
let billing_type = quota_snapshot
.map(|quota| quota.billing_type.clone())
.or_else(|| provider.billing_type.clone());
let monthly_quota_usd = quota_snapshot
.and_then(|quota| quota.monthly_quota_usd)
.or(provider.monthly_quota_usd);
let monthly_used_usd = quota_snapshot
.map(|quota| quota.monthly_used_usd)
.or(provider.monthly_used_usd);
let quota_reset_day = quota_snapshot
.and_then(|quota| quota.quota_reset_day)
.or(provider.quota_reset_day);
let quota_last_reset_at = quota_snapshot
.and_then(|quota| quota.quota_last_reset_at_unix_secs)
.or(provider.quota_last_reset_at_unix_secs)
.and_then(unix_secs_to_rfc3339);
let quota_expires_at = quota_snapshot
.and_then(|quota| quota.quota_expires_at_unix_secs)
.or(provider.quota_expires_at_unix_secs)
.and_then(unix_secs_to_rfc3339);
json!({
"id": provider.id.clone(),
@@ -142,16 +163,12 @@ pub(crate) fn build_admin_provider_summary_value(
"keep_priority_on_conversion": provider.keep_priority_on_conversion,
"enable_format_conversion": provider.enable_format_conversion,
"is_active": provider.is_active,
"billing_type": quota_snapshot.map(|quota| quota.billing_type.clone()),
"monthly_quota_usd": quota_snapshot.and_then(|quota| quota.monthly_quota_usd),
"monthly_used_usd": quota_snapshot.map(|quota| quota.monthly_used_usd),
"quota_reset_day": quota_snapshot.and_then(|quota| quota.quota_reset_day),
"quota_last_reset_at": quota_snapshot
.and_then(|quota| quota.quota_last_reset_at_unix_secs)
.and_then(unix_secs_to_rfc3339),
"quota_expires_at": quota_snapshot
.and_then(|quota| quota.quota_expires_at_unix_secs)
.and_then(unix_secs_to_rfc3339),
"billing_type": billing_type,
"monthly_quota_usd": monthly_quota_usd,
"monthly_used_usd": monthly_used_usd,
"quota_reset_day": quota_reset_day,
"quota_last_reset_at": quota_last_reset_at,
"quota_expires_at": quota_expires_at,
"max_retries": provider.max_retries,
"proxy": provider.proxy.clone(),
"stream_first_byte_timeout": provider.stream_first_byte_timeout_secs,

9
apps/aether-gateway/src/handlers/admin/provider/write/keys/create.rs
View File

@@ -10,6 +10,7 @@ use crate::handlers::admin::shared::{
use aether_data_contracts::repository::provider_catalog::{
StoredProviderCatalogKey, StoredProviderCatalogProvider,
};
use aether_provider_transport::provider_types::provider_type_is_fixed;
use serde_json::json;
use std::time::{SystemTime, UNIX_EPOCH};
use uuid::Uuid;
@@ -129,6 +130,8 @@ pub(crate) async fn build_admin_create_provider_key_record(
.ok()
.map(|duration| duration.as_secs())
.unwrap_or(0);
let inherits_provider_api_formats =
auth_type == "oauth" && provider_type_is_fixed(&provider.provider_type);
let mut key = StoredProviderCatalogKey::new(
Uuid::new_v4().to_string(),
provider.id.clone(),
@@ -139,7 +142,11 @@ pub(crate) async fn build_admin_create_provider_key_record(
)
.map_err(|err| err.to_string())?
.with_transport_fields(
Some(json!(api_formats)),
if inherits_provider_api_formats {
None
} else {
Some(json!(api_formats))
},
encrypted_api_key,
encrypted_auth_config,
normalize_json_object(payload.rate_multipliers, "rate_multipliers")?,

9
apps/aether-gateway/src/handlers/admin/provider/write/keys/payload.rs
View File

@@ -1,4 +1,5 @@
use crate::handlers::admin::request::AdminAppState;
use crate::provider_key_auth::provider_key_effective_api_formats;
use aether_data_contracts::repository::provider_catalog::{
ProviderCatalogKeyListOrder, ProviderCatalogKeyListQuery,
};
@@ -29,6 +30,11 @@ pub(crate) async fn build_admin_provider_keys_payload(
})
.await
.ok()?;
let endpoints = state
.list_provider_catalog_endpoints_by_provider_ids(std::slice::from_ref(&provider.id))
.await
.ok()
.unwrap_or_default();
let now_unix_secs = SystemTime::now()
.duration_since(UNIX_EPOCH)
.ok()
@@ -39,9 +45,12 @@ pub(crate) async fn build_admin_provider_keys_payload(
.items
.into_iter()
.map(|key| {
let api_formats =
provider_key_effective_api_formats(&key, &provider.provider_type, &endpoints);
state.build_admin_provider_key_response(
&key,
&provider.provider_type,
&api_formats,
now_unix_secs,
)
})

21
apps/aether-gateway/src/handlers/admin/provider/write/keys/update.rs
View File

@@ -7,9 +7,11 @@ use crate::handlers::admin::shared::{
decrypt_catalog_secret_with_fallbacks, encrypt_catalog_secret_with_fallbacks, json_string_list,
normalize_json_object, normalize_string_list, parse_catalog_auth_config_json,
};
use crate::provider_key_auth::provider_key_is_oauth_managed;
use aether_data_contracts::repository::provider_catalog::{
StoredProviderCatalogKey, StoredProviderCatalogProvider,
};
use aether_provider_transport::provider_types::provider_type_is_fixed;
use serde_json::json;
use std::time::{SystemTime, UNIX_EPOCH};
@@ -35,6 +37,9 @@ pub(crate) async fn build_admin_update_provider_key_record(
.auth_type
.as_deref()
.is_some_and(|_| target_auth_type != current_auth_type);
let managed_fixed_oauth_key = provider_type_is_fixed(&provider.provider_type)
&& (provider_key_is_oauth_managed(existing, &provider.provider_type)
|| target_auth_type.eq_ignore_ascii_case("oauth"));
let api_key_present = fields.contains("api_key");
let api_key_value = payload
@@ -187,11 +192,19 @@ pub(crate) async fn build_admin_update_provider_key_record(
if fields.contains("api_formats") {
let api_formats = normalize_string_list(payload.api_formats)
.ok_or_else(|| "api_formats 为必填字段".to_string())?;
validate_vertex_api_formats(&provider.provider_type, &target_auth_type, &api_formats)?;
updated.api_formats = Some(json!(api_formats));
if managed_fixed_oauth_key {
updated.api_formats = None;
} else {
validate_vertex_api_formats(&provider.provider_type, &target_auth_type, &api_formats)?;
updated.api_formats = Some(json!(api_formats));
}
} else if payload.auth_type.is_some() {
let api_formats = json_string_list(existing.api_formats.as_ref());
validate_vertex_api_formats(&provider.provider_type, &target_auth_type, &api_formats)?;
if managed_fixed_oauth_key {
updated.api_formats = None;
} else {
let api_formats = json_string_list(existing.api_formats.as_ref());
validate_vertex_api_formats(&provider.provider_type, &target_auth_type, &api_formats)?;
}
}
updated.auth_type = target_auth_type;

6
apps/aether-gateway/src/handlers/admin/provider/write/provider.rs
View File

@@ -1,7 +1,13 @@
mod create;
mod endpoint;
mod template;
mod update;
pub(crate) use self::create::build_admin_create_provider_record;
pub(crate) use self::endpoint::build_admin_fixed_provider_endpoint_record;
pub(crate) use self::template::{
apply_admin_fixed_provider_endpoint_template_overrides,
reconcile_admin_fixed_provider_template_endpoints,
reconcile_admin_fixed_provider_template_keys,
};
pub(crate) use self::update::build_admin_update_provider_record;

97
apps/aether-gateway/src/handlers/admin/provider/write/provider/endpoint.rs
View File

@@ -3,30 +3,65 @@ use crate::handlers::public::normalize_admin_base_url;
use aether_data_contracts::repository::provider_catalog::{
StoredProviderCatalogEndpoint, StoredProviderCatalogProvider,
};
use serde_json::json;
use aether_provider_transport::provider_types::{
FixedProviderEndpointTemplate, FixedProviderTemplate,
};
use std::time::{SystemTime, UNIX_EPOCH};
use uuid::Uuid;
pub(crate) fn build_admin_fixed_provider_endpoint_record(
#[derive(Debug, Clone, PartialEq)]
pub(crate) struct AdminFixedProviderEndpointDefaults {
pub(crate) api_format: String,
pub(crate) api_family: String,
pub(crate) endpoint_kind: String,
pub(crate) is_active: bool,
pub(crate) base_url: String,
pub(crate) header_rules: Option<serde_json::Value>,
pub(crate) body_rules: Option<serde_json::Value>,
pub(crate) max_retries: Option<i32>,
pub(crate) custom_path: Option<String>,
pub(crate) config: Option<serde_json::Value>,
pub(crate) format_acceptance_config: Option<serde_json::Value>,
pub(crate) proxy: Option<serde_json::Value>,
}
pub(crate) fn build_admin_fixed_provider_endpoint_defaults(
provider: &StoredProviderCatalogProvider,
api_format: &str,
base_url: &str,
) -> Result<StoredProviderCatalogEndpoint, String> {
template: &FixedProviderTemplate,
endpoint_template: &FixedProviderEndpointTemplate,
) -> Result<AdminFixedProviderEndpointDefaults, String> {
let (normalized_api_format, api_family, endpoint_kind) =
admin_endpoint_signature_parts(api_format)
.ok_or_else(|| format!("无效的 api_format: {api_format}"))?;
admin_endpoint_signature_parts(endpoint_template.api_format)
.ok_or_else(|| format!("无效的 api_format: {}", endpoint_template.api_format))?;
let body_rules = admin_default_body_rules_for_signature(
normalized_api_format,
Some(provider.provider_type.as_str()),
)
.and_then(|(_, rules)| (!rules.is_empty()).then_some(serde_json::Value::Array(rules)));
let endpoint_config = if provider.provider_type == "codex"
&& matches!(normalized_api_format, "openai:cli" | "openai:image")
{
Some(json!({ "upstream_stream_policy": "force_stream" }))
} else {
None
};
Ok(AdminFixedProviderEndpointDefaults {
api_format: normalized_api_format.to_string(),
api_family: api_family.to_string(),
endpoint_kind: endpoint_kind.to_string(),
is_active: true,
base_url: normalize_admin_base_url(template.base_url)?,
header_rules: None,
body_rules,
max_retries: Some(provider.max_retries.unwrap_or(2)),
custom_path: endpoint_template.custom_path.map(ToOwned::to_owned),
config: fixed_provider_endpoint_default_config(endpoint_template),
format_acceptance_config: None,
proxy: None,
})
}
pub(crate) fn build_admin_fixed_provider_endpoint_record(
provider: &StoredProviderCatalogProvider,
template: &FixedProviderTemplate,
endpoint_template: &FixedProviderEndpointTemplate,
) -> Result<StoredProviderCatalogEndpoint, String> {
let defaults =
build_admin_fixed_provider_endpoint_defaults(provider, template, endpoint_template)?;
let now_unix_secs = SystemTime::now()
.duration_since(UNIX_EPOCH)
.ok()
@@ -36,22 +71,32 @@ pub(crate) fn build_admin_fixed_provider_endpoint_record(
StoredProviderCatalogEndpoint::new(
Uuid::new_v4().to_string(),
provider.id.clone(),
normalized_api_format.to_string(),
Some(api_family.to_string()),
Some(endpoint_kind.to_string()),
true,
defaults.api_format,
Some(defaults.api_family),
Some(defaults.endpoint_kind),
defaults.is_active,
)
.map_err(|err| err.to_string())?
.with_timestamps(Some(now_unix_secs), Some(now_unix_secs))
.with_transport_fields(
normalize_admin_base_url(base_url)?,
None,
body_rules,
Some(provider.max_retries.unwrap_or(2)),
None,
endpoint_config,
None,
None,
defaults.base_url,
defaults.header_rules,
defaults.body_rules,
defaults.max_retries,
defaults.custom_path,
defaults.config,
defaults.format_acceptance_config,
defaults.proxy,
)
.map_err(|err| err.to_string())
}
fn fixed_provider_endpoint_default_config(
endpoint_template: &FixedProviderEndpointTemplate,
) -> Option<serde_json::Value> {
let mut config = serde_json::Map::new();
for default in endpoint_template.config_defaults {
config.insert(default.key.to_string(), default.value.to_json_value());
}
(!config.is_empty()).then_some(serde_json::Value::Object(config))
}

530
apps/aether-gateway/src/handlers/admin/provider/write/provider/template.rs Normal file
View File

@@ -0,0 +1,530 @@
use super::endpoint::{
build_admin_fixed_provider_endpoint_defaults, build_admin_fixed_provider_endpoint_record,
};
use crate::handlers::admin::request::AdminAppState;
use crate::provider_key_auth::provider_key_is_oauth_managed;
use crate::GatewayError;
use aether_data_contracts::repository::provider_catalog::{
StoredProviderCatalogEndpoint, StoredProviderCatalogKey, StoredProviderCatalogProvider,
};
use aether_provider_transport::provider_types::{
fixed_provider_template, FixedProviderEndpointTemplate, FixedProviderTemplate,
};
use serde_json::{json, Map, Value};
use std::collections::{BTreeMap, BTreeSet};
const FIXED_PROVIDER_TEMPLATE_METADATA_KEY: &str = "_aether_fixed_provider_template";
const OVERRIDE_BODY_RULES: &str = "body_rules";
const OVERRIDE_FORMAT_ACCEPTANCE_CONFIG: &str = "format_acceptance_config";
const OVERRIDE_HEADER_RULES: &str = "header_rules";
const OVERRIDE_IS_ACTIVE: &str = "is_active";
const OVERRIDE_MAX_RETRIES: &str = "max_retries";
const OVERRIDE_PROXY: &str = "proxy";
#[derive(Debug, Clone, Default, PartialEq, Eq)]
struct FixedProviderEndpointMetadata {
provider_type: String,
item_key: String,
version: u32,
retired: bool,
overrides: BTreeSet<String>,
config_keys: BTreeSet<String>,
}
pub(crate) async fn reconcile_admin_fixed_provider_template_endpoints(
state: &AdminAppState<'_>,
provider: &StoredProviderCatalogProvider,
) -> Result<(), GatewayError> {
let Some(template) = state.fixed_provider_template(&provider.provider_type) else {
return Ok(());
};
let existing_endpoints = state
.list_provider_catalog_endpoints_by_provider_ids(std::slice::from_ref(&provider.id))
.await?;
let mut matched_endpoint_ids = BTreeSet::new();
for endpoint_template in template.endpoints {
let existing_endpoint = existing_endpoints
.iter()
.find(|endpoint| endpoint_matches_fixed_provider_template(endpoint, endpoint_template));
match existing_endpoint {
Some(existing_endpoint) => {
matched_endpoint_ids.insert(existing_endpoint.id.clone());
let updated = reconcile_fixed_provider_endpoint(
provider,
existing_endpoint,
template,
endpoint_template,
)
.map_err(GatewayError::Internal)?;
if updated != *existing_endpoint {
let Some(_) = state.update_provider_catalog_endpoint(&updated).await? else {
return Err(GatewayError::Internal(
"provider catalog endpoint writer unavailable".to_string(),
));
};
}
}
None => {
let mut created = build_admin_fixed_provider_endpoint_record(
provider,
template,
endpoint_template,
)
.map_err(GatewayError::Internal)?;
let metadata =
managed_fixed_provider_endpoint_metadata(template, endpoint_template);
upsert_fixed_provider_endpoint_metadata(&mut created, &metadata);
let Some(_) = state.create_provider_catalog_endpoint(&created).await? else {
return Err(GatewayError::Internal(
"provider catalog endpoint writer unavailable".to_string(),
));
};
}
}
}
for existing_endpoint in &existing_endpoints {
if matched_endpoint_ids.contains(&existing_endpoint.id) {
continue;
}
let Some(metadata) = fixed_provider_endpoint_metadata(existing_endpoint) else {
continue;
};
if metadata.retired && !existing_endpoint.is_active {
continue;
}
let mut retired = existing_endpoint.clone();
let mut retired_metadata = metadata;
retired.is_active = false;
retired_metadata.retired = true;
upsert_fixed_provider_endpoint_metadata(&mut retired, &retired_metadata);
if retired != *existing_endpoint {
retired.updated_at_unix_secs = Some(current_unix_secs());
let Some(_) = state.update_provider_catalog_endpoint(&retired).await? else {
return Err(GatewayError::Internal(
"provider catalog endpoint writer unavailable".to_string(),
));
};
}
}
Ok(())
}
pub(crate) async fn reconcile_admin_fixed_provider_template_keys(
state: &AdminAppState<'_>,
provider: &StoredProviderCatalogProvider,
) -> Result<(), GatewayError> {
let Some(_) = state.fixed_provider_template(&provider.provider_type) else {
return Ok(());
};
let existing_keys = state
.list_provider_catalog_keys_by_provider_ids(std::slice::from_ref(&provider.id))
.await?;
for existing_key in existing_keys {
let Some(updated_key) = reconcile_fixed_provider_key(provider, &existing_key) else {
continue;
};
let Some(_) = state.update_provider_catalog_key(&updated_key).await? else {
return Err(GatewayError::Internal(
"provider catalog key writer unavailable".to_string(),
));
};
}
Ok(())
}
pub(crate) fn apply_admin_fixed_provider_endpoint_template_overrides(
provider: &StoredProviderCatalogProvider,
existing_endpoint: &StoredProviderCatalogEndpoint,
updated_endpoint: &mut StoredProviderCatalogEndpoint,
) -> Result<(), String> {
let Some(template) = fixed_provider_template(&provider.provider_type) else {
return Ok(());
};
let Some(endpoint_template) =
resolve_fixed_provider_endpoint_template(template, existing_endpoint, updated_endpoint)
else {
return Ok(());
};
let defaults =
build_admin_fixed_provider_endpoint_defaults(provider, template, endpoint_template)?;
let mut metadata = fixed_provider_endpoint_metadata(existing_endpoint)
.unwrap_or_else(|| managed_fixed_provider_endpoint_metadata(template, endpoint_template));
let mut overrides = metadata.overrides.clone();
sync_override_if_changed(
&mut overrides,
OVERRIDE_HEADER_RULES,
&existing_endpoint.header_rules,
&updated_endpoint.header_rules,
&defaults.header_rules,
);
sync_override_if_changed(
&mut overrides,
OVERRIDE_BODY_RULES,
&existing_endpoint.body_rules,
&updated_endpoint.body_rules,
&defaults.body_rules,
);
sync_override_if_changed(
&mut overrides,
OVERRIDE_MAX_RETRIES,
&existing_endpoint.max_retries,
&updated_endpoint.max_retries,
&defaults.max_retries,
);
sync_override_if_changed(
&mut overrides,
OVERRIDE_IS_ACTIVE,
&existing_endpoint.is_active,
&updated_endpoint.is_active,
&defaults.is_active,
);
sync_override_if_changed(
&mut overrides,
OVERRIDE_PROXY,
&existing_endpoint.proxy,
&updated_endpoint.proxy,
&defaults.proxy,
);
sync_override_if_changed(
&mut overrides,
OVERRIDE_FORMAT_ACCEPTANCE_CONFIG,
&existing_endpoint.format_acceptance_config,
&updated_endpoint.format_acceptance_config,
&defaults.format_acceptance_config,
);
let current_config_defaults = fixed_provider_endpoint_config_defaults(endpoint_template);
let config = endpoint_config_without_metadata(updated_endpoint.config.as_ref());
let existing_config = endpoint_config_without_metadata(existing_endpoint.config.as_ref());
let current_config_keys = current_config_defaults
.keys()
.cloned()
.collect::<BTreeSet<_>>();
let mut tracked_config_keys = metadata.config_keys.clone();
tracked_config_keys.extend(current_config_keys.iter().cloned());
for key in tracked_config_keys {
let before = existing_config.get(&key);
let actual = config.get(&key);
let desired = current_config_defaults.get(&key);
sync_override_if_changed(
&mut overrides,
&config_override_key(&key),
&before.cloned(),
&actual.cloned(),
&desired.cloned(),
);
}
metadata.provider_type = template.provider_type.to_string();
metadata.item_key = endpoint_template.item_key.to_string();
metadata.version = template.version;
metadata.retired = false;
metadata.overrides = overrides;
metadata.config_keys = current_config_keys;
updated_endpoint.config = materialize_endpoint_config(config, &metadata);
Ok(())
}
fn reconcile_fixed_provider_endpoint(
provider: &StoredProviderCatalogProvider,
existing_endpoint: &StoredProviderCatalogEndpoint,
template: &FixedProviderTemplate,
endpoint_template: &FixedProviderEndpointTemplate,
) -> Result<StoredProviderCatalogEndpoint, String> {
let defaults =
build_admin_fixed_provider_endpoint_defaults(provider, template, endpoint_template)?;
let mut updated = existing_endpoint.clone();
let metadata = fixed_provider_endpoint_metadata(existing_endpoint)
.unwrap_or_else(|| managed_fixed_provider_endpoint_metadata(template, endpoint_template));
updated.api_format = defaults.api_format.clone();
updated.api_family = Some(defaults.api_family.clone());
updated.endpoint_kind = Some(defaults.endpoint_kind.clone());
updated.base_url = defaults.base_url;
updated.custom_path = defaults.custom_path;
if !metadata.overrides.contains(OVERRIDE_HEADER_RULES) {
updated.header_rules = defaults.header_rules;
}
if !metadata.overrides.contains(OVERRIDE_BODY_RULES) {
updated.body_rules = defaults.body_rules;
}
if !metadata.overrides.contains(OVERRIDE_MAX_RETRIES) {
updated.max_retries = defaults.max_retries;
}
if !metadata.overrides.contains(OVERRIDE_IS_ACTIVE) {
updated.is_active = defaults.is_active;
}
if !metadata.overrides.contains(OVERRIDE_PROXY) {
updated.proxy = defaults.proxy;
}
if !metadata
.overrides
.contains(OVERRIDE_FORMAT_ACCEPTANCE_CONFIG)
{
updated.format_acceptance_config = defaults.format_acceptance_config;
}
let mut config = endpoint_config_without_metadata(updated.config.as_ref());
let current_config_defaults = fixed_provider_endpoint_config_defaults(endpoint_template);
let current_config_keys = current_config_defaults
.keys()
.cloned()
.collect::<BTreeSet<_>>();
for old_key in metadata.config_keys.difference(&current_config_keys) {
if !metadata
.overrides
.contains(config_override_key(old_key.as_str()).as_str())
{
config.remove(old_key);
}
}
for (key, value) in &current_config_defaults {
if !metadata
.overrides
.contains(config_override_key(key.as_str()).as_str())
{
config.insert(key.clone(), value.clone());
}
}
let mut next_metadata = metadata;
next_metadata.provider_type = template.provider_type.to_string();
next_metadata.item_key = endpoint_template.item_key.to_string();
next_metadata.version = template.version;
next_metadata.retired = false;
next_metadata.config_keys = current_config_keys;
updated.config = materialize_endpoint_config(config, &next_metadata);
if updated != *existing_endpoint {
updated.updated_at_unix_secs = Some(current_unix_secs());
}
Ok(updated)
}
fn resolve_fixed_provider_endpoint_template<'a>(
template: &'a FixedProviderTemplate,
existing_endpoint: &StoredProviderCatalogEndpoint,
updated_endpoint: &StoredProviderCatalogEndpoint,
) -> Option<&'a FixedProviderEndpointTemplate> {
if let Some(metadata) = fixed_provider_endpoint_metadata(existing_endpoint) {
if let Some(item) = template
.endpoints
.iter()
.find(|item| item.item_key == metadata.item_key)
{
return Some(item);
}
}
template.endpoints.iter().find(|item| {
item.api_format
.eq_ignore_ascii_case(updated_endpoint.api_format.trim())
|| item
.api_format
.eq_ignore_ascii_case(existing_endpoint.api_format.trim())
})
}
fn endpoint_matches_fixed_provider_template(
endpoint: &StoredProviderCatalogEndpoint,
endpoint_template: &FixedProviderEndpointTemplate,
) -> bool {
if let Some(metadata) = fixed_provider_endpoint_metadata(endpoint) {
if metadata.item_key == endpoint_template.item_key {
return true;
}
}
endpoint
.api_format
.trim()
.eq_ignore_ascii_case(endpoint_template.api_format)
}
fn fixed_provider_endpoint_metadata(
endpoint: &StoredProviderCatalogEndpoint,
) -> Option<FixedProviderEndpointMetadata> {
let config = endpoint.config.as_ref()?.as_object()?;
let metadata = config
.get(FIXED_PROVIDER_TEMPLATE_METADATA_KEY)?
.as_object()?;
let provider_type = metadata
.get("provider_type")
.and_then(Value::as_str)
.map(str::trim)
.filter(|value| !value.is_empty())?;
let item_key = metadata
.get("item_key")
.and_then(Value::as_str)
.map(str::trim)
.filter(|value| !value.is_empty())?;
if !metadata
.get("managed")
.and_then(Value::as_bool)
.unwrap_or(false)
{
return None;
}
Some(FixedProviderEndpointMetadata {
provider_type: provider_type.to_string(),
item_key: item_key.to_string(),
version: metadata
.get("version")
.and_then(Value::as_u64)
.and_then(|value| u32::try_from(value).ok())
.unwrap_or(0),
retired: metadata
.get("retired")
.and_then(Value::as_bool)
.unwrap_or(false),
overrides: metadata
.get("overrides")
.and_then(Value::as_array)
.map(|items| {
items
.iter()
.filter_map(Value::as_str)
.map(ToOwned::to_owned)
.collect()
})
.unwrap_or_default(),
config_keys: metadata
.get("config_keys")
.and_then(Value::as_array)
.map(|items| {
items
.iter()
.filter_map(Value::as_str)
.map(ToOwned::to_owned)
.collect()
})
.unwrap_or_default(),
})
}
fn managed_fixed_provider_endpoint_metadata(
template: &FixedProviderTemplate,
endpoint_template: &FixedProviderEndpointTemplate,
) -> FixedProviderEndpointMetadata {
FixedProviderEndpointMetadata {
provider_type: template.provider_type.to_string(),
item_key: endpoint_template.item_key.to_string(),
version: template.version,
retired: false,
overrides: BTreeSet::new(),
config_keys: fixed_provider_endpoint_config_defaults(endpoint_template)
.into_keys()
.collect(),
}
}
fn upsert_fixed_provider_endpoint_metadata(
endpoint: &mut StoredProviderCatalogEndpoint,
metadata: &FixedProviderEndpointMetadata,
) {
let config = endpoint_config_without_metadata(endpoint.config.as_ref());
endpoint.config = materialize_endpoint_config(config, metadata);
}
fn endpoint_config_without_metadata(config: Option<&Value>) -> Map<String, Value> {
let mut config = config
.and_then(Value::as_object)
.cloned()
.unwrap_or_default();
config.remove(FIXED_PROVIDER_TEMPLATE_METADATA_KEY);
config
}
fn materialize_endpoint_config(
mut config: Map<String, Value>,
metadata: &FixedProviderEndpointMetadata,
) -> Option<Value> {
config.insert(
FIXED_PROVIDER_TEMPLATE_METADATA_KEY.to_string(),
json!({
"managed": true,
"provider_type": metadata.provider_type,
"item_key": metadata.item_key,
"version": metadata.version,
"retired": metadata.retired,
"overrides": metadata.overrides.iter().cloned().collect::<Vec<_>>(),
"config_keys": metadata.config_keys.iter().cloned().collect::<Vec<_>>(),
}),
);
Some(Value::Object(config))
}
fn fixed_provider_endpoint_config_defaults(
endpoint_template: &FixedProviderEndpointTemplate,
) -> BTreeMap<String, Value> {
endpoint_template
.config_defaults
.iter()
.map(|item| (item.key.to_string(), item.value.to_json_value()))
.collect()
}
fn config_override_key(key: &str) -> String {
format!("config.{key}")
}
fn current_unix_secs() -> u64 {
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.ok()
.map(|duration| duration.as_secs())
.unwrap_or(0)
}
fn reconcile_fixed_provider_key(
provider: &StoredProviderCatalogProvider,
existing_key: &StoredProviderCatalogKey,
) -> Option<StoredProviderCatalogKey> {
if !provider_key_is_oauth_managed(existing_key, &provider.provider_type)
|| existing_key.api_formats.is_none()
{
return None;
}
let mut updated = existing_key.clone();
updated.api_formats = None;
updated.updated_at_unix_secs = Some(current_unix_secs());
Some(updated)
}
fn sync_override<T>(overrides: &mut BTreeSet<String>, key: &str, actual: &T, desired: &T)
where
T: PartialEq,
{
if actual == desired {
overrides.remove(key);
} else {
overrides.insert(key.to_string());
}
}
fn sync_override_if_changed<T>(
overrides: &mut BTreeSet<String>,
key: &str,
before: &T,
actual: &T,
desired: &T,
) where
T: PartialEq,
{
if before == actual {
return;
}
sync_override(overrides, key, actual, desired);
}

8
apps/aether-gateway/src/handlers/admin/request/provider/builders.rs
View File

@@ -36,12 +36,14 @@ impl<'a> AdminAppState<'a> {
&self,
key: &aether_data_contracts::repository::provider_catalog::StoredProviderCatalogKey,
provider_type: &str,
api_formats: &[String],
now_unix_secs: u64,
) -> serde_json::Value {
crate::handlers::admin::shared::build_admin_provider_key_response(
self.app,
key,
provider_type,
api_formats,
now_unix_secs,
)
}
@@ -274,6 +276,7 @@ impl<'a> AdminAppState<'a> {
String,
> {
use crate::api::ai::admin_endpoint_signature_parts;
use crate::handlers::admin::provider::write::provider::apply_admin_fixed_provider_endpoint_template_overrides;
use crate::handlers::public::{admin_requested_force_stream, normalize_admin_base_url};
use aether_admin::provider::endpoints as admin_provider_endpoints_pure;
let (fields, payload) = patch.into_parts();
@@ -346,6 +349,11 @@ impl<'a> AdminAppState<'a> {
.ok_or_else(|| format!("无效的 api_format: {}", updated.api_format))?;
updated.api_family = Some(api_family.to_string());
updated.endpoint_kind = Some(endpoint_kind.to_string());
apply_admin_fixed_provider_endpoint_template_overrides(
provider,
existing_endpoint,
&mut updated,
)?;
updated.updated_at_unix_secs = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.ok()

10
apps/aether-gateway/src/handlers/admin/request/provider/catalog.rs
View File

@@ -113,6 +113,16 @@ impl<'a> AdminAppState<'a> {
.await
}
pub(crate) async fn read_provider_quota_snapshots(
&self,
provider_ids: &[String],
) -> Result<
Vec<aether_data_contracts::repository::quota::StoredProviderQuotaSnapshot>,
GatewayError,
> {
self.app.read_provider_quota_snapshots(provider_ids).await
}
pub(crate) async fn update_provider_catalog_key_health_state(
&self,
key_id: &str,

6
apps/aether-gateway/src/handlers/admin/request/provider/oauth.rs
View File

@@ -465,6 +465,7 @@ impl<'a> AdminAppState<'a> {
pub(crate) async fn create_provider_oauth_catalog_key(
&self,
provider_id: &str,
provider_type: &str,
name: &str,
access_token: &str,
auth_config: &serde_json::Map<String, serde_json::Value>,
@@ -478,6 +479,7 @@ impl<'a> AdminAppState<'a> {
crate::handlers::admin::provider::oauth::provisioning::create_provider_oauth_catalog_key(
self,
provider_id,
provider_type,
name,
access_token,
auth_config,
@@ -491,8 +493,10 @@ impl<'a> AdminAppState<'a> {
pub(crate) async fn update_existing_provider_oauth_catalog_key(
&self,
existing_key: &aether_data_contracts::repository::provider_catalog::StoredProviderCatalogKey,
provider_type: &str,
access_token: &str,
auth_config: &serde_json::Map<String, serde_json::Value>,
api_formats: &[String],
proxy: Option<serde_json::Value>,
expires_at_unix_secs: Option<u64>,
) -> Result<
@@ -502,8 +506,10 @@ impl<'a> AdminAppState<'a> {
crate::handlers::admin::provider::oauth::provisioning::update_existing_provider_oauth_catalog_key(
self,
existing_key,
provider_type,
access_token,
auth_config,
api_formats,
proxy,
expires_at_unix_secs,
)

2
apps/aether-gateway/src/handlers/admin/request/provider/transport.rs
View File

@@ -94,7 +94,7 @@ impl<'a> AdminAppState<'a> {
pub(crate) fn fixed_provider_template(
&self,
provider_type: &str,
) -> Option<(&'static str, &'static [&'static str])> {
) -> Option<&'static crate::provider_transport::provider_types::FixedProviderTemplate> {
crate::provider_transport::provider_types::fixed_provider_template(provider_type)
}