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

Merge pull request #681 from zhefox/main

Browse Source 
Codex 重置功能和显示缓存修复以及批量key的导入和管理功能
This commit is contained in:
ZheFox
2026-07-16 19:48:33 +08:00
committed by GitHub
parent 9ea84f9748 6664c2dbb8
commit 3f5f65eb9a
19 changed files with 2216 additions and 131 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/aether-gateway/src/handlers/admin/provider/oauth/quota/codex/mod.rs
+64 -20
View File
@@ -51,14 +51,6 @@ fn merge_codex_quota_metadata(
serde_json::Value::Object(merged)
}
fn codex_reset_credits_available_count(metadata: &Map<String, Value>) -> Option<u64> {
metadata
.get("reset_credits")
.and_then(Value::as_object)
.and_then(|reset_credits| reset_credits.get("available_count"))
.and_then(aether_admin::provider::quota::coerce_json_u64)
}
fn truncate_codex_reset_credit_detail_error(message: impl Into<String>) -> String {
let message = message.into();
let mut sanitized = message.replace('\n', " ");
@@ -85,11 +77,7 @@ fn merge_codex_reset_credit_detail_metadata(
.cloned()
.unwrap_or_default();
let has_usage_available_count = reset_credits.contains_key("available_count");
for (key, value) in detail_reset_credits {
if key == "available_count" && has_usage_available_count {
continue;
}
reset_credits.insert(key.clone(), value.clone());
}
codex_metadata.insert("reset_credits".to_string(), Value::Object(reset_credits));
@@ -97,6 +85,7 @@ fn merge_codex_reset_credit_detail_metadata(
fn mark_codex_reset_credit_detail_failed(
codex_metadata: &mut Map<String, Value>,
updated_at_unix_secs: u64,
detail_error: impl Into<String>,
) {
let mut reset_credits = codex_metadata
@@ -104,6 +93,7 @@ fn mark_codex_reset_credit_detail_failed(
.and_then(Value::as_object)
.cloned()
.unwrap_or_default();
reset_credits.insert("updated_at".to_string(), json!(updated_at_unix_secs));
reset_credits.insert("detail_source".to_string(), json!("wham_readonly"));
reset_credits.insert("detail_status".to_string(), json!("failed"));
reset_credits.insert(
@@ -124,16 +114,11 @@ async fn enrich_codex_reset_credit_details(
codex_metadata: &mut Map<String, Value>,
now_unix_secs: u64,
) -> Result<(), GatewayError> {
let available_count = codex_reset_credits_available_count(codex_metadata).unwrap_or(0);
if available_count == 0 {
return Ok(());
}
let request_spec = match build_codex_reset_credits_request_spec(transport, resolved_oauth_auth)
{
Ok(request_spec) => request_spec,
Err(message) => {
mark_codex_reset_credit_detail_failed(codex_metadata, message);
mark_codex_reset_credit_detail_failed(codex_metadata, now_unix_secs, message);
return Ok(());
}
};
@@ -146,6 +131,7 @@ async fn enrich_codex_reset_credit_details(
ProviderQuotaExecutionOutcome::Failure(detail) => {
mark_codex_reset_credit_detail_failed(
codex_metadata,
now_unix_secs,
format!("reset credit detail 请求执行失败: {detail}"),
);
return Ok(());
@@ -157,6 +143,7 @@ async fn enrich_codex_reset_credit_details(
.unwrap_or_else(|| format!("HTTP {}", result.status_code));
mark_codex_reset_credit_detail_failed(
codex_metadata,
now_unix_secs,
format!(
"reset credit detail 返回状态码 {}: {detail}",
result.status_code
@@ -170,7 +157,11 @@ async fn enrich_codex_reset_credit_details(
.as_ref()
.and_then(|body| body.json_body.as_ref())
else {
mark_codex_reset_credit_detail_failed(codex_metadata, "无法解析 reset credit detail 响应");
mark_codex_reset_credit_detail_failed(
codex_metadata,
now_unix_secs,
"无法解析 reset credit detail 响应",
);
return Ok(());
};
if let Some(detail_metadata) =
@@ -178,7 +169,11 @@ async fn enrich_codex_reset_credit_details(
{
merge_codex_reset_credit_detail_metadata(codex_metadata, &detail_metadata);
} else {
mark_codex_reset_credit_detail_failed(codex_metadata, "reset credit detail 响应为空");
mark_codex_reset_credit_detail_failed(
codex_metadata,
now_unix_secs,
"reset credit detail 响应为空",
);
}
Ok(())
@@ -776,3 +771,52 @@ pub(crate) async fn refresh_codex_provider_quota_locally(
"auto_removed_hard_banned": auto_removed_hard_banned_count,
})))
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn codex_reset_credit_detail_count_overrides_usage_count() {
let mut metadata = json!({
"reset_credits": {
"available_count": 0,
"detail_source": "wham_usage"
}
})
.as_object()
.cloned()
.expect("metadata object");
let detail = json!({
"reset_credits": {
"available_count": 2,
"detail_source": "wham_readonly"
}
});
merge_codex_reset_credit_detail_metadata(&mut metadata, &detail);
assert_eq!(
metadata
.get("reset_credits")
.and_then(Value::as_object)
.and_then(|credits| credits.get("available_count")),
Some(&json!(2u64))
);
}
#[test]
fn codex_reset_credit_detail_failure_records_attempt_time() {
let mut metadata = Map::new();
mark_codex_reset_credit_detail_failed(&mut metadata, 1_777_000_000, "request failed");
assert_eq!(
metadata
.get("reset_credits")
.and_then(Value::as_object)
.and_then(|credits| credits.get("updated_at")),
Some(&json!(1_777_000_000u64))
);
}
}
apps/aether-gateway/src/handlers/admin/provider/pool_admin/batch_routes/import.rs
-7
View File
@@ -55,13 +55,6 @@ pub(super) async fn build_admin_pool_batch_import_response(
}
};
if payload.keys.len() > 500 {
return Ok(build_admin_pool_error_response(
http::StatusCode::BAD_REQUEST,
"keys length must be less than or equal to 500",
));
}
state
.build_admin_pool_batch_import_response(&provider_id, payload)
.await
apps/aether-gateway/src/handlers/admin/request/provider/tasks.rs
+157 -21
View File
@@ -12,7 +12,8 @@ use axum::{
response::{IntoResponse, Response},
Json,
};
use serde_json::json;
use serde_json::{json, Map, Value};
use std::collections::BTreeSet;
impl<'a> AdminAppState<'a> {
pub(crate) async fn clear_admin_provider_pool_cooldown(&self, provider_id: &str, key_id: &str) {
@@ -105,9 +106,9 @@ impl<'a> AdminAppState<'a> {
let existing_keys = self
.list_provider_catalog_keys_by_provider_ids(std::slice::from_ref(&provider.id))
.await?;
let api_formats =
let available_api_formats =
admin_provider_pool_pure::admin_pool_resolved_api_formats(&endpoints, &existing_keys);
if api_formats.is_empty() {
if available_api_formats.is_empty() {
return Ok((
http::StatusCode::BAD_REQUEST,
Json(json!({ "detail": "Provider 没有可用 endpoint 或现有 key，无法推断 api_formats" })),
@@ -115,8 +116,74 @@ impl<'a> AdminAppState<'a> {
.into_response());
}
let proxy =
admin_provider_pool_pure::admin_pool_key_proxy_value(payload.proxy_node_id.as_deref());
let requested_api_formats = payload
.api_formats
.iter()
.map(|value| value.trim().to_string())
.filter(|value| !value.is_empty())
.collect::<BTreeSet<_>>()
.into_iter()
.collect::<Vec<_>>();
let available_api_format_set = available_api_formats
.iter()
.cloned()
.collect::<BTreeSet<_>>();
let api_formats = if requested_api_formats.is_empty() {
available_api_formats.clone()
} else {
if let Some(unsupported) = requested_api_formats
.iter()
.find(|value| !available_api_format_set.contains(*value))
{
return Ok((
http::StatusCode::BAD_REQUEST,
Json(json!({ "detail": format!("Provider 不支持 api_format: {unsupported}") })),
)
.into_response());
}
requested_api_formats
};
let mut settings_map = match payload.settings {
Some(Value::Object(map)) => map,
Some(_) => {
return Ok((
http::StatusCode::BAD_REQUEST,
Json(json!({ "detail": "settings payload must be an object" })),
)
.into_response());
}
None => Map::new(),
};
if let Some(proxy_node_id) = payload.proxy_node_id {
settings_map
.entry("proxy_node_id".to_string())
.or_insert(Value::String(proxy_node_id));
}
let shared_settings = (!settings_map.is_empty()).then_some(Value::Object(settings_map));
if let Some(settings) = shared_settings.as_ref() {
if let Err(detail) =
admin_provider_pool_pure::validate_admin_pool_key_settings_payload(settings)
{
return Ok((
http::StatusCode::BAD_REQUEST,
Json(json!({ "detail": detail })),
)
.into_response());
}
}
let mut known_names = existing_keys
.iter()
.map(|key| key.name.trim().to_string())
.filter(|name| !name.is_empty())
.collect::<BTreeSet<_>>();
let mut known_api_keys = existing_keys
.iter()
.filter_map(|key| key.encrypted_api_key.as_deref())
.filter_map(|ciphertext| self.decrypt_catalog_secret_with_fallbacks(ciphertext))
.filter(|value| value != "__placeholder__")
.collect::<BTreeSet<_>>();
let mut imported = 0usize;
let skipped = 0usize;
let mut errors = Vec::new();
@@ -136,6 +203,79 @@ impl<'a> AdminAppState<'a> {
continue;
}
let name = item.name.trim();
if name.is_empty() {
errors.push(json!({
"index": index,
"reason": "name is empty",
}));
continue;
}
if known_names.contains(name) {
errors.push(json!({
"index": index,
"reason": "该名称已存在于当前 Provider 或本次导入中",
}));
continue;
}
let auth_type = item.auth_type.trim().to_ascii_lowercase();
let auth_type = if auth_type.is_empty() {
"api_key".to_string()
} else {
auth_type
};
if !matches!(auth_type.as_str(), "api_key" | "bearer") {
errors.push(json!({
"index": index,
"reason": "auth_type must be api_key or bearer",
}));
continue;
}
let requested_item_api_formats = item
.api_formats
.iter()
.map(|value| value.trim().to_string())
.filter(|value| !value.is_empty())
.collect::<BTreeSet<_>>()
.into_iter()
.collect::<Vec<_>>();
let item_api_formats = if requested_item_api_formats.is_empty() {
api_formats.clone()
} else {
if let Some(unsupported) = requested_item_api_formats
.iter()
.find(|value| !available_api_format_set.contains(*value))
{
errors.push(json!({
"index": index,
"reason": format!("Provider 不支持 api_format: {unsupported}"),
}));
continue;
}
requested_item_api_formats
};
let item_settings = match admin_provider_pool_pure::resolve_admin_pool_key_settings(
shared_settings.as_ref(),
item.settings.as_ref(),
) {
Ok(value) => value,
Err(detail) => {
errors.push(json!({
"index": index,
"reason": detail,
}));
continue;
}
};
if known_api_keys.contains(api_key) {
errors.push(json!({
"index": index,
"reason": "该 API Key 已存在于当前 Provider 或本次导入中",
}));
continue;
}
let Some(encrypted_api_key) = self.encrypt_catalog_secret_with_fallbacks(api_key)
else {
errors.push(json!({
@@ -144,26 +284,15 @@ impl<'a> AdminAppState<'a> {
}));
continue;
};
let auth_type = item.auth_type.trim().to_ascii_lowercase();
let auth_type = if auth_type.is_empty() {
"api_key".to_string()
} else {
auth_type
};
let name = item.name.trim();
let record = match admin_provider_pool_pure::build_admin_pool_batch_import_key_record(
uuid::Uuid::new_v4().to_string(),
provider.id.clone(),
if name.is_empty() {
format!("imported-{index}")
} else {
name.to_string()
},
name.to_string(),
auth_type,
api_formats.clone(),
item_api_formats,
encrypted_api_key,
proxy.clone(),
None,
item_settings.as_ref(),
now_unix_secs,
) {
Ok(value) => value,
@@ -175,7 +304,6 @@ impl<'a> AdminAppState<'a> {
continue;
}
};
let Some(_) = self.create_provider_catalog_key(&record).await? else {
return Ok((
http::StatusCode::SERVICE_UNAVAILABLE,
@@ -185,6 +313,8 @@ impl<'a> AdminAppState<'a> {
)
.into_response());
};
known_names.insert(name.to_string());
known_api_keys.insert(api_key.to_string());
imported += 1;
}
@@ -473,6 +603,12 @@ impl<'a> AdminAppState<'a> {
AdminPoolBatchActionKind::Disable => key.is_active = false,
AdminPoolBatchActionKind::ClearProxy => key.proxy = None,
AdminPoolBatchActionKind::SetProxy => key.proxy = plan.proxy_payload.clone(),
AdminPoolBatchActionKind::UpdateSettings => {
if let Some(settings) = plan.settings_payload.as_ref() {
admin_provider_pool_pure::apply_admin_pool_key_settings(&mut key, settings)
.map_err(GatewayError::Internal)?;
}
}
AdminPoolBatchActionKind::RegenerateFingerprint => {
key.fingerprint =
Some(aether_provider_transport::claude_code::generate_random_fingerprint())
apps/aether-gateway/src/handlers/shared/catalog.rs
+83 -1
View File
@@ -2054,6 +2054,29 @@ fn quota_snapshot_has_materialized_data(
})
}
fn codex_upstream_metadata_is_at_least_as_fresh(
quota_snapshot: Option<&Map<String, Value>>,
upstream_metadata: Option<&Value>,
) -> bool {
let Some(metadata) = provider_quota_metadata_bucket(upstream_metadata, "codex") else {
return false;
};
let Some(metadata_updated_at) = metadata
.get("updated_at")
.and_then(admin_provider_quota_pure::coerce_json_u64)
else {
return false;
};
let snapshot_updated_at = quota_snapshot.and_then(|quota| {
quota
.get("updated_at")
.or_else(|| quota.get("observed_at"))
.and_then(admin_provider_quota_pure::coerce_json_u64)
});
snapshot_updated_at.is_none_or(|updated_at| metadata_updated_at >= updated_at)
}
fn windsurf_quota_snapshot_has_stale_cooldown(quota_snapshot: &Map<String, Value>) -> bool {
let code = quota_snapshot
.get("code")
@@ -2121,8 +2144,15 @@ pub(crate) fn provider_key_status_snapshot_payload(
.and_then(Value::as_object)
.and_then(|snapshot| snapshot.get("quota"))
.and_then(Value::as_object);
let refresh_codex_snapshot = provider_type.trim().eq_ignore_ascii_case("codex")
&& codex_upstream_metadata_is_at_least_as_fresh(
quota_snapshot,
key.upstream_metadata.as_ref(),
);
let payload = if quota_snapshot_has_materialized_data(quota_snapshot, provider_type) {
let payload = if quota_snapshot_has_materialized_data(quota_snapshot, provider_type)
&& !refresh_codex_snapshot
{
status_snapshot
.cloned()
.unwrap_or_else(default_provider_key_status_snapshot)
@@ -3512,6 +3542,58 @@ mod tests {
);
}
#[test]
fn provider_key_status_snapshot_payload_restores_complete_codex_cache() {
let mut key = sample_catalog_key();
key.upstream_metadata = Some(json!({
"codex": {
"updated_at": 200u64,
"plan_type": "plus",
"primary_used_percent": 89.0,
"primary_reset_at": 1_900_000_000u64,
"spark_primary_used_percent": 40.0,
"spark_primary_reset_at": 1_900_100_000u64,
"reset_credits": {
"available_count": 3,
"updated_at": 200u64,
"detail_status": "available",
"credits": []
}
}
}));
key.status_snapshot = Some(json!({
"quota": {
"version": 2,
"provider_type": "codex",
"updated_at": 200u64,
"windows": [{
"code": "weekly",
"used_ratio": 1.0,
"reset_at": 1_900_000_000u64
}]
}
}));
let payload = provider_key_status_snapshot_payload(&key, "codex");
let windows = payload["quota"]["windows"]
.as_array()
.expect("quota windows should exist");
assert_eq!(payload.pointer("/quota/plan_type"), Some(&json!("plus")));
assert_eq!(
payload.pointer("/quota/reset_credits/available_count"),
Some(&json!(3u64))
);
assert!(windows.iter().any(|window| window["code"] == "spark_5h"));
assert_eq!(
windows
.iter()
.find(|window| window["code"] == "weekly")
.and_then(|window| window.get("used_ratio")),
Some(&json!(0.89))
);
}
#[test]
fn sync_provider_key_quota_status_snapshot_preserves_codex_usage_state() {
let current_status_snapshot = json!({
apps/aether-gateway/src/tests/control/admin/endpoints/quota.rs
+31 -27
View File
@@ -70,7 +70,7 @@ async fn gateway_refreshes_admin_provider_quota_locally_for_codex_with_trusted_a
}),
);
let seen_execution_runtime = Arc::new(Mutex::new(None::<SeenExecutionRuntimeRequest>));
let seen_execution_runtime = Arc::new(Mutex::new(Vec::<SeenExecutionRuntimeRequest>::new()));
let seen_execution_runtime_clone = Arc::clone(&seen_execution_runtime);
let execution_runtime = Router::new().route(
"/v1/execute/sync",
@@ -83,21 +83,22 @@ async fn gateway_refreshes_admin_provider_quota_locally_for_codex_with_trusted_a
.expect("body should read"),
)
.expect("plan should parse");
*seen_execution_runtime_inner
seen_execution_runtime_inner
.lock()
.expect("mutex should lock") = Some(SeenExecutionRuntimeRequest {
url: plan.url.clone(),
authorization: plan
.headers
.get("authorization")
.cloned()
.unwrap_or_default(),
provider_api_format: plan.provider_api_format.clone(),
total_ms: plan
.timeouts
.as_ref()
.and_then(|timeouts| timeouts.total_ms),
});
.expect("mutex should lock")
.push(SeenExecutionRuntimeRequest {
url: plan.url.clone(),
authorization: plan
.headers
.get("authorization")
.cloned()
.unwrap_or_default(),
provider_api_format: plan.provider_api_format.clone(),
total_ms: plan
.timeouts
.as_ref()
.and_then(|timeouts| timeouts.total_ms),
});
let result = aether_contracts::ExecutionResult {
request_id: plan.request_id,
candidate_id: None,
@@ -223,24 +224,24 @@ async fn gateway_refreshes_admin_provider_quota_locally_for_codex_with_trusted_a
);
assert_eq!(*upstream_hits.lock().expect("mutex should lock"), 0);
let seen_execution_runtime_request = seen_execution_runtime
let seen_execution_runtime_requests = seen_execution_runtime
.lock()
.expect("mutex should lock")
.clone()
.expect("execution runtime request should be captured");
.clone();
assert_eq!(seen_execution_runtime_requests.len(), 2);
assert_eq!(
seen_execution_runtime_request.url,
seen_execution_runtime_requests[0].url,
"https://chatgpt.com/backend-api/wham/usage"
);
assert_eq!(
seen_execution_runtime_request.authorization,
"Bearer sk-codex-123"
seen_execution_runtime_requests[1].url,
"https://chatgpt.com/backend-api/wham/rate-limit-reset-credits"
);
assert_eq!(
seen_execution_runtime_request.provider_api_format,
"openai:responses"
);
assert_eq!(seen_execution_runtime_request.total_ms, Some(30_000));
for request in seen_execution_runtime_requests {
assert_eq!(request.authorization, "Bearer sk-codex-123");
assert_eq!(request.provider_api_format, "openai:responses");
assert_eq!(request.total_ms, Some(30_000));
}
let reloaded = provider_catalog_repository
.list_keys_by_ids(&["key-codex-a".to_string()])
@@ -679,7 +680,10 @@ async fn gateway_refreshes_admin_provider_quota_locally_for_requested_codex_keys
.lock()
.expect("mutex should lock")
.clone(),
vec!["Bearer sk-codex-a".to_string()]
vec![
"Bearer sk-codex-a".to_string(),
"Bearer sk-codex-a".to_string(),
]
);
let reloaded = provider_catalog_repository