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

feat: combine usage quota and pool stats updates

Browse Source
This commit is contained in:
fawney19
2026-05-07 01:26:42 +08:00
parent 012f8bcdf7
commit 1f5b294bd0
35 changed files with 3472 additions and 122 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
apps/aether-gateway/src/admin_api.rs
View File

@@ -3,12 +3,12 @@ pub(crate) use crate::handlers::admin::{
build_internal_control_error_response, create_provider_oauth_catalog_key,
find_duplicate_provider_oauth_key, maybe_build_local_admin_pool_response,
maybe_build_local_admin_response, provider_oauth_runtime_endpoint_for_provider,
refresh_antigravity_provider_quota_locally, refresh_codex_provider_quota_locally,
refresh_kiro_provider_quota_locally, refresh_provider_oauth_account_state_after_update,
update_existing_provider_oauth_catalog_key, AdminAppState,
AdminGatewayProviderTransportSnapshot, AdminLocalOAuthRefreshError, AdminRequestContext,
AdminRouteRequest, AdminRouteResponse, AdminRouteResult, AdminStatsTimeRange,
AdminStatsUsageFilter,
refresh_antigravity_provider_quota_locally, refresh_chatgpt_web_provider_quota_locally,
refresh_codex_provider_quota_locally, refresh_kiro_provider_quota_locally,
refresh_provider_oauth_account_state_after_update, update_existing_provider_oauth_catalog_key,
AdminAppState, AdminGatewayProviderTransportSnapshot, AdminLocalOAuthRefreshError,
AdminRequestContext, AdminRouteRequest, AdminRouteResponse, AdminRouteResult,
AdminStatsTimeRange, AdminStatsUsageFilter,
};
use crate::handlers::admin::{

19
apps/aether-gateway/src/data/state/runtime.rs
View File

@@ -30,8 +30,9 @@ use super::{
WalletLookupKey, WalletMutationOutcome,
};
use aether_data_contracts::repository::usage::{
PendingUsageCleanupSummary, StoredUsageDailySummary, UsageAuditListQuery, UsageCleanupSummary,
UsageCleanupWindow, UsageDailyHeatmapQuery,
PendingUsageCleanupSummary, ProviderApiKeyWindowUsageRequest,
StoredProviderApiKeyWindowUsageSummary, StoredUsageDailySummary, UsageAuditListQuery,
UsageCleanupSummary, UsageCleanupWindow, UsageDailyHeatmapQuery,
};
use aether_video_tasks_core::read_data_backed_video_task_response;
@@ -1366,6 +1367,20 @@ impl GatewayDataState {
}
}
pub(crate) async fn summarize_usage_by_provider_api_key_windows(
&self,
requests: &[ProviderApiKeyWindowUsageRequest],
) -> Result<Vec<StoredProviderApiKeyWindowUsageSummary>, DataLayerError> {
match &self.usage_reader {
Some(repository) => {
repository
.summarize_usage_by_provider_api_key_windows(requests)
.await
}
None => Ok(Vec::new()),
}
}
pub(crate) async fn list_users_by_ids(
&self,
user_ids: &[String],

1
apps/aether-gateway/src/handlers/admin/mod.rs
View File

@@ -26,6 +26,7 @@ pub(crate) use self::provider::oauth::provisioning::{
create_provider_oauth_catalog_key, update_existing_provider_oauth_catalog_key,
};
pub(crate) use self::provider::oauth::quota::antigravity::refresh_antigravity_provider_quota_locally;
pub(crate) use self::provider::oauth::quota::chatgpt_web::refresh_chatgpt_web_provider_quota_locally;
pub(crate) use self::provider::oauth::quota::codex::refresh_codex_provider_quota_locally;
pub(crate) use self::provider::oauth::quota::kiro::refresh_kiro_provider_quota_locally;
pub(crate) use self::provider::oauth::runtime::{

13
apps/aether-gateway/src/handlers/admin/provider/endpoint_keys/quota.rs
View File

@@ -14,6 +14,7 @@ use serde_json::json;
use std::collections::BTreeSet;
use super::super::oauth::quota::antigravity::refresh_antigravity_provider_quota_locally;
use super::super::oauth::quota::chatgpt_web::refresh_chatgpt_web_provider_quota_locally;
use super::super::oauth::quota::codex::refresh_codex_provider_quota_locally;
use super::super::oauth::quota::kiro::refresh_kiro_provider_quota_locally;
use super::super::oauth::quota::shared::normalize_string_id_list;
@@ -110,6 +111,13 @@ pub(super) async fn maybe_handle(
})
.cloned()
.or_else(|| endpoints.into_iter().find(|endpoint| endpoint.is_active)),
"chatgpt_web" => endpoints.into_iter().find(|endpoint| {
endpoint.is_active
&& endpoint
.api_format
.trim()
.eq_ignore_ascii_case("openai:image")
}),
_ => return Ok(None),
};
@@ -118,6 +126,7 @@ pub(super) async fn maybe_handle(
"codex" => "找不到有效的 openai:responses 端点",
"antigravity" => "找不到有效的 gemini:generate_content 端点",
"kiro" => "找不到有效的 Kiro 端点",
"chatgpt_web" => "找不到有效的 openai:image 端点",
_ => "找不到有效端点",
};
return Ok(Some(
@@ -198,6 +207,10 @@ pub(super) async fn maybe_handle(
refresh_antigravity_provider_quota_locally(state, &provider, &endpoint, keys, None)
.await?
}
"chatgpt_web" => {
refresh_chatgpt_web_provider_quota_locally(state, &provider, &endpoint, keys, None)
.await?
}
_ => None,
}) else {
return Ok(None);

521
apps/aether-gateway/src/handlers/admin/provider/oauth/quota/chatgpt_web.rs Normal file
View File

@@ -0,0 +1,521 @@
use super::shared::{
build_quota_snapshot_payload, default_provider_quota_execution_timeouts,
execute_provider_quota_plan, extract_execution_error_message,
persist_provider_quota_refresh_state, quota_refresh_success_invalid_state,
ProviderQuotaExecutionOutcome,
};
use crate::handlers::admin::provider::shared::payloads::{
OAUTH_ACCOUNT_BLOCK_PREFIX, OAUTH_EXPIRED_PREFIX,
};
use crate::handlers::admin::request::{AdminAppState, AdminGatewayProviderTransportSnapshot};
use crate::GatewayError;
use aether_admin::provider::quota::parse_chatgpt_web_conversation_init_response;
use aether_contracts::{
ExecutionPlan, ProxySnapshot, RequestBody, EXECUTION_REQUEST_ACCEPT_INVALID_CERTS_HEADER,
};
use aether_data_contracts::repository::provider_catalog::{
StoredProviderCatalogEndpoint, StoredProviderCatalogKey, StoredProviderCatalogProvider,
};
use serde_json::json;
use std::collections::BTreeMap;
use std::time::{SystemTime, UNIX_EPOCH};
const CHATGPT_WEB_DEFAULT_BASE_URL: &str = "https://chatgpt.com";
const CHATGPT_WEB_CONVERSATION_INIT_PATH: &str = "/backend-api/conversation/init";
const CHATGPT_WEB_USER_AGENT: &str = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/143.0.0.0 Safari/537.36 Edg/143.0.0.0";
const CHATGPT_WEB_CLIENT_VERSION: &str = "prod-be885abbfcfe7b1f511e88b3003d9ee44757fbad";
const CHATGPT_WEB_BUILD_NUMBER: &str = "5955942";
const CHATGPT_WEB_SEC_CH_UA: &str =
r#""Microsoft Edge";v="143", "Chromium";v="143", "Not A(Brand";v="24""#;
const PLACEHOLDER_API_KEY: &str = "__placeholder__";
const CHATGPT_WEB_FREE_IMAGE_QUOTA_LIMIT: f64 = 25.0;
fn chatgpt_web_base_url(endpoint: &StoredProviderCatalogEndpoint) -> String {
let base_url = endpoint.base_url.trim().trim_end_matches('/');
if base_url.is_empty() {
CHATGPT_WEB_DEFAULT_BASE_URL.to_string()
} else {
base_url.to_string()
}
}
fn build_chatgpt_web_quota_headers(
authorization: (String, String),
base_url: &str,
) -> BTreeMap<String, String> {
let device_id = uuid::Uuid::new_v4().to_string();
let session_id = uuid::Uuid::new_v4().to_string();
let mut headers = BTreeMap::from([
("accept".to_string(), "application/json".to_string()),
("content-type".to_string(), "application/json".to_string()),
("user-agent".to_string(), CHATGPT_WEB_USER_AGENT.to_string()),
("origin".to_string(), base_url.to_string()),
("referer".to_string(), format!("{base_url}/")),
(
"accept-language".to_string(),
"zh-CN,zh;q=0.9,en;q=0.8,en-US;q=0.7".to_string(),
),
("cache-control".to_string(), "no-cache".to_string()),
("pragma".to_string(), "no-cache".to_string()),
("priority".to_string(), "u=1, i".to_string()),
("sec-ch-ua".to_string(), CHATGPT_WEB_SEC_CH_UA.to_string()),
("sec-ch-ua-arch".to_string(), r#""x86""#.to_string()),
("sec-ch-ua-bitness".to_string(), r#""64""#.to_string()),
("sec-ch-ua-mobile".to_string(), "?0".to_string()),
("sec-ch-ua-model".to_string(), r#""""#.to_string()),
("sec-ch-ua-platform".to_string(), r#""Windows""#.to_string()),
(
"sec-ch-ua-platform-version".to_string(),
r#""19.0.0""#.to_string(),
),
("sec-fetch-dest".to_string(), "empty".to_string()),
("sec-fetch-mode".to_string(), "cors".to_string()),
("sec-fetch-site".to_string(), "same-origin".to_string()),
("oai-device-id".to_string(), device_id),
("oai-session-id".to_string(), session_id),
("oai-language".to_string(), "zh-CN".to_string()),
(
"oai-client-version".to_string(),
CHATGPT_WEB_CLIENT_VERSION.to_string(),
),
(
"oai-client-build-number".to_string(),
CHATGPT_WEB_BUILD_NUMBER.to_string(),
),
(
"x-openai-target-path".to_string(),
CHATGPT_WEB_CONVERSATION_INIT_PATH.to_string(),
),
(
"x-openai-target-route".to_string(),
CHATGPT_WEB_CONVERSATION_INIT_PATH.to_string(),
),
(
EXECUTION_REQUEST_ACCEPT_INVALID_CERTS_HEADER.to_string(),
"true".to_string(),
),
]);
headers.insert(authorization.0.to_ascii_lowercase(), authorization.1);
headers
}
fn chatgpt_web_auth_config(
transport: &AdminGatewayProviderTransportSnapshot,
) -> Option<serde_json::Value> {
transport
.key
.decrypted_auth_config
.as_deref()
.map(str::trim)
.filter(|value| !value.is_empty())
.and_then(|value| serde_json::from_str::<serde_json::Value>(value).ok())
}
fn chatgpt_web_auth_config_string(
auth_config: Option<&serde_json::Value>,
fields: &[&str],
) -> Option<String> {
let object = auth_config.and_then(serde_json::Value::as_object)?;
fields.iter().find_map(|field| {
object
.get(*field)
.and_then(serde_json::Value::as_str)
.map(str::trim)
.filter(|value| !value.is_empty())
.map(ToOwned::to_owned)
})
}
fn enrich_chatgpt_web_quota_metadata(
metadata: &mut serde_json::Value,
auth_config: Option<&serde_json::Value>,
) {
let Some(object) = metadata.as_object_mut() else {
return;
};
for (target, fields) in [
("plan_type", &["plan_type", "tier", "plan"][..]),
("email", &["email"][..]),
("account_id", &["account_id", "accountId"][..]),
("account_user_id", &["account_user_id", "accountUserId"][..]),
("user_id", &["user_id", "userId"][..]),
] {
if object.contains_key(target) {
continue;
}
if let Some(value) = chatgpt_web_auth_config_string(auth_config, fields) {
object.insert(target.to_string(), json!(value));
}
}
}
fn chatgpt_web_json_number(value: Option<&serde_json::Value>) -> Option<f64> {
let value = value?;
if let Some(number) = value.as_f64() {
return number.is_finite().then_some(number);
}
value
.as_str()
.map(str::trim)
.filter(|value| !value.is_empty())
.and_then(|value| value.parse::<f64>().ok())
.filter(|value| value.is_finite())
}
fn chatgpt_web_json_string(value: Option<&serde_json::Value>) -> Option<&str> {
value
.and_then(serde_json::Value::as_str)
.map(str::trim)
.filter(|value| !value.is_empty())
}
fn existing_chatgpt_web_image_quota_limit(
upstream_metadata: Option<&serde_json::Value>,
) -> Option<f64> {
upstream_metadata
.and_then(serde_json::Value::as_object)
.and_then(|metadata| metadata.get("chatgpt_web"))
.and_then(serde_json::Value::as_object)
.and_then(|bucket| chatgpt_web_json_number(bucket.get("image_quota_total")))
.filter(|value| *value > 0.0)
}
fn infer_chatgpt_web_image_quota_limit(
plan_type: Option<&str>,
remaining: Option<f64>,
existing_limit: Option<f64>,
) -> Option<f64> {
let normalized_plan = plan_type.unwrap_or_default().trim().to_ascii_lowercase();
if normalized_plan == "free" {
return Some(CHATGPT_WEB_FREE_IMAGE_QUOTA_LIMIT);
}
if let Some(existing_limit) = existing_limit.filter(|value| *value > 0.0) {
return Some(existing_limit);
}
remaining.filter(|value| *value > 0.0)
}
fn normalize_chatgpt_web_image_quota_limit(
metadata: &mut serde_json::Value,
upstream_metadata: Option<&serde_json::Value>,
) {
let existing_limit = existing_chatgpt_web_image_quota_limit(upstream_metadata);
let Some(object) = metadata.as_object_mut() else {
return;
};
let remaining = chatgpt_web_json_number(object.get("image_quota_remaining"));
let explicit_limit =
chatgpt_web_json_number(object.get("image_quota_total")).filter(|value| *value > 0.0);
let plan_type = chatgpt_web_json_string(object.get("plan_type"));
let is_free_plan = plan_type.is_some_and(|value| value.trim().eq_ignore_ascii_case("free"));
let limit = if is_free_plan {
Some(CHATGPT_WEB_FREE_IMAGE_QUOTA_LIMIT)
} else {
explicit_limit
.or_else(|| infer_chatgpt_web_image_quota_limit(plan_type, remaining, existing_limit))
};
if let Some(limit) = limit {
object.insert("image_quota_total".to_string(), json!(limit));
if !object.contains_key("image_quota_used") {
if let Some(remaining) = remaining {
object.insert(
"image_quota_used".to_string(),
json!((limit - remaining).max(0.0)),
);
} else if object
.get("image_quota_blocked")
.and_then(serde_json::Value::as_bool)
== Some(true)
{
object.insert("image_quota_used".to_string(), json!(limit));
}
}
}
}
async fn resolve_chatgpt_web_quota_auth(
state: &AdminAppState<'_>,
transport: &AdminGatewayProviderTransportSnapshot,
) -> Result<Option<(String, String)>, GatewayError> {
if let Some(auth) = state.resolve_local_oauth_header_auth(transport).await? {
return Ok(Some(auth));
}
let decrypted_key = transport.key.decrypted_api_key.trim();
if decrypted_key.is_empty() || decrypted_key == PLACEHOLDER_API_KEY {
return Ok(None);
}
Ok(Some((
"authorization".to_string(),
format!("Bearer {decrypted_key}"),
)))
}
async fn execute_chatgpt_web_quota_plan(
state: &AdminAppState<'_>,
transport: &AdminGatewayProviderTransportSnapshot,
endpoint: &StoredProviderCatalogEndpoint,
authorization: (String, String),
proxy_override: Option<&ProxySnapshot>,
) -> Result<ProviderQuotaExecutionOutcome, GatewayError> {
let base_url = chatgpt_web_base_url(endpoint);
let proxy = match proxy_override {
Some(proxy) => Some(proxy.clone()),
None => {
state
.resolve_transport_proxy_snapshot_with_tunnel_affinity(transport)
.await
}
};
let timeouts = state
.resolve_transport_execution_timeouts(transport)
.or(Some(default_provider_quota_execution_timeouts(
proxy.as_ref(),
)));
let plan = ExecutionPlan {
request_id: format!("chatgpt-web-quota:{}", transport.key.id),
candidate_id: None,
provider_name: Some("chatgpt_web".to_string()),
provider_id: transport.provider.id.clone(),
endpoint_id: transport.endpoint.id.clone(),
key_id: transport.key.id.clone(),
method: "POST".to_string(),
url: format!("{base_url}{CHATGPT_WEB_CONVERSATION_INIT_PATH}"),
headers: build_chatgpt_web_quota_headers(authorization, base_url.as_str()),
content_type: Some("application/json".to_string()),
content_encoding: None,
body: RequestBody::from_json(json!({
"gizmo_id": serde_json::Value::Null,
"requested_default_model": serde_json::Value::Null,
"conversation_id": serde_json::Value::Null,
"timezone_offset_min": -480,
"system_hints": ["picture_v2"],
})),
stream: false,
client_api_format: "openai:image".to_string(),
provider_api_format: "chatgpt_web:conversation_init".to_string(),
model_name: Some("chatgpt-web-conversation-init".to_string()),
proxy,
transport_profile: state.resolve_transport_profile(transport),
timeouts,
};
execute_provider_quota_plan(state, transport, plan, "chatgpt_web").await
}
fn chatgpt_web_quota_invalid_reason(status_code: u16, upstream_message: Option<&str>) -> String {
let message = upstream_message.unwrap_or_default().trim();
let detail = if message.is_empty() {
match status_code {
401 => "ChatGPT Web Token 无效或已过期",
403 => "ChatGPT Web 账户访问受限",
_ => "ChatGPT Web 请求失败",
}
} else {
message
};
match status_code {
401 => format!("{OAUTH_EXPIRED_PREFIX}{detail}"),
403 => format!("{OAUTH_ACCOUNT_BLOCK_PREFIX}{detail}"),
_ => detail.to_string(),
}
}
pub(crate) async fn refresh_chatgpt_web_provider_quota_locally(
state: &AdminAppState<'_>,
provider: &StoredProviderCatalogProvider,
endpoint: &StoredProviderCatalogEndpoint,
keys: Vec<StoredProviderCatalogKey>,
proxy_override: Option<ProxySnapshot>,
) -> Result<Option<serde_json::Value>, GatewayError> {
let mut results = Vec::new();
let mut success_count = 0usize;
let mut failed_count = 0usize;
for key in keys {
let transport = match state
.read_provider_transport_snapshot(&provider.id, &endpoint.id, &key.id)
.await?
{
Some(transport) => transport,
None => {
failed_count += 1;
results.push(json!({
"key_id": key.id,
"key_name": key.name,
"status": "error",
"message": "Provider transport snapshot unavailable",
}));
continue;
}
};
let authorization = match resolve_chatgpt_web_quota_auth(state, &transport).await? {
Some(auth) => auth,
None => {
failed_count += 1;
results.push(json!({
"key_id": key.id,
"key_name": key.name,
"status": "error",
"message": "缺少 ChatGPT Web OAuth 认证信息，请先导入/刷新 Token",
}));
continue;
}
};
let result = match execute_chatgpt_web_quota_plan(
state,
&transport,
endpoint,
authorization,
proxy_override.as_ref(),
)
.await?
{
ProviderQuotaExecutionOutcome::Response(result) => result,
ProviderQuotaExecutionOutcome::Failure(detail) => {
failed_count += 1;
results.push(json!({
"key_id": key.id,
"key_name": key.name,
"status": "error",
"message": format!("conversation/init 请求执行失败: {detail}"),
"status_code": 502,
}));
continue;
}
};
let now_unix_secs = SystemTime::now()
.duration_since(UNIX_EPOCH)
.ok()
.map(|duration| duration.as_secs())
.unwrap_or(0);
let mut metadata_update = None::<serde_json::Value>;
let (mut oauth_invalid_at_unix_secs, mut oauth_invalid_reason) = (
key.oauth_invalid_at_unix_secs,
key.oauth_invalid_reason.clone(),
);
let mut status = "error".to_string();
let mut message = None::<String>;
if result.status_code == 200 {
if let Some(body_json) = result
.body
.as_ref()
.and_then(|body| body.json_body.as_ref())
{
if let Some(mut metadata) =
parse_chatgpt_web_conversation_init_response(body_json, now_unix_secs)
{
let auth_config = chatgpt_web_auth_config(&transport);
enrich_chatgpt_web_quota_metadata(&mut metadata, auth_config.as_ref());
normalize_chatgpt_web_image_quota_limit(
&mut metadata,
key.upstream_metadata.as_ref(),
);
metadata_update = Some(json!({ "chatgpt_web": metadata }));
(oauth_invalid_at_unix_secs, oauth_invalid_reason) =
quota_refresh_success_invalid_state(&key);
status = "success".to_string();
} else {
status = "no_metadata".to_string();
message = Some("响应中未包含 ChatGPT Web 生图限额信息".to_string());
}
} else {
status = "no_metadata".to_string();
message = Some("响应中未包含 ChatGPT Web 生图限额信息".to_string());
}
} else {
let err_msg = extract_execution_error_message(&result);
message = Some(match err_msg.as_deref() {
Some(detail) if !detail.is_empty() => {
format!(
"conversation/init 返回状态码 {}: {}",
result.status_code, detail
)
}
_ => format!("conversation/init 返回状态码 {}", result.status_code),
});
if matches!(result.status_code, 401 | 403) {
oauth_invalid_at_unix_secs = Some(now_unix_secs);
oauth_invalid_reason = Some(chatgpt_web_quota_invalid_reason(
result.status_code,
err_msg.as_deref(),
));
status = if result.status_code == 401 {
"auth_invalid".to_string()
} else {
"forbidden".to_string()
};
}
}
if !persist_provider_quota_refresh_state(
state,
&key.id,
metadata_update.as_ref(),
oauth_invalid_at_unix_secs,
oauth_invalid_reason,
None,
)
.await?
{
failed_count += 1;
results.push(json!({
"key_id": key.id,
"key_name": key.name,
"status": "error",
"message": "Key 状态写入失败",
}));
continue;
}
if status == "success" {
success_count += 1;
} else {
failed_count += 1;
}
let mut payload = serde_json::Map::new();
payload.insert("key_id".to_string(), json!(key.id));
payload.insert("key_name".to_string(), json!(key.name));
payload.insert("status".to_string(), json!(status));
if let Some(message) = message {
payload.insert("message".to_string(), json!(message));
}
if result.status_code != 200 {
payload.insert("status_code".to_string(), json!(result.status_code));
}
if let Some(metadata) = metadata_update
.as_ref()
.and_then(|value| value.get("chatgpt_web"))
.cloned()
{
payload.insert("metadata".to_string(), metadata);
}
if let Some(quota_snapshot) = build_quota_snapshot_payload(
"chatgpt_web",
key.status_snapshot.as_ref(),
metadata_update.as_ref(),
) {
payload.insert("quota_snapshot".to_string(), quota_snapshot);
}
results.push(serde_json::Value::Object(payload));
}
Ok(Some(json!({
"success": success_count,
"failed": failed_count,
"total": success_count + failed_count,
"results": results,
"message": format!("已处理 {} 个 Key", success_count + failed_count),
"auto_removed": 0,
})))
}

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

@@ -1,4 +1,5 @@
pub(crate) mod antigravity;
pub(crate) mod chatgpt_web;
pub(crate) mod codex;
pub(crate) mod kiro;
pub(crate) mod shared;

2
apps/aether-gateway/src/handlers/admin/provider/oauth/quota/shared.rs
View File

@@ -130,7 +130,7 @@ pub(crate) async fn persist_provider_quota_refresh_state(
metadata_update,
));
quota_snapshot_provider_type = metadata_update.as_object().and_then(|object| {
["codex", "kiro", "antigravity", "gemini_cli"]
["codex", "kiro", "antigravity", "gemini_cli", "chatgpt_web"]
.into_iter()
.find(|provider_type| object.contains_key(*provider_type))
});

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

@@ -1,4 +1,5 @@
use super::quota::antigravity::refresh_antigravity_provider_quota_locally;
use super::quota::chatgpt_web::refresh_chatgpt_web_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;
@@ -72,7 +73,10 @@ pub(crate) async fn refresh_provider_oauth_account_state_after_update(
proxy_override: Option<&ProxySnapshot>,
) -> Result<(bool, Option<String>), GatewayError> {
let provider_type = provider.provider_type.trim().to_ascii_lowercase();
if !matches!(provider_type.as_str(), "codex" | "kiro" | "antigravity") {
if !matches!(
provider_type.as_str(),
"codex" | "kiro" | "antigravity" | "chatgpt_web"
) {
return Ok((false, None));
}
@@ -127,6 +131,16 @@ pub(crate) async fn refresh_provider_oauth_account_state_after_update(
)
.await?
}
"chatgpt_web" => {
refresh_chatgpt_web_provider_quota_locally(
state,
provider,
&endpoint,
vec![key],
proxy_override,
)
.await?
}
_ => None,
};
let Some(payload) = payload else {

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

@@ -9,7 +9,11 @@ use aether_admin::provider::quota as admin_provider_quota_pure;
use aether_data_contracts::repository::provider_catalog::{
StoredProviderCatalogEndpoint, StoredProviderCatalogKey,
};
use aether_data_contracts::repository::usage::{
ProviderApiKeyWindowUsageRequest, StoredProviderApiKeyWindowUsageSummary,
};
use serde_json::json;
use std::collections::BTreeMap;
fn admin_pool_string_list(value: Option<&serde_json::Value>) -> Option<Vec<String>> {
let values = value
@@ -289,6 +293,124 @@ fn admin_pool_quota_window<'a>(
})
}
pub(super) type AdminPoolCodexWindowUsageByKey =
BTreeMap<(String, String), StoredProviderApiKeyWindowUsageSummary>;
fn admin_pool_provider_type_is_codex(provider_type: &str) -> bool {
provider_type.trim().eq_ignore_ascii_case("codex")
}
fn admin_pool_codex_window_usage_code(
window: &serde_json::Map<String, serde_json::Value>,
) -> Option<&'static str> {
let code = window
.get("code")
.and_then(serde_json::Value::as_str)
.map(str::trim)?;
if code.eq_ignore_ascii_case("5h") {
Some("5h")
} else if code.eq_ignore_ascii_case("weekly") {
Some("weekly")
} else {
None
}
}
fn admin_pool_codex_window_usage_bounds(
window: &serde_json::Map<String, serde_json::Value>,
) -> Option<(u64, u64)> {
let reset_at = admin_pool_json_to_u64(window.get("reset_at"))?;
let window_minutes = admin_pool_json_to_u64(window.get("window_minutes"))?;
let window_seconds = window_minutes.checked_mul(60)?;
let start = reset_at.checked_sub(window_seconds)?;
(start < reset_at).then_some((start, reset_at))
}
pub(super) fn build_admin_pool_codex_window_usage_requests(
provider_type: &str,
keys: &[StoredProviderCatalogKey],
) -> Vec<ProviderApiKeyWindowUsageRequest> {
if !admin_pool_provider_type_is_codex(provider_type) {
return Vec::new();
}
let mut requests = Vec::new();
for key in keys {
let status_snapshot = provider_key_status_snapshot_payload(key, provider_type);
let Some(windows) = status_snapshot
.get("quota")
.and_then(serde_json::Value::as_object)
.and_then(|quota| quota.get("windows"))
.and_then(serde_json::Value::as_array)
else {
continue;
};
for window in windows.iter().filter_map(serde_json::Value::as_object) {
let Some(window_code) = admin_pool_codex_window_usage_code(window) else {
continue;
};
let Some((start_unix_secs, end_unix_secs)) =
admin_pool_codex_window_usage_bounds(window)
else {
continue;
};
requests.push(ProviderApiKeyWindowUsageRequest {
provider_api_key_id: key.id.clone(),
window_code: window_code.to_string(),
start_unix_secs,
end_unix_secs,
});
}
}
requests
}
fn admin_pool_codex_window_usage_payload(
usage: &StoredProviderApiKeyWindowUsageSummary,
) -> serde_json::Value {
json!({
"request_count": usage.request_count,
"total_tokens": usage.total_tokens,
"total_cost_usd": format!("{:.8}", usage.total_cost_usd),
})
}
fn admin_pool_attach_codex_window_usage(
status_snapshot: &mut serde_json::Value,
key_id: &str,
usage_by_key: &AdminPoolCodexWindowUsageByKey,
) {
if usage_by_key.is_empty() {
return;
}
let Some(windows) = status_snapshot
.get_mut("quota")
.and_then(serde_json::Value::as_object_mut)
.and_then(|quota| quota.get_mut("windows"))
.and_then(serde_json::Value::as_array_mut)
else {
return;
};
for window in windows
.iter_mut()
.filter_map(serde_json::Value::as_object_mut)
{
let Some(window_code) = admin_pool_codex_window_usage_code(window) else {
continue;
};
let lookup_key = (key_id.to_string(), window_code.to_string());
if let Some(usage) = usage_by_key.get(&lookup_key) {
window.insert(
"usage".to_string(),
admin_pool_codex_window_usage_payload(usage),
);
}
}
}
fn admin_pool_quota_window_used_percent(
window: &serde_json::Map<String, serde_json::Value>,
) -> Option<f64> {
@@ -481,6 +603,43 @@ fn admin_pool_build_kiro_account_quota_from_snapshot(
}
}
fn admin_pool_build_chatgpt_web_account_quota_from_snapshot(
quota_snapshot: &serde_json::Map<String, serde_json::Value>,
) -> Option<String> {
let now_unix_secs = chrono::Utc::now().timestamp().max(0) as u64;
let window = admin_pool_quota_window(quota_snapshot, "image_gen")
.or_else(|| admin_pool_quota_windows(quota_snapshot).into_iter().next())?;
let remaining_value = admin_pool_json_to_f64(window.get("remaining_value"));
let limit_value = admin_pool_json_to_f64(window.get("limit_value"));
let remaining_percent = admin_pool_json_to_f64(window.get("remaining_ratio"))
.map(|value| (value * 100.0).clamp(0.0, 100.0))
.or_else(|| {
admin_pool_json_to_f64(window.get("used_ratio"))
.map(|value| ((1.0 - value) * 100.0).clamp(0.0, 100.0))
});
let reset_seconds =
admin_pool_quota_window_reset_seconds(quota_snapshot, window, now_unix_secs);
let mut text = match (remaining_value, limit_value, remaining_percent) {
(Some(remaining), Some(limit), _) if limit > 0.0 => Some(format!(
"生图剩余 {}/{}",
admin_pool_format_quota_value(remaining),
admin_pool_format_quota_value(limit),
)),
(Some(remaining), _, _) => Some(format!(
"生图剩余 {}",
admin_pool_format_quota_value(remaining),
)),
(_, _, Some(percent)) => Some(format!("生图剩余 {}", admin_pool_format_percent(percent))),
_ => None,
}?;
if let Some(reset_text) = reset_seconds.and_then(admin_pool_format_reset_after) {
text.push_str(&format!(" ({reset_text})"));
}
Some(text)
}
fn admin_pool_build_antigravity_account_quota_from_snapshot(
quota_snapshot: &serde_json::Map<String, serde_json::Value>,
) -> Option<String> {
@@ -617,6 +776,13 @@ fn admin_pool_build_account_quota(
return Some(account_quota);
}
}
"chatgpt_web" => {
if let Some(account_quota) =
admin_pool_build_chatgpt_web_account_quota_from_snapshot(quota_snapshot)
{
return Some(account_quota);
}
}
"antigravity" => {
if let Some(account_quota) =
admin_pool_build_antigravity_account_quota_from_snapshot(quota_snapshot)
@@ -760,6 +926,7 @@ pub(super) fn build_admin_pool_key_payload(
key: &StoredProviderCatalogKey,
runtime: &AdminProviderPoolRuntimeState,
pool_config: Option<AdminProviderPoolConfig>,
codex_window_usage_by_key: &AdminPoolCodexWindowUsageByKey,
) -> serde_json::Value {
let cooldown_reason = runtime.cooldown_reason_by_key.get(&key.id).cloned();
let cooldown_ttl_seconds = cooldown_reason
@@ -777,7 +944,14 @@ pub(super) fn build_admin_pool_key_payload(
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, provider_type);
let mut status_snapshot = provider_key_status_snapshot_payload(key, provider_type);
if admin_pool_provider_type_is_codex(provider_type) {
admin_pool_attach_codex_window_usage(
&mut status_snapshot,
&key.id,
codex_window_usage_by_key,
);
}
let account_snapshot = status_snapshot
.get("account")
.and_then(serde_json::Value::as_object);

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

@@ -253,6 +253,26 @@ pub(super) async fn build_admin_pool_list_keys_response(
}
_ => AdminProviderPoolRuntimeState::default(),
};
let codex_window_usage_requests =
pool_payloads::build_admin_pool_codex_window_usage_requests(&provider.provider_type, &keys);
let codex_window_usage_by_key: pool_payloads::AdminPoolCodexWindowUsageByKey =
if codex_window_usage_requests.is_empty() {
pool_payloads::AdminPoolCodexWindowUsageByKey::new()
} else {
state
.app()
.summarize_usage_by_provider_api_key_windows(&codex_window_usage_requests)
.await?
.into_iter()
.map(|usage| {
(
(usage.provider_api_key_id.clone(), usage.window_code.clone()),
usage,
)
})
.collect()
};
let items = keys
.into_iter()
.map(|key| {
@@ -263,6 +283,7 @@ pub(super) async fn build_admin_pool_list_keys_response(
&key,
&runtime,
pool_config.clone(),
&codex_window_usage_by_key,
)
})
.collect::<Vec<_>>();

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

@@ -425,6 +425,31 @@ fn quota_window_reset_seconds(
.map(|(observed_at, reset_at)| reset_at.saturating_sub(observed_at))
}
fn chatgpt_web_image_quota_limit(
metadata: &Map<String, Value>,
remaining: Option<f64>,
) -> Option<f64> {
let plan_type = metadata
.get("plan_type")
.and_then(Value::as_str)
.map(str::trim)
.filter(|value| !value.is_empty())
.map(|value| value.to_ascii_lowercase());
if plan_type.as_deref() == Some("free") {
return Some(25.0);
}
let explicit_limit = metadata
.get("image_quota_total")
.and_then(admin_provider_quota_pure::coerce_json_f64)
.filter(|value| *value > 0.0);
if let Some(limit) = explicit_limit {
return Some(limit);
}
remaining.filter(|value| *value > 0.0)
}
fn model_quota_window_snapshot(
model_name: &str,
item: &Map<String, Value>,
@@ -813,6 +838,98 @@ fn build_kiro_quota_status_snapshot(
}))
}
fn build_chatgpt_web_quota_status_snapshot(
upstream_metadata: Option<&Value>,
source: &str,
) -> Option<Value> {
let metadata = provider_quota_metadata_bucket(upstream_metadata, "chatgpt_web")?;
let observed_at_unix_secs = provider_quota_timestamp_unix_secs(metadata.get("updated_at"));
let remaining = metadata
.get("image_quota_remaining")
.and_then(admin_provider_quota_pure::coerce_json_f64);
let limit = chatgpt_web_image_quota_limit(metadata, remaining);
let used = metadata
.get("image_quota_used")
.and_then(admin_provider_quota_pure::coerce_json_f64)
.or_else(|| {
limit
.zip(remaining)
.map(|(limit, remaining)| (limit - remaining).max(0.0))
});
let reset_at = provider_quota_timestamp_unix_secs(metadata.get("image_quota_reset_at"));
let reset_seconds = quota_window_reset_seconds(observed_at_unix_secs, reset_at);
let plan_type = metadata
.get("plan_type")
.and_then(Value::as_str)
.map(str::trim)
.filter(|value| !value.is_empty())
.map(|value| value.to_ascii_lowercase());
let image_blocked = metadata
.get("image_quota_blocked")
.and_then(admin_provider_quota_pure::coerce_json_bool)
== Some(true);
let usage_ratio = used
.zip(limit)
.and_then(|(used, limit)| (limit > 0.0).then_some((used / limit).clamp(0.0, 1.0)));
let remaining_ratio = remaining.zip(limit).and_then(|(remaining, limit)| {
(limit > 0.0).then_some((remaining / limit).clamp(0.0, 1.0))
});
let mut windows = Vec::new();
if remaining.is_some()
|| limit.is_some()
|| used.is_some()
|| reset_at.is_some()
|| image_blocked
{
windows.push(json!({
"code": "image_gen",
"label": "生图",
"scope": "account",
"unit": "count",
"used_ratio": usage_ratio,
"remaining_ratio": remaining_ratio,
"used_value": used,
"remaining_value": remaining,
"limit_value": limit,
"reset_at": reset_at,
"reset_seconds": reset_seconds,
"is_exhausted": image_blocked || remaining.is_some_and(|value| value <= 0.0),
}));
}
if windows.is_empty() && plan_type.is_none() && observed_at_unix_secs.is_none() {
return None;
}
let exhausted = image_blocked
|| remaining.is_some_and(|value| value <= 0.0)
|| usage_ratio.is_some_and(|value| value >= 1.0 - 1e-6);
let reason = if exhausted {
Some("生图额度已耗尽")
} else {
None
};
Some(json!({
"version": 2,
"provider_type": "chatgpt_web",
"code": if exhausted { "exhausted" } else { "ok" },
"label": if exhausted { Some("额度耗尽") } else { None::<&str> },
"reason": reason,
"freshness": "fresh",
"source": source,
"observed_at": observed_at_unix_secs,
"exhausted": exhausted,
"usage_ratio": usage_ratio,
"updated_at": observed_at_unix_secs,
"reset_at": reset_at,
"reset_seconds": reset_seconds,
"plan_type": plan_type,
"windows": windows,
}))
}
fn build_antigravity_quota_status_snapshot(
upstream_metadata: Option<&Value>,
source: &str,
@@ -993,6 +1110,7 @@ pub(crate) fn sync_provider_key_quota_status_snapshot(
let quota = match normalized_provider_type.as_str() {
"codex" => build_codex_quota_status_snapshot(upstream_metadata, source),
"kiro" => build_kiro_quota_status_snapshot(upstream_metadata, source),
"chatgpt_web" => build_chatgpt_web_quota_status_snapshot(upstream_metadata, source),
"antigravity" => build_antigravity_quota_status_snapshot(upstream_metadata, source),
"gemini_cli" => build_gemini_cli_quota_status_snapshot(upstream_metadata, source),
_ => None,
@@ -1784,6 +1902,42 @@ mod tests {
assert_eq!(window.get("reset_seconds"), Some(&json!(3_600u64)));
}
#[test]
fn provider_key_status_snapshot_payload_backfills_chatgpt_web_image_quota() {
let mut key = sample_catalog_key();
key.upstream_metadata = Some(json!({
"chatgpt_web": {
"updated_at": 1_778_067_246u64,
"plan_type": "free",
"image_quota_remaining": 24.0,
"image_quota_reset_at": 1_778_157_172u64
}
}));
let payload = provider_key_status_snapshot_payload(&key, "chatgpt_web");
let quota = payload
.get("quota")
.and_then(Value::as_object)
.expect("quota snapshot should be object");
let window = quota
.get("windows")
.and_then(Value::as_array)
.and_then(|windows| windows.first())
.and_then(Value::as_object)
.expect("image quota window should exist");
assert_eq!(quota.get("provider_type"), Some(&json!("chatgpt_web")));
assert_eq!(quota.get("code"), Some(&json!("ok")));
assert_eq!(quota.get("plan_type"), Some(&json!("free")));
assert_eq!(quota.get("reset_at"), Some(&json!(1_778_157_172u64)));
assert_eq!(quota.get("usage_ratio"), Some(&json!(0.04)));
assert_eq!(window.get("code"), Some(&json!("image_gen")));
assert_eq!(window.get("remaining_value"), Some(&json!(24.0)));
assert_eq!(window.get("limit_value"), Some(&json!(25.0)));
assert_eq!(window.get("used_value"), Some(&json!(1.0)));
assert_eq!(window.get("remaining_ratio"), Some(&json!(0.96)));
}
#[test]
fn provider_key_status_snapshot_payload_preserves_existing_materialized_quota_snapshot() {
let mut key = sample_catalog_key();

11
apps/aether-gateway/src/maintenance/runtime/pool_quota_probe.rs
View File

@@ -10,8 +10,8 @@ use tracing::{debug, info, warn};
use crate::admin_api::{
admin_provider_pool_config, provider_oauth_runtime_endpoint_for_provider,
refresh_antigravity_provider_quota_locally, refresh_codex_provider_quota_locally,
refresh_kiro_provider_quota_locally, AdminAppState,
refresh_antigravity_provider_quota_locally, refresh_chatgpt_web_provider_quota_locally,
refresh_codex_provider_quota_locally, refresh_kiro_provider_quota_locally, AdminAppState,
};
use crate::{AppState, GatewayError};
@@ -94,7 +94,7 @@ fn now_unix_secs() -> u64 {
fn provider_supports_quota_probe(provider_type: &str) -> bool {
matches!(
provider_type.trim().to_ascii_lowercase().as_str(),
"codex" | "kiro" | "antigravity"
"codex" | "kiro" | "antigravity" | "chatgpt_web"
)
}
@@ -116,6 +116,7 @@ fn extract_quota_updated_at(provider_type: &str, upstream_metadata: Option<&Valu
"codex" => "codex",
"kiro" => "kiro",
"antigravity" => "antigravity",
"chatgpt_web" => "chatgpt_web",
_ => return None,
};
let bucket = metadata.get(bucket_name)?.as_object()?;
@@ -382,6 +383,10 @@ async fn refresh_provider_probe_keys(
refresh_antigravity_provider_quota_locally(admin_state, provider, endpoint, keys, None)
.await
}
"chatgpt_web" => {
refresh_chatgpt_web_provider_quota_locally(admin_state, provider, endpoint, keys, None)
.await
}
_ => Ok(None),
}
}

10
apps/aether-gateway/src/state/runtime/usage_queries.rs
View File

@@ -345,6 +345,16 @@ impl AppState {
.map_err(|err| GatewayError::Internal(err.to_string()))
}
pub(crate) async fn summarize_usage_by_provider_api_key_windows(
&self,
requests: &[usage::ProviderApiKeyWindowUsageRequest],
) -> Result<Vec<usage::StoredProviderApiKeyWindowUsageSummary>, GatewayError> {
self.data
.summarize_usage_by_provider_api_key_windows(requests)
.await
.map_err(|err| GatewayError::Internal(err.to_string()))
}
pub(crate) async fn list_users_by_ids(
&self,
user_ids: &[String],

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

@@ -2,7 +2,9 @@ use std::sync::{Arc, Mutex};
use aether_crypto::{encrypt_python_fernet_plaintext, DEVELOPMENT_ENCRYPTION_KEY};
use aether_data::repository::provider_catalog::InMemoryProviderCatalogReadRepository;
use aether_data::repository::usage::InMemoryUsageReadRepository;
use aether_data_contracts::repository::provider_catalog::ProviderCatalogReadRepository;
use aether_data_contracts::repository::usage::StoredRequestUsageAudit;
use axum::body::{to_bytes, Body, Bytes};
use axum::routing::{any, get, post};
use axum::{extract::Request, Router};
@@ -21,6 +23,54 @@ use crate::constants::{
use crate::control::resolve_public_request_context;
use crate::data::GatewayDataState;
fn sample_pool_usage_row(
request_id: &str,
provider_api_key_id: &str,
created_at_unix_secs: i64,
total_tokens: i32,
total_cost_usd: f64,
) -> StoredRequestUsageAudit {
StoredRequestUsageAudit::new(
format!("usage-{request_id}"),
request_id.to_string(),
Some("user-codex".to_string()),
Some("api-key-codex".to_string()),
Some("codex-user".to_string()),
Some("codex-api-key".to_string()),
"codex".to_string(),
"gpt-5-codex".to_string(),
None,
Some("provider-codex".to_string()),
Some("endpoint-codex".to_string()),
Some(provider_api_key_id.to_string()),
Some("responses".to_string()),
Some("openai:responses".to_string()),
Some("openai".to_string()),
Some("responses".to_string()),
Some("openai:responses".to_string()),
Some("openai".to_string()),
Some("responses".to_string()),
false,
false,
total_tokens,
0,
total_tokens,
total_cost_usd,
total_cost_usd,
Some(200),
None,
None,
Some(240),
Some(80),
"completed".to_string(),
"settled".to_string(),
created_at_unix_secs,
created_at_unix_secs + 1,
Some(created_at_unix_secs + 2),
)
.expect("usage row should build")
}
fn trusted_admin_headers() -> HeaderMap {
let mut headers = HeaderMap::new();
headers.insert(GATEWAY_HEADER, HeaderValue::from_static("rust-phase3b"));
@@ -793,6 +843,233 @@ async fn gateway_sorts_admin_pool_keys_by_imported_and_last_used_time() {
assert_eq!(last_used_names, vec!["active", "old", "fresh"]);
}
#[tokio::test]
async fn gateway_pool_list_adds_codex_cycle_usage_to_quota_windows() {
const RESET_AT: u64 = 1_711_000_000;
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
}
})),
);
provider.provider_type = "codex".to_string();
let mut usage_key = sample_key(
"key-codex-cycle",
"provider-codex",
"openai:responses",
"oauth-placeholder",
);
usage_key.name = "codex cycle usage".to_string();
usage_key.auth_type = "oauth".to_string();
usage_key.request_count = Some(4);
usage_key.total_tokens = 999;
usage_key.total_cost_usd = 9.99;
usage_key.status_snapshot = Some(json!({
"quota": {
"version": 2,
"provider_type": "codex",
"code": "ok",
"label": serde_json::Value::Null,
"reason": serde_json::Value::Null,
"freshness": "fresh",
"source": "response_headers",
"observed_at": RESET_AT,
"exhausted": false,
"usage_ratio": 0.0,
"updated_at": RESET_AT,
"reset_seconds": serde_json::Value::Null,
"plan_type": "plus",
"windows": [
{
"code": "weekly",
"label": "周",
"scope": "account",
"unit": "percent",
"used_ratio": 0.0,
"remaining_ratio": 1.0,
"reset_at": RESET_AT,
"reset_seconds": 604_800,
"window_minutes": 10_080
},
{
"code": "5h",
"label": "5H",
"scope": "account",
"unit": "percent",
"used_ratio": 0.0,
"remaining_ratio": 1.0,
"reset_at": RESET_AT,
"reset_seconds": 18_000,
"window_minutes": 300
}
]
}
}));
let mut zero_key = sample_key(
"key-codex-zero",
"provider-codex",
"openai:responses",
"oauth-placeholder",
);
zero_key.name = "codex zero usage".to_string();
zero_key.auth_type = "oauth".to_string();
zero_key.status_snapshot = usage_key.status_snapshot.clone();
let mut invalid_key = sample_key(
"key-codex-invalid",
"provider-codex",
"openai:responses",
"oauth-placeholder",
);
invalid_key.name = "codex invalid window".to_string();
invalid_key.auth_type = "oauth".to_string();
invalid_key.status_snapshot = Some(json!({
"quota": {
"version": 2,
"provider_type": "codex",
"code": "ok",
"windows": [
{
"code": "weekly",
"label": "周",
"reset_at": serde_json::Value::Null,
"window_minutes": 10_080
},
{
"code": "5h",
"label": "5H",
"reset_at": RESET_AT,
"window_minutes": serde_json::Value::Null
}
]
}
}));
let provider_catalog_repository = Arc::new(InMemoryProviderCatalogReadRepository::seed(
vec![provider],
Vec::new(),
vec![usage_key, zero_key, invalid_key],
));
let usage_repository = Arc::new(InMemoryUsageReadRepository::seed(vec![
sample_pool_usage_row(
"req-5h-a",
"key-codex-cycle",
RESET_AT as i64 - 60,
100,
0.10,
),
sample_pool_usage_row(
"req-5h-b",
"key-codex-cycle",
RESET_AT as i64 - 17_999,
125,
0.20,
),
sample_pool_usage_row(
"req-weekly-only",
"key-codex-cycle",
RESET_AT as i64 - 18_001,
150,
0.30,
),
sample_pool_usage_row(
"req-before-weekly",
"key-codex-cycle",
RESET_AT as i64 - 604_801,
200,
0.40,
),
]));
let state = AppState::new()
.expect("gateway should build")
.with_data_state_for_tests(
GatewayDataState::with_provider_catalog_and_usage_reader_for_tests(
provider_catalog_repository,
usage_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");
fn key_by_id<'a>(keys: &'a [serde_json::Value], key_id: &str) -> &'a serde_json::Value {
keys.iter()
.find(|key| key["key_id"] == json!(key_id))
.expect("key payload should exist")
}
fn window_by_code<'a>(key_payload: &'a serde_json::Value, code: &str) -> &'a serde_json::Value {
key_payload["status_snapshot"]["quota"]["windows"]
.as_array()
.expect("quota windows should be array")
.iter()
.find(|window| window["code"] == json!(code))
.expect("quota window should exist")
}
let usage_key_payload = key_by_id(keys, "key-codex-cycle");
let five_hour_window = window_by_code(usage_key_payload, "5h");
let weekly_window = window_by_code(usage_key_payload, "weekly");
assert_eq!(five_hour_window["usage"]["request_count"], json!(2));
assert_eq!(five_hour_window["usage"]["total_tokens"], json!(225));
assert_eq!(
five_hour_window["usage"]["total_cost_usd"],
json!("0.30000000")
);
assert_eq!(weekly_window["usage"]["request_count"], json!(3));
assert_eq!(weekly_window["usage"]["total_tokens"], json!(375));
assert_eq!(
weekly_window["usage"]["total_cost_usd"],
json!("0.60000000")
);
assert_eq!(usage_key_payload["request_count"], json!(4));
assert_eq!(usage_key_payload["total_tokens"], json!(999));
assert_eq!(usage_key_payload["total_cost_usd"], json!("9.99000000"));
let zero_key_payload = key_by_id(keys, "key-codex-zero");
assert_eq!(
window_by_code(zero_key_payload, "5h")["usage"]["request_count"],
json!(0)
);
assert_eq!(
window_by_code(zero_key_payload, "weekly")["usage"]["total_tokens"],
json!(0)
);
let invalid_key_payload = key_by_id(keys, "key-codex-invalid");
assert!(window_by_code(invalid_key_payload, "5h")
.get("usage")
.is_none());
assert!(window_by_code(invalid_key_payload, "weekly")
.get("usage")
.is_none());
}
#[tokio::test]
async fn gateway_marks_account_blocked_pool_key_in_list_keys_response() {
let mut provider = sample_provider("provider-codex", "codex", 10).with_transport_fields(