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

fix(kiro,pool,model): 对齐 Kiro 管理链路并修复全局模型删除行为 (#305)

Browse Source 
* feat(pool): 号池支持跳过额度耗尽账号

- 新增 pool_advanced.skip_exhausted_accounts 开关及高级设置 UI, 默认关闭并兼容旧配置
- 为 Codex/Kiro 增加额度耗尽判定, 接入请求侧候选跳过并新增 account_quota_exhausted skip reason
- 号池列表将额度耗尽账号标记为 blocked/额度耗尽, 并补充前后端相关测试

* fix(kiro): 对齐账号管理与 provider-query 的 Rust 行为

- 修复 Kiro 单条导入误走 import-refresh-token 的前端分流, 并为误用路径返回明确错误提示
- 为 Kiro 导入与本地请求链补齐 bearer 兼容, 同步放开账号启停等 Key 更新操作的 auth_type 校验
- 实现 Kiro provider-query 本地模型测试与 failover 执行链, 并修复结果弹窗在无 trace 时无法展示 attempts/响应体的问题

* fix(model): 删除全局模型时级联清理关联提供商模型

- 对齐 Python 版本删除逻辑, GlobalModel 删除前先在事务内清理关联的 Provider Model 记录
- 修复已绑定 Provider 的模型在 Rust SQL 仓库下会被外键约束拦住、无法正常删除的问题
- 增加管理端回归测试, 覆盖绑定 Provider Model 的 GlobalModel 删除场景

* fix(kiro,ci): 恢复 Kiro OAuth 持久化并修复 Rust CI

* Fix oauth-managed provider key semantics

---------

Co-authored-by: fawney19 <elky0401@gmail.com>
This commit is contained in:
Entropy.Xu
2026-04-17 12:57:06 +08:00
committed by GitHub
parent 96a25d058b
commit ac1a126756
43 changed files with 2909 additions and 168 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -10,6 +10,7 @@ use super::shared::{
};
use crate::handlers::admin::provider::shared::paths::admin_provider_oauth_complete_key_id;
use crate::handlers::admin::request::{AdminAppState, AdminRequestContext};
use crate::provider_key_auth::provider_key_is_oauth_managed;
use crate::GatewayError;
use axum::{
body::{Body, Bytes},
@@ -76,12 +77,6 @@ pub(super) async fn handle_admin_provider_oauth_complete_key(
"Key 不存在",
));
};
if !key.auth_type.eq_ignore_ascii_case("oauth") {
return Ok(build_internal_control_error_response(
http::StatusCode::BAD_REQUEST,
"该 Key 不是 oauth 认证类型",
));
}
if !state_data.provider_id.trim().is_empty() && state_data.provider_id != key.provider_id {
return Ok(build_internal_control_error_response(
http::StatusCode::BAD_REQUEST,
@@ -102,6 +97,12 @@ pub(super) async fn handle_admin_provider_oauth_complete_key(
));
};
let provider_type = provider.provider_type.trim().to_ascii_lowercase();
if !provider_key_is_oauth_managed(&key, provider_type.as_str()) {
return Ok(build_internal_control_error_response(
http::StatusCode::BAD_REQUEST,
"该 Key 不是 OAuth 管理账号",
));
}
if !is_fixed_provider_type_for_provider_oauth(&provider_type) {
return Ok(build_internal_control_error_response(
http::StatusCode::BAD_REQUEST,

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

@@ -94,6 +94,12 @@ pub(super) async fn handle_admin_provider_oauth_import_refresh_token(
"该 Provider 不是固定类型，无法使用 provider-oauth",
));
}
if provider_type == "kiro" {
return Ok(build_internal_control_error_response(
http::StatusCode::BAD_REQUEST,
"Kiro 不支持单条 Refresh Token 导入，请使用批量导入或设备授权。",
));
}
let Some(template) = admin_provider_oauth_template(&provider_type) else {
return Ok(build_admin_provider_oauth_backend_unavailable_response());
};

14
apps/aether-gateway/src/handlers/admin/provider/oauth/dispatch/refresh/request.rs
View File

@@ -4,6 +4,7 @@ use super::helpers::{self, RefreshDispatch, RefreshRequestContext};
use super::response;
use crate::handlers::admin::provider::shared::paths::admin_provider_oauth_refresh_key_id;
use crate::handlers::admin::request::{AdminAppState, AdminRequestContext};
use crate::provider_key_auth::provider_key_is_oauth_managed;
use crate::GatewayError;
use axum::http;
@@ -28,13 +29,6 @@ pub(super) async fn parse_admin_provider_oauth_refresh_request(
"Key 不存在",
)));
};
if !key.auth_type.eq_ignore_ascii_case("oauth") {
return Ok(RefreshDispatch::Respond(response::control_error_response(
http::StatusCode::BAD_REQUEST,
"该 Key 不是 oauth 认证类型",
)));
}
let Some(encrypted_auth_config) = key.encrypted_auth_config.as_deref() else {
return Ok(RefreshDispatch::Respond(response::control_error_response(
http::StatusCode::BAD_REQUEST,
@@ -69,6 +63,12 @@ pub(super) async fn parse_admin_provider_oauth_refresh_request(
)));
};
let provider_type = provider.provider_type.trim().to_ascii_lowercase();
if !provider_key_is_oauth_managed(&key, provider_type.as_str()) {
return Ok(RefreshDispatch::Respond(response::control_error_response(
http::StatusCode::BAD_REQUEST,
"该 Key 不是 OAuth 管理账号",
)));
}
if !is_fixed_provider_type_for_provider_oauth(&provider_type) {
return Ok(RefreshDispatch::Respond(response::control_error_response(
http::StatusCode::BAD_REQUEST,

14
apps/aether-gateway/src/handlers/admin/provider/oauth/dispatch/start.rs
View File

@@ -8,6 +8,7 @@ use crate::handlers::admin::provider::shared::paths::{
admin_provider_oauth_start_key_id, admin_provider_oauth_start_provider_id,
};
use crate::handlers::admin::request::{AdminAppState, AdminRequestContext};
use crate::provider_key_auth::provider_key_is_oauth_managed;
use crate::GatewayError;
use axum::{
body::Body,
@@ -37,13 +38,6 @@ pub(super) async fn handle_admin_provider_oauth_start_key(
"Key 不存在",
));
};
if !key.auth_type.eq_ignore_ascii_case("oauth") {
return Ok(build_internal_control_error_response(
http::StatusCode::BAD_REQUEST,
"该 Key 不是 oauth 认证类型",
));
}
let provider_id = key.provider_id.clone();
let provider = state
.read_provider_catalog_providers_by_ids(std::slice::from_ref(&provider_id))
@@ -57,6 +51,12 @@ pub(super) async fn handle_admin_provider_oauth_start_key(
));
};
let provider_type = provider.provider_type.trim().to_ascii_lowercase();
if !provider_key_is_oauth_managed(&key, provider_type.as_str()) {
return Ok(build_internal_control_error_response(
http::StatusCode::BAD_REQUEST,
"该 Key 不是 OAuth 管理账号",
));
}
if !is_fixed_provider_type_for_provider_oauth(&provider_type) {
return Ok(build_internal_control_error_response(
http::StatusCode::BAD_REQUEST,

8
apps/aether-gateway/src/handlers/admin/provider/oauth/duplicates.rs
View File

@@ -1,4 +1,5 @@
use crate::handlers::admin::request::AdminAppState;
use crate::provider_key_auth::provider_key_is_oauth_managed;
use aether_data_contracts::repository::provider_catalog::StoredProviderCatalogKey;
fn normalize_codex_plan_group_for_provider_oauth(
@@ -141,8 +142,13 @@ pub(crate) async fn find_duplicate_provider_oauth_key(
.await
.map_err(|err| format!("{err:?}"))?;
let provider_type = auth_config
.get("provider_type")
.and_then(serde_json::Value::as_str)
.unwrap_or_default()
.to_string();
for existing_key in existing_keys.into_iter().filter(|key| {
key.auth_type.trim().eq_ignore_ascii_case("oauth")
provider_key_is_oauth_managed(key, provider_type.as_str())
&& exclude_key_id.is_none_or(|exclude| key.id != exclude)
}) {
let Some(existing_auth_config) = state.parse_catalog_auth_config_json(&existing_key) else {

12
apps/aether-gateway/src/handlers/admin/provider/oauth/quota/codex/mod.rs
View File

@@ -18,6 +18,7 @@ use super::shared::{
should_auto_remove_structured_reason, ProviderQuotaExecutionOutcome,
};
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,
@@ -55,11 +56,12 @@ pub(crate) async fn refresh_codex_provider_quota_locally(
}
};
let resolved_oauth_auth = if key.auth_type.trim().eq_ignore_ascii_case("oauth") {
state.resolve_local_oauth_header_auth(&transport).await?
} else {
None
};
let resolved_oauth_auth =
if provider_key_is_oauth_managed(&key, provider.provider_type.as_str()) {
state.resolve_local_oauth_header_auth(&transport).await?
} else {
None
};
let headers = match build_codex_refresh_headers(&transport, resolved_oauth_auth) {
Ok(headers) => headers,

4
apps/aether-gateway/src/handlers/admin/provider/oauth/runtime.rs
View File

@@ -2,6 +2,7 @@ use super::quota::antigravity::refresh_antigravity_provider_quota_locally;
use super::quota::codex::refresh_codex_provider_quota_locally;
use super::quota::kiro::refresh_kiro_provider_quota_locally;
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, StoredProviderCatalogProvider,
@@ -85,6 +86,9 @@ pub(crate) async fn refresh_provider_oauth_account_state_after_update(
else {
return Ok((false, None));
};
if provider_type == "kiro" && !provider_key_is_oauth_managed(&key, provider_type.as_str()) {
return Ok((false, None));
}
let payload = match provider_type.as_str() {
"codex" => {

59
apps/aether-gateway/src/handlers/admin/provider/pool/config.rs
View File

@@ -53,6 +53,7 @@ pub(crate) fn admin_provider_pool_config(
let Some(pool_advanced) = raw_pool_advanced.as_object() else {
return Some(AdminProviderPoolConfig {
lru_enabled: false,
skip_exhausted_accounts: false,
cost_window_seconds: 18_000,
cost_limit_per_key_tokens: None,
});
@@ -60,6 +61,10 @@ pub(crate) fn admin_provider_pool_config(
Some(AdminProviderPoolConfig {
lru_enabled: admin_provider_pool_lru_enabled(pool_advanced),
skip_exhausted_accounts: pool_advanced
.get("skip_exhausted_accounts")
.and_then(Value::as_bool)
.unwrap_or(false),
cost_window_seconds: pool_advanced
.get("cost_window_seconds")
.and_then(json_u64)
@@ -70,3 +75,57 @@ pub(crate) fn admin_provider_pool_config(
.and_then(json_u64),
})
}
#[cfg(test)]
mod tests {
use super::admin_provider_pool_config;
use aether_data_contracts::repository::provider_catalog::StoredProviderCatalogProvider;
use serde_json::json;
fn sample_provider(config: serde_json::Value) -> StoredProviderCatalogProvider {
StoredProviderCatalogProvider::new(
"provider-1".to_string(),
"provider-1".to_string(),
Some("https://example.com".to_string()),
"codex".to_string(),
)
.expect("provider should build")
.with_transport_fields(
true,
false,
false,
None,
None,
None,
None,
None,
Some(config),
)
}
#[test]
fn defaults_skip_exhausted_accounts_to_false() {
let provider = sample_provider(json!({ "pool_advanced": {} }));
let config = admin_provider_pool_config(&provider).expect("pool config should exist");
assert!(!config.skip_exhausted_accounts);
}
#[test]
fn parses_skip_exhausted_accounts_from_pool_advanced() {
let provider = sample_provider(json!({
"pool_advanced": {
"skip_exhausted_accounts": true,
"lru_enabled": true,
"cost_window_seconds": 7200,
"cost_limit_per_key_tokens": 12000
}
}));
let config = admin_provider_pool_config(&provider).expect("pool config should exist");
assert!(config.skip_exhausted_accounts);
assert!(config.lru_enabled);
assert_eq!(config.cost_window_seconds, 7200);
assert_eq!(config.cost_limit_per_key_tokens, Some(12_000));
}
}

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

@@ -3,6 +3,8 @@ 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 aether_admin::provider::pool as admin_provider_pool_pure;
use aether_data_contracts::repository::provider_catalog::StoredProviderCatalogKey;
use serde_json::json;
@@ -122,10 +124,11 @@ fn admin_pool_normalize_oauth_plan_type(value: &str, provider_type: &str) -> Opt
}
fn admin_pool_derive_oauth_expires_at(
provider_type: &str,
key: &StoredProviderCatalogKey,
auth_config: Option<&serde_json::Map<String, serde_json::Value>>,
) -> Option<u64> {
if !key.auth_type.trim().eq_ignore_ascii_case("oauth") {
if !provider_key_auth_semantics(key, provider_type).oauth_managed() {
return None;
}
@@ -144,7 +147,7 @@ fn admin_pool_derive_oauth_plan_type(
provider_type: &str,
auth_config: Option<&serde_json::Map<String, serde_json::Value>>,
) -> Option<String> {
if !key.auth_type.trim().eq_ignore_ascii_case("oauth") {
if !provider_key_auth_semantics(key, provider_type).oauth_managed() {
return None;
}
@@ -551,6 +554,7 @@ fn admin_pool_scheduling_payload(
cooldown_ttl_seconds: Option<u64>,
health_score: f64,
circuit_breaker_open: bool,
account_quota_exhausted: bool,
) -> (String, String, String, Vec<serde_json::Value>) {
if !key.is_active {
return (
@@ -567,6 +571,21 @@ fn admin_pool_scheduling_payload(
})],
);
}
if account_quota_exhausted {
return (
"blocked".to_string(),
"account_quota_exhausted".to_string(),
"额度耗尽".to_string(),
vec![json!({
"code": "account_quota_exhausted",
"label": "额度耗尽",
"blocking": true,
"source": "quota",
"ttl_seconds": serde_json::Value::Null,
"detail": serde_json::Value::Null,
})],
);
}
if let Some(reason) = cooldown_reason {
return (
"degraded".to_string(),
@@ -632,6 +651,9 @@ pub(super) fn build_admin_pool_key_payload(
.and_then(|_| runtime.cooldown_ttl_by_key.get(&key.id).copied());
let health_score = admin_pool_health_score(key);
let circuit_breaker_open = admin_pool_circuit_breaker_open(key);
let auth_semantics = provider_key_auth_semantics(key, provider_type);
let account_quota_exhausted = pool_config.is_some_and(|config| config.skip_exhausted_accounts)
&& admin_provider_pool_pure::admin_pool_key_account_quota_exhausted(key, provider_type);
let (scheduling_status, scheduling_reason, scheduling_label, scheduling_reasons) =
admin_pool_scheduling_payload(
key,
@@ -639,9 +661,11 @@ pub(super) fn build_admin_pool_key_payload(
cooldown_ttl_seconds,
health_score,
circuit_breaker_open,
account_quota_exhausted,
);
let auth_config = state.parse_catalog_auth_config_json(key);
let oauth_expires_at = admin_pool_derive_oauth_expires_at(key, auth_config.as_ref());
let oauth_expires_at =
admin_pool_derive_oauth_expires_at(provider_type, key, auth_config.as_ref());
let oauth_plan_type =
admin_pool_derive_oauth_plan_type(key, provider_type, auth_config.as_ref());
let status_snapshot = provider_key_status_snapshot_payload(key);
@@ -656,15 +680,29 @@ pub(super) fn build_admin_pool_key_payload(
.and_then(serde_json::Value::as_object);
let quota_updated_at =
admin_pool_json_to_u64(quota_snapshot.and_then(|item| item.get("updated_at")));
let oauth_invalid_at =
let oauth_invalid_at = if auth_semantics.can_show_oauth_metadata() {
admin_pool_json_to_u64(oauth_snapshot.and_then(|item| item.get("invalid_at")))
.or(key.oauth_invalid_at_unix_secs);
let oauth_account_id = admin_pool_trimmed_string_from_map(auth_config.as_ref(), "account_id");
let oauth_account_name =
admin_pool_trimmed_string_from_map(auth_config.as_ref(), "account_name");
let oauth_account_user_id =
admin_pool_trimmed_string_from_map(auth_config.as_ref(), "account_user_id");
let oauth_organizations = admin_pool_oauth_organizations(auth_config.as_ref());
.or(key.oauth_invalid_at_unix_secs)
} else {
None
};
let oauth_account_id = auth_semantics
.can_show_oauth_metadata()
.then(|| admin_pool_trimmed_string_from_map(auth_config.as_ref(), "account_id"))
.flatten();
let oauth_account_name = auth_semantics
.can_show_oauth_metadata()
.then(|| admin_pool_trimmed_string_from_map(auth_config.as_ref(), "account_name"))
.flatten();
let oauth_account_user_id = auth_semantics
.can_show_oauth_metadata()
.then(|| admin_pool_trimmed_string_from_map(auth_config.as_ref(), "account_user_id"))
.flatten();
let oauth_organizations = if auth_semantics.can_show_oauth_metadata() {
admin_pool_oauth_organizations(auth_config.as_ref())
} else {
Vec::new()
};
let account_status_code = admin_pool_trimmed_string_from_map(account_snapshot, "code");
let account_status_label =
admin_pool_trimmed_string(account_snapshot.and_then(|item| item.get("label")));
@@ -686,11 +724,38 @@ pub(super) fn build_admin_pool_key_payload(
payload.insert("key_name".to_string(), json!(key.name));
payload.insert("is_active".to_string(), json!(key.is_active));
payload.insert("auth_type".to_string(), json!(key.auth_type));
payload.insert(
"credential_kind".to_string(),
json!(auth_semantics.credential_kind().as_str()),
);
payload.insert(
"runtime_auth_kind".to_string(),
json!(auth_semantics.runtime_auth_kind().as_str()),
);
payload.insert(
"oauth_managed".to_string(),
json!(auth_semantics.oauth_managed()),
);
payload.insert(
"can_refresh_oauth".to_string(),
json!(auth_semantics.can_refresh_oauth()),
);
payload.insert(
"can_export_oauth".to_string(),
json!(auth_semantics.can_export_oauth()),
);
payload.insert(
"can_edit_oauth".to_string(),
json!(auth_semantics.can_edit_oauth()),
);
payload.insert("oauth_expires_at".to_string(), json!(oauth_expires_at));
payload.insert("oauth_invalid_at".to_string(), json!(oauth_invalid_at));
payload.insert(
"oauth_invalid_reason".to_string(),
json!(key.oauth_invalid_reason),
json!(auth_semantics
.can_show_oauth_metadata()
.then_some(key.oauth_invalid_reason.clone())
.flatten()),
);
payload.insert("oauth_plan_type".to_string(), json!(oauth_plan_type));
payload.insert("oauth_account_id".to_string(), json!(oauth_account_id));

26
apps/aether-gateway/src/handlers/admin/provider/pool_admin/read_routes/resolve_selection.rs
View File

@@ -3,6 +3,7 @@ use super::{
AdminPoolResolveSelectionRequest, ADMIN_POOL_PROVIDER_CATALOG_READER_UNAVAILABLE_DETAIL,
};
use crate::handlers::admin::request::{AdminAppState, AdminRequestContext};
use crate::provider_key_auth::provider_key_auth_semantics;
use crate::GatewayError;
use aether_admin::provider::pool as admin_provider_pool_pure;
use axum::{
@@ -11,6 +12,7 @@ use axum::{
response::{IntoResponse, Response},
Json,
};
use serde_json::json;
pub(super) async fn build_admin_pool_resolve_selection_response(
state: &AdminAppState<'_>,
@@ -88,5 +90,27 @@ pub(super) async fn build_admin_pool_resolve_selection_response(
.then_with(|| left.name.cmp(&right.name))
});
Ok(Json(admin_provider_pool_pure::build_admin_pool_selection_payload(&keys)).into_response())
let items = keys
.iter()
.map(|key| {
let auth_semantics = provider_key_auth_semantics(key, &provider_type);
json!({
"key_id": key.id,
"key_name": key.name,
"auth_type": key.auth_type,
"credential_kind": auth_semantics.credential_kind().as_str(),
"runtime_auth_kind": auth_semantics.runtime_auth_kind().as_str(),
"oauth_managed": auth_semantics.oauth_managed(),
"can_refresh_oauth": auth_semantics.can_refresh_oauth(),
"can_export_oauth": auth_semantics.can_export_oauth(),
"can_edit_oauth": auth_semantics.can_edit_oauth(),
})
})
.collect::<Vec<_>>();
Ok(Json(json!({
"total": items.len(),
"items": items,
}))
.into_response())
}

3
apps/aether-gateway/src/handlers/admin/provider/pool_admin/selection.rs
View File

@@ -1,4 +1,5 @@
use crate::handlers::admin::request::AdminAppState;
use crate::provider_key_auth::provider_key_is_oauth_managed;
use aether_admin::provider::pool as admin_provider_pool_pure;
use aether_data_contracts::repository::provider_catalog::StoredProviderCatalogKey;
@@ -44,7 +45,7 @@ fn admin_pool_derive_oauth_plan_type(
}
};
if key.auth_type.trim() != "oauth" {
if !provider_key_is_oauth_managed(key, provider_type) {
return None;
}

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

@@ -1,17 +1,29 @@
use super::payload::{
provider_query_extract_api_key_id, provider_query_extract_force_refresh,
provider_query_extract_provider_id,
provider_query_extract_model, provider_query_extract_provider_id,
provider_query_extract_request_id,
};
use super::response::{
build_admin_provider_query_bad_request_response, build_admin_provider_query_not_found_response,
ADMIN_PROVIDER_QUERY_API_KEY_NOT_FOUND_DETAIL, ADMIN_PROVIDER_QUERY_NO_ACTIVE_API_KEY_DETAIL,
ADMIN_PROVIDER_QUERY_API_KEY_NOT_FOUND_DETAIL, ADMIN_PROVIDER_QUERY_MODEL_REQUIRED_DETAIL,
ADMIN_PROVIDER_QUERY_NO_ACTIVE_API_KEY_DETAIL, ADMIN_PROVIDER_QUERY_NO_LOCAL_MODELS_DETAIL,
ADMIN_PROVIDER_QUERY_PROVIDER_ID_REQUIRED_DETAIL,
ADMIN_PROVIDER_QUERY_PROVIDER_NOT_FOUND_DETAIL,
};
use crate::ai_pipeline::{maybe_build_sync_finalize_outcome, GatewayControlDecision};
use crate::execution_runtime;
use crate::handlers::admin::request::AdminAppState;
use crate::model_fetch::ModelFetchRuntimeState;
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,
KiroProviderHeadersInput, KIRO_ENVELOPE_NAME,
};
use crate::usage::GatewaySyncReportRequest;
use crate::{AppState, GatewayError};
use aether_admin::provider::pool as admin_provider_pool_pure;
use aether_contracts::{ExecutionPlan, RequestBody};
use aether_data_contracts::repository::global_models::AdminProviderModelListQuery;
use aether_data_contracts::repository::provider_catalog::{
StoredProviderCatalogEndpoint, StoredProviderCatalogKey, StoredProviderCatalogProvider,
};
@@ -19,9 +31,16 @@ use aether_model_fetch::{
aggregate_models_for_cache, fetch_models_from_transports, json_string_list,
preset_models_for_provider, selected_models_fetch_endpoints,
};
use axum::{body::Body, http::Response, response::IntoResponse, Json};
use axum::{
body::{to_bytes, Body},
http::{HeaderMap, HeaderName, HeaderValue},
response::{IntoResponse, Response},
Json,
};
use base64::Engine as _;
use serde_json::{json, Value};
use std::collections::BTreeSet;
use std::collections::{BTreeMap, BTreeSet};
use uuid::Uuid;
pub(crate) const ADMIN_PROVIDER_QUERY_LOCAL_TEST_MODEL_MESSAGE: &str =
"Rust local provider-query model test is not configured";
@@ -32,7 +51,10 @@ const ADMIN_PROVIDER_QUERY_NO_ACTIVE_ENDPOINT_DETAIL: &str =
const ADMIN_PROVIDER_QUERY_NO_MODELS_FROM_ENDPOINT_DETAIL: &str =
"No models returned from any endpoint";
const ADMIN_PROVIDER_QUERY_NO_MODELS_FROM_KEY_DETAIL: &str = "No models returned from any key";
const ADMIN_PROVIDER_QUERY_NO_ACTIVE_TEST_CANDIDATE_DETAIL: &str =
"No active endpoint or API key found";
const ANTIGRAVITY_PROVIDER_CACHE_KEY_PREFIX: &str = "upstream_models_provider:";
const DEFAULT_PROVIDER_QUERY_TEST_MESSAGE: &str = "Hello! This is a test message.";
#[derive(Debug)]
struct ProviderQueryKeyFetchResult {
@@ -42,6 +64,47 @@ struct ProviderQueryKeyFetchResult {
has_success: bool,
}
#[derive(Debug, Clone)]
struct ProviderQueryTestCandidate {
endpoint: StoredProviderCatalogEndpoint,
key: StoredProviderCatalogKey,
effective_model: String,
}
#[derive(Debug, Clone)]
struct ProviderQueryTestAttempt {
candidate_index: usize,
endpoint_api_format: String,
endpoint_base_url: String,
key_name: String,
key_id: String,
auth_type: String,
effective_model: String,
status: &'static str,
skip_reason: Option<String>,
error_message: Option<String>,
status_code: Option<u16>,
latency_ms: Option<u64>,
request_url: Option<String>,
request_headers: Option<BTreeMap<String, String>>,
request_body: Option<Value>,
response_headers: Option<BTreeMap<String, String>>,
response_body: Option<Value>,
}
#[derive(Debug, Clone)]
struct ProviderQueryExecutionOutcome {
status: &'static str,
error_message: Option<String>,
status_code: Option<u16>,
latency_ms: Option<u64>,
request_url: String,
request_headers: BTreeMap<String, String>,
request_body: Value,
response_headers: BTreeMap<String, String>,
response_body: Option<Value>,
}
fn provider_query_provider_payload(provider: &StoredProviderCatalogProvider) -> Value {
json!({
"id": provider.id.clone(),
@@ -50,6 +113,753 @@ fn provider_query_provider_payload(provider: &StoredProviderCatalogProvider) ->
})
}
fn provider_query_test_mode(payload: &Value) -> &str {
payload
.get("mode")
.and_then(Value::as_str)
.map(str::trim)
.filter(|value| !value.is_empty())
.unwrap_or("global")
}
fn provider_query_extract_endpoint_id(payload: &Value) -> Option<String> {
payload
.get("endpoint_id")
.and_then(Value::as_str)
.map(str::trim)
.filter(|value| !value.is_empty())
.map(ToOwned::to_owned)
}
fn provider_query_extract_api_format(payload: &Value) -> Option<String> {
payload
.get("api_format")
.and_then(Value::as_str)
.map(str::trim)
.filter(|value| !value.is_empty())
.map(ToOwned::to_owned)
}
fn provider_query_extract_message(payload: &Value) -> Option<String> {
payload
.get("message")
.and_then(Value::as_str)
.map(str::trim)
.filter(|value| !value.is_empty())
.map(ToOwned::to_owned)
}
fn provider_query_extract_request_body(payload: &Value) -> Option<Value> {
payload
.get("request_body")
.filter(|value| value.is_object())
.cloned()
}
fn provider_query_extract_request_headers(payload: &Value) -> HeaderMap {
let mut headers = HeaderMap::new();
let Some(values) = payload.get("request_headers").and_then(Value::as_object) else {
return headers;
};
for (key, value) in values {
let key = key.trim();
if key.is_empty() {
continue;
}
let Some(value) = (match value {
Value::String(value) => Some(value.trim().to_string()),
Value::Bool(value) => Some(value.to_string()),
Value::Number(value) => Some(value.to_string()),
other => serde_json::to_string(other).ok(),
}) else {
continue;
};
if value.is_empty() {
continue;
}
let Ok(name) = HeaderName::from_bytes(key.as_bytes()) else {
continue;
};
let Ok(value) = HeaderValue::from_str(&value) else {
continue;
};
headers.insert(name, value);
}
headers
}
fn provider_query_build_test_request_body(payload: &Value, model: &str) -> Value {
if let Some(mut body) = provider_query_extract_request_body(payload) {
if let Some(object) = body.as_object_mut() {
object.insert("model".to_string(), Value::String(model.to_string()));
}
return body;
}
json!({
"model": model,
"messages": [{
"role": "user",
"content": provider_query_extract_message(payload)
.unwrap_or_else(|| DEFAULT_PROVIDER_QUERY_TEST_MESSAGE.to_string())
}],
"max_tokens": 30,
"temperature": 0.7,
"stream": true,
})
}
fn provider_query_select_kiro_endpoint<'a>(
endpoints: &'a [StoredProviderCatalogEndpoint],
endpoint_id: Option<&str>,
api_format: Option<&str>,
) -> Result<Option<&'a StoredProviderCatalogEndpoint>, &'static str> {
if let Some(endpoint_id) = endpoint_id {
let endpoint = endpoints.iter().find(|endpoint| endpoint.id == endpoint_id);
return endpoint
.ok_or("Endpoint not found")
.map(|endpoint| Some(endpoint));
}
if let Some(api_format) = api_format {
let endpoint = endpoints.iter().find(|endpoint| {
endpoint.is_active && endpoint.api_format.trim().eq_ignore_ascii_case(api_format)
});
return Ok(endpoint);
}
Ok(endpoints.iter().find(|endpoint| endpoint.is_active))
}
fn provider_query_key_supports_endpoint(
key: &StoredProviderCatalogKey,
endpoint_api_format: &str,
) -> bool {
let formats = json_string_list(key.api_formats.as_ref());
formats.is_empty()
|| formats
.iter()
.any(|value| value.eq_ignore_ascii_case(endpoint_api_format))
}
fn provider_query_test_key_sort_key(
provider_type: &str,
key: &StoredProviderCatalogKey,
endpoint_api_format: &str,
) -> (u8, u8, i32, u64, i32) {
let quota_exhausted =
admin_provider_pool_pure::admin_pool_key_account_quota_exhausted(key, provider_type);
let circuit_open = key
.circuit_breaker_by_format
.as_ref()
.and_then(Value::as_object)
.and_then(|value| value.get(endpoint_api_format))
.and_then(Value::as_object)
.and_then(|value| value.get("open"))
.and_then(Value::as_bool)
.unwrap_or(false);
let health_score = key
.health_by_format
.as_ref()
.and_then(Value::as_object)
.and_then(|value| value.get(endpoint_api_format))
.and_then(Value::as_object)
.and_then(|value| value.get("health_score"))
.and_then(Value::as_f64)
.unwrap_or(1.0);
let consecutive_failures = key
.health_by_format
.as_ref()
.and_then(Value::as_object)
.and_then(|value| value.get(endpoint_api_format))
.and_then(Value::as_object)
.and_then(|value| value.get("consecutive_failures"))
.and_then(Value::as_u64)
.unwrap_or(0);
let normalized_health = (health_score.clamp(0.0, 1.0) * 1000.0).round() as i32;
(
if quota_exhausted { 1 } else { 0 },
if circuit_open { 1 } else { 0 },
-normalized_health,
consecutive_failures,
key.internal_priority,
)
}
async fn provider_query_resolve_global_effective_model(
state: &AdminAppState<'_>,
provider_id: &str,
requested_model: &str,
) -> Result<String, GatewayError> {
let models = state
.list_admin_provider_models(&AdminProviderModelListQuery {
provider_id: provider_id.to_string(),
is_active: Some(true),
offset: 0,
limit: 1024,
})
.await
.unwrap_or_default();
Ok(models
.into_iter()
.find(|model| {
model.is_available
&& model
.global_model_name
.as_deref()
.is_some_and(|value| value.eq_ignore_ascii_case(requested_model))
})
.map(|model| model.provider_model_name)
.unwrap_or_else(|| requested_model.to_string()))
}
async fn provider_query_build_kiro_test_candidates(
state: &AdminAppState<'_>,
provider: &StoredProviderCatalogProvider,
payload: &Value,
) -> Result<Vec<ProviderQueryTestCandidate>, Response<Body>> {
let provider_ids = vec![provider.id.clone()];
let endpoints = state
.app()
.list_provider_catalog_endpoints_by_provider_ids(&provider_ids)
.await
.map_err(|_| {
build_admin_provider_query_bad_request_response(
ADMIN_PROVIDER_QUERY_NO_ACTIVE_API_KEY_DETAIL,
)
})?;
let endpoint = match provider_query_select_kiro_endpoint(
&endpoints,
provider_query_extract_endpoint_id(payload).as_deref(),
provider_query_extract_api_format(payload).as_deref(),
) {
Ok(Some(endpoint)) => endpoint.clone(),
Ok(None) => {
return Err(build_admin_provider_query_not_found_response(
ADMIN_PROVIDER_QUERY_NO_ACTIVE_API_KEY_DETAIL,
));
}
Err("Endpoint not found") => {
return Err(build_admin_provider_query_not_found_response(
ADMIN_PROVIDER_QUERY_API_KEY_NOT_FOUND_DETAIL,
));
}
Err(_) => {
return Err(build_admin_provider_query_not_found_response(
ADMIN_PROVIDER_QUERY_NO_ACTIVE_API_KEY_DETAIL,
));
}
};
let all_keys = state
.app()
.list_provider_catalog_keys_by_provider_ids(&provider_ids)
.await
.map_err(|_| {
build_admin_provider_query_bad_request_response(
ADMIN_PROVIDER_QUERY_NO_ACTIVE_API_KEY_DETAIL,
)
})?;
let selected_key_id = provider_query_extract_api_key_id(payload);
if let Some(api_key_id) = selected_key_id.as_deref() {
let Some(key) = all_keys.iter().find(|key| key.id == api_key_id) else {
return Err(build_admin_provider_query_not_found_response(
ADMIN_PROVIDER_QUERY_API_KEY_NOT_FOUND_DETAIL,
));
};
if !key.is_active || !provider_query_key_supports_endpoint(key, &endpoint.api_format) {
return Err(build_admin_provider_query_not_found_response(
ADMIN_PROVIDER_QUERY_NO_ACTIVE_TEST_CANDIDATE_DETAIL,
));
}
}
let requested_model = provider_query_extract_model(payload).ok_or_else(|| {
build_admin_provider_query_bad_request_response(ADMIN_PROVIDER_QUERY_MODEL_REQUIRED_DETAIL)
})?;
let effective_model = if provider_query_test_mode(payload).eq_ignore_ascii_case("direct") {
requested_model.clone()
} else {
provider_query_resolve_global_effective_model(state, &provider.id, &requested_model)
.await
.unwrap_or(requested_model.clone())
};
let mut keys = all_keys
.into_iter()
.filter(|key| key.is_active)
.filter(|key| {
selected_key_id
.as_deref()
.is_none_or(|value| value == key.id.as_str())
})
.filter(|key| provider_query_key_supports_endpoint(key, &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)
});
let candidates = keys
.into_iter()
.map(|key| ProviderQueryTestCandidate {
endpoint: endpoint.clone(),
key,
effective_model: effective_model.clone(),
})
.collect::<Vec<_>>();
if candidates.is_empty() {
return Err(build_admin_provider_query_not_found_response(
ADMIN_PROVIDER_QUERY_NO_ACTIVE_TEST_CANDIDATE_DETAIL,
));
}
Ok(candidates)
}
fn provider_query_decode_execution_body(
result: &aether_contracts::ExecutionResult,
) -> Option<Vec<u8>> {
result
.body
.as_ref()
.and_then(|body| body.body_bytes_b64.as_deref())
.and_then(|value| base64::engine::general_purpose::STANDARD.decode(value).ok())
}
fn provider_query_extract_error_message(
result: &aether_contracts::ExecutionResult,
) -> Option<String> {
result
.body
.as_ref()
.and_then(|body| body.json_body.as_ref())
.and_then(Value::as_object)
.and_then(|value| {
value
.get("error")
.and_then(Value::as_object)
.and_then(|error| error.get("message"))
.and_then(Value::as_str)
.or_else(|| value.get("message").and_then(Value::as_str))
})
.map(str::trim)
.filter(|value| !value.is_empty())
.map(ToOwned::to_owned)
.or_else(|| {
provider_query_decode_execution_body(result)
.and_then(|bytes| String::from_utf8(bytes).ok())
.map(|value| value.trim().to_string())
.filter(|value| !value.is_empty())
})
.or_else(|| {
result
.error
.as_ref()
.map(|error| error.message.trim().to_string())
.filter(|value| !value.is_empty())
})
}
async fn provider_query_finalize_kiro_result(
route_path: &str,
trace_id: &str,
requested_model: &str,
endpoint_api_format: &str,
effective_model: &str,
original_request_body: &Value,
result: &aether_contracts::ExecutionResult,
) -> Result<Option<Value>, GatewayError> {
let decision = GatewayControlDecision::synthetic(
route_path,
Some("admin_proxy".to_string()),
Some("provider_query_manage".to_string()),
Some("test_model_failover".to_string()),
Some(endpoint_api_format.to_string()),
);
let payload = GatewaySyncReportRequest {
trace_id: trace_id.to_string(),
report_kind: "claude_cli_sync_finalize".to_string(),
report_context: Some(json!({
"client_api_format": endpoint_api_format,
"provider_api_format": endpoint_api_format,
"model": requested_model,
"mapped_model": effective_model,
"needs_conversion": false,
"has_envelope": true,
"envelope_name": KIRO_ENVELOPE_NAME,
"original_request_body": original_request_body,
})),
status_code: result.status_code,
headers: result.headers.clone(),
body_json: result.body.as_ref().and_then(|body| body.json_body.clone()),
client_body_json: None,
body_base64: result
.body
.as_ref()
.and_then(|body| body.body_bytes_b64.clone()),
telemetry: result.telemetry.clone(),
};
let Some(outcome) = maybe_build_sync_finalize_outcome(trace_id, &decision, &payload)? else {
return Ok(None);
};
let bytes = to_bytes(outcome.response.into_body(), usize::MAX)
.await
.map_err(|err| GatewayError::Internal(err.to_string()))?;
serde_json::from_slice::<Value>(&bytes)
.map(Some)
.map_err(|err| GatewayError::Internal(err.to_string()))
}
async fn provider_query_execute_kiro_test_candidate(
state: &AdminAppState<'_>,
provider: &StoredProviderCatalogProvider,
candidate: &ProviderQueryTestCandidate,
payload: &Value,
route_path: &str,
trace_id: &str,
requested_model: &str,
) -> Result<ProviderQueryExecutionOutcome, GatewayError> {
let Some(transport) = state
.read_provider_transport_snapshot(&provider.id, &candidate.endpoint.id, &candidate.key.id)
.await?
else {
return Ok(ProviderQueryExecutionOutcome {
status: "skipped",
error_message: None,
status_code: None,
latency_ms: None,
request_url: String::new(),
request_headers: BTreeMap::new(),
request_body: Value::Null,
response_headers: BTreeMap::new(),
response_body: None,
});
};
if !supports_local_kiro_request_transport_with_network(&transport) {
return Ok(ProviderQueryExecutionOutcome {
status: "skipped",
error_message: None,
status_code: None,
latency_ms: None,
request_url: String::new(),
request_headers: BTreeMap::new(),
request_body: Value::Null,
response_headers: BTreeMap::new(),
response_body: None,
});
}
let Some(kiro_auth) = state
.resolve_local_oauth_kiro_request_auth(&transport)
.await?
else {
return Ok(ProviderQueryExecutionOutcome {
status: "failed",
error_message: Some("oauth auth failed".to_string()),
status_code: None,
latency_ms: None,
request_url: String::new(),
request_headers: BTreeMap::new(),
request_body: Value::Null,
response_headers: BTreeMap::new(),
response_body: None,
});
};
let request_body = provider_query_build_test_request_body(payload, &candidate.effective_model);
let provider_request_body = match build_kiro_provider_request_body(
&request_body,
&candidate.effective_model,
&kiro_auth.auth_config,
transport.endpoint.body_rules.as_ref(),
) {
Some(body) => body,
None => {
return Ok(ProviderQueryExecutionOutcome {
status: "failed",
error_message: Some("provider request body build failed".to_string()),
status_code: None,
latency_ms: None,
request_url: String::new(),
request_headers: BTreeMap::new(),
request_body,
response_headers: BTreeMap::new(),
response_body: None,
});
}
};
let mut synthetic_request = http::Request::builder()
.uri(route_path)
.body(())
.map_err(|err| GatewayError::Internal(err.to_string()))?;
*synthetic_request.headers_mut() = provider_query_extract_request_headers(payload);
let (parts, _) = synthetic_request.into_parts();
let request_url = build_kiro_generate_assistant_response_url(
&transport.endpoint.base_url,
parts.uri.query(),
Some(kiro_auth.auth_config.effective_api_region()),
)
.ok_or_else(|| GatewayError::Internal("kiro request url is unavailable".to_string()))?;
let request_headers = build_kiro_provider_headers(KiroProviderHeadersInput {
headers: &parts.headers,
provider_request_body: &provider_request_body,
original_request_body: &request_body,
header_rules: transport.endpoint.header_rules.as_ref(),
auth_header: kiro_auth.name,
auth_value: &kiro_auth.value,
auth_config: &kiro_auth.auth_config,
machine_id: kiro_auth.machine_id.as_str(),
})
.ok_or_else(|| GatewayError::Internal("kiro request headers are unavailable".to_string()))?;
let plan = ExecutionPlan {
request_id: trace_id.to_string(),
candidate_id: Some(format!("provider-query-{}", candidate.key.id)),
provider_name: Some(provider.name.clone()),
provider_id: provider.id.clone(),
endpoint_id: candidate.endpoint.id.clone(),
key_id: candidate.key.id.clone(),
method: "POST".to_string(),
url: request_url.clone(),
headers: request_headers.clone(),
content_type: Some("application/json".to_string()),
content_encoding: None,
body: RequestBody::from_json(provider_request_body.clone()),
stream: true,
client_api_format: candidate.endpoint.api_format.clone(),
provider_api_format: candidate.endpoint.api_format.clone(),
model_name: Some(candidate.effective_model.clone()),
proxy: state
.resolve_transport_proxy_snapshot_with_tunnel_affinity(&transport)
.await,
tls_profile: state.resolve_transport_tls_profile(&transport),
timeouts: state.resolve_transport_execution_timeouts(&transport),
};
let result = state
.execute_execution_runtime_sync_plan(Some(trace_id), &plan)
.await?;
let response_body = if result.status_code < 400 {
provider_query_finalize_kiro_result(
route_path,
trace_id,
requested_model,
candidate.endpoint.api_format.as_str(),
candidate.effective_model.as_str(),
&request_body,
&result,
)
.await?
} else {
result.body.as_ref().and_then(|body| body.json_body.clone())
};
let error_message = if result.status_code >= 400 {
provider_query_extract_error_message(&result)
} else if response_body.is_none()
&& provider_query_decode_execution_body(&result)
.is_some_and(|body| crate::ai_pipeline::stream_body_contains_error_event(&body))
{
Some("Kiro upstream returned embedded stream error".to_string())
} else {
None
};
Ok(ProviderQueryExecutionOutcome {
status: if error_message.is_some() {
"failed"
} else {
"success"
},
error_message,
status_code: Some(result.status_code),
latency_ms: result.telemetry.as_ref().and_then(|value| value.elapsed_ms),
request_url,
request_headers,
request_body: provider_request_body,
response_headers: result.headers,
response_body,
})
}
fn provider_query_test_attempt_payload(
candidate_index: usize,
candidate: &ProviderQueryTestCandidate,
execution: &ProviderQueryExecutionOutcome,
) -> Value {
json!({
"candidate_index": candidate_index,
"retry_index": 0,
"endpoint_api_format": candidate.endpoint.api_format,
"endpoint_base_url": candidate.endpoint.base_url,
"key_name": provider_query_key_display_name(&candidate.key),
"key_id": candidate.key.id,
"auth_type": candidate.key.auth_type,
"effective_model": candidate.effective_model,
"status": execution.status,
"skip_reason": Value::Null,
"error_message": execution.error_message,
"status_code": execution.status_code,
"latency_ms": execution.latency_ms,
"request_url": execution.request_url,
"request_headers": execution.request_headers,
"request_body": execution.request_body,
"response_headers": execution.response_headers,
"response_body": execution.response_body,
})
}
async fn build_admin_provider_query_kiro_failover_response(
state: &AdminAppState<'_>,
payload: &Value,
route_path: &str,
) -> Result<Response<Body>, GatewayError> {
let Some(provider_id) = provider_query_extract_provider_id(payload) else {
return Ok(build_admin_provider_query_bad_request_response(
ADMIN_PROVIDER_QUERY_PROVIDER_ID_REQUIRED_DETAIL,
));
};
let requested_model = match provider_query_extract_model(payload) {
Some(model) => model,
None => {
return Ok(build_admin_provider_query_bad_request_response(
ADMIN_PROVIDER_QUERY_MODEL_REQUIRED_DETAIL,
));
}
};
let Some(provider) = state
.app()
.read_provider_catalog_providers_by_ids(std::slice::from_ref(&provider_id))
.await?
.into_iter()
.find(|item| item.id == provider_id)
else {
return Ok(build_admin_provider_query_not_found_response(
ADMIN_PROVIDER_QUERY_PROVIDER_NOT_FOUND_DETAIL,
));
};
if !provider.provider_type.trim().eq_ignore_ascii_case("kiro") {
return Ok(build_admin_provider_query_test_model_failover_response(
provider_id,
super::payload::provider_query_extract_failover_models(payload),
));
}
let candidates =
match provider_query_build_kiro_test_candidates(state, &provider, payload).await {
Ok(candidates) => candidates,
Err(response) => return Ok(response),
};
let trace_id = provider_query_extract_request_id(payload)
.unwrap_or_else(|| format!("provider-query-test-{}", Uuid::new_v4().simple()));
let mut attempts = Vec::new();
let mut total_attempts = 0usize;
let mut success_body = None;
for (candidate_index, candidate) in candidates.iter().enumerate() {
let execution = provider_query_execute_kiro_test_candidate(
state,
&provider,
candidate,
payload,
route_path,
&trace_id,
&requested_model,
)
.await?;
if execution.status != "skipped" {
total_attempts += 1;
}
let is_success = execution.status == "success";
let response_body = execution.response_body.clone();
attempts.push(provider_query_test_attempt_payload(
candidate_index,
candidate,
&execution,
));
if is_success {
success_body = response_body;
break;
}
}
let success = success_body.is_some();
let error = if success {
Value::Null
} else {
attempts
.iter()
.rev()
.find_map(|attempt| {
attempt
.get("error_message")
.cloned()
.filter(|value| !value.is_null())
})
.unwrap_or_else(|| json!(ADMIN_PROVIDER_QUERY_NO_LOCAL_MODELS_DETAIL))
};
Ok(Json(json!({
"success": success,
"model": requested_model,
"provider": provider_query_provider_payload(&provider),
"attempts": attempts,
"total_candidates": candidates.len(),
"total_attempts": total_attempts,
"data": success_body.as_ref().map(|body| json!({
"stream": true,
"response": body,
})),
"error": error,
}))
.into_response())
}
pub(crate) async fn build_admin_provider_query_test_model_local_response(
state: &AdminAppState<'_>,
payload: &Value,
) -> Result<Response<Body>, GatewayError> {
let response = build_admin_provider_query_kiro_failover_response(
state,
payload,
"/api/admin/provider-query/test-model",
)
.await?;
let body = to_bytes(response.into_body(), usize::MAX)
.await
.map_err(|err| GatewayError::Internal(err.to_string()))?;
let parsed: Value =
serde_json::from_slice(&body).map_err(|err| GatewayError::Internal(err.to_string()))?;
Ok(Json(json!({
"success": parsed.get("success").cloned().unwrap_or(Value::Bool(false)),
"error": parsed.get("error").cloned().unwrap_or(Value::Null),
"data": parsed.get("data").cloned().unwrap_or(Value::Null),
"provider": parsed.get("provider").cloned().unwrap_or(Value::Null),
"model": parsed.get("model").cloned().unwrap_or(Value::Null),
}))
.into_response())
}
pub(crate) async fn build_admin_provider_query_test_model_failover_local_response(
state: &AdminAppState<'_>,
payload: &Value,
) -> Result<Response<Body>, GatewayError> {
build_admin_provider_query_kiro_failover_response(
state,
payload,
"/api/admin/provider-query/test-model-failover",
)
.await
}
fn provider_query_key_display_name(key: &StoredProviderCatalogKey) -> String {
let trimmed = key.name.trim();
if trimmed.is_empty() {

1
apps/aether-gateway/src/handlers/admin/provider/shared/support.rs
View File

@@ -13,6 +13,7 @@ pub(crate) const ADMIN_PROVIDER_OAUTH_DATA_UNAVAILABLE_DETAIL: &str =
#[derive(Debug, Clone, Copy)]
pub(crate) struct AdminProviderPoolConfig {
pub(crate) lru_enabled: bool,
pub(crate) skip_exhausted_accounts: bool,
pub(crate) cost_window_seconds: u64,
pub(crate) cost_limit_per_key_tokens: Option<u64>,
}

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

@@ -13,8 +13,8 @@ pub(crate) fn normalize_provider_type_input(value: &str) -> Result<String, Strin
pub(crate) fn normalize_auth_type(value: Option<&str>) -> Result<String, String> {
let auth_type = value.unwrap_or("api_key").trim().to_ascii_lowercase();
match auth_type.as_str() {
"api_key" | "service_account" | "oauth" => Ok(auth_type),
_ => Err("auth_type 仅支持 api_key / service_account / oauth".to_string()),
"api_key" | "service_account" | "oauth" | "bearer" => Ok(auth_type),
_ => Err("auth_type 仅支持 api_key / service_account / oauth / bearer".to_string()),
}
}
@@ -63,7 +63,7 @@ pub(crate) fn validate_vertex_api_formats(
#[cfg(test)]
mod tests {
use super::normalize_pool_advanced_config;
use super::{normalize_auth_type, normalize_pool_advanced_config};
use serde_json::json;
#[test]
@@ -85,4 +85,12 @@ mod tests {
"pool_advanced 必须是 JSON 对象"
);
}
#[test]
fn normalize_auth_type_supports_bearer() {
assert_eq!(
normalize_auth_type(Some("bearer")).expect("bearer should normalize"),
"bearer"
);
}
}

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

@@ -8,17 +8,37 @@ fn normalize_reveal_auth_type(value: &str) -> &str {
use crate::handlers::admin::request::AdminAppState;
use crate::handlers::admin::shared::parse_catalog_auth_config_json;
use crate::provider_key_auth::provider_key_auth_semantics;
use aether_data_contracts::repository::provider_catalog::StoredProviderCatalogKey;
use chrono::{SecondsFormat, Utc};
use serde_json::json;
fn reveal_provider_type_from_auth_config(
auth_config: Option<&serde_json::Map<String, serde_json::Value>>,
) -> String {
auth_config
.and_then(|value| value.get("provider_type"))
.and_then(serde_json::Value::as_str)
.map(str::trim)
.filter(|value| !value.is_empty())
.unwrap_or_default()
.to_string()
}
pub(crate) fn build_admin_reveal_key_payload(
state: &AdminAppState<'_>,
key: &StoredProviderCatalogKey,
) -> Result<serde_json::Value, String> {
let auth_type = normalize_reveal_auth_type(&key.auth_type);
let parsed_auth_config = state.parse_catalog_auth_config_json(key);
let provider_type = reveal_provider_type_from_auth_config(parsed_auth_config.as_ref());
let auth_semantics = provider_key_auth_semantics(key, provider_type.as_str());
let auth_type = if auth_semantics.oauth_managed() {
"oauth"
} else {
normalize_reveal_auth_type(&key.auth_type)
};
if matches!(auth_type, "service_account") {
if let Some(auth_config) = state.parse_catalog_auth_config_json(key) {
if let Some(auth_config) = parsed_auth_config {
return Ok(json!({
"auth_type": auth_type,
"auth_config": auth_config,
@@ -92,11 +112,6 @@ pub(crate) async fn build_admin_export_key_payload(
state: &AdminAppState<'_>,
key: &StoredProviderCatalogKey,
) -> Result<serde_json::Value, String> {
let auth_type = normalize_reveal_auth_type(&key.auth_type);
if auth_type != "oauth" {
return Err("仅 OAuth 类型的 Key 支持导出".to_string());
}
let ciphertext = key
.encrypted_auth_config
.as_deref()
@@ -136,6 +151,9 @@ pub(crate) async fn build_admin_export_key_payload(
.map(|provider| provider.provider_type)
.unwrap_or_default()
};
if !provider_key_auth_semantics(key, provider_type.as_str()).can_export_oauth() {
return Err("仅 OAuth 管理账号支持导出".to_string());
}
let mut payload =
provider_oauth_export_payload(&provider_type, &auth_config, key.upstream_metadata.as_ref());

54
apps/aether-gateway/src/handlers/admin/request/provider/routes/query.rs
View File

@@ -1,7 +1,9 @@
use crate::handlers::admin::provider::query::{
models::{
build_admin_provider_query_models_response,
build_admin_provider_query_test_model_failover_local_response,
build_admin_provider_query_test_model_failover_response,
build_admin_provider_query_test_model_local_response,
build_admin_provider_query_test_model_response,
},
payload::{
@@ -78,10 +80,25 @@ impl<'a> AdminAppState<'a> {
ADMIN_PROVIDER_QUERY_MODEL_REQUIRED_DETAIL,
)));
};
Ok(Some(build_admin_provider_query_test_model_response(
provider_id,
model,
)))
let provider_type = self
.app()
.read_provider_catalog_providers_by_ids(std::slice::from_ref(&provider_id))
.await?
.into_iter()
.find(|provider| provider.id == provider_id)
.map(|provider| provider.provider_type)
.unwrap_or_default();
if provider_type.trim().eq_ignore_ascii_case("kiro") {
Ok(Some(
build_admin_provider_query_test_model_local_response(self, &payload)
.await?,
))
} else {
Ok(Some(build_admin_provider_query_test_model_response(
provider_id,
model,
)))
}
}
"test_model_failover" => {
let Some(provider_id) = provider_query_extract_provider_id(&payload) else {
@@ -107,12 +124,29 @@ impl<'a> AdminAppState<'a> {
ADMIN_PROVIDER_QUERY_FAILOVER_MODELS_REQUIRED_DETAIL,
)));
}
Ok(Some(
build_admin_provider_query_test_model_failover_response(
provider_id,
failover_models,
),
))
let provider_type = self
.app()
.read_provider_catalog_providers_by_ids(std::slice::from_ref(&provider_id))
.await?
.into_iter()
.find(|provider| provider.id == provider_id)
.map(|provider| provider.provider_type)
.unwrap_or_default();
if provider_type.trim().eq_ignore_ascii_case("kiro") {
Ok(Some(
build_admin_provider_query_test_model_failover_local_response(
self, &payload,
)
.await?,
))
} else {
Ok(Some(
build_admin_provider_query_test_model_failover_response(
provider_id,
failover_models,
),
))
}
}
_ => Ok(Some(
build_admin_provider_query_models_response(self, &payload).await?,

115
apps/aether-gateway/src/handlers/shared/catalog.rs
View File

@@ -1,4 +1,5 @@
use crate::handlers::shared::{json_string_list, unix_secs_to_rfc3339};
use crate::provider_key_auth::provider_key_auth_semantics;
use crate::AppState;
#[cfg(test)]
use aether_crypto::DEVELOPMENT_ENCRYPTION_KEY;
@@ -273,7 +274,7 @@ fn derive_catalog_oauth_plan_type(
provider_type: &str,
auth_config: Option<&serde_json::Map<String, serde_json::Value>>,
) -> Option<String> {
if !key.auth_type.trim().eq_ignore_ascii_case("oauth") {
if !provider_key_auth_semantics(key, provider_type).oauth_managed() {
return None;
}
@@ -339,13 +340,18 @@ pub(crate) fn build_admin_provider_key_response(
} else {
0.0
};
let auth_semantics = provider_key_auth_semantics(key, provider_type);
let auth_config = parse_catalog_auth_config_json(state, key);
let oauth_organizations = auth_config
.as_ref()
.and_then(|config| config.get("organizations"))
.and_then(serde_json::Value::as_array)
.cloned()
.unwrap_or_default();
let oauth_organizations = if auth_semantics.can_show_oauth_metadata() {
auth_config
.as_ref()
.and_then(|config| config.get("organizations"))
.and_then(serde_json::Value::as_array)
.cloned()
.unwrap_or_default()
} else {
Vec::new()
};
let oauth_plan_type = derive_catalog_oauth_plan_type(key, provider_type, auth_config.as_ref());
let (
health_score,
@@ -387,6 +393,30 @@ pub(crate) fn build_admin_provider_key_response(
);
payload.insert("api_key_plain".to_string(), serde_json::Value::Null);
payload.insert("auth_type".to_string(), json!(key.auth_type));
payload.insert(
"credential_kind".to_string(),
json!(auth_semantics.credential_kind().as_str()),
);
payload.insert(
"runtime_auth_kind".to_string(),
json!(auth_semantics.runtime_auth_kind().as_str()),
);
payload.insert(
"oauth_managed".to_string(),
json!(auth_semantics.oauth_managed()),
);
payload.insert(
"can_refresh_oauth".to_string(),
json!(auth_semantics.can_refresh_oauth()),
);
payload.insert(
"can_export_oauth".to_string(),
json!(auth_semantics.can_export_oauth()),
);
payload.insert(
"can_edit_oauth".to_string(),
json!(auth_semantics.can_edit_oauth()),
);
payload.insert("name".to_string(), json!(key.name));
payload.insert("rate_multipliers".to_string(), json!(key.rate_multipliers));
payload.insert(
@@ -410,40 +440,59 @@ pub(crate) fn build_admin_provider_key_response(
payload.insert("capabilities".to_string(), json!(key.capabilities));
payload.insert(
"oauth_expires_at".to_string(),
json!(key.expires_at_unix_secs),
json!(auth_semantics
.can_show_oauth_metadata()
.then_some(key.expires_at_unix_secs)
.flatten()),
);
payload.insert(
"oauth_email".to_string(),
auth_config
.as_ref()
.and_then(|config| config.get("email"))
.cloned()
.unwrap_or(serde_json::Value::Null),
if auth_semantics.can_show_oauth_metadata() {
auth_config
.as_ref()
.and_then(|config| config.get("email"))
.cloned()
.unwrap_or(serde_json::Value::Null)
} else {
serde_json::Value::Null
},
);
payload.insert("oauth_plan_type".to_string(), json!(oauth_plan_type));
payload.insert(
"oauth_account_id".to_string(),
auth_config
.as_ref()
.and_then(|config| config.get("account_id"))
.cloned()
.unwrap_or(serde_json::Value::Null),
if auth_semantics.can_show_oauth_metadata() {
auth_config
.as_ref()
.and_then(|config| config.get("account_id"))
.cloned()
.unwrap_or(serde_json::Value::Null)
} else {
serde_json::Value::Null
},
);
payload.insert(
"oauth_account_name".to_string(),
auth_config
.as_ref()
.and_then(|config| config.get("account_name"))
.cloned()
.unwrap_or(serde_json::Value::Null),
if auth_semantics.can_show_oauth_metadata() {
auth_config
.as_ref()
.and_then(|config| config.get("account_name"))
.cloned()
.unwrap_or(serde_json::Value::Null)
} else {
serde_json::Value::Null
},
);
payload.insert(
"oauth_account_user_id".to_string(),
auth_config
.as_ref()
.and_then(|config| config.get("account_user_id"))
.cloned()
.unwrap_or(serde_json::Value::Null),
if auth_semantics.can_show_oauth_metadata() {
auth_config
.as_ref()
.and_then(|config| config.get("account_user_id"))
.cloned()
.unwrap_or(serde_json::Value::Null)
} else {
serde_json::Value::Null
},
);
payload.insert(
"oauth_organizations".to_string(),
@@ -451,11 +500,17 @@ pub(crate) fn build_admin_provider_key_response(
);
payload.insert(
"oauth_invalid_at".to_string(),
json!(key.oauth_invalid_at_unix_secs),
json!(auth_semantics
.can_show_oauth_metadata()
.then_some(key.oauth_invalid_at_unix_secs)
.flatten()),
);
payload.insert(
"oauth_invalid_reason".to_string(),
json!(key.oauth_invalid_reason),
json!(auth_semantics
.can_show_oauth_metadata()
.then_some(key.oauth_invalid_reason.clone())
.flatten()),
);
payload.insert(
"status_snapshot".to_string(),

1
apps/aether-gateway/src/lib.rs
View File

@@ -48,6 +48,7 @@ mod log_ids;
mod maintenance;
pub(crate) mod middleware;
mod model_fetch;
mod provider_key_auth;
pub(crate) use aether_provider_transport as provider_transport;
mod query;
mod rate_limit;

230
apps/aether-gateway/src/provider_key_auth.rs Normal file
View File

@@ -0,0 +1,230 @@
use aether_data_contracts::repository::provider_catalog::StoredProviderCatalogKey;
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum ProviderKeyCredentialKind {
RawSecret,
OAuthSession,
ServiceAccount,
}
impl ProviderKeyCredentialKind {
pub(crate) const fn as_str(self) -> &'static str {
match self {
Self::RawSecret => "raw_secret",
Self::OAuthSession => "oauth_session",
Self::ServiceAccount => "service_account",
}
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum ProviderKeyRuntimeAuthKind {
ApiKey,
Bearer,
ServiceAccount,
Unknown,
}
impl ProviderKeyRuntimeAuthKind {
pub(crate) const fn as_str(self) -> &'static str {
match self {
Self::ApiKey => "api_key",
Self::Bearer => "bearer",
Self::ServiceAccount => "service_account",
Self::Unknown => "unknown",
}
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) struct ProviderKeyAuthSemantics {
credential_kind: ProviderKeyCredentialKind,
runtime_auth_kind: ProviderKeyRuntimeAuthKind,
oauth_managed: bool,
}
impl ProviderKeyAuthSemantics {
pub(crate) const fn credential_kind(self) -> ProviderKeyCredentialKind {
self.credential_kind
}
pub(crate) const fn runtime_auth_kind(self) -> ProviderKeyRuntimeAuthKind {
self.runtime_auth_kind
}
pub(crate) const fn oauth_managed(self) -> bool {
self.oauth_managed
}
pub(crate) const fn can_refresh_oauth(self) -> bool {
self.oauth_managed
}
pub(crate) const fn can_export_oauth(self) -> bool {
self.oauth_managed
}
pub(crate) const fn can_edit_oauth(self) -> bool {
self.oauth_managed
}
pub(crate) const fn can_show_oauth_metadata(self) -> bool {
self.oauth_managed
}
}
fn normalized_auth_type(key: &StoredProviderCatalogKey) -> String {
key.auth_type.trim().to_ascii_lowercase()
}
fn key_has_auth_config(key: &StoredProviderCatalogKey) -> bool {
key.encrypted_auth_config
.as_deref()
.map(str::trim)
.is_some_and(|value| !value.is_empty())
}
fn provider_uses_bearer_oauth_runtime(provider_type: &str) -> bool {
matches!(
provider_type.trim().to_ascii_lowercase().as_str(),
"claude_code" | "codex" | "gemini_cli" | "antigravity" | "kiro"
)
}
fn provider_key_is_legacy_kiro_oauth_session(
key: &StoredProviderCatalogKey,
provider_type: &str,
auth_type: &str,
) -> bool {
provider_type.trim().eq_ignore_ascii_case("kiro")
&& auth_type.eq_ignore_ascii_case("bearer")
&& key_has_auth_config(key)
}
pub(crate) fn provider_key_auth_semantics(
key: &StoredProviderCatalogKey,
provider_type: &str,
) -> ProviderKeyAuthSemantics {
let auth_type = normalized_auth_type(key);
let oauth_managed = auth_type == "oauth"
|| provider_key_is_legacy_kiro_oauth_session(key, provider_type, &auth_type);
let credential_kind = if oauth_managed {
ProviderKeyCredentialKind::OAuthSession
} else if matches!(auth_type.as_str(), "service_account" | "vertex_ai") {
ProviderKeyCredentialKind::ServiceAccount
} else {
ProviderKeyCredentialKind::RawSecret
};
let runtime_auth_kind = match credential_kind {
ProviderKeyCredentialKind::OAuthSession => {
if provider_uses_bearer_oauth_runtime(provider_type) {
ProviderKeyRuntimeAuthKind::Bearer
} else {
ProviderKeyRuntimeAuthKind::Unknown
}
}
ProviderKeyCredentialKind::ServiceAccount => ProviderKeyRuntimeAuthKind::ServiceAccount,
ProviderKeyCredentialKind::RawSecret => match auth_type.as_str() {
"bearer" => ProviderKeyRuntimeAuthKind::Bearer,
"api_key" => ProviderKeyRuntimeAuthKind::ApiKey,
_ => ProviderKeyRuntimeAuthKind::Unknown,
},
};
ProviderKeyAuthSemantics {
credential_kind,
runtime_auth_kind,
oauth_managed,
}
}
pub(crate) fn provider_key_is_oauth_managed(
key: &StoredProviderCatalogKey,
provider_type: &str,
) -> bool {
provider_key_auth_semantics(key, provider_type).oauth_managed()
}
#[cfg(test)]
mod tests {
use super::{
provider_key_auth_semantics, ProviderKeyCredentialKind, ProviderKeyRuntimeAuthKind,
};
use aether_data_contracts::repository::provider_catalog::StoredProviderCatalogKey;
fn sample_key(auth_type: &str) -> StoredProviderCatalogKey {
StoredProviderCatalogKey::new(
"key-1".to_string(),
"provider-1".to_string(),
"key-1".to_string(),
auth_type.to_string(),
None,
true,
)
.expect("key should build")
}
#[test]
fn recognizes_oauth_managed_key() {
let semantics = provider_key_auth_semantics(&sample_key("oauth"), "codex");
assert!(semantics.oauth_managed());
assert_eq!(
semantics.credential_kind(),
ProviderKeyCredentialKind::OAuthSession
);
assert_eq!(
semantics.runtime_auth_kind(),
ProviderKeyRuntimeAuthKind::Bearer
);
}
#[test]
fn recognizes_legacy_kiro_bearer_key_with_auth_config_as_oauth_managed() {
let mut key = sample_key("bearer");
key.encrypted_auth_config = Some("ciphertext".to_string());
let semantics = provider_key_auth_semantics(&key, "kiro");
assert!(semantics.oauth_managed());
assert_eq!(
semantics.credential_kind(),
ProviderKeyCredentialKind::OAuthSession
);
assert_eq!(
semantics.runtime_auth_kind(),
ProviderKeyRuntimeAuthKind::Bearer
);
}
#[test]
fn keeps_plain_bearer_key_as_raw_secret() {
let semantics = provider_key_auth_semantics(&sample_key("bearer"), "kiro");
assert!(!semantics.oauth_managed());
assert_eq!(
semantics.credential_kind(),
ProviderKeyCredentialKind::RawSecret
);
assert_eq!(
semantics.runtime_auth_kind(),
ProviderKeyRuntimeAuthKind::Bearer
);
}
#[test]
fn recognizes_service_account_key() {
let semantics = provider_key_auth_semantics(&sample_key("service_account"), "vertex_ai");
assert!(!semantics.oauth_managed());
assert_eq!(
semantics.credential_kind(),
ProviderKeyCredentialKind::ServiceAccount
);
assert_eq!(
semantics.runtime_auth_kind(),
ProviderKeyRuntimeAuthKind::ServiceAccount
);
}
}

78
apps/aether-gateway/src/scheduler/candidate/runtime.rs
View File

@@ -1,5 +1,6 @@
use std::collections::{BTreeMap, BTreeSet};
use aether_admin::provider::pool as admin_provider_pool_pure;
use aether_data_contracts::repository::candidates::StoredRequestCandidate;
use aether_data_contracts::repository::provider_catalog::StoredProviderCatalogKey;
use aether_scheduler_core::{
@@ -20,6 +21,7 @@ pub(super) struct CandidateRuntimeSelectionSnapshot {
pub(super) provider_concurrent_limits: BTreeMap<String, usize>,
pub(super) provider_key_rpm_states: BTreeMap<String, StoredProviderCatalogKey>,
provider_quota_blocks_requests: BTreeMap<String, bool>,
key_account_quota_exhausted: BTreeMap<String, bool>,
provider_key_rpm_reset_ats: BTreeMap<String, Option<u64>>,
}
@@ -30,7 +32,14 @@ pub(super) async fn read_candidate_runtime_selection_snapshot(
) -> Result<CandidateRuntimeSelectionSnapshot, GatewayError> {
let recent_candidates = state.read_recent_request_candidates(128).await?;
let provider_concurrent_limits = read_provider_concurrent_limits(state, candidates).await?;
let provider_skip_exhausted_accounts =
read_provider_skip_exhausted_account_map(state, candidates).await?;
let provider_key_rpm_states = read_provider_key_rpm_states(state, candidates).await?;
let key_account_quota_exhausted = read_key_account_quota_exhaustion_map(
candidates,
&provider_key_rpm_states,
&provider_skip_exhausted_accounts,
);
let provider_quota_blocks_requests =
read_provider_quota_block_map(state, candidates, now_unix_secs).await?;
let provider_key_rpm_reset_ats =
@@ -41,6 +50,7 @@ pub(super) async fn read_candidate_runtime_selection_snapshot(
provider_concurrent_limits,
provider_key_rpm_states,
provider_quota_blocks_requests,
key_account_quota_exhausted,
provider_key_rpm_reset_ats,
})
}
@@ -89,6 +99,11 @@ pub(super) fn is_candidate_selectable(
.get(candidate.provider_id.as_str())
.copied()
.unwrap_or(false),
account_quota_exhausted: snapshot
.key_account_quota_exhausted
.get(candidate.key_id.as_str())
.copied()
.unwrap_or(false),
rpm_reset_at: snapshot
.provider_key_rpm_reset_ats
.get(candidate.key_id.as_str())
@@ -122,6 +137,11 @@ pub(super) fn current_candidate_runtime_skip_reason(
now_unix_secs,
cached_affinity_target,
provider_quota_blocks_requests,
account_quota_exhausted: snapshot
.key_account_quota_exhausted
.get(candidate.key_id.as_str())
.copied()
.unwrap_or(false),
rpm_reset_at,
})
}
@@ -192,6 +212,64 @@ async fn read_provider_quota_block_map(
Ok(quota_blocks)
}
async fn read_provider_skip_exhausted_account_map(
state: &(impl SchedulerRuntimeState + ?Sized),
candidates: &[SchedulerMinimalCandidateSelectionCandidate],
) -> Result<BTreeMap<String, bool>, GatewayError> {
let provider_ids = candidates
.iter()
.map(|candidate| candidate.provider_id.clone())
.collect::<BTreeSet<_>>()
.into_iter()
.collect::<Vec<_>>();
if provider_ids.is_empty() {
return Ok(BTreeMap::new());
}
let providers = state
.read_provider_catalog_providers_by_ids(&provider_ids)
.await?;
Ok(providers
.into_iter()
.map(|provider| {
let skip_exhausted_accounts = provider
.config
.as_ref()
.and_then(|value| value.get("pool_advanced"))
.and_then(serde_json::Value::as_object)
.and_then(|value| value.get("skip_exhausted_accounts"))
.and_then(serde_json::Value::as_bool)
.unwrap_or(false);
(provider.id, skip_exhausted_accounts)
})
.collect())
}
fn read_key_account_quota_exhaustion_map(
candidates: &[SchedulerMinimalCandidateSelectionCandidate],
provider_key_rpm_states: &BTreeMap<String, StoredProviderCatalogKey>,
provider_skip_exhausted_accounts: &BTreeMap<String, bool>,
) -> BTreeMap<String, bool> {
candidates
.iter()
.map(|candidate| {
let exhausted = provider_skip_exhausted_accounts
.get(candidate.provider_id.as_str())
.copied()
.unwrap_or(false)
&& provider_key_rpm_states
.get(candidate.key_id.as_str())
.is_some_and(|key| {
admin_provider_pool_pure::admin_pool_key_account_quota_exhausted(
key,
candidate.provider_type.as_str(),
)
});
(candidate.key_id.clone(), exhausted)
})
.collect()
}
fn read_provider_key_rpm_reset_at_map(
state: &(impl SchedulerRuntimeState + ?Sized),
candidates: &[SchedulerMinimalCandidateSelectionCandidate],

247
apps/aether-gateway/src/scheduler/candidate/tests/selection.rs
View File

@@ -1140,6 +1140,253 @@ async fn exposes_runtime_skipped_candidates_with_skip_reasons() {
assert_eq!(skipped[0].skip_reason, "key_circuit_open");
}
#[tokio::test]
async fn skips_codex_candidate_when_account_quota_is_exhausted_and_pool_flag_enabled() {
let mut first = sample_row();
first.provider_id = "provider-codex".to_string();
first.provider_name = "codex".to_string();
first.provider_type = "codex".to_string();
first.endpoint_id = "endpoint-codex".to_string();
first.endpoint_api_format = "openai:cli".to_string();
first.key_id = "key-codex".to_string();
first.key_name = "codex-exhausted".to_string();
first.key_auth_type = "oauth".to_string();
first.key_api_formats = Some(vec!["openai:cli".to_string()]);
first.key_global_priority_by_format = Some(serde_json::json!({"openai:cli": 1}));
let mut second = sample_row();
second.provider_id = "provider-openai".to_string();
second.provider_name = "openai".to_string();
second.endpoint_id = "endpoint-openai".to_string();
second.endpoint_api_format = "openai:cli".to_string();
second.key_id = "key-openai".to_string();
second.key_name = "fallback".to_string();
second.key_api_formats = Some(vec!["openai:cli".to_string()]);
second.key_global_priority_by_format = Some(serde_json::json!({"openai:cli": 2}));
let candidates = Arc::new(InMemoryMinimalCandidateSelectionReadRepository::seed(vec![
first, second,
]));
let mut codex_provider = sample_provider("provider-codex", None);
codex_provider.provider_type = "codex".to_string();
codex_provider.config = Some(serde_json::json!({
"pool_advanced": {
"skip_exhausted_accounts": true
}
}));
let provider_catalog = Arc::new(InMemoryProviderCatalogReadRepository::seed(
vec![codex_provider, sample_provider("provider-openai", None)],
Vec::new(),
vec![
{
let mut key = sample_key("key-codex", "provider-codex", Some(10));
key.auth_type = "oauth".to_string();
key.upstream_metadata = Some(serde_json::json!({
"codex": {
"secondary_used_percent": 100.0
}
}));
key
},
sample_key("key-openai", "provider-openai", Some(10)),
],
));
let quotas = Arc::new(InMemoryProviderQuotaRepository::seed(vec![]));
let request_candidates = Arc::new(InMemoryRequestCandidateRepository::seed(vec![]));
let state = AppState::new()
.expect("state should build")
.with_data_state_for_tests(
GatewayDataState::with_candidate_selection_provider_catalog_quota_and_request_candidates_for_tests(
candidates,
provider_catalog,
quotas,
request_candidates,
),
);
let (selected, skipped) = collect_selectable_candidates_with_skip_reasons(
state.data.as_ref(),
&state,
"openai:cli",
"gpt-4.1",
false,
None,
100,
)
.await
.expect("selection should succeed");
assert_eq!(selected.len(), 1);
assert_eq!(selected[0].provider_id, "provider-openai");
assert_eq!(skipped.len(), 1);
assert_eq!(skipped[0].candidate.provider_id, "provider-codex");
assert_eq!(skipped[0].skip_reason, "account_quota_exhausted");
}
#[tokio::test]
async fn keeps_codex_candidate_selectable_when_exhausted_account_flag_is_disabled() {
let mut first = sample_row();
first.provider_id = "provider-codex".to_string();
first.provider_name = "codex".to_string();
first.provider_type = "codex".to_string();
first.endpoint_id = "endpoint-codex".to_string();
first.endpoint_api_format = "openai:cli".to_string();
first.key_id = "key-codex".to_string();
first.key_name = "codex-exhausted".to_string();
first.key_auth_type = "oauth".to_string();
first.key_api_formats = Some(vec!["openai:cli".to_string()]);
first.key_global_priority_by_format = Some(serde_json::json!({"openai:cli": 1}));
let mut second = sample_row();
second.provider_id = "provider-openai".to_string();
second.provider_name = "openai".to_string();
second.endpoint_id = "endpoint-openai".to_string();
second.endpoint_api_format = "openai:cli".to_string();
second.key_id = "key-openai".to_string();
second.key_name = "fallback".to_string();
second.key_api_formats = Some(vec!["openai:cli".to_string()]);
second.key_global_priority_by_format = Some(serde_json::json!({"openai:cli": 2}));
let candidates = Arc::new(InMemoryMinimalCandidateSelectionReadRepository::seed(vec![
first, second,
]));
let mut codex_provider = sample_provider("provider-codex", None);
codex_provider.provider_type = "codex".to_string();
codex_provider.config = Some(serde_json::json!({
"pool_advanced": {}
}));
let provider_catalog = Arc::new(InMemoryProviderCatalogReadRepository::seed(
vec![codex_provider, sample_provider("provider-openai", None)],
Vec::new(),
vec![
{
let mut key = sample_key("key-codex", "provider-codex", Some(10));
key.auth_type = "oauth".to_string();
key.upstream_metadata = Some(serde_json::json!({
"codex": {
"secondary_used_percent": 100.0
}
}));
key
},
sample_key("key-openai", "provider-openai", Some(10)),
],
));
let quotas = Arc::new(InMemoryProviderQuotaRepository::seed(vec![]));
let request_candidates = Arc::new(InMemoryRequestCandidateRepository::seed(vec![]));
let state = AppState::new()
.expect("state should build")
.with_data_state_for_tests(
GatewayDataState::with_candidate_selection_provider_catalog_quota_and_request_candidates_for_tests(
candidates,
provider_catalog,
quotas,
request_candidates,
),
);
let (selected, skipped) = collect_selectable_candidates_with_skip_reasons(
state.data.as_ref(),
&state,
"openai:cli",
"gpt-4.1",
false,
None,
100,
)
.await
.expect("selection should succeed");
assert_eq!(selected.len(), 2);
assert!(selected
.iter()
.any(|candidate| candidate.provider_id == "provider-codex"));
assert!(skipped.is_empty());
}
#[tokio::test]
async fn skips_kiro_candidate_when_account_quota_is_exhausted_and_pool_flag_enabled() {
let mut first = sample_row();
first.provider_id = "provider-kiro".to_string();
first.provider_name = "kiro".to_string();
first.provider_type = "kiro".to_string();
first.endpoint_id = "endpoint-kiro".to_string();
first.endpoint_api_format = "claude:cli".to_string();
first.key_id = "key-kiro".to_string();
first.key_name = "kiro-exhausted".to_string();
first.key_auth_type = "oauth".to_string();
first.key_api_formats = Some(vec!["claude:cli".to_string()]);
first.key_global_priority_by_format = Some(serde_json::json!({"claude:cli": 1}));
let mut second = sample_row();
second.provider_id = "provider-openai".to_string();
second.provider_name = "openai".to_string();
second.endpoint_id = "endpoint-openai".to_string();
second.endpoint_api_format = "claude:cli".to_string();
second.key_id = "key-openai".to_string();
second.key_name = "fallback".to_string();
second.key_api_formats = Some(vec!["claude:cli".to_string()]);
second.key_global_priority_by_format = Some(serde_json::json!({"claude:cli": 2}));
let candidates = Arc::new(InMemoryMinimalCandidateSelectionReadRepository::seed(vec![
first, second,
]));
let mut kiro_provider = sample_provider("provider-kiro", None);
kiro_provider.provider_type = "kiro".to_string();
kiro_provider.config = Some(serde_json::json!({
"pool_advanced": {
"skip_exhausted_accounts": true
}
}));
let provider_catalog = Arc::new(InMemoryProviderCatalogReadRepository::seed(
vec![kiro_provider, sample_provider("provider-openai", None)],
Vec::new(),
vec![
{
let mut key = sample_key("key-kiro", "provider-kiro", Some(10));
key.auth_type = "oauth".to_string();
key.upstream_metadata = Some(serde_json::json!({
"kiro": {
"remaining": 0
}
}));
key
},
sample_key("key-openai", "provider-openai", Some(10)),
],
));
let quotas = Arc::new(InMemoryProviderQuotaRepository::seed(vec![]));
let request_candidates = Arc::new(InMemoryRequestCandidateRepository::seed(vec![]));
let state = AppState::new()
.expect("state should build")
.with_data_state_for_tests(
GatewayDataState::with_candidate_selection_provider_catalog_quota_and_request_candidates_for_tests(
candidates,
provider_catalog,
quotas,
request_candidates,
),
);
let (selected, skipped) = collect_selectable_candidates_with_skip_reasons(
state.data.as_ref(),
&state,
"claude:cli",
"gpt-4.1",
false,
None,
100,
)
.await
.expect("selection should succeed");
assert_eq!(selected.len(), 1);
assert_eq!(selected[0].provider_id, "provider-openai");
assert_eq!(skipped.len(), 1);
assert_eq!(skipped[0].candidate.provider_id, "provider-kiro");
assert_eq!(skipped[0].skip_reason, "account_quota_exhausted");
}
#[tokio::test]
async fn same_priority_candidates_prefer_healthier_provider_key_before_id_order() {
let mut first = sample_row();

76
apps/aether-gateway/src/tests/control/admin/models/global.rs
View File

@@ -1027,6 +1027,82 @@ async fn gateway_batch_deletes_admin_global_models_locally_with_trusted_admin_pr
upstream_handle.abort();
}
#[tokio::test]
async fn gateway_deletes_admin_global_model_with_bound_provider_models_locally() {
let upstream_hits = Arc::new(Mutex::new(0usize));
let upstream_hits_clone = Arc::clone(&upstream_hits);
let upstream = Router::new().route(
"/api/admin/models/global/global-gpt-5",
any(move |_request: Request| {
let upstream_hits_inner = Arc::clone(&upstream_hits_clone);
async move {
*upstream_hits_inner.lock().expect("mutex should lock") += 1;
(StatusCode::OK, Body::from("unexpected upstream hit"))
}
}),
);
let global_model_repository = Arc::new(
InMemoryGlobalModelReadRepository::seed(Vec::new())
.with_admin_global_models(vec![sample_admin_global_model(
"global-gpt-5",
"gpt-5",
"GPT 5",
)])
.with_admin_provider_models(vec![sample_admin_provider_model(
"model-openai-gpt5",
"provider-openai",
"global-gpt-5",
"gpt-5-upstream",
)]),
);
let (upstream_url, upstream_handle) = start_server(upstream).await;
let gateway = build_router_with_state(
AppState::new()
.expect("gateway should build")
.with_data_state_for_tests(
GatewayDataState::disabled()
.with_global_model_repository_for_tests(global_model_repository.clone()),
),
);
let (gateway_url, gateway_handle) = start_server(gateway).await;
let response = reqwest::Client::new()
.delete(format!(
"{gateway_url}/api/admin/models/global/global-gpt-5"
))
.header(crate::constants::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")
.send()
.await
.expect("request should succeed");
assert_eq!(response.status(), StatusCode::NO_CONTENT);
assert_eq!(*upstream_hits.lock().expect("mutex should lock"), 0);
let deleted = global_model_repository
.get_admin_global_model_by_id("global-gpt-5")
.await
.expect("model lookup should succeed");
assert!(deleted.is_none());
let provider_models = global_model_repository
.list_admin_provider_models(&AdminProviderModelListQuery {
provider_id: "provider-openai".to_string(),
is_active: None,
offset: 0,
limit: 20,
})
.await
.expect("provider models should read");
assert!(provider_models.is_empty());
gateway_handle.abort();
upstream_handle.abort();
}
#[tokio::test]
async fn gateway_assigns_admin_global_model_to_providers_locally_with_trusted_admin_principal() {
let upstream_hits = Arc::new(Mutex::new(0usize));

94
apps/aether-gateway/src/tests/control/admin/oauth.rs
View File

@@ -1792,6 +1792,51 @@ async fn gateway_imports_admin_provider_oauth_refresh_token_locally_with_trusted
upstream_handle.abort();
}
#[tokio::test]
async fn gateway_rejects_kiro_single_refresh_token_import_with_clear_error() {
let provider_catalog_repository = Arc::new(InMemoryProviderCatalogReadRepository::seed(
vec![{
let mut provider = sample_provider("provider-kiro", "kiro", 10);
provider.provider_type = "kiro".to_string();
provider
}],
vec![sample_endpoint(
"endpoint-kiro-chat",
"provider-kiro",
"kiro:generateAssistantResponse",
"https://service.kiro.dev",
)],
Vec::new(),
));
let state = AppState::new()
.expect("gateway should build")
.with_data_state_for_tests(GatewayDataState::with_provider_catalog_reader_for_tests(
provider_catalog_repository,
));
let response = local_admin_provider_oauth_response(
&state,
http::Method::POST,
"/api/admin/provider-oauth/providers/provider-kiro/import-refresh-token",
Some(json!({
"refresh_token": "kiro-refresh-token"
})),
)
.await;
assert_eq!(response.status(), StatusCode::BAD_REQUEST);
let payload: serde_json::Value = serde_json::from_slice(
&to_bytes(response.into_body(), usize::MAX)
.await
.expect("body should read"),
)
.expect("json body should parse");
assert_eq!(
payload["detail"],
json!("Kiro 不支持单条 Refresh Token 导入，请使用批量导入或设备授权。")
);
}
#[tokio::test]
async fn gateway_imports_admin_provider_oauth_refresh_token_via_execution_runtime_proxy_node() {
let execution_plans = Arc::new(Mutex::new(Vec::<ExecutionPlan>::new()));
@@ -2382,20 +2427,37 @@ async fn gateway_batch_imports_admin_provider_oauth_kiro_via_execution_runtime_p
.lock()
.expect("mutex should lock")
.push(plan.clone());
assert_eq!(plan.request_id, "kiro_batch_refresh:social");
assert_eq!(plan.url, "https://oauth.example/refreshToken");
assert_eq!(
plan.proxy
.as_ref()
.and_then(|proxy| proxy.node_id.as_deref()),
Some("proxy-node-kiro-batch-runtime")
);
assert_eq!(
plan.headers
.get(EXECUTION_REQUEST_FOLLOW_REDIRECTS_HEADER)
.map(String::as_str),
Some("true")
);
if plan.request_id == "kiro_batch_refresh:social" {
assert_eq!(plan.url, "https://oauth.example/refreshToken");
assert_eq!(
plan.headers
.get(EXECUTION_REQUEST_FOLLOW_REDIRECTS_HEADER)
.map(String::as_str),
Some("true")
);
return Json(json!({
"request_id": plan.request_id,
"status_code": 200,
"headers": {
"content-type": "application/json"
},
"body": {
"json_body": {
"accessToken": sample_kiro_device_access_token("kiro-runtime@example.com"),
"refreshToken": "kiro-runtime-refresh-token-new",
"expiresIn": 1800,
}
}
}));
}
assert_eq!(plan.request_id, "kiro-quota:key-kiro-batch-runtime");
Json(json!({
"request_id": plan.request_id,
"status_code": 200,
@@ -2404,9 +2466,17 @@ async fn gateway_batch_imports_admin_provider_oauth_kiro_via_execution_runtime_p
},
"body": {
"json_body": {
"accessToken": sample_kiro_device_access_token("kiro-runtime@example.com"),
"refreshToken": "kiro-runtime-refresh-token-new",
"expiresIn": 1800,
"subscriptionInfo": {
"subscriptionTitle": "KIRO PRO+"
},
"usageBreakdownList": [{
"currentUsageWithPrecision": 5.0,
"usageLimitWithPrecision": 20.0,
"nextDateReset": 1_900_000_000u64
}],
"desktopUserInfo": {
"email": "kiro-runtime@example.com"
}
}
}
}))
@@ -2496,7 +2566,7 @@ async fn gateway_batch_imports_admin_provider_oauth_kiro_via_execution_runtime_p
{
let plans = execution_plans.lock().expect("mutex should lock");
assert_eq!(plans.len(), 1);
assert_eq!(plans.len(), 2);
}
let stored_key = provider_catalog_repository

215
apps/aether-gateway/src/tests/control/admin/pool.rs
View File

@@ -1109,6 +1109,162 @@ async fn gateway_formats_codex_quota_countdown_from_reset_after_seconds() {
);
}
#[tokio::test]
async fn gateway_marks_exhausted_codex_pool_key_as_blocked_when_flag_enabled() {
let mut provider = sample_provider("provider-codex", "codex", 10).with_transport_fields(
true,
false,
true,
None,
None,
None,
None,
None,
Some(json!({
"pool_advanced": {
"enabled": true,
"skip_exhausted_accounts": true
}
})),
);
provider.provider_type = "codex".to_string();
let mut key = sample_key(
"key-codex-exhausted",
"provider-codex",
"openai:cli",
"oauth-placeholder",
);
key.name = "codex exhausted".to_string();
key.auth_type = "oauth".to_string();
key.upstream_metadata = Some(json!({
"codex": {
"secondary_used_percent": 100.0,
"plan_type": "plus"
}
}));
let provider_catalog_repository = Arc::new(InMemoryProviderCatalogReadRepository::seed(
vec![provider],
Vec::new(),
vec![key],
));
let state = AppState::new()
.expect("gateway should build")
.with_data_state_for_tests(GatewayDataState::with_provider_catalog_reader_for_tests(
provider_catalog_repository,
));
let response = local_admin_pool_response(
&state,
http::Method::GET,
"/api/admin/pool/provider-codex/keys?page=1&page_size=50&status=all",
None,
)
.await;
assert_eq!(response.status(), StatusCode::OK);
let payload: serde_json::Value = serde_json::from_slice(
&to_bytes(response.into_body(), usize::MAX)
.await
.expect("body should read"),
)
.expect("json body should parse");
let keys = payload["keys"].as_array().expect("keys should be array");
assert_eq!(keys[0]["scheduling_status"], json!("blocked"));
assert_eq!(
keys[0]["scheduling_reason"],
json!("account_quota_exhausted")
);
assert_eq!(keys[0]["scheduling_label"], json!("额度耗尽"));
assert_eq!(
keys[0]["scheduling_reasons"][0],
json!({
"code": "account_quota_exhausted",
"label": "额度耗尽",
"blocking": true,
"source": "quota",
"ttl_seconds": serde_json::Value::Null,
"detail": serde_json::Value::Null,
})
);
assert_eq!(keys[0]["account_quota"], json!("5H剩余 0.0%"));
}
#[tokio::test]
async fn gateway_marks_exhausted_kiro_pool_key_as_blocked_when_flag_enabled() {
let mut provider = sample_provider("provider-kiro", "kiro", 10).with_transport_fields(
true,
false,
true,
None,
None,
None,
None,
None,
Some(json!({
"pool_advanced": {
"enabled": true,
"skip_exhausted_accounts": true
}
})),
);
provider.provider_type = "kiro".to_string();
let mut key = sample_key(
"key-kiro-exhausted",
"provider-kiro",
"claude:cli",
"oauth-placeholder",
);
key.name = "kiro exhausted".to_string();
key.auth_type = "oauth".to_string();
key.upstream_metadata = Some(json!({
"kiro": {
"remaining": 0.0,
"usage_limit": 100.0,
"current_usage": 100.0
}
}));
let provider_catalog_repository = Arc::new(InMemoryProviderCatalogReadRepository::seed(
vec![provider],
Vec::new(),
vec![key],
));
let state = AppState::new()
.expect("gateway should build")
.with_data_state_for_tests(GatewayDataState::with_provider_catalog_reader_for_tests(
provider_catalog_repository,
));
let response = local_admin_pool_response(
&state,
http::Method::GET,
"/api/admin/pool/provider-kiro/keys?page=1&page_size=50&status=all",
None,
)
.await;
assert_eq!(response.status(), StatusCode::OK);
let payload: serde_json::Value = serde_json::from_slice(
&to_bytes(response.into_body(), usize::MAX)
.await
.expect("body should read"),
)
.expect("json body should parse");
let keys = payload["keys"].as_array().expect("keys should be array");
assert_eq!(keys[0]["scheduling_status"], json!("blocked"));
assert_eq!(
keys[0]["scheduling_reason"],
json!("account_quota_exhausted")
);
assert_eq!(keys[0]["scheduling_label"], json!("额度耗尽"));
assert_eq!(keys[0]["account_quota"], json!("剩余 0/100"));
}
#[tokio::test]
async fn gateway_codex_quota_resets_to_full_after_countdown_elapsed() {
let mut provider = sample_provider("provider-codex", "codex", 10).with_transport_fields(
@@ -1427,6 +1583,65 @@ async fn gateway_handles_admin_pool_resolve_selection_locally_with_trusted_admin
upstream_handle.abort();
}
#[tokio::test]
async fn gateway_resolve_selection_marks_legacy_kiro_bearer_keys_as_oauth_managed() {
let mut provider = sample_provider("provider-kiro", "kiro", 10);
provider.provider_type = "kiro".to_string();
let mut key = sample_key(
"key-kiro-legacy",
"provider-kiro",
"kiro:generateAssistantResponse",
"kiro-access-token",
);
key.auth_type = "bearer".to_string();
key.encrypted_auth_config = Some(
encrypt_python_fernet_plaintext(
DEVELOPMENT_ENCRYPTION_KEY,
r#"{"provider_type":"kiro","email":"legacy-kiro@example.com","refresh_token":"legacy-kiro-refresh-token"}"#,
)
.expect("auth config ciphertext should build"),
);
let provider_catalog_repository = Arc::new(InMemoryProviderCatalogReadRepository::seed(
vec![provider],
Vec::new(),
vec![key],
));
let state = AppState::new()
.expect("gateway should build")
.with_data_state_for_tests(
GatewayDataState::with_provider_catalog_reader_for_tests(provider_catalog_repository)
.with_encryption_key_for_tests(DEVELOPMENT_ENCRYPTION_KEY),
);
let response = local_admin_pool_response(
&state,
http::Method::POST,
"/api/admin/pool/provider-kiro/keys/resolve-selection",
Some(json!({})),
)
.await;
assert_eq!(response.status(), StatusCode::OK);
let payload: serde_json::Value = serde_json::from_slice(
&to_bytes(response.into_body(), usize::MAX)
.await
.expect("body should read"),
)
.expect("json body should parse");
assert_eq!(payload["total"], json!(1));
let items = payload["items"].as_array().expect("items should be array");
assert_eq!(items.len(), 1);
assert_eq!(items[0]["key_id"], json!("key-kiro-legacy"));
assert_eq!(items[0]["auth_type"], json!("bearer"));
assert_eq!(items[0]["credential_kind"], json!("oauth_session"));
assert_eq!(items[0]["runtime_auth_kind"], json!("bearer"));
assert_eq!(items[0]["oauth_managed"], json!(true));
assert_eq!(items[0]["can_refresh_oauth"], json!(true));
assert_eq!(items[0]["can_export_oauth"], json!(true));
assert_eq!(items[0]["can_edit_oauth"], json!(true));
}
#[tokio::test]
async fn gateway_handles_admin_pool_batch_action_locally_with_trusted_admin_principal() {
let upstream_hits = Arc::new(Mutex::new(0usize));

322
apps/aether-gateway/src/tests/control/admin/provider_query.rs
View File

@@ -7,6 +7,7 @@ use aether_data_contracts::repository::provider_catalog::StoredProviderCatalogEn
use axum::body::Body;
use axum::routing::any;
use axum::{extract::Request, Json, Router};
use base64::Engine as _;
use http::StatusCode;
use serde_json::json;
@@ -20,6 +21,56 @@ use crate::constants::{
};
use crate::data::GatewayDataState;
fn crc32(data: &[u8]) -> u32 {
let mut crc = 0xffff_ffffu32;
for &byte in data {
crc ^= byte as u32;
for _ in 0..8 {
let mask = if crc & 1 == 1 { 0xedb8_8320 } else { 0 };
crc = (crc >> 1) ^ mask;
}
}
!crc
}
fn encode_string_header(name: &str, value: &str) -> Vec<u8> {
let mut out = Vec::new();
out.push(name.len() as u8);
out.extend_from_slice(name.as_bytes());
out.push(7);
out.extend_from_slice(&(value.len() as u16).to_be_bytes());
out.extend_from_slice(value.as_bytes());
out
}
fn encode_frame(headers: Vec<u8>, payload: Vec<u8>) -> Vec<u8> {
let total_len = 12 + headers.len() + payload.len() + 4;
let header_len = headers.len();
let mut out = Vec::with_capacity(total_len);
out.extend_from_slice(&(total_len as u32).to_be_bytes());
out.extend_from_slice(&(header_len as u32).to_be_bytes());
let prelude_crc = crc32(&out[..8]);
out.extend_from_slice(&prelude_crc.to_be_bytes());
out.extend_from_slice(&headers);
out.extend_from_slice(&payload);
let message_crc = crc32(&out);
out.extend_from_slice(&message_crc.to_be_bytes());
out
}
fn encode_kiro_event_frame(event_type: &str, payload: serde_json::Value) -> Vec<u8> {
let mut headers = encode_string_header(":message-type", "event");
headers.extend_from_slice(&encode_string_header(":event-type", event_type));
let payload = serde_json::to_vec(&payload).expect("payload should encode");
encode_frame(headers, payload)
}
fn encode_kiro_exception_frame(exception_type: &str) -> Vec<u8> {
let mut headers = encode_string_header(":message-type", "exception");
headers.extend_from_slice(&encode_string_header(":exception-type", exception_type));
encode_frame(headers, Vec::new())
}
async fn assert_admin_provider_query_route(
path: &str,
request_payload: serde_json::Value,
@@ -676,6 +727,277 @@ async fn gateway_handles_admin_provider_query_test_model_failover_locally_with_t
.await;
}
#[tokio::test]
async fn gateway_handles_admin_provider_query_test_model_for_kiro_locally() {
let execution_runtime = Router::new().route(
"/v1/execute/sync",
any(move |Json(plan): Json<ExecutionPlan>| async move {
assert_eq!(plan.provider_id, "provider-kiro");
assert_eq!(plan.endpoint_id, "endpoint-kiro-cli");
assert_eq!(plan.key_id, "key-kiro-primary");
assert_eq!(plan.provider_api_format, "claude:cli");
assert_eq!(plan.model_name.as_deref(), Some("claude-sonnet-4-upstream"));
Json(json!({
"request_id": plan.request_id,
"candidate_id": plan.candidate_id,
"status_code": 200,
"headers": {
"content-type": "application/vnd.amazon.eventstream"
},
"body": {
"body_bytes_b64": base64::engine::general_purpose::STANDARD.encode(
[
encode_kiro_event_frame("assistantResponseEvent", json!({"content": "Hello from Kiro"})),
encode_kiro_exception_frame("ContentLengthExceededException"),
]
.concat()
)
},
"telemetry": {
"elapsed_ms": 42
}
}))
}),
);
let (execution_runtime_url, execution_runtime_handle) = start_server(execution_runtime).await;
let mut provider = sample_provider("provider-kiro", "Kiro", 10);
provider.provider_type = "kiro".to_string();
let mut key = sample_key(
"key-kiro-primary",
"provider-kiro",
"claude:cli",
"__placeholder__",
);
key.auth_type = "oauth".to_string();
key.encrypted_auth_config = Some(
aether_crypto::encrypt_python_fernet_plaintext(
DEVELOPMENT_ENCRYPTION_KEY,
r#"{
"provider_type":"kiro",
"auth_method":"idc",
"access_token":"cached-kiro-token",
"refresh_token":"rrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrr",
"machine_id":"123e4567-e89b-12d3-a456-426614174000",
"api_region":"us-east-1",
"client_id":"client-id",
"client_secret":"client-secret"
}"#,
)
.expect("auth config should encrypt"),
);
let provider_catalog_repository = Arc::new(InMemoryProviderCatalogReadRepository::seed(
vec![provider],
vec![StoredProviderCatalogEndpoint::new(
"endpoint-kiro-cli".to_string(),
"provider-kiro".to_string(),
"claude:cli".to_string(),
Some("claude".to_string()),
Some("cli".to_string()),
true,
)
.expect("endpoint should build")
.with_transport_fields(
"https://q.{region}.amazonaws.com".to_string(),
None,
None,
None,
None,
None,
None,
None,
)
.expect("endpoint transport should build")],
vec![key],
));
let gateway = build_router_with_state(
build_state_with_execution_runtime_override(execution_runtime_url)
.with_data_state_for_tests(GatewayDataState::with_provider_transport_reader_for_tests(
provider_catalog_repository,
DEVELOPMENT_ENCRYPTION_KEY.to_string(),
)),
);
let (gateway_url, gateway_handle) = start_server(gateway).await;
let response = reqwest::Client::new()
.post(format!("{gateway_url}/api/admin/provider-query/test-model"))
.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!({
"provider_id": "provider-kiro",
"model_name": "claude-sonnet-4-upstream",
"api_format": "claude:cli"
}))
.send()
.await
.expect("request should succeed");
assert_eq!(response.status(), StatusCode::OK);
let payload: serde_json::Value = response.json().await.expect("json body should parse");
assert_eq!(payload["success"], json!(true));
assert_eq!(payload["provider"]["id"], json!("provider-kiro"));
assert_eq!(payload["model"], json!("claude-sonnet-4-upstream"));
assert_eq!(
payload["data"]["response"]["content"][0]["text"],
json!("Hello from Kiro")
);
gateway_handle.abort();
execution_runtime_handle.abort();
}
#[tokio::test]
async fn gateway_handles_admin_provider_query_test_model_failover_for_kiro_locally() {
let execution_runtime = Router::new().route(
"/v1/execute/sync",
any(move |Json(plan): Json<ExecutionPlan>| async move {
let payload = if plan.key_id == "key-kiro-first" {
json!({
"request_id": plan.request_id,
"candidate_id": plan.candidate_id,
"status_code": 429,
"headers": {
"content-type": "application/json"
},
"body": {
"json_body": {
"message": "too many requests"
}
},
"telemetry": {
"elapsed_ms": 11
}
})
} else {
json!({
"request_id": plan.request_id,
"candidate_id": plan.candidate_id,
"status_code": 200,
"headers": {
"content-type": "application/vnd.amazon.eventstream"
},
"body": {
"body_bytes_b64": base64::engine::general_purpose::STANDARD.encode(
[
encode_kiro_event_frame("assistantResponseEvent", json!({"content": "Recovered from failover"})),
encode_kiro_exception_frame("ContentLengthExceededException"),
]
.concat()
)
},
"telemetry": {
"elapsed_ms": 27
}
})
};
Json(payload)
}),
);
let (execution_runtime_url, execution_runtime_handle) = start_server(execution_runtime).await;
let mut provider = sample_provider("provider-kiro", "Kiro", 10);
provider.provider_type = "kiro".to_string();
let build_key = |id: &str| {
let mut key = sample_key(id, "provider-kiro", "claude:cli", "__placeholder__");
key.auth_type = "oauth".to_string();
key.encrypted_auth_config = Some(
aether_crypto::encrypt_python_fernet_plaintext(
DEVELOPMENT_ENCRYPTION_KEY,
r#"{
"provider_type":"kiro",
"auth_method":"idc",
"access_token":"cached-kiro-token",
"refresh_token":"rrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrr",
"machine_id":"123e4567-e89b-12d3-a456-426614174000",
"api_region":"us-east-1",
"client_id":"client-id",
"client_secret":"client-secret"
}"#,
)
.expect("auth config should encrypt"),
);
key
};
let provider_catalog_repository = Arc::new(InMemoryProviderCatalogReadRepository::seed(
vec![provider],
vec![StoredProviderCatalogEndpoint::new(
"endpoint-kiro-cli".to_string(),
"provider-kiro".to_string(),
"claude:cli".to_string(),
Some("claude".to_string()),
Some("cli".to_string()),
true,
)
.expect("endpoint should build")
.with_transport_fields(
"https://q.{region}.amazonaws.com".to_string(),
None,
None,
None,
None,
None,
None,
None,
)
.expect("endpoint transport should build")],
vec![build_key("key-kiro-first"), build_key("key-kiro-second")],
));
let gateway = build_router_with_state(
build_state_with_execution_runtime_override(execution_runtime_url)
.with_data_state_for_tests(GatewayDataState::with_provider_transport_reader_for_tests(
provider_catalog_repository,
DEVELOPMENT_ENCRYPTION_KEY.to_string(),
)),
);
let (gateway_url, gateway_handle) = start_server(gateway).await;
let response = reqwest::Client::new()
.post(format!(
"{gateway_url}/api/admin/provider-query/test-model-failover"
))
.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!({
"provider_id": "provider-kiro",
"mode": "direct",
"model_name": "claude-sonnet-4-upstream",
"failover_models": ["claude-sonnet-4-upstream"],
"api_format": "claude:cli",
"request_id": "provider-test-kiro"
}))
.send()
.await
.expect("request should succeed");
assert_eq!(response.status(), StatusCode::OK);
let payload: serde_json::Value = response.json().await.expect("json body should parse");
assert_eq!(payload["success"], json!(true));
assert_eq!(payload["total_candidates"], json!(2));
assert_eq!(payload["total_attempts"], json!(2));
let attempts = payload["attempts"]
.as_array()
.expect("attempts should be an array");
assert_eq!(attempts.len(), 2);
assert_eq!(attempts[0]["status"], json!("failed"));
assert_eq!(attempts[0]["status_code"], json!(429));
assert_eq!(attempts[1]["status"], json!("success"));
assert_eq!(
payload["data"]["response"]["content"][0]["text"],
json!("Recovered from failover")
);
gateway_handle.abort();
execution_runtime_handle.abort();
}
#[tokio::test]
async fn gateway_handles_admin_provider_query_test_model_failover_with_single_model_name_alias() {
assert_admin_provider_query_route(