Merge branch 'review-pr-612'

This commit is contained in:
elky
2026-05-30 21:26:18 +08:00
6 changed files with 351 additions and 46 deletions
@@ -943,6 +943,10 @@ fn admin_pool_scheduling_payload(
account_status_reason: Option<&str>,
account_status_source: Option<&str>,
account_quota_exhausted: bool,
oauth_status_code: Option<&str>,
oauth_status_label: Option<&str>,
oauth_status_reason: Option<&str>,
oauth_status_source: Option<&str>,
) -> (String, String, String, Vec<serde_json::Value>) {
if !key.is_active {
return (
@@ -989,6 +993,26 @@ fn admin_pool_scheduling_payload(
})],
);
}
if matches!(oauth_status_code, Some("invalid" | "expired")) {
let label = oauth_status_label.unwrap_or(if oauth_status_code == Some("expired") {
"已过期"
} else {
"已失效"
});
return (
"blocked".to_string(),
oauth_status_code.unwrap_or("invalid").to_string(),
label.to_string(),
vec![json!({
"code": oauth_status_code.unwrap_or("invalid"),
"label": label,
"blocking": true,
"source": oauth_status_source.unwrap_or("oauth"),
"ttl_seconds": serde_json::Value::Null,
"detail": oauth_status_reason,
})],
);
}
if let Some(reason) = cooldown_reason {
return (
"degraded".to_string(),
@@ -1105,6 +1129,13 @@ pub(super) fn build_admin_pool_key_payload(
.unwrap_or(false);
let account_status_source =
admin_pool_trimmed_string(account_snapshot.and_then(|item| item.get("source")));
let oauth_status_code = admin_pool_trimmed_string_from_map(oauth_snapshot, "code");
let oauth_status_label =
admin_pool_trimmed_string(oauth_snapshot.and_then(|item| item.get("label")));
let oauth_status_reason =
admin_pool_trimmed_string(oauth_snapshot.and_then(|item| item.get("reason")));
let oauth_status_source =
admin_pool_trimmed_string(oauth_snapshot.and_then(|item| item.get("source")));
let (scheduling_status, scheduling_reason, scheduling_label, scheduling_reasons) =
admin_pool_scheduling_payload(
key,
@@ -1116,6 +1147,10 @@ pub(super) fn build_admin_pool_key_payload(
account_status_reason.as_deref(),
account_status_source.as_deref(),
account_quota_exhausted,
oauth_status_code.as_deref(),
oauth_status_label.as_deref(),
oauth_status_reason.as_deref(),
oauth_status_source.as_deref(),
);
let mut payload = serde_json::Map::new();
@@ -2,13 +2,15 @@ use super::{
admin_pool_provider_id_from_path, admin_provider_pool_config, build_admin_pool_error_response,
parse_admin_pool_key_sort, parse_admin_pool_page, parse_admin_pool_page_size,
parse_admin_pool_quick_selectors, parse_admin_pool_search, parse_admin_pool_status_filter,
pool_payloads, pool_selection, read_admin_provider_pool_cooldown_key_ids,
read_admin_provider_pool_runtime_state, AdminPoolKeySort, AdminPoolKeySortDirection,
AdminPoolKeySortField, AdminProviderPoolRuntimeState, ProviderCatalogKeyListOrder,
ProviderCatalogKeyListQuery, ADMIN_POOL_PROVIDER_CATALOG_READER_UNAVAILABLE_DETAIL,
pool_payloads, pool_selection, read_admin_provider_pool_runtime_state, AdminPoolKeySort,
AdminPoolKeySortDirection, AdminPoolKeySortField, AdminProviderPoolRuntimeState,
ProviderCatalogKeyListOrder, ProviderCatalogKeyListQuery,
ADMIN_POOL_PROVIDER_CATALOG_READER_UNAVAILABLE_DETAIL,
};
use crate::ai_serving::{provider_key_pool_score_id, provider_key_pool_score_scope};
use crate::handlers::admin::provider::shared::support::AdminProviderPoolConfig;
use crate::handlers::admin::request::{AdminAppState, AdminRequestContext};
use crate::handlers::admin::shared::provider_key_status_snapshot_payload;
use crate::GatewayError;
use aether_admin::provider::pool as admin_provider_pool_pure;
use aether_data_contracts::repository::pool_scores::{
@@ -24,7 +26,7 @@ use axum::{
response::{IntoResponse, Response},
Json,
};
use serde_json::json;
use serde_json::{json, Value};
use std::{
cmp::Ordering,
collections::BTreeMap,
@@ -290,6 +292,163 @@ fn admin_pool_repository_key_order(sort: AdminPoolKeySort) -> ProviderCatalogKey
}
}
fn admin_pool_trimmed_string(value: Option<&Value>) -> Option<String> {
value
.and_then(Value::as_str)
.map(str::trim)
.filter(|value| !value.is_empty())
.map(ToOwned::to_owned)
}
fn admin_pool_account_code_status_filter(code: &str) -> Option<&'static str> {
match code.trim().to_ascii_lowercase().as_str() {
"oauth_token_invalid" => Some("invalid"),
"account_banned" | "account_suspended" => Some("account_banned"),
"account_disabled" => Some("account_disabled"),
"workspace_deactivated" => Some("workspace_deactivated"),
"account_forbidden" => Some("account_forbidden"),
"account_verification" => Some("account_verification"),
"account_quarantined" => Some("account_quarantined"),
"account_blocked" => Some("account_blocked"),
_ => None,
}
}
fn admin_pool_label_status_filter(label: &str) -> Option<&'static str> {
match label.trim() {
"已失效" | "Token 失效" | "Token失效" => Some("invalid"),
"已过期" => Some("expired"),
"账号封禁" | "账号已封禁" | "封禁" => Some("account_banned"),
"额度耗尽" => Some("quota_exhausted"),
"访问受限" | "账号访问受限" => Some("account_forbidden"),
"账号停用" => Some("account_disabled"),
"工作区停用" | "工作区已停用" => Some("workspace_deactivated"),
"需要验证" => Some("account_verification"),
"账号隔离" => Some("account_quarantined"),
"账号异常" => Some("account_blocked"),
"速率受限" => Some("rate_limited"),
"超限" => Some("cost_exhausted"),
"冷却中" => Some("cooldown"),
"已禁用" | "禁用" | "停用" => Some("inactive"),
"可用" => Some("available"),
_ => None,
}
}
fn admin_pool_oauth_status_filter(
oauth_snapshot: Option<&serde_json::Map<String, Value>>,
) -> Option<&'static str> {
let oauth_snapshot = oauth_snapshot?;
let code = admin_pool_trimmed_string(oauth_snapshot.get("code"))
.unwrap_or_default()
.to_ascii_lowercase();
match code.as_str() {
"invalid" => Some("invalid"),
"expired" => Some("expired"),
_ => admin_pool_trimmed_string(oauth_snapshot.get("label"))
.as_deref()
.and_then(admin_pool_label_status_filter),
}
}
fn admin_pool_account_status_filter(
account_snapshot: Option<&serde_json::Map<String, Value>>,
) -> Option<&'static str> {
let account_snapshot = account_snapshot?;
let code = admin_pool_trimmed_string(account_snapshot.get("code"));
let label = admin_pool_trimmed_string(account_snapshot.get("label"));
let blocked = account_snapshot
.get("blocked")
.and_then(Value::as_bool)
.unwrap_or(false);
if let Some(code) = code
.as_deref()
.and_then(admin_pool_account_code_status_filter)
{
return Some(code);
}
if let Some(label) = label.as_deref().and_then(admin_pool_label_status_filter) {
return Some(label);
}
blocked.then_some("account_blocked")
}
fn admin_pool_quota_status_filter(
quota_snapshot: Option<&serde_json::Map<String, Value>>,
) -> Option<&'static str> {
let quota_snapshot = quota_snapshot?;
let code = admin_pool_trimmed_string(quota_snapshot.get("code"))
.unwrap_or_default()
.to_ascii_lowercase();
match code.as_str() {
"banned" => Some("account_banned"),
"forbidden" => Some("account_forbidden"),
"quarantined" => Some("account_quarantined"),
"rate_limited" => Some("rate_limited"),
"exhausted" => Some("quota_exhausted"),
_ => admin_pool_trimmed_string(quota_snapshot.get("label"))
.as_deref()
.and_then(admin_pool_label_status_filter),
}
}
fn admin_pool_key_cost_exhausted(
key: &StoredProviderCatalogKey,
pool_config: Option<&AdminProviderPoolConfig>,
runtime: &AdminProviderPoolRuntimeState,
) -> bool {
let Some(limit) = pool_config
.and_then(|config| config.cost_limit_per_key_tokens)
.filter(|limit| *limit > 0)
else {
return false;
};
runtime
.cost_window_usage_by_key
.get(&key.id)
.copied()
.unwrap_or(0)
>= limit
}
fn admin_pool_key_visible_status_filter(
key: &StoredProviderCatalogKey,
provider_type: &str,
pool_config: Option<&AdminProviderPoolConfig>,
runtime: &AdminProviderPoolRuntimeState,
) -> &'static str {
let status_snapshot = provider_key_status_snapshot_payload(key, provider_type);
let account_snapshot = status_snapshot.get("account").and_then(Value::as_object);
let quota_snapshot = status_snapshot.get("quota").and_then(Value::as_object);
let oauth_snapshot = status_snapshot.get("oauth").and_then(Value::as_object);
if let Some(status) = admin_pool_account_status_filter(account_snapshot) {
return status;
}
if let Some(status) = admin_pool_quota_status_filter(quota_snapshot) {
return status;
}
if let Some(status) = admin_pool_oauth_status_filter(oauth_snapshot) {
return status;
}
if pool_config.is_some_and(|config| config.skip_exhausted_accounts)
&& admin_provider_pool_pure::admin_pool_key_account_quota_exhausted(key, provider_type)
{
return "quota_exhausted";
}
if !key.is_active {
return "inactive";
}
if runtime.cooldown_reason_by_key.contains_key(&key.id) {
return "cooldown";
}
if admin_pool_key_cost_exhausted(key, pool_config, runtime) {
return "cost_exhausted";
}
"available"
}
pub(super) async fn build_admin_pool_list_keys_response(
state: &AdminAppState<'_>,
request_context: &AdminRequestContext<'_>,
@@ -365,16 +524,12 @@ pub(super) async fn build_admin_pool_list_keys_response(
let page_offset = page.saturating_sub(1).saturating_mul(page_size);
let sort_by_score = matches!(sort.field, AdminPoolKeySortField::Score);
let (keys, total, preloaded_pool_scores_by_key_id) = if status == "cooldown" {
let cooldown_key_ids =
read_admin_provider_pool_cooldown_key_ids(state.runtime_state(), &provider.id).await;
let mut keys = if cooldown_key_ids.is_empty() {
Vec::new()
} else {
state
.read_provider_catalog_keys_by_ids(&cooldown_key_ids)
.await?
};
let (keys, total, preloaded_pool_scores_by_key_id) = if status != "all" {
let mut keys = state
.list_provider_catalog_keys_by_provider_ids(std::slice::from_ref(&provider.id))
.await?
.into_iter()
.collect::<Vec<_>>();
if let Some(keyword) = search.as_ref() {
keys.retain(|key| {
pool_selection::admin_pool_matches_search(
@@ -397,6 +552,30 @@ pub(super) async fn build_admin_pool_list_keys_response(
})
});
}
let key_ids = keys.iter().map(|key| key.id.clone()).collect::<Vec<_>>();
let runtime = match pool_config.as_ref() {
Some(pool_config) if !key_ids.is_empty() => {
read_admin_provider_pool_runtime_state(
state.runtime_state(),
&provider.id,
&key_ids,
pool_config,
None,
)
.await
}
_ => AdminProviderPoolRuntimeState::default(),
};
keys.retain(|key| {
admin_pool_key_visible_status_filter(
key,
&provider.provider_type,
pool_config.as_ref(),
&runtime,
) == status
});
let preloaded_pool_scores_by_key_id = if sort_by_score {
let key_ids = keys.iter().map(|key| key.id.clone()).collect::<Vec<_>>();
let scores = read_admin_pool_scores_by_key_id(state, &provider.id, &key_ids)
@@ -420,11 +599,6 @@ pub(super) async fn build_admin_pool_list_keys_response(
.list_provider_catalog_keys_by_provider_ids(std::slice::from_ref(&provider.id))
.await?
.into_iter()
.filter(|key| match status.as_str() {
"active" => key.is_active,
"inactive" => !key.is_active,
_ => true,
})
.filter(|key| {
pool_selection::admin_pool_matches_search(
state,
@@ -467,11 +641,7 @@ pub(super) async fn build_admin_pool_list_keys_response(
.list_provider_catalog_key_page(&ProviderCatalogKeyListQuery {
provider_id: provider.id.clone(),
search: search.clone(),
is_active: match status.as_str() {
"active" => Some(true),
"inactive" => Some(false),
_ => None,
},
is_active: None,
offset: page_offset,
limit: page_size,
order: admin_pool_repository_key_order(sort),
@@ -107,8 +107,29 @@ pub(crate) fn parse_admin_pool_status_filter(query: Option<&str>) -> Result<Stri
.trim()
.to_ascii_lowercase();
match value.as_str() {
"all" | "active" | "inactive" | "cooldown" => Ok(value),
_ => Err("status must be one of: all, active, cooldown, inactive".to_string()),
"all"
| "available"
| "cooldown"
| "inactive"
| "invalid"
| "expired"
| "account_banned"
| "quota_exhausted"
| "account_forbidden"
| "account_disabled"
| "workspace_deactivated"
| "account_verification"
| "account_quarantined"
| "account_blocked"
| "rate_limited"
| "cost_exhausted" => Ok(value),
// Backward compatibility for older links/localStorage. This filter used to
// be called "active", but the table status column displays the exact
// scheduling/account state. Treat it as the visible "可用" bucket.
"active" => Ok("available".to_string()),
_ => Err(
"status must be one of: all, available, cooldown, inactive, invalid, expired, account_banned, quota_exhausted, account_forbidden, account_disabled, workspace_deactivated, account_verification, account_quarantined, account_blocked, rate_limited, cost_exhausted".to_string(),
),
}
}