From 73658238e545f2efba503f28f273bd71fa552331 Mon Sep 17 00:00:00 2001 From: zyycn Date: Sun, 5 Jul 2026 16:25:01 +0800 Subject: [PATCH] fix: refine admin usage reporting Co-authored-by: Codex --- backend/src/admin/accounts/quota_view.rs | 73 ++++++++- backend/src/admin/accounts/routes.rs | 152 +++++++++++++++++- backend/src/admin/monitoring/dashboard.rs | 52 +++--- .../admin/monitoring/usage_record_routes.rs | 24 +-- backend/src/infra/format.rs | 21 +-- backend/src/upstream/accounts/quota/mod.rs | 14 +- backend/src/upstream/transport/websocket.rs | 31 ++-- .../src/upstream/transport/websocket_pool.rs | 5 +- backend/tests/admin/accounts/list.rs | 21 ++- backend/tests/admin/accounts/quota.rs | 40 +++++ backend/tests/admin/monitoring/dashboard.rs | 90 ++++++++++- .../admin/monitoring/usage_record_routes.rs | 14 +- .../upstream/transport/websocket_pool.rs | 86 ++++++++++ frontend/src/api/modules/dashboard.ts | 3 +- frontend/src/layout/components/AppSidebar.vue | 2 +- .../components/AccountQuotaSummaryCell.vue | 60 +++++-- frontend/src/views/accounts/constants.ts | 3 +- frontend/src/views/accounts/index.vue | 11 -- .../components/AccountOverviewCard.vue | 8 +- .../dashboard/components/RequestTrendCard.vue | 30 ++-- .../dashboard/composables/useDashboard.ts | 60 +++++-- frontend/src/views/dashboard/index.vue | 2 + .../views/usage/components/UsageCostCell.vue | 18 ++- .../views/usage/components/UsageTokenCell.vue | 18 +-- .../usage/composables/useUsageRecordsTable.ts | 2 +- frontend/src/views/usage/constants.ts | 8 - 26 files changed, 692 insertions(+), 156 deletions(-) diff --git a/backend/src/admin/accounts/quota_view.rs b/backend/src/admin/accounts/quota_view.rs index 5c94be582..6e973ceb5 100644 --- a/backend/src/admin/accounts/quota_view.rs +++ b/backend/src/admin/accounts/quota_view.rs @@ -1,11 +1,13 @@ //! 管理端账号 quota 展示视图。 -use chrono::{DateTime, Utc}; +use std::collections::HashMap; + +use chrono::{DateTime, Duration, Utc}; use serde::Serialize; use serde_json::Value; use crate::infra::{ - format::format_percent, + format::{format_compact_percent, format_tokens}, time::{china_datetime, china_relative_time}, }; @@ -29,8 +31,33 @@ struct AdminAccountQuotaWindowData { label_display: String, used_percent: Option, used_percent_display: String, + token_usage_display: String, reset_at_display: String, window_used_display: String, + #[serde(skip)] + reset_at: Option>, +} + +#[derive(Debug, Clone)] +pub(crate) struct AdminAccountQuotaUsageWindow { + pub(crate) key: String, + pub(crate) start: DateTime, + pub(crate) end: DateTime, +} + +#[derive(Debug, Clone, Copy, Default)] +pub(crate) struct AdminAccountQuotaWindowTokenUsage { + total_tokens: u64, +} + +impl AdminAccountQuotaWindowTokenUsage { + pub(crate) fn add_tokens(&mut self, tokens: u64) { + self.total_tokens = self.total_tokens.saturating_add(tokens); + } + + fn display(self) -> String { + format_tokens(self.total_tokens) + } } impl Default for AdminAccountQuotaData { @@ -48,6 +75,40 @@ impl AdminAccountQuotaData { .iter() .any(|window| window.used_percent.is_some_and(|percent| percent >= 80.0)) } + + pub(crate) fn usage_windows(&self) -> Vec { + self.windows + .iter() + .filter_map(AdminAccountQuotaWindowData::usage_window) + .collect() + } + + pub(crate) fn apply_token_usage( + &mut self, + usage_by_window: &HashMap, + ) { + for window in &mut self.windows { + window.token_usage_display = + match (window.usage_window(), usage_by_window.get(&window.key)) { + (Some(_), Some(usage)) => usage.display(), + (Some(_), None) => AdminAccountQuotaWindowTokenUsage::default().display(), + (None, _) => "-".to_string(), + }; + } + } +} + +impl AdminAccountQuotaWindowData { + fn usage_window(&self) -> Option { + let end = self.reset_at?; + let seconds = i64::try_from(self.window_seconds?).ok()?; + let start = end.checked_sub_signed(Duration::seconds(seconds))?; + (start <= end).then(|| AdminAccountQuotaUsageWindow { + key: self.key.clone(), + start, + end, + }) + } } pub(crate) fn quota_data( @@ -111,11 +172,13 @@ fn push_monthly_quota_window( window_seconds, label_display: "月限额".to_string(), used_percent, - used_percent_display: used_percent.map_or_else(|| "-".to_string(), format_percent), + used_percent_display: used_percent.map_or_else(|| "-".to_string(), format_compact_percent), + token_usage_display: "-".to_string(), reset_at_display: reset_at .as_ref() .map_or_else(|| "-".to_string(), china_datetime), window_used_display: quota_window_used_display(reset_at, window_seconds), + reset_at, }); true } @@ -198,11 +261,13 @@ fn push_quota_window( window_seconds, label_display, used_percent, - used_percent_display: used_percent.map_or_else(|| "-".to_string(), format_percent), + used_percent_display: used_percent.map_or_else(|| "-".to_string(), format_compact_percent), + token_usage_display: "-".to_string(), reset_at_display: reset_at .as_ref() .map_or_else(|| "-".to_string(), china_datetime), window_used_display: quota_window_used_display(reset_at, window_seconds), + reset_at, }); } diff --git a/backend/src/admin/accounts/routes.rs b/backend/src/admin/accounts/routes.rs index 4744a56b9..7081ee34f 100644 --- a/backend/src/admin/accounts/routes.rs +++ b/backend/src/admin/accounts/routes.rs @@ -10,7 +10,7 @@ use axum::{ response::{IntoResponse, Response}, Json, }; -use std::collections::HashMap; +use std::collections::{HashMap, HashSet}; use chrono::{DateTime, Utc}; use serde::{Deserialize, Serialize}; @@ -25,7 +25,10 @@ use crate::{ }, admin::update_payload::{parse_editable_update, EditableUpdateMessages}, admin::{ - accounts::quota_view::{quota_data, AdminAccountQuotaData}, + accounts::quota_view::{ + quota_data, AdminAccountQuotaData, AdminAccountQuotaUsageWindow, + AdminAccountQuotaWindowTokenUsage, + }, accounts::service::{ AdminAccountError, AdminAccountHealthCheck, AdminAccountMetadata, AdminAccountRefreshOutcome, AdminAccountRefreshResult, AdminAccountUpdate, @@ -38,7 +41,10 @@ use crate::{ format_cost, format_percent, format_plain_number, format_tokens, nonnegative_i64_to_u64, }, json::{clamp_limit, clamp_page, total_pages, Page}, - time::{china_datetime, china_quarter_hour_start, china_relative_time, china_rfc3339}, + time::{ + china_datetime, china_quarter_hour_start, china_relative_time, china_rfc3339, + parse_rfc3339_utc, + }, }, runtime::state::AppState, upstream::accounts::model::AccountStatus, @@ -297,6 +303,12 @@ struct AccountModelUsageRecord { last_used_at: Option>, } +#[derive(Debug, Clone)] +struct AccountQuotaUsageWindow { + account_id: String, + window: AdminAccountQuotaUsageWindow, +} + impl AccountListStats { fn data_for(&self, account: AdminAccountMetadata) -> AdminAccountData { let account_id = account.id.clone(); @@ -642,7 +654,8 @@ pub(crate) async fn account_quota( let quota = data.get("quota").cloned().unwrap_or(Value::Null); let raw = data.get("raw").cloned().unwrap_or(Value::Null); let quota_json = quota.to_string(); - let quota_data = quota_data("a_json, Some(Utc::now())); + let mut quota_data = quota_data("a_json, Some(Utc::now())); + apply_account_quota_window_token_usage(&state, &account_id, &mut quota_data).await; let plan_type = quota .get("plan_type") .and_then(Value::as_str) @@ -858,6 +871,8 @@ async fn account_list_stats( .list_by_account_ids(&account_ids) .await .unwrap_or_default(); + let mut quota_by_account = quota_by_account.clone(); + apply_quota_window_token_usage(state, accounts, &mut quota_by_account).await; let models_by_account = list_current_window_model_usage(state, &usage_records).await; AccountListStats { @@ -865,11 +880,138 @@ async fn account_list_stats( .into_iter() .map(|usage| (usage.account_id.clone(), usage)) .collect(), - quota_by_account: quota_by_account.clone(), + quota_by_account, models_by_account, } } +async fn apply_account_quota_window_token_usage( + state: &AppState, + account_id: &str, + quota: &mut AdminAccountQuotaData, +) { + let windows = quota + .usage_windows() + .into_iter() + .map(|window| AccountQuotaUsageWindow { + account_id: account_id.to_string(), + window, + }) + .collect::>(); + let usage_by_account = quota_window_token_usage_by_account(state, &windows).await; + let empty_usage = HashMap::new(); + quota.apply_token_usage(usage_by_account.get(account_id).unwrap_or(&empty_usage)); +} + +async fn apply_quota_window_token_usage( + state: &AppState, + accounts: &[AdminAccountMetadata], + quota_by_account: &mut HashMap, +) { + let windows = accounts + .iter() + .filter_map(|account| { + quota_by_account + .get(&account.id) + .map(|quota| (account.id.as_str(), quota)) + }) + .flat_map(|(account_id, quota)| { + quota + .usage_windows() + .into_iter() + .map(move |window| AccountQuotaUsageWindow { + account_id: account_id.to_string(), + window, + }) + }) + .collect::>(); + let usage_by_account = quota_window_token_usage_by_account(state, &windows).await; + + for account in accounts { + if let Some(quota) = quota_by_account.get_mut(&account.id) { + let empty_usage = HashMap::new(); + quota.apply_token_usage(usage_by_account.get(&account.id).unwrap_or(&empty_usage)); + } + } +} + +async fn quota_window_token_usage_by_account( + state: &AppState, + windows: &[AccountQuotaUsageWindow], +) -> HashMap> { + let Some(min_start) = windows.iter().map(|window| window.window.start).min() else { + return HashMap::new(); + }; + let Some(max_end) = windows.iter().map(|window| window.window.end).max() else { + return HashMap::new(); + }; + + let account_ids = windows + .iter() + .map(|window| window.account_id.as_str()) + .collect::>(); + let mut windows_by_account = HashMap::<&str, Vec<&AdminAccountQuotaUsageWindow>>::new(); + for window in windows { + windows_by_account + .entry(window.account_id.as_str()) + .or_default() + .push(&window.window); + } + let mut builder = QueryBuilder::::new( + "select + account_id, + bucket_start, + coalesce(sum(input_tokens), 0) as input_tokens, + coalesce(sum(output_tokens), 0) as output_tokens + from usage_time_buckets + where account_id in (", + ); + let mut separated = builder.separated(", "); + for account_id in &account_ids { + separated.push_bind(*account_id); + } + separated.push_unseparated(")"); + builder.push(" and bucket_start >= "); + builder.push_bind(min_start.to_rfc3339()); + builder.push(" and bucket_start <= "); + builder.push_bind(max_end.to_rfc3339()); + builder.push(" group by account_id, bucket_start"); + + let rows = builder + .build() + .fetch_all(state.services.background_tasks.accounts.pool()) + .await + .unwrap_or_default(); + let mut usage_by_account = + HashMap::>::new(); + + for row in rows { + let account_id: String = row.get("account_id"); + let Ok(bucket_start) = parse_rfc3339_utc(row.get::<&str, _>("bucket_start")) else { + continue; + }; + let total_tokens = nonnegative_i64_to_u64(row.get::("input_tokens")) + .saturating_add(nonnegative_i64_to_u64(row.get::("output_tokens"))); + let Some(account_windows) = windows_by_account.get(account_id.as_str()) else { + continue; + }; + + for window in account_windows + .iter() + .filter(|window| bucket_start >= window.start && bucket_start <= window.end) + { + usage_by_account + .entry(account_id.clone()) + .or_default() + .entry(window.key.clone()) + .or_default() + .add_tokens(total_tokens); + } + } + + usage_by_account +} + async fn account_summary_data( state: &AppState, quota_by_account: &HashMap, diff --git a/backend/src/admin/monitoring/dashboard.rs b/backend/src/admin/monitoring/dashboard.rs index a3dcf0311..c1a87c0bd 100644 --- a/backend/src/admin/monitoring/dashboard.rs +++ b/backend/src/admin/monitoring/dashboard.rs @@ -219,10 +219,10 @@ struct UsageWindow { errors: u64, first_token_latency_sum: u64, first_token_latency_count: u64, + max_first_token_bucket_latency: u64, + min_first_token_bucket_latency: u64, latency_sum: u64, latency_count: u64, - max_latency: u64, - min_latency: u64, } impl UsageWindow { @@ -243,6 +243,14 @@ impl UsageWindow { .checked_div(self.first_token_latency_count) } + fn max_first_token_bucket_latency(self) -> Option { + (self.max_first_token_bucket_latency > 0).then_some(self.max_first_token_bucket_latency) + } + + fn min_first_token_bucket_latency(self) -> Option { + (self.min_first_token_bucket_latency > 0).then_some(self.min_first_token_bucket_latency) + } + fn avg_latency(self) -> Option { self.latency_sum.checked_div(self.latency_count) } @@ -252,6 +260,7 @@ impl UsageWindow { pub(crate) async fn dashboard_summary( State(state): State, _auth: AdminAuth, + Query(query): Query, ) -> Result { let accounts = state .services @@ -288,7 +297,7 @@ pub(crate) async fn dashboard_summary( let now = Utc::now(); let time_buckets = dashboard_time_buckets(&state, now).await; let pool_summary = account_pool_summary(&accounts); - let trend = dashboard_trend_data(&time_buckets, DashboardTrendKind::Usage); + let trend = dashboard_trend_data(&time_buckets, query.kind.unwrap_or_default()); let quota_used_by_account = account_quota_used_percent_by_id(&state, &usage_records).await; let settings = state.services.settings.current(); let recent_usage_records = recent_events.into_iter().take(10).collect::>(); @@ -483,7 +492,9 @@ fn dashboard_trend_data_at( let points = buckets .iter() .map(|(bucket_start, bucket)| { - let latency = bucket.avg_latency().unwrap_or(0); + let latency = bucket.avg_first_token_latency().unwrap_or(0); + let max_latency = bucket.max_first_token_bucket_latency().unwrap_or(0); + let min_latency = bucket.min_first_token_bucket_latency().unwrap_or(0); let tokens = bucket.tokens(); let success_rate = if bucket.requests > 0 { ((bucket.requests - bucket.errors) as f64 / bucket.requests as f64 * 1000.0).round() @@ -507,10 +518,10 @@ fn dashboard_trend_data_at( errors_value: bucket.errors, latency: format_optional_duration_ms(Some(latency)), latency_value: latency, - max_latency: format_optional_duration_ms(Some(bucket.max_latency)), - max_latency_value: bucket.max_latency, - min_latency: format_optional_duration_ms(Some(bucket.min_latency)), - min_latency_value: bucket.min_latency, + max_latency: format_optional_duration_ms(Some(max_latency)), + max_latency_value: max_latency, + min_latency: format_optional_duration_ms(Some(min_latency)), + min_latency_value: min_latency, success_rate: format_percent(success_rate), success_rate_value: success_rate, } @@ -772,20 +783,23 @@ fn apply_bucket(window: &mut UsageWindow, record: &AdminUsageTimeBucketRecord) { window.errors += nonnegative_i64_to_u64(record.error_count); window.first_token_latency_sum += nonnegative_i64_to_u64(record.first_token_latency_sum); window.first_token_latency_count += nonnegative_i64_to_u64(record.first_token_latency_count); - window.latency_sum += nonnegative_i64_to_u64(record.latency_sum); - window.latency_count += nonnegative_i64_to_u64(record.latency_count); - let max_latency = nonnegative_i64_to_u64(record.max_latency_ms); - if max_latency > 0 { - window.max_latency = window.max_latency.max(max_latency); - } - let min_latency = nonnegative_i64_to_u64(record.min_latency_ms); - if min_latency > 0 { - window.min_latency = if window.min_latency == 0 { - min_latency + let bucket_first_token_latency = first_token_bucket_latency(record); + if let Some(latency) = bucket_first_token_latency { + window.max_first_token_bucket_latency = window.max_first_token_bucket_latency.max(latency); + window.min_first_token_bucket_latency = if window.min_first_token_bucket_latency == 0 { + latency } else { - window.min_latency.min(min_latency) + window.min_first_token_bucket_latency.min(latency) }; } + window.latency_sum += nonnegative_i64_to_u64(record.latency_sum); + window.latency_count += nonnegative_i64_to_u64(record.latency_count); +} + +fn first_token_bucket_latency(record: &AdminUsageTimeBucketRecord) -> Option { + let sum = nonnegative_i64_to_u64(record.first_token_latency_sum); + let count = nonnegative_i64_to_u64(record.first_token_latency_count); + sum.checked_div(count).filter(|latency| *latency > 0) } fn account_pool_summary(accounts: &[Account]) -> AccountPoolDiagnostics { diff --git a/backend/src/admin/monitoring/usage_record_routes.rs b/backend/src/admin/monitoring/usage_record_routes.rs index a0259b072..f5b997fca 100644 --- a/backend/src/admin/monitoring/usage_record_routes.rs +++ b/backend/src/admin/monitoring/usage_record_routes.rs @@ -29,8 +29,8 @@ use crate::{ }, infra::{ format::{ - format_compact_number, format_duration_ms, format_duration_ms_f64, format_multiplier, - format_number, format_percent, format_precise_cost, format_rate, format_token_price, + format_compact_number, format_cost, format_duration_ms, format_duration_ms_f64, + format_multiplier, format_number, format_percent, format_rate, format_token_price, format_tokens, }, json::{clamp_limit, clamp_page, NumberedPage, Page}, @@ -607,11 +607,11 @@ fn usage_cost_details( service_tier: cost.service_tier, service_tier_display: cost.service_tier_display, multiplier: cost.multiplier, - input_cost_display: format_precise_cost(cost.input_cost), - output_cost_display: format_precise_cost(cost.output_cost), - cache_read_cost_display: format_precise_cost(cost.cache_read_cost), - total_cost_display: format_precise_cost(cost.total_cost), - original_cost_display: format_precise_cost(cost.original_cost), + input_cost_display: format_cost(cost.input_cost), + output_cost_display: format_cost(cost.output_cost), + cache_read_cost_display: format_cost(cost.cache_read_cost), + total_cost_display: format_cost(cost.total_cost), + original_cost_display: format_cost(cost.original_cost), input_price_display: format_token_price(cost.input_price_per_mtoken), output_price_display: format_token_price(cost.output_price_per_mtoken), cache_read_price_display: format_token_price(cost.cache_read_price_per_mtoken), @@ -673,9 +673,9 @@ impl UsageRecordBreakdownData { total_tokens_value: item.total_tokens, total_tokens_total: format_tokens(total_tokens_total), total_tokens_total_value: total_tokens_total, - cost: format_precise_cost(item.cost), - actual_cost: format_precise_cost(item.actual_cost), - account_cost: format_precise_cost(item.account_cost), + cost: format_cost(item.cost), + actual_cost: format_cost(item.actual_cost), + account_cost: format_cost(item.account_cost), average_latency_ms: format_duration_ms_f64(item.average_latency_ms), } } @@ -717,8 +717,8 @@ impl From for UsageRecordTrendPointData { total_tokens_value: point.total_tokens, cache_hit_rate: format_percent(cache_hit_rate_value), cache_hit_rate_value, - cost: format_precise_cost(point.cost), - actual_cost: format_precise_cost(point.actual_cost), + cost: format_cost(point.cost), + actual_cost: format_cost(point.actual_cost), average_latency_ms: format_duration_ms_f64(point.average_latency_ms), average_latency_ms_value: point.average_latency_ms, } diff --git a/backend/src/infra/format.rs b/backend/src/infra/format.rs index dec187ae3..072930b4c 100644 --- a/backend/src/infra/format.rs +++ b/backend/src/infra/format.rs @@ -63,6 +63,15 @@ pub fn format_percent(value: f64) -> String { } } +/// 展示百分比，并去掉无意义的尾随 0。 +pub fn format_compact_percent(value: f64) -> String { + if value.is_finite() { + format!("{}%", trim_trailing_zeroes(&format!("{value:.1}"))) + } else { + "—".to_string() + } +} + /// 将 0 到 1 的比例展示为百分比。 pub fn format_rate(value: Option) -> String { value.map_or_else(|| "—".to_string(), |value| format_percent(value * 100.0)) @@ -86,15 +95,6 @@ pub fn format_cost(value: f64) -> String { format!("${value:.2}") } -/// 使用六位小数展示美元成本。 -pub fn format_precise_cost(value: f64) -> String { - if value.is_finite() { - format!("${value:.6}") - } else { - "—".to_string() - } -} - /// 展示每百万 token 的美元单价。 pub fn format_token_price(value: f64) -> String { if value.is_finite() { @@ -160,7 +160,8 @@ fn format_compact_scaled(value: f64, unit: &str) -> String { format!("{}{unit}", trim_trailing_zeroes(&rounded)) } -fn trim_trailing_zeroes(value: &str) -> String { +/// 去掉小数字符串中无意义的尾随 0 和小数点。 +pub fn trim_trailing_zeroes(value: &str) -> String { value .trim_end_matches('0') .trim_end_matches('.') diff --git a/backend/src/upstream/accounts/quota/mod.rs b/backend/src/upstream/accounts/quota/mod.rs index e9ec5b999..90beede39 100644 --- a/backend/src/upstream/accounts/quota/mod.rs +++ b/backend/src/upstream/accounts/quota/mod.rs @@ -132,6 +132,8 @@ fn window_from_rate_limit(window: Option<&Value>) -> Value { "reset_at": window.get("reset_at").and_then(positive_i64), "window_minutes": window_minutes, "limit_reached": used_percent >= 100.0, + "used": first_present(window, &["used_tokens", "used", "used_credits", "usage", "consumed_tokens", "consumed"]), + "limit": first_present(window, &["limit_tokens", "limit", "limit_credits", "quota", "total_tokens", "total"]), }) } @@ -201,8 +203,10 @@ fn monthly_limit_from_snapshots(snapshots: &[Value]) -> Option { "reset_at": bucket.get("reset_at").cloned().unwrap_or(Value::Null), "window_minutes": bucket.get("window_minutes").cloned().unwrap_or(Value::Null), "limit_reached": bucket.get("limit_reached").cloned().unwrap_or(Value::Bool(false)), - "used_credits": Value::Null, - "limit_credits": Value::Null, + "used": bucket.get("used").cloned().unwrap_or(Value::Null), + "limit": bucket.get("limit").cloned().unwrap_or(Value::Null), + "used_credits": bucket.get("used_credits").cloned().unwrap_or(Value::Null), + "limit_credits": bucket.get("limit_credits").cloned().unwrap_or(Value::Null), })); } } @@ -214,6 +218,12 @@ fn remaining_percent(used_percent: f64) -> i64 { (100.0 - used_percent.clamp(0.0, 100.0)).round() as i64 } +fn first_present(value: &Value, keys: &[&str]) -> Value { + keys.iter() + .find_map(|key| value.get(*key).filter(|value| !value.is_null()).cloned()) + .unwrap_or(Value::Null) +} + fn rate_limit_blocked(rate_limit: &Value) -> bool { rate_limit .get("limit_reached") diff --git a/backend/src/upstream/transport/websocket.rs b/backend/src/upstream/transport/websocket.rs index 0f320e136..3d1338a1b 100644 --- a/backend/src/upstream/transport/websocket.rs +++ b/backend/src/upstream/transport/websocket.rs @@ -10,8 +10,9 @@ use std::{ use bytes::Bytes; use chrono::Utc; -use futures::{channel::mpsc, SinkExt, Stream, StreamExt}; +use futures::{SinkExt, Stream, StreamExt}; use thiserror::Error; +use tokio::sync::mpsc; use tokio::{sync::Mutex, time::timeout}; use tokio_tungstenite::{connect_async_tls_with_config, Connector}; use tungstenite::{ @@ -876,8 +877,12 @@ fn stream_websocket_response( .await; }); + let body = futures::stream::unfold(rx, |mut rx| async move { + rx.recv().await.map(|item| (item, rx)) + }); + CodexWebSocketStreamingExchange { - body: Box::pin(rx), + body: Box::pin(body), turn_state: response_metadata.turn_state, set_cookie_headers: response_metadata.set_cookie_headers, rate_limit_headers: response_metadata.rate_limit_headers, @@ -894,7 +899,7 @@ async fn forward_websocket_response_stream( prefetched_text: Option, rate_limit_header_updates: CodexWebSocketRateLimitHeaderUpdates, turn_state_update: CodexWebSocketTurnStateUpdate, - mut tx: mpsc::Sender>, + tx: mpsc::Sender>, ) { let mut pool_return = pool_return; let mut prefetched_text = prefetched_text; @@ -903,12 +908,20 @@ async fn forward_websocket_response_stream( let raw = if let Some(text) = prefetched_text.take() { text } else { - let message = match next_websocket_message( - &mut websocket, - receive_idle_timeout(saw_response_frame), - ) - .await - { + let message = tokio::select! { + biased; + // 客户端断开下游 SSE 流：立即丢弃连接并释放池 slot， + // 不再傻等上游 idle 超时（否则同会话后续请求会一直 bypass/busy）。 + () = tx.closed() => { + discard_stream_websocket(websocket, pool_return).await; + return; + } + message = next_websocket_message( + &mut websocket, + receive_idle_timeout(saw_response_frame), + ) => message, + }; + let message = match message { Ok(message) => message, Err(error) => { discard_stream_websocket(websocket, pool_return).await; diff --git a/backend/src/upstream/transport/websocket_pool.rs b/backend/src/upstream/transport/websocket_pool.rs index c235587d1..bec280af8 100644 --- a/backend/src/upstream/transport/websocket_pool.rs +++ b/backend/src/upstream/transport/websocket_pool.rs @@ -17,7 +17,6 @@ const DEFAULT_MAX_AGE: Duration = Duration::from_mins(55); const DEFAULT_MAINTENANCE_INTERVAL: Duration = Duration::from_secs(25); const DEFAULT_PING_INTERVAL: Duration = Duration::from_secs(25); const DEFAULT_PING_TIMEOUT: Duration = Duration::from_secs(5); -const DEFAULT_LIVENESS_TIMEOUT: Duration = Duration::from_millis(62_500); /// WebSocket 连接池 key。 #[derive(Debug, Clone, PartialEq, Eq, Hash)] @@ -99,7 +98,9 @@ impl Default for CodexWebSocketPoolConfig { maintenance_interval: Some(DEFAULT_MAINTENANCE_INTERVAL), ping_interval: Some(DEFAULT_PING_INTERVAL), ping_timeout: DEFAULT_PING_TIMEOUT, - liveness_timeout: Some(DEFAULT_LIVENESS_TIMEOUT), + // idle 连接不设失活截断：靠 ping/pong 保活，只在 max_age（55 分钟） + // 或 ping 失败时关闭，最大化跨轮复用（对齐 Codex CLI 的长连接策略）。 + liveness_timeout: None, } } } diff --git a/backend/tests/admin/accounts/list.rs b/backend/tests/admin/accounts/list.rs index d883c9476..de43b378a 100644 --- a/backend/tests/admin/accounts/list.rs +++ b/backend/tests/admin/accounts/list.rs @@ -89,8 +89,8 @@ async fn admin_accounts_list_should_include_usage_quota_and_model_stats() { window_input_tokens: 120_000, window_output_tokens: 4_000, window_cached_tokens: 40_000, - window_started_at: "2026-06-20T00:00:00Z", - window_reset_at: "2026-06-27T00:00:00Z", + window_started_at: "2026-06-22T12:51:00Z", + window_reset_at: "2026-06-29T12:51:00Z", limit_window_seconds: 604_800, last_used_at: "2026-06-23T08:50:13Z", }, @@ -217,6 +217,20 @@ async fn admin_accounts_list_should_include_usage_quota_and_model_stats() { sqlx::query( "insert into usage_time_buckets (bucket_start, account_id, model, request_count, input_tokens, output_tokens, cached_tokens, updated_at) values (?, ?, ?, ?, ?, ?, ?, ?)", ) + .bind("2026-06-22T10:00:00+00:00") + .bind("acct_stats") + .bind("gpt-five-hour-window") + .bind(1) + .bind(2_000) + .bind(500) + .bind(100) + .bind("2026-06-22T10:00:00Z") + .execute(&pool) + .await + .unwrap(); + sqlx::query( + "insert into usage_time_buckets (bucket_start, account_id, model, request_count, input_tokens, output_tokens, cached_tokens, updated_at) values (?, ?, ?, ?, ?, ?, ?, ?)", + ) .bind("2026-06-19T23:45:00+00:00") .bind("acct_stats") .bind("gpt-outside-window") @@ -253,10 +267,13 @@ async fn admin_accounts_list_should_include_usage_quota_and_model_stats() { assert_eq!(item["quota"]["windows"].as_array().unwrap().len(), 3); assert_eq!(item["quota"]["windows"][0]["labelDisplay"], "月限额"); assert_eq!(item["quota"]["windows"][0]["group"], "monthly"); + assert_eq!(item["quota"]["windows"][0]["tokenUsageDisplay"], "1.12M"); assert_eq!(item["quota"]["windows"][1]["labelDisplay"], "5小时限额"); assert_eq!(item["quota"]["windows"][1]["group"], "shortTerm"); + assert_eq!(item["quota"]["windows"][1]["tokenUsageDisplay"], "2.5K"); assert_eq!(item["quota"]["windows"][2]["labelDisplay"], "周限额"); assert_eq!(item["quota"]["windows"][2]["group"], "shortTerm"); + assert_eq!(item["quota"]["windows"][2]["tokenUsageDisplay"], "124K"); assert!(item["quota"]["windows"][2]["windowUsedDisplay"] .as_str() .is_some_and(|value| value.contains(" / 7.0d"))); diff --git a/backend/tests/admin/accounts/quota.rs b/backend/tests/admin/accounts/quota.rs index bd2893fd0..032e0271e 100644 --- a/backend/tests/admin/accounts/quota.rs +++ b/backend/tests/admin/accounts/quota.rs @@ -18,6 +18,11 @@ async fn admin_account_quota_should_send_usage_cookie() { "used_percent": 12, "reset_at": 1_800_000_000, "limit_window_seconds": 18_000 + }, + "secondary_window": { + "used_percent": 34, + "reset_at": 1_800_604_800, + "limit_window_seconds": 604_800 } } }))) @@ -46,6 +51,29 @@ async fn admin_account_quota_should_send_usage_cookie() { }, ) .await; + sqlx::query( + "insert into usage_time_buckets ( + bucket_start, + account_id, + model, + request_count, + input_tokens, + output_tokens, + cached_tokens, + updated_at + ) values (?, ?, ?, ?, ?, ?, ?, ?)", + ) + .bind("2027-01-15T04:00:00+00:00") + .bind("acct_quota_cookie") + .bind("gpt-5") + .bind(1) + .bind(1_500) + .bind(500) + .bind(300) + .bind("2027-01-15T04:00:00Z") + .execute(&pool) + .await + .unwrap(); SqliteCookieStore::new(pool.clone()) .capture_set_cookie( "acct_quota_cookie", @@ -87,4 +115,16 @@ async fn admin_account_quota_should_send_usage_cookie() { body["data"]["quotaData"]["windows"][0]["group"], "shortTerm" ); + assert_eq!( + body["data"]["quotaData"]["windows"][0]["tokenUsageDisplay"], + "2K" + ); + assert_eq!( + body["data"]["quotaData"]["windows"][1]["labelDisplay"], + "周限额" + ); + assert_eq!( + body["data"]["quotaData"]["windows"][1]["tokenUsageDisplay"], + "0" + ); } diff --git a/backend/tests/admin/monitoring/dashboard.rs b/backend/tests/admin/monitoring/dashboard.rs index 5d04e03be..20492809c 100644 --- a/backend/tests/admin/monitoring/dashboard.rs +++ b/backend/tests/admin/monitoring/dashboard.rs @@ -14,7 +14,7 @@ use codex_proxy_rs::{ }, infra::{ database::connect_sqlite, - time::{china_day_start, china_hour}, + time::{china_day_start, china_hour, china_hour_start}, }, proxy::dispatch::session_affinity::SqliteSessionAffinityStore, runtime::{ @@ -155,7 +155,7 @@ async fn dashboard_trend_should_bucket_usage_by_china_hour() { .await .unwrap(); - let body = dashboard_trend(app).await; + let body = dashboard_trend(app, "usage").await; let point = body["data"]["points"] .as_array() .unwrap() @@ -167,6 +167,78 @@ async fn dashboard_trend_should_bucket_usage_by_china_hour() { assert_eq!(point["tokensValue"], 10); } +#[tokio::test] +async fn dashboard_latency_trend_should_use_first_token_latency() { + let (app, store, _pool, _dir) = dashboard_test_app( + "dashboard-trend-first-token-latency.sqlite", + crate::support::fingerprint::test_fingerprint(), + ) + .await; + let now = Utc::now(); + let current_hour_start = china_hour_start(now); + let sample_hour_start = if now - current_hour_start > Duration::minutes(31) { + current_hour_start + } else { + current_hour_start - Duration::hours(1) + }; + let mut slow_completion = + usage_record_with_tokens(sample_hour_start + Duration::minutes(1), 10); + slow_completion.latency_ms = Some(10_000); + slow_completion.metadata["firstTokenMs"] = json!(100_i64); + let mut fast_completion = + usage_record_with_tokens(sample_hour_start + Duration::minutes(16), 10); + fast_completion.latency_ms = Some(1_000); + fast_completion.metadata["firstTokenMs"] = json!(500_i64); + store.append(&slow_completion).await.unwrap(); + store.append(&fast_completion).await.unwrap(); + + let body = dashboard_trend(app, "latency").await; + let point = body["data"]["points"] + .as_array() + .unwrap() + .iter() + .find(|point| point["requestsValue"] == 2) + .expect("latency trend should include the inserted usage bucket"); + let summary = body["data"]["summary"].as_array().unwrap(); + + assert_eq!(body["data"]["kind"], "latency"); + assert_eq!(point["latencyValue"], 300); + assert_eq!(point["latency"], "300 ms"); + assert_eq!(point["maxLatencyValue"], 500); + assert_eq!(point["maxLatency"], "500 ms"); + assert_eq!(point["minLatencyValue"], 100); + assert_eq!(point["minLatency"], "100 ms"); + assert_eq!(summary[0]["value"], "300 ms"); + assert_eq!(summary[1]["value"], "500 ms"); + assert_eq!(summary[2]["value"], "100 ms"); +} + +#[tokio::test] +async fn dashboard_summary_should_return_requested_trend_kind() { + let (app, store, _pool, _dir) = dashboard_test_app( + "dashboard-summary-requested-trend-kind.sqlite", + crate::support::fingerprint::test_fingerprint(), + ) + .await; + let now = Utc::now(); + let mut record = usage_record_with_tokens(now - Duration::seconds(1), 10); + record.latency_ms = Some(10_000); + record.metadata["firstTokenMs"] = json!(250_i64); + store.append(&record).await.unwrap(); + + let body = dashboard_summary_with_kind(app, "latency").await; + + assert_eq!(body["data"]["trend"]["kind"], "latency"); + assert!( + body["data"]["trend"]["summary"] + .as_array() + .unwrap() + .iter() + .any(|item| item["value"] == "250 ms"), + "summary should include first-token latency trend values: {body:?}" + ); +} + #[tokio::test] async fn dashboard_summary_should_calculate_priority_cost_from_event_service_tier() { let cost = dashboard_summary_total_cost_for_usage_record( @@ -408,11 +480,19 @@ async fn dashboard_test_app( } async fn dashboard_summary(app: axum::Router) -> Value { + dashboard_summary_uri(app, "/api/admin/dashboard/summary").await +} + +async fn dashboard_summary_with_kind(app: axum::Router, kind: &str) -> Value { + dashboard_summary_uri(app, &format!("/api/admin/dashboard/summary?kind={kind}")).await +} + +async fn dashboard_summary_uri(app: axum::Router, uri: &str) -> Value { let response = app .oneshot( Request::builder() .method("GET") - .uri("/api/admin/dashboard/summary") + .uri(uri) .header("cookie", "cpr_admin_session=session_1") .header("x-request-id", "req_dashboard_summary") .body(Body::empty()) @@ -424,12 +504,12 @@ async fn dashboard_summary(app: axum::Router) -> Value { response_json(response).await } -async fn dashboard_trend(app: axum::Router) -> Value { +async fn dashboard_trend(app: axum::Router, kind: &str) -> Value { let response = app .oneshot( Request::builder() .method("GET") - .uri("/api/admin/dashboard/trend?kind=usage") + .uri(format!("/api/admin/dashboard/trend?kind={kind}")) .header("cookie", "cpr_admin_session=session_1") .header("x-request-id", "req_dashboard_trend") .body(Body::empty()) diff --git a/backend/tests/admin/monitoring/usage_record_routes.rs b/backend/tests/admin/monitoring/usage_record_routes.rs index 3f572f256..b211e884a 100644 --- a/backend/tests/admin/monitoring/usage_record_routes.rs +++ b/backend/tests/admin/monitoring/usage_record_routes.rs @@ -183,10 +183,10 @@ async fn admin_usage_records_should_return_table_display_fields() { assert_eq!(row["tokenDetails"]["reasoningTokens"], 12); assert_eq!(row["tokenDetails"]["totalTokens"], 1200); assert_eq!(row["tokenDetails"]["inputTokensDisplay"], "1,000"); - assert_eq!(row["costDetails"]["inputCostDisplay"], "$0.004500"); - assert_eq!(row["costDetails"]["outputCostDisplay"], "$0.006000"); - assert_eq!(row["costDetails"]["cacheReadCostDisplay"], "$0.000050"); - assert_eq!(row["costDetails"]["totalCostDisplay"], "$0.010550"); + assert_eq!(row["costDetails"]["inputCostDisplay"], "$0.00"); + assert_eq!(row["costDetails"]["outputCostDisplay"], "$0.01"); + assert_eq!(row["costDetails"]["cacheReadCostDisplay"], "$0.00"); + assert_eq!(row["costDetails"]["totalCostDisplay"], "$0.01"); assert_eq!( row["costDetails"]["inputPriceDisplay"], "$5.0000 / 1M Token" @@ -343,7 +343,7 @@ async fn admin_usage_record_insight_cards_should_aggregate_filtered_usage_dimens assert_eq!(requested_body["data"][0]["totalTokensValue"], 42); assert_eq!(requested_body["data"][0]["totalTokensTotal"], "48"); assert_eq!(requested_body["data"][0]["totalTokensTotalValue"], 48); - assert_eq!(requested_body["data"][0]["actualCost"], "$0.000483"); + assert_eq!(requested_body["data"][0]["actualCost"], "$0.00"); let upstream_models = app .clone() @@ -422,8 +422,8 @@ async fn admin_usage_record_insight_cards_should_aggregate_filtered_usage_dimens assert_eq!(token_body["data"][0]["outputTokens"], "13"); assert_eq!(token_body["data"][0]["cacheCreationTokens"], "0"); assert_eq!(token_body["data"][0]["cachedTokens"], "6"); - assert_eq!(token_body["data"][0]["actualCost"], "$0.000538"); - assert_eq!(token_body["data"][0]["cost"], "$0.000538"); + assert_eq!(token_body["data"][0]["actualCost"], "$0.00"); + assert_eq!(token_body["data"][0]["cost"], "$0.00"); let latency_trend = app .oneshot( diff --git a/backend/tests/upstream/transport/websocket_pool.rs b/backend/tests/upstream/transport/websocket_pool.rs index 537ca52bc..09dd4e919 100644 --- a/backend/tests/upstream/transport/websocket_pool.rs +++ b/backend/tests/upstream/transport/websocket_pool.rs @@ -246,6 +246,92 @@ async fn websocket_pool_should_bypass_busy_key_with_one_shot_connections() { ); } +#[tokio::test] +async fn websocket_pool_should_release_slot_when_client_drops_stream() { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + let server = tokio::spawn(async move { + // 第一个连接：发一帧后保持沉默（模拟上游不再发帧、也不发 terminal）。 + // slot 只能靠客户端断开来释放，隔离验证 tx.closed() 机制。 + let (first_stream, _) = listener.accept().await.unwrap(); + let mut first_websocket = accept_codex_test_websocket(first_stream).await; + let _first_message = first_websocket.next().await.unwrap().unwrap(); + first_websocket + .send(Message::Text( + json!({ + "type": "response.output_text.delta", + "delta": "streaming has begun" + }) + .to_string() + .into(), + )) + .await + .unwrap(); + + // 第二个连接：客户端断开释放 slot 后，同 key 请求应新建连接。 + let (second_stream, _) = listener.accept().await.unwrap(); + let mut second_websocket = accept_codex_test_websocket(second_stream).await; + let _second_message = second_websocket.next().await.unwrap().unwrap(); + second_websocket + .send(Message::Text( + completed_websocket_response("resp_released_second", 2, 1).into(), + )) + .await + .unwrap(); + second_websocket.close(None).await.unwrap(); + drop(first_websocket); + }); + let pool = Arc::new(CodexWebSocketPool::default()); + let backend = CodexBackendClient::new( + reqwest::Client::builder().no_proxy().build().unwrap(), + format!("http://{addr}"), + crate::support::fingerprint::test_fingerprint(), + ) + .with_websocket_pool(pool); + let request = pooled_websocket_request("conversation-drop"); + + // 起流式请求：slot 变 Busy。 + let mut stream = backend + .create_response_stream( + &request, + request_context("req_drop_first", Some("chatgpt-account")), + ) + .await + .expect("first pooled websocket stream should start") + .body; + let first_chunk = stream + .next() + .await + .expect("stream should yield an initial chunk") + .expect("stream chunk should be valid"); + assert!(std::str::from_utf8(&first_chunk) + .unwrap() + .contains("streaming has begun")); + + // 客户端断开：drop stream → rx 被 drop → tx.closed() 完成 → + // 代理丢弃上游连接并释放 slot（不再等 idle 超时）。 + drop(stream); + tokio::time::sleep(Duration::from_millis(200)).await; + + // 同 key 的后续请求：slot 已释放 → 新建连接（new），而非 bypass(busy)。 + let second = backend + .create_response( + &request, + request_context("req_drop_second", Some("chatgpt-account")), + ) + .await + .expect("second request should succeed after slot release"); + server.await.unwrap(); + + assert!(second.body.contains("resp_released_second")); + let decision = second.websocket_pool_decision.unwrap(); + assert_eq!( + decision.kind(), + "new", + "client-drop must release the pool slot so the next same-key request builds a fresh connection instead of bypassing as busy" + ); +} + #[tokio::test] async fn websocket_pool_should_bypass_new_keys_after_account_cap() { let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); diff --git a/frontend/src/api/modules/dashboard.ts b/frontend/src/api/modules/dashboard.ts index f1a32ca56..63def2d10 100644 --- a/frontend/src/api/modules/dashboard.ts +++ b/frontend/src/api/modules/dashboard.ts @@ -1,9 +1,10 @@ import request from '../request' -export function getDashboardSummary() { +export function getDashboardSummary(params?: any) { return request({ url: '/api/admin/dashboard/summary', method: 'GET', + params, }) } diff --git a/frontend/src/layout/components/AppSidebar.vue b/frontend/src/layout/components/AppSidebar.vue index aba032be5..92c078040 100644 --- a/frontend/src/layout/components/AppSidebar.vue +++ b/frontend/src/layout/components/AppSidebar.vue @@ -313,7 +313,7 @@ watch( -
+
0" :class="summaryClass">
-
- +
+ {{ window.labelDisplay }} - - {{ window.usedPercentDisplay }} + + {{ quotaWindowUsageText(window) }}
diff --git a/frontend/src/views/accounts/constants.ts b/frontend/src/views/accounts/constants.ts index 581b70415..eba97cb60 100644 --- a/frontend/src/views/accounts/constants.ts +++ b/frontend/src/views/accounts/constants.ts @@ -20,8 +20,7 @@ export const accountColumns = [ { key: 'identity', label: '邮箱', flex: 2.6, minWidth: '260px', cellClass: relaxedCellClass }, { key: 'status', label: '状态', flex: 0.8, cellClass: relaxedCellClass }, { key: 'planType', label: '套餐', flex: 0.8, cellClass: relaxedCellClass }, - { key: 'usage', label: '用量', flex: 1.4, minWidth: '210px', cellClass: relaxedCellClass }, - { key: 'quota', label: '额度', flex: 0.8, minWidth: '112px', cellClass: relaxedCellClass }, + { key: 'usage', label: '用量', flex: 1.8, minWidth: '250px', cellClass: relaxedCellClass }, { key: 'updatedAtDisplay', label: '最后使用', diff --git a/frontend/src/views/accounts/index.vue b/frontend/src/views/accounts/index.vue index f412857de..b761b1470 100644 --- a/frontend/src/views/accounts/index.vue +++ b/frontend/src/views/accounts/index.vue @@ -107,11 +107,6 @@ const { } = useAccountsTable(accounts, selectedIds) bindAccountLoader(loadAccounts) - -function quotaTokenDisplay(account: any) { - const display = account?.usage?.totalTokensDisplay - return typeof display === 'string' && display.trim() ? display : '0' -} - -