diff --git a/src/admin/monitoring/dashboard.rs b/src/admin/monitoring/dashboard.rs index 840d93487..31fcbd29f 100644 --- a/src/admin/monitoring/dashboard.rs +++ b/src/admin/monitoring/dashboard.rs @@ -265,11 +265,7 @@ pub(crate) async fn dashboard_summary( let settings = state.services.settings.current(); let recent_usage_records = recent_events.into_iter().take(10).collect::>(); let account_emails = account_email_map(&state, &recent_usage_records).await?; - let dashboard_usage_records = usage_record_items( - recent_usage_records, - &account_emails, - &settings.model_aliases, - ); + let dashboard_usage_records = usage_record_items(recent_usage_records, &account_emails); Ok(AdminResponse::new( StatusCode::OK, diff --git a/src/admin/monitoring/usage_record_store.rs b/src/admin/monitoring/usage_record_store.rs index f7ce2fcf7..10d2ebd65 100644 --- a/src/admin/monitoring/usage_record_store.rs +++ b/src/admin/monitoring/usage_record_store.rs @@ -187,12 +187,38 @@ impl SqliteUsageRecordStore { usage_summary(&self.pool, &filter).await } - /// 聚合使用记录分析数据。 - pub async fn insights( + /// 按模型来源聚合使用记录分布。 + pub async fn model_distribution( &self, filter: UsageRecordFilter, - ) -> SqliteUsageRecordStoreResult { - usage_insights(&self.pool, &filter).await + source: UsageRecordModelSource, + ) -> SqliteUsageRecordStoreResult> { + usage_model_distribution(&self.pool, &filter, source, 8).await + } + + /// 按端点来源聚合使用记录分布。 + pub async fn endpoint_distribution( + &self, + filter: UsageRecordFilter, + source: UsageRecordEndpointSource, + ) -> SqliteUsageRecordStoreResult> { + usage_endpoint_distribution(&self.pool, &filter, source, 8).await + } + + /// 聚合 Token 使用趋势。 + pub async fn token_trend( + &self, + filter: UsageRecordFilter, + ) -> SqliteUsageRecordStoreResult> { + usage_trend(&self.pool, &filter).await + } + + /// 聚合延迟趋势。 + pub async fn latency_trend( + &self, + filter: UsageRecordFilter, + ) -> SqliteUsageRecordStoreResult> { + usage_trend(&self.pool, &filter).await } /// 清空使用记录。 @@ -483,88 +509,60 @@ fn nonnegative_i64(value: Option) -> u64 { value.unwrap_or_default().max(0).cast_unsigned() } -async fn usage_insights( +async fn usage_model_distribution( pool: &SqlitePool, filter: &UsageRecordFilter, -) -> SqliteUsageRecordStoreResult { - Ok(UsageRecordInsights { - models: usage_breakdown( - pool, - filter, - "coalesce(nullif(trim(json_extract(metadata_json, '$.requestedModel')), ''), nullif(trim(model), ''), '未知模型')", + source: UsageRecordModelSource, + limit: u32, +) -> SqliteUsageRecordStoreResult> { + let (group_expr, group_alias) = match source { + UsageRecordModelSource::Requested => ( + "coalesce(nullif(trim(json_extract(metadata_json, '$.requestedModel')), ''), nullif(trim(usage_records.model), ''), '未知模型')", "requested_model", - 8, - ) - .await?, - upstream_models: usage_model_breakdown(pool, filter, 8).await?, - model_mappings: usage_breakdown( - pool, - filter, + ), + UsageRecordModelSource::Upstream => ( + "coalesce( + nullif(trim(json_extract(usage_records.metadata_json, '$.upstreamModel')), ''), + nullif(trim(json_extract(usage_records.metadata_json, '$.requestedModel')), ''), + nullif(trim(usage_records.model), ''), + '未知模型' + )", + "model", + ), + UsageRecordModelSource::Mapping => ( "case when coalesce(nullif(trim(json_extract(metadata_json, '$.requestedModel')), ''), nullif(trim(usage_records.model), '')) is not null then coalesce(nullif(trim(json_extract(metadata_json, '$.requestedModel')), ''), nullif(trim(usage_records.model), ''), '未知模型') || ' -> ' || - coalesce(nullif(trim(json_extract(metadata_json, '$.upstreamModel')), ''), nullif(trim(alias.value), ''), nullif(trim(usage_records.model), ''), '未知模型') + coalesce(nullif(trim(json_extract(metadata_json, '$.upstreamModel')), ''), nullif(trim(json_extract(metadata_json, '$.requestedModel')), ''), nullif(trim(usage_records.model), ''), '未知模型') else '未知模型' end", "model_mapping", - 8, - ) - .await?, - endpoints: usage_breakdown( - pool, - filter, - "coalesce(nullif(route, ''), '未知端点')", - "endpoint", - 8, - ) - .await?, - upstream_endpoints: usage_breakdown( - pool, - filter, - "coalesce(nullif(trim(json_extract(metadata_json, '$.upstreamEndpoint')), ''), nullif(trim(json_extract(metadata_json, '$.upstreamRoute')), ''), nullif(trim(json_extract(metadata_json, '$.upstreamPath')), ''), nullif(route, ''), '未知端点')", - "upstream_endpoint", - 8, - ) - .await?, - endpoint_paths: usage_breakdown( - pool, - filter, - "coalesce(nullif(route, ''), '未知端点') || ' -> ' || coalesce(nullif(trim(json_extract(metadata_json, '$.upstreamEndpoint')), ''), nullif(trim(json_extract(metadata_json, '$.upstreamRoute')), ''), nullif(trim(json_extract(metadata_json, '$.upstreamPath')), ''), nullif(route, ''), '未知端点')", - "endpoint_path", - 8, - ) - .await?, - types: usage_breakdown( - pool, - filter, - "case when json_extract(metadata_json, '$.stream') = 1 then '流式' when json_extract(metadata_json, '$.stream') = 0 then '普通' else coalesce(nullif(transport, ''), '未知') end", - "record_type", - 8, - ) - .await?, - trend: usage_trend(pool, filter).await?, - }) + ), + }; + usage_breakdown(pool, filter, group_expr, group_alias, limit).await } -async fn usage_model_breakdown( +async fn usage_endpoint_distribution( pool: &SqlitePool, filter: &UsageRecordFilter, + source: UsageRecordEndpointSource, limit: u32, ) -> SqliteUsageRecordStoreResult> { - usage_breakdown( - pool, - filter, - "coalesce( - nullif(trim(json_extract(usage_records.metadata_json, '$.upstreamModel')), ''), - nullif(trim(alias.value), ''), - nullif(trim(usage_records.model), ''), - '未知模型' - )", - "model", - limit, - ) - .await + let (group_expr, group_alias) = match source { + UsageRecordEndpointSource::Inbound => { + ("coalesce(nullif(route, ''), '未知端点')", "endpoint") + } + UsageRecordEndpointSource::Upstream => ( + "coalesce(nullif(trim(json_extract(metadata_json, '$.upstreamEndpoint')), ''), nullif(trim(json_extract(metadata_json, '$.upstreamRoute')), ''), nullif(trim(json_extract(metadata_json, '$.upstreamPath')), ''), nullif(route, ''), '未知端点')", + "upstream_endpoint", + ), + UsageRecordEndpointSource::Path => ( + "coalesce(nullif(route, ''), '未知端点') || ' -> ' || coalesce(nullif(trim(json_extract(metadata_json, '$.upstreamEndpoint')), ''), nullif(trim(json_extract(metadata_json, '$.upstreamRoute')), ''), nullif(trim(json_extract(metadata_json, '$.upstreamPath')), ''), nullif(route, ''), '未知端点')", + "endpoint_path", + ), + }; + usage_breakdown(pool, filter, group_expr, group_alias, limit).await } async fn usage_breakdown( @@ -585,15 +583,13 @@ async fn usage_breakdown( sum(coalesce(cast(json_extract(metadata_json, '$.usage.cachedTokens') as integer), cast(json_extract(metadata_json, '$.cachedTokens') as integer), 0)) as cached_tokens, coalesce( nullif(trim(json_extract(metadata_json, '$.upstreamModel')), ''), - nullif(trim(alias.value), ''), + nullif(trim(json_extract(metadata_json, '$.requestedModel')), ''), nullif(trim(usage_records.model), ''), '未知模型' ) as billing_model, coalesce(nullif(trim(json_extract(metadata_json, '$.billingServiceTier')), ''), nullif(trim(json_extract(metadata_json, '$.serviceTier')), ''), nullif(trim(json_extract(metadata_json, '$.usage.serviceTier')), '')) as service_tier, avg(case when latency_ms > 0 then latency_ms else null end) as average_latency_ms - from usage_records - left join runtime_settings on runtime_settings.id = 1 - left join json_each(runtime_settings.model_aliases_json) as alias on alias.key = usage_records.model", + from usage_records", ); push_filter(&mut builder, filter, None)?; builder.push(" group by "); @@ -677,15 +673,13 @@ async fn usage_trend( coalesce(cast(json_extract(metadata_json, '$.usage.cachedTokens') as integer), cast(json_extract(metadata_json, '$.cachedTokens') as integer), 0) as cached_tokens, coalesce( nullif(trim(json_extract(metadata_json, '$.upstreamModel')), ''), - nullif(trim(alias.value), ''), + nullif(trim(json_extract(metadata_json, '$.requestedModel')), ''), nullif(trim(usage_records.model), ''), '未知模型' ) as billing_model, coalesce(nullif(trim(json_extract(metadata_json, '$.billingServiceTier')), ''), nullif(trim(json_extract(metadata_json, '$.serviceTier')), ''), nullif(trim(json_extract(metadata_json, '$.usage.serviceTier')), '')) as service_tier, latency_ms - from usage_records - left join runtime_settings on runtime_settings.id = 1 - left join json_each(runtime_settings.model_aliases_json) as alias on alias.key = usage_records.model", + from usage_records", ); push_filter(&mut builder, filter, None)?; builder.push(" order by date asc, created_at asc"); @@ -1010,25 +1004,26 @@ pub struct UsageRecordSummary { pub average_latency_ms: Option, } -/// 使用记录分析。 -#[derive(Debug, Clone, Default, PartialEq)] -pub struct UsageRecordInsights { - /// 模型分布。 - pub models: Vec, - /// 上游模型分布。 - pub upstream_models: Vec, - /// 模型映射分布。 - pub model_mappings: Vec, - /// 端点分布。 - pub endpoints: Vec, - /// 上游端点分布。 - pub upstream_endpoints: Vec, - /// 端点路径分布。 - pub endpoint_paths: Vec, - /// 请求类型分布。 - pub types: Vec, - /// Token 趋势。 - pub trend: Vec, +/// 模型分布来源。 +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum UsageRecordModelSource { + /// 客户端请求模型。 + Requested, + /// 实际上游模型。 + Upstream, + /// 请求模型到上游模型的映射。 + Mapping, +} + +/// 端点分布来源。 +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum UsageRecordEndpointSource { + /// 入站端点。 + Inbound, + /// 上游端点。 + Upstream, + /// 入站到上游路径映射。 + Path, } /// 使用记录分布项。 @@ -1206,13 +1201,48 @@ impl AdminUsageRecordService { .map_err(|_| AdminUsageRecordError::List) } - /// 聚合使用记录分析数据。 - pub async fn insights( + /// 按模型来源聚合使用记录分布。 + pub async fn model_distribution( &self, filter: AdminUsageRecordFilter, - ) -> Result { + source: UsageRecordModelSource, + ) -> Result, AdminUsageRecordError> { self.store - .insights(filter.into()) + .model_distribution(filter.into(), source) + .await + .map_err(|_| AdminUsageRecordError::List) + } + + /// 按端点来源聚合使用记录分布。 + pub async fn endpoint_distribution( + &self, + filter: AdminUsageRecordFilter, + source: UsageRecordEndpointSource, + ) -> Result, AdminUsageRecordError> { + self.store + .endpoint_distribution(filter.into(), source) + .await + .map_err(|_| AdminUsageRecordError::List) + } + + /// 聚合 Token 趋势。 + pub async fn token_trend( + &self, + filter: AdminUsageRecordFilter, + ) -> Result, AdminUsageRecordError> { + self.store + .token_trend(filter.into()) + .await + .map_err(|_| AdminUsageRecordError::List) + } + + /// 聚合延迟趋势。 + pub async fn latency_trend( + &self, + filter: AdminUsageRecordFilter, + ) -> Result, AdminUsageRecordError> { + self.store + .latency_trend(filter.into()) .await .map_err(|_| AdminUsageRecordError::List) } diff --git a/src/admin/monitoring/usage_records.rs b/src/admin/monitoring/usage_records.rs index dfb9e67e6..7de0c45c7 100644 --- a/src/admin/monitoring/usage_records.rs +++ b/src/admin/monitoring/usage_records.rs @@ -1,6 +1,6 @@ //! 使用记录 HTTP 处理器。 -use std::collections::{BTreeMap, HashMap}; +use std::collections::HashMap; use axum::{ extract::{Query, State}, @@ -17,7 +17,8 @@ use crate::{ usage_record::{UsageRecord, UsageRecordLevel}, usage_record_store::{ AdminUsageRecordError, AdminUsageRecordFilter, UsageRecordBreakdown, - UsageRecordInsights, UsageRecordSummary, UsageRecordTrendPoint, + UsageRecordEndpointSource, UsageRecordModelSource, UsageRecordSummary, + UsageRecordTrendPoint, }, }, admin::{ @@ -56,6 +57,14 @@ pub(crate) struct UsageRecordsQuery { end_time: Option, } +#[derive(Debug, Deserialize)] +#[serde(rename_all = "camelCase")] +pub(crate) struct UsageRecordDistributionQuery { + #[serde(flatten)] + records: UsageRecordsQuery, + source: Option, +} + #[derive(Debug, Deserialize)] #[serde(rename_all = "camelCase")] pub(crate) struct UsageRecordDetailQuery { @@ -142,19 +151,6 @@ struct UsageRecordSummaryData { average_latency_ms_display: String, } -#[derive(Debug, Serialize)] -#[serde(rename_all = "camelCase")] -struct UsageRecordInsightsData { - models: Vec, - upstream_models: Vec, - model_mappings: Vec, - endpoints: Vec, - upstream_endpoints: Vec, - endpoint_paths: Vec, - types: Vec, - trend: Vec, -} - #[derive(Debug, Serialize)] #[serde(rename_all = "camelCase")] struct UsageRecordBreakdownData { @@ -213,9 +209,8 @@ pub(crate) async fn usage_records( { Ok(page) => { let account_emails = account_email_map(&state, &page.items).await?; - let model_aliases = state.services.settings.current().model_aliases.clone(); let page = NumberedPage { - items: usage_record_items(page.items, &account_emails, &model_aliases), + items: usage_record_items(page.items, &account_emails), total: page.total, page: page.page, page_size: page.page_size, @@ -237,9 +232,8 @@ pub(crate) async fn usage_records( { Ok(page) => { let account_emails = account_email_map(&state, &page.items).await?; - let model_aliases = state.services.settings.current().model_aliases.clone(); let page = Page { - items: usage_record_items(page.items, &account_emails, &model_aliases), + items: usage_record_items(page.items, &account_emails), next_cursor: page.next_cursor, }; Ok(AdminResponse::new( @@ -268,18 +262,101 @@ pub(crate) async fn usage_records_summary( } } -/// `GET /api/admin/usage/records/insights` -pub(crate) async fn usage_records_insights( +/// `GET /api/admin/usage/records/insights/models` +pub(crate) async fn usage_records_model_distribution( + State(state): State, + headers: HeaderMap, + Query(query): Query, +) -> Result { + require_admin_auth(&state, &headers).await?; + let source = model_source_from_query(query.source)?; + let filter = filter_from_query(query.records)?; + match state + .services + .usage_records + .model_distribution(filter, source) + .await + { + Ok(items) => Ok(AdminResponse::new( + StatusCode::OK, + AdminEnvelope::ok( + items + .into_iter() + .map(UsageRecordBreakdownData::from) + .collect::>(), + ), + )), + Err(error) => Err(log_error(&error)), + } +} + +/// `GET /api/admin/usage/records/insights/endpoints` +pub(crate) async fn usage_records_endpoint_distribution( + State(state): State, + headers: HeaderMap, + Query(query): Query, +) -> Result { + require_admin_auth(&state, &headers).await?; + let source = endpoint_source_from_query(query.source)?; + let filter = filter_from_query(query.records)?; + match state + .services + .usage_records + .endpoint_distribution(filter, source) + .await + { + Ok(items) => Ok(AdminResponse::new( + StatusCode::OK, + AdminEnvelope::ok( + items + .into_iter() + .map(UsageRecordBreakdownData::from) + .collect::>(), + ), + )), + Err(error) => Err(log_error(&error)), + } +} + +/// `GET /api/admin/usage/records/insights/token-trend` +pub(crate) async fn usage_records_token_trend( State(state): State, headers: HeaderMap, Query(query): Query, ) -> Result { require_admin_auth(&state, &headers).await?; let filter = filter_from_query(query)?; - match state.services.usage_records.insights(filter).await { - Ok(insights) => Ok(AdminResponse::new( + match state.services.usage_records.token_trend(filter).await { + Ok(points) => Ok(AdminResponse::new( StatusCode::OK, - AdminEnvelope::ok(UsageRecordInsightsData::from(insights)), + AdminEnvelope::ok( + points + .into_iter() + .map(UsageRecordTrendPointData::from) + .collect::>(), + ), + )), + Err(error) => Err(log_error(&error)), + } +} + +/// `GET /api/admin/usage/records/insights/latency-trend` +pub(crate) async fn usage_records_latency_trend( + State(state): State, + headers: HeaderMap, + Query(query): Query, +) -> Result { + require_admin_auth(&state, &headers).await?; + let filter = filter_from_query(query)?; + match state.services.usage_records.latency_trend(filter).await { + Ok(points) => Ok(AdminResponse::new( + StatusCode::OK, + AdminEnvelope::ok( + points + .into_iter() + .map(UsageRecordTrendPointData::from) + .collect::>(), + ), )), Err(error) => Err(log_error(&error)), } @@ -295,10 +372,9 @@ pub(crate) async fn usage_record_detail( match state.services.usage_records.get(&query.id).await { Ok(Some(log)) => { let account_emails = account_email_map(&state, std::slice::from_ref(&log)).await?; - let model_aliases = state.services.settings.current().model_aliases.clone(); Ok(AdminResponse::new( StatusCode::OK, - AdminEnvelope::ok(usage_record_data(log, &account_emails, &model_aliases)), + AdminEnvelope::ok(usage_record_data(log, &account_emails)), )) } Ok(None) => Err(AdminError::new( @@ -349,6 +425,34 @@ fn filter_from_query(query: UsageRecordsQuery) -> Result) -> Result { + match non_empty(value).as_deref().unwrap_or("requested") { + "requested" => Ok(UsageRecordModelSource::Requested), + "upstream" => Ok(UsageRecordModelSource::Upstream), + "mapping" => Ok(UsageRecordModelSource::Mapping), + _ => Err(AdminError::new( + StatusCode::BAD_REQUEST, + 40003, + "Invalid model distribution source", + )), + } +} + +fn endpoint_source_from_query( + value: Option, +) -> Result { + match non_empty(value).as_deref().unwrap_or("inbound") { + "inbound" => Ok(UsageRecordEndpointSource::Inbound), + "upstream" => Ok(UsageRecordEndpointSource::Upstream), + "path" => Ok(UsageRecordEndpointSource::Path), + _ => Err(AdminError::new( + StatusCode::BAD_REQUEST, + 40004, + "Invalid endpoint distribution source", + )), + } +} + fn optional_datetime( value: Option, ) -> Result>, AdminError> { @@ -440,25 +544,24 @@ pub(crate) async fn account_email_map( pub(crate) fn usage_record_items( items: Vec, account_emails: &HashMap, - model_aliases: &BTreeMap, ) -> Vec { items .into_iter() - .map(|record| usage_record_data(record, account_emails, model_aliases)) + .map(|record| usage_record_data(record, account_emails)) .collect() } fn usage_record_data( record: UsageRecord, account_emails: &HashMap, - model_aliases: &BTreeMap, ) -> UsageRecordData { let account_email = record .account_id .as_deref() .and_then(|account_id| account_emails.get(account_id)) - .cloned(); - let (requested_model, upstream_model) = usage_record_models(&record, model_aliases); + .cloned() + .or_else(|| metadata_string(&record.metadata, &["accountEmail", "account_email"])); + let (requested_model, upstream_model) = usage_record_models(&record); let client_ip = metadata_string(&record.metadata, &["clientIp", "ipAddress", "ip_address"]); let user_agent = metadata_string(&record.metadata, &["userAgent", "user_agent"]); let reasoning_effort = @@ -494,30 +597,17 @@ fn usage_record_data( } } -fn usage_record_models( - record: &UsageRecord, - model_aliases: &BTreeMap, -) -> (Option, Option) { +fn usage_record_models(record: &UsageRecord) -> (Option, Option) { let stored_model = record .model .as_deref() .map(str::trim) .filter(|model| !model.is_empty()); - let alias_target = stored_model - .and_then(|model| model_aliases.get(model)) - .map(String::as_str) - .map(str::trim) - .filter(|model| !model.is_empty()); - let requested_model = metadata_string(&record.metadata, &["requestedModel"]) - .or_else(|| alias_target.and_then(|_| stored_model.map(ToString::to_string))); + .or_else(|| stored_model.map(ToString::to_string)); let upstream_model = metadata_string(&record.metadata, &["upstreamModel"]) - .or_else(|| alias_target.map(ToString::to_string)) - .or_else(|| { - requested_model - .as_ref() - .and_then(|_| stored_model.map(ToString::to_string)) - }); + .or_else(|| requested_model.clone()) + .or_else(|| stored_model.map(ToString::to_string)); (requested_model, upstream_model) } @@ -740,53 +830,6 @@ impl From for UsageRecordSummaryData { } } -impl From for UsageRecordInsightsData { - fn from(insights: UsageRecordInsights) -> Self { - Self { - models: insights - .models - .into_iter() - .map(UsageRecordBreakdownData::from) - .collect(), - upstream_models: insights - .upstream_models - .into_iter() - .map(UsageRecordBreakdownData::from) - .collect(), - model_mappings: insights - .model_mappings - .into_iter() - .map(UsageRecordBreakdownData::from) - .collect(), - endpoints: insights - .endpoints - .into_iter() - .map(UsageRecordBreakdownData::from) - .collect(), - upstream_endpoints: insights - .upstream_endpoints - .into_iter() - .map(UsageRecordBreakdownData::from) - .collect(), - endpoint_paths: insights - .endpoint_paths - .into_iter() - .map(UsageRecordBreakdownData::from) - .collect(), - types: insights - .types - .into_iter() - .map(UsageRecordBreakdownData::from) - .collect(), - trend: insights - .trend - .into_iter() - .map(UsageRecordTrendPointData::from) - .collect(), - } - } -} - impl From for UsageRecordBreakdownData { fn from(item: UsageRecordBreakdown) -> Self { Self { diff --git a/src/admin/router.rs b/src/admin/router.rs index 0682e74b1..fcc8509d4 100644 --- a/src/admin/router.rs +++ b/src/admin/router.rs @@ -18,8 +18,9 @@ use super::{ monitoring::{ dashboard::{dashboard_summary, dashboard_trend}, usage_records::{ - clear_usage_records, usage_record_detail, usage_records, usage_records_insights, - usage_records_summary, + clear_usage_records, usage_record_detail, usage_records, + usage_records_endpoint_distribution, usage_records_latency_trend, + usage_records_model_distribution, usage_records_summary, usage_records_token_trend, }, }, settings::routes::{ @@ -68,8 +69,20 @@ pub fn router() -> Router { get(usage_records_summary), ) .route( - "/api/admin/usage/records/insights", - get(usage_records_insights), + "/api/admin/usage/records/insights/models", + get(usage_records_model_distribution), + ) + .route( + "/api/admin/usage/records/insights/endpoints", + get(usage_records_endpoint_distribution), + ) + .route( + "/api/admin/usage/records/insights/token-trend", + get(usage_records_token_trend), + ) + .route( + "/api/admin/usage/records/insights/latency-trend", + get(usage_records_latency_trend), ) .route("/api/admin/usage/records/delete", post(clear_usage_records)) .route("/api/admin/usage/records/detail", get(usage_record_detail)) diff --git a/src/proxy/dispatch/chat.rs b/src/proxy/dispatch/chat.rs index c3edb9412..796a8addb 100644 --- a/src/proxy/dispatch/chat.rs +++ b/src/proxy/dispatch/chat.rs @@ -384,7 +384,7 @@ impl ChatDispatchService { request_id, account_id: &account_id, route: "/v1/chat/completions", - model: &request.model, + model: &display_model, requested_model: Some(requested_model), client_ip: request.client_ip.as_deref(), client_user_agent: request.client_user_agent.as_deref(), diff --git a/src/proxy/dispatch/responses.rs b/src/proxy/dispatch/responses.rs index 3bb5e8731..9388bb591 100644 --- a/src/proxy/dispatch/responses.rs +++ b/src/proxy/dispatch/responses.rs @@ -58,6 +58,7 @@ use crate::{ pool::{AccountAcquireRequest, RuntimeAccountPoolService}, }, upstream::{ + models::ModelCatalog, protocol::{ events::{extract_sse_usage, extract_usage, TokenUsage}, responses::{ @@ -386,6 +387,7 @@ impl ResponseDispatchService { let started_at = Instant::now(); let catalog = self.models.catalog().await; let parsed_model = catalog.parse_model_name(requested_model); + let display_model = ModelCatalog::build_display_model_name(&parsed_model); apply_response_model_options(&mut request, &parsed_model); let tuple_schema = request.tuple_schema.clone(); let image_generation_requested = request.expects_image_generation(); @@ -833,6 +835,7 @@ impl ResponseDispatchService { usage_records: &self.usage_records, request_id, account_id: &release_account_id, + account_email: account.email.as_deref(), route, model: requested_model, started_at, @@ -876,7 +879,7 @@ impl ResponseDispatchService { request_id, account_id: &account.id, route, - model: &request.model, + model: &display_model, requested_model: Some(requested_model), client_ip: request.client_ip.as_deref(), client_user_agent: request.client_user_agent.as_deref(), @@ -988,6 +991,7 @@ impl ResponseDispatchService { let started_at = Instant::now(); let catalog = self.models.catalog().await; let parsed_model = catalog.parse_model_name(requested_model); + let display_model = ModelCatalog::build_display_model_name(&parsed_model); apply_response_model_options(&mut request, &parsed_model); request.stream = true; let tuple_schema = request.tuple_schema.clone(); @@ -1222,6 +1226,7 @@ impl ResponseDispatchService { usage_records: &self.usage_records, request_id, account_id: &release_account_id, + account_email: account.email.as_deref(), route, model: requested_model, started_at, @@ -1334,7 +1339,7 @@ impl ResponseDispatchService { request_id, account_id: &release_account_id, route, - model: &request.model, + model: &display_model, requested_model, started_at, transport, @@ -1358,6 +1363,7 @@ impl ResponseDispatchService { request_id: request_id.to_string(), route: route.to_string(), model: request.model.clone(), + display_model: display_model.clone(), requested_model: requested_model.to_string(), client_ip: request.client_ip.clone(), request, @@ -1492,6 +1498,7 @@ impl ResponseDispatchService { usage_records: &self.usage_records, request_id, account_id: &release_account_id, + account_email: account.email.as_deref(), route, model: requested_model, started_at, @@ -1516,6 +1523,7 @@ impl ResponseDispatchService { let started_at = Instant::now(); let catalog = self.models.catalog().await; let parsed_model = catalog.parse_model_name(requested_model); + let display_model = ModelCatalog::build_display_model_name(&parsed_model); request.model = parsed_model.model_id; let mut excluded_account_ids = Vec::new(); let mut rate_limited_count = 0usize; @@ -1756,7 +1764,7 @@ impl ResponseDispatchService { request_id, account_id: &account.id, route: "/v1/responses/compact", - model: &request.model, + model: &display_model, requested_model: Some(requested_model), client_ip: request.client_ip.as_deref(), client_user_agent: request.client_user_agent.as_deref(), @@ -2046,6 +2054,7 @@ struct LiveResponseStreamContext { request_id: String, route: String, model: String, + display_model: String, requested_model: String, client_ip: Option, request: CodexResponsesRequest, @@ -2169,7 +2178,7 @@ fn update_first_token_ms( first_token_ms: &mut Option, ) { if first_token_ms.is_none() && response_body_has_first_token(body_bytes) { - *first_token_ms = Some(elapsed_millis_i64(context.started_at)); + *first_token_ms = Some(elapsed_millis_i64(context.started_at).max(1)); } } @@ -2180,23 +2189,78 @@ fn response_body_has_first_token(body_bytes: &[u8]) -> bool { }; parse_sse_events(&body[..event_end + 2]).is_ok_and(|events| { events.iter().any(|event| { - serde_json::from_str::(&event.data).is_ok_and(|value| { - let event_type = event - .event - .as_deref() - .or_else(|| value.get("type").and_then(Value::as_str)); - matches!( - event_type, - Some("response.output_text.delta" | "response.reasoning_summary_text.delta") - ) && value - .get("delta") - .and_then(Value::as_str) - .is_some_and(|delta| !delta.is_empty()) - }) + serde_json::from_str::(&event.data) + .is_ok_and(|value| response_event_has_first_token(event.event.as_deref(), &value)) }) }) } +fn response_event_has_first_token(event_name: Option<&str>, value: &Value) -> bool { + match response_stream_event_type(event_name, value) { + Some( + "response.output_text.delta" + | "response.reasoning_summary_text.delta" + | "response.reasoning_text.delta" + | "response.function_call_arguments.delta" + | "response.custom_tool_call_input.delta", + ) => non_empty_string_field(value, &["delta"]), + Some("response.output_text.done" | "response.reasoning_text.done") => { + non_empty_string_field(value, &["text"]) + } + Some("response.output_item.added" | "response.output_item.done") => value + .get("item") + .is_some_and(output_item_has_generated_content), + Some("response.completed") => value + .get("response") + .is_some_and(completed_response_has_generated_content), + _ => false, + } +} + +fn response_stream_event_type<'a>( + event_name: Option<&'a str>, + value: &'a Value, +) -> Option<&'a str> { + event_name.or_else(|| value.get("type").and_then(Value::as_str)) +} + +fn completed_response_has_generated_content(response: &Value) -> bool { + non_empty_string_field(response, &["output_text"]) + || response + .get("output") + .and_then(Value::as_array) + .is_some_and(|items| items.iter().any(output_item_has_generated_content)) + || response + .pointer("/usage/output_tokens") + .and_then(Value::as_i64) + .is_some_and(|tokens| tokens > 0) +} + +fn output_item_has_generated_content(item: &Value) -> bool { + non_empty_string_field(item, &["content", "text", "arguments", "input"]) + || item + .get("content") + .and_then(Value::as_array) + .is_some_and(|parts| parts.iter().any(output_content_part_has_text)) + || item + .get("summary") + .and_then(Value::as_array) + .is_some_and(|parts| parts.iter().any(output_content_part_has_text)) +} + +fn output_content_part_has_text(part: &Value) -> bool { + non_empty_string_field(part, &["text", "content"]) +} + +fn non_empty_string_field(value: &Value, fields: &[&str]) -> bool { + fields.iter().any(|field| { + value + .get(field) + .and_then(Value::as_str) + .is_some_and(|text| !text.trim().is_empty()) + }) +} + async fn send_live_response_stream_chunk( sender: &mpsc::Sender>, body_bytes: &mut Vec, @@ -2534,6 +2598,7 @@ struct ResponseUpstreamErrorEventRecord<'a> { usage_records: &'a AdminUsageRecordService, request_id: &'a str, account_id: &'a str, + account_email: Option<&'a str>, route: &'a str, model: &'a str, started_at: Instant, @@ -2626,6 +2691,18 @@ async fn record_response_upstream_error_event(record: ResponseUpstreamErrorEvent event.model = Some(record.model.to_string()); event.status_code = Some(i64::from(upstream_error_http_status(record.error))); event.latency_ms = Some(elapsed_millis_i64(record.started_at)); + if let Some(object) = metadata.as_object_mut() { + if let Some(account_email) = record + .account_email + .map(str::trim) + .filter(|value| !value.is_empty()) + { + object.insert( + "accountEmail".to_string(), + Value::String(account_email.to_string()), + ); + } + } event.metadata = metadata; if let Err(error) = record.usage_records.record(event).await { tracing::warn!(account_id = %record.account_id, error = %error, "failed to record upstream error event"); @@ -2852,13 +2929,13 @@ async fn record_live_response_stream_event( event.request_id = Some(context.request_id.clone()); event.account_id = Some(context.account_id.clone()); event.route = Some(context.route.clone()); - event.model = Some(context.model.clone()); + event.model = Some(context.display_model.clone()); event.status_code = Some(status_code); event.latency_ms = Some(elapsed_millis_i64(context.started_at)); enrich_usage_record_identity( &mut metadata, Some(&context.requested_model), - &context.model, + &context.display_model, context.client_ip.as_deref(), context.request.client_user_agent.as_deref(), reasoning_effort_from_request(&context.request), @@ -3269,3 +3346,46 @@ fn enrich_response_dispatch_error_metadata(metadata: &mut Value, error: &Respons fn elapsed_millis_i64(started_at: Instant) -> i64 { started_at.elapsed().as_millis().min(i64::MAX as u128) as i64 } + +#[cfg(test)] +mod tests { + use super::response_body_has_first_token; + + #[test] + fn first_token_detection_should_match_text_delta() { + let body = br#"event: response.output_text.delta +data: {"delta":"hello"} + +"#; + + assert!(response_body_has_first_token(body)); + } + + #[test] + fn first_token_detection_should_match_output_item_done_text() { + let body = br#"event: response.output_item.done +data: {"item":{"type":"message","role":"assistant","content":[{"type":"output_text","text":"hello"}]}} + +"#; + + assert!(response_body_has_first_token(body)); + } + + #[test] + fn first_token_detection_should_fallback_to_completed_usage_output_tokens() { + let body = br#"event: response.completed +data: {"response":{"id":"resp_1","status":"completed","usage":{"input_tokens":3,"output_tokens":4}}} + +"#; + + assert!(response_body_has_first_token(body)); + } + + #[test] + fn first_token_detection_should_ignore_incomplete_frames() { + let body = br#"event: response.output_text.delta +data: {"delta":"hello"}"#; + + assert!(!response_body_has_first_token(body)); + } +} diff --git a/tests/admin/monitoring/usage_records_routes.rs b/tests/admin/monitoring/usage_records_routes.rs index 58ccbfb50..3104f77d8 100644 --- a/tests/admin/monitoring/usage_records_routes.rs +++ b/tests/admin/monitoring/usage_records_routes.rs @@ -116,15 +116,8 @@ async fn admin_usage_records_should_return_numbered_page_metadata() { #[tokio::test] async fn admin_usage_records_should_return_table_display_fields() { - let (app, store, pool, _dir) = admin_usage_records_test_app_with_config( - "admin-usage-records-display-fields.sqlite", - |config| { - config - .model_aliases - .insert("client-alias".to_string(), "gpt-5.5".to_string()); - }, - ) - .await; + let (app, store, pool, _dir) = + admin_usage_records_test_app_with_pool("admin-usage-records-display-fields.sqlite").await; let now = Utc::now().to_rfc3339(); sqlx::query("insert into accounts (id, email, access_token, status, added_at, updated_at) values (?, ?, ?, ?, ?, ?)") .bind("acct_display") @@ -140,9 +133,11 @@ async fn admin_usage_records_should_return_table_display_fields() { let mut event = UsageRecord::new("request", UsageRecordLevel::Info, "display fields"); event.id = "usage_display_fields".to_string(); event.account_id = Some("acct_display".to_string()); - event.model = Some("client-alias".to_string()); + event.model = Some("gpt-5.5".to_string()); event.latency_ms = Some(12_345); event.metadata = json!({ + "requestedModel": "client-alias", + "upstreamModel": "gpt-5.5", "clientIp": "203.0.113.8", "userAgent": "codex-tui/0.142.2 (Ubuntu 24.4.0; aarch64) xterm-256color", "reasoningEffort": "xhigh", @@ -201,6 +196,38 @@ async fn admin_usage_records_should_return_table_display_fields() { assert_eq!(row["latencyMsDisplay"], "12.3 s"); } +#[tokio::test] +async fn admin_usage_records_should_use_record_account_email_snapshot_when_account_is_missing() { + let (app, store, _dir) = + admin_usage_records_test_app("admin-usage-records-email-snapshot.sqlite").await; + let mut event = UsageRecord::new("v1.response", UsageRecordLevel::Error, "upstream failed"); + event.id = "usage_email_snapshot".to_string(); + event.account_id = Some("acct_deleted".to_string()); + event.metadata = json!({ + "accountEmail": "deleted@example.com", + "failed": true + }); + store.append(&event).await.unwrap(); + + let response = app + .oneshot( + Request::builder() + .method("GET") + .uri("/api/admin/usage/records?page=1&pageSize=20") + .header("cookie", "cpr_admin_session=session_1") + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + let body = response_json(response).await; + + assert_eq!( + body["data"]["items"][0]["accountEmail"], + "deleted@example.com" + ); +} + #[tokio::test] async fn admin_usage_records_summary_should_aggregate_filtered_usage() { let (app, store, _dir) = @@ -256,13 +283,13 @@ async fn admin_usage_records_summary_should_aggregate_filtered_usage() { } #[tokio::test] -async fn admin_usage_records_insights_should_aggregate_filtered_usage_dimensions() { - let (app, store, pool, _dir) = +async fn admin_usage_record_insight_cards_should_aggregate_filtered_usage_dimensions() { + let (app, store, _pool, _dir) = admin_usage_records_test_app_with_pool("admin-usage-records-insights.sqlite").await; let mut first = UsageRecord::new("request", UsageRecordLevel::Info, "insight success"); first.id = "usage_insight_success".to_string(); first.route = Some("/v1/responses".to_string()); - first.model = Some("client-alias".to_string()); + first.model = Some("gpt-5.5".to_string()); first.latency_ms = Some(240); first.metadata = json!({ "stream": true, @@ -277,14 +304,6 @@ async fn admin_usage_records_insights_should_aggregate_filtered_usage_dimensions } }); store.append(&first).await.unwrap(); - sqlx::query( - "insert into runtime_settings (id, model_aliases_json, refresh_margin_seconds, refresh_concurrency, max_concurrent_per_account, request_interval_ms, rotation_strategy, updated_at) values (1, ?, 300, 2, 3, 50, 'least_used', ?)", - ) - .bind(r#"{"client-alias":"gpt-5.5"}"#) - .bind(chrono::Utc::now().to_rfc3339()) - .execute(&pool) - .await - .unwrap(); let mut second = UsageRecord::new("request", UsageRecordLevel::Error, "insight failure"); second.id = "usage_insight_failure".to_string(); @@ -305,40 +324,75 @@ async fn admin_usage_records_insights_should_aggregate_filtered_usage_dimensions }); store.append(&second).await.unwrap(); - let response = app + let requested_models = app + .clone() .oneshot( Request::builder() .method("GET") - .uri("/api/admin/usage/records/insights?search=insight") + .uri("/api/admin/usage/records/insights/models?source=requested&search=insight") .header("cookie", "cpr_admin_session=session_1") - .header("x-request-id", "req_usage_records_insights") .body(Body::empty()) .unwrap(), ) .await .unwrap(); - let body = response_json(response).await; + let requested_body = response_json(requested_models).await; + assert_eq!(requested_body["code"], 200); + assert_eq!(requested_body["data"][0]["name"], "client-alias"); + assert_eq!(requested_body["data"][0]["totalTokens"], 42); + assert_eq!(requested_body["data"][0]["actualCostDisplay"], "$0.000483"); - assert_eq!(body["code"], 200); - assert_eq!(body["data"]["models"][0]["name"], "client-alias"); - assert_eq!(body["data"]["models"][0]["totalTokens"], 42); - assert_eq!(body["data"]["models"][0]["actualCostDisplay"], "$0.000483"); - assert_eq!(body["data"]["upstreamModels"][0]["name"], "gpt-5.5"); + let upstream_models = app + .clone() + .oneshot( + Request::builder() + .method("GET") + .uri("/api/admin/usage/records/insights/models?source=upstream&search=insight") + .header("cookie", "cpr_admin_session=session_1") + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); assert_eq!( - body["data"]["modelMappings"][0]["name"], - "client-alias -> gpt-5.5" + response_json(upstream_models).await["data"][0]["name"], + "gpt-5.5" ); + + let mappings = app + .clone() + .oneshot( + Request::builder() + .method("GET") + .uri("/api/admin/usage/records/insights/models?source=mapping&search=insight") + .header("cookie", "cpr_admin_session=session_1") + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + let mapping_body = response_json(mappings).await; + assert_eq!(mapping_body["data"][0]["name"], "client-alias -> gpt-5.5"); assert_eq!( - body["data"]["modelMappings"][1]["name"], + mapping_body["data"][1]["name"], "gpt-5.5-mini -> gpt-5.5-mini" ); - assert_eq!(body["data"]["endpoints"].as_array().unwrap().len(), 2); - assert_eq!( - body["data"]["upstreamEndpoints"].as_array().unwrap().len(), - 2 - ); - assert_eq!(body["data"]["endpointPaths"].as_array().unwrap().len(), 2); - let endpoint_paths = body["data"]["endpointPaths"] + + let endpoints = app + .clone() + .oneshot( + Request::builder() + .method("GET") + .uri("/api/admin/usage/records/insights/endpoints?source=path&search=insight") + .header("cookie", "cpr_admin_session=session_1") + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + let endpoint_body = response_json(endpoints).await; + assert_eq!(endpoint_body["data"].as_array().unwrap().len(), 2); + let endpoint_paths = endpoint_body["data"] .as_array() .unwrap() .iter() @@ -346,14 +400,185 @@ async fn admin_usage_records_insights_should_aggregate_filtered_usage_dimensions .collect::>(); assert!(endpoint_paths.contains(&"/v1/responses -> /backend/responses")); assert!(endpoint_paths.contains(&"/v1/chat/completions -> /backend/chat")); - assert_eq!(body["data"]["types"].as_array().unwrap().len(), 2); - assert_eq!(body["data"]["trend"][0]["inputTokens"], 35); - assert_eq!(body["data"]["trend"][0]["outputTokens"], 13); - assert_eq!(body["data"]["trend"][0]["cacheCreationTokens"], 0); - assert_eq!(body["data"]["trend"][0]["cachedTokens"], 6); - assert_eq!(body["data"]["trend"][0]["actualCostDisplay"], "$0.000538"); - assert_eq!(body["data"]["trend"][0]["costDisplay"], "$0.000538"); - assert_eq!(body["data"]["trend"][0]["averageLatencyMs"], 180.0); + + let token_trend = app + .clone() + .oneshot( + Request::builder() + .method("GET") + .uri("/api/admin/usage/records/insights/token-trend?search=insight") + .header("cookie", "cpr_admin_session=session_1") + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + let token_body = response_json(token_trend).await; + assert_eq!(token_body["data"][0]["inputTokens"], 35); + 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]["actualCostDisplay"], "$0.000538"); + assert_eq!(token_body["data"][0]["costDisplay"], "$0.000538"); + + let latency_trend = app + .oneshot( + Request::builder() + .method("GET") + .uri("/api/admin/usage/records/insights/latency-trend?search=insight") + .header("cookie", "cpr_admin_session=session_1") + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!( + response_json(latency_trend).await["data"][0]["averageLatencyMs"], + 180.0 + ); +} + +#[tokio::test] +async fn admin_usage_records_insight_card_endpoints_should_return_source_specific_data() { + let (app, store, _pool, _dir) = + admin_usage_records_test_app_with_pool("admin-usage-records-insight-cards.sqlite").await; + let mut first = UsageRecord::new("request", UsageRecordLevel::Info, "card success"); + first.id = "usage_card_success".to_string(); + first.route = Some("/v1/responses".to_string()); + first.model = Some("gpt-5.5".to_string()); + first.latency_ms = Some(240); + first.metadata = json!({ + "stream": true, + "requestedModel": "client-alias", + "upstreamModel": "gpt-5.5", + "upstreamEndpoint": "/backend/responses", + "usage": { + "inputTokens": 30, + "outputTokens": 12, + "cachedTokens": 6 + } + }); + store.append(&first).await.unwrap(); + + let mut second = UsageRecord::new("request", UsageRecordLevel::Info, "card mini"); + second.id = "usage_card_mini".to_string(); + second.route = Some("/v1/chat/completions".to_string()); + second.model = Some("gpt-5.5-mini".to_string()); + second.latency_ms = Some(120); + second.metadata = json!({ + "stream": false, + "requestedModel": "gpt-5.5-mini", + "upstreamModel": "gpt-5.5-mini", + "upstreamEndpoint": "/backend/chat", + "usage": { + "inputTokens": 5, + "outputTokens": 1 + } + }); + store.append(&second).await.unwrap(); + + let upstream_models = app + .clone() + .oneshot( + Request::builder() + .method("GET") + .uri("/api/admin/usage/records/insights/models?source=upstream&search=card") + .header("cookie", "cpr_admin_session=session_1") + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + let upstream_body = response_json(upstream_models).await; + assert_eq!(upstream_body["code"], 200); + assert_eq!(upstream_body["data"][0]["name"], "gpt-5.5"); + assert_eq!(upstream_body["data"][0]["totalTokens"], 42); + + let mappings = app + .clone() + .oneshot( + Request::builder() + .method("GET") + .uri("/api/admin/usage/records/insights/models?source=mapping&search=card") + .header("cookie", "cpr_admin_session=session_1") + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!( + response_json(mappings).await["data"][0]["name"], + "client-alias -> gpt-5.5" + ); + + let endpoints = app + .clone() + .oneshot( + Request::builder() + .method("GET") + .uri("/api/admin/usage/records/insights/endpoints?source=path&search=card") + .header("cookie", "cpr_admin_session=session_1") + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + let endpoint_body = response_json(endpoints).await; + let endpoint_paths = endpoint_body["data"] + .as_array() + .unwrap() + .iter() + .map(|item| item["name"].as_str().unwrap()) + .collect::>(); + assert!(endpoint_paths.contains(&"/v1/responses -> /backend/responses")); + assert!(endpoint_paths.contains(&"/v1/chat/completions -> /backend/chat")); + + let token_trend = app + .clone() + .oneshot( + Request::builder() + .method("GET") + .uri("/api/admin/usage/records/insights/token-trend?search=card") + .header("cookie", "cpr_admin_session=session_1") + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + let token_body = response_json(token_trend).await; + assert_eq!(token_body["data"][0]["inputTokens"], 35); + assert_eq!(token_body["data"][0]["cachedTokens"], 6); + + let latency_trend = app + .clone() + .oneshot( + Request::builder() + .method("GET") + .uri("/api/admin/usage/records/insights/latency-trend?search=card") + .header("cookie", "cpr_admin_session=session_1") + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!( + response_json(latency_trend).await["data"][0]["averageLatencyMs"], + 180.0 + ); + + let invalid_source = app + .oneshot( + Request::builder() + .method("GET") + .uri("/api/admin/usage/records/insights/models?source=actual") + .header("cookie", "cpr_admin_session=session_1") + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(invalid_source.status(), StatusCode::BAD_REQUEST); + assert_eq!(response_json(invalid_source).await["code"], 40003); } #[tokio::test] diff --git a/tests/proxy/dispatch/chat_upstream/usage_logging.rs b/tests/proxy/dispatch/chat_upstream/usage_logging.rs index 31d1c1395..54bb750ff 100644 --- a/tests/proxy/dispatch/chat_upstream/usage_logging.rs +++ b/tests/proxy/dispatch/chat_upstream/usage_logging.rs @@ -132,6 +132,9 @@ async fn responses_usage_record_should_store_resolved_upstream_model_after_alias let event = latest_response_usage_record(&pool).await; assert_eq!(event.request_id.as_deref(), Some("req_alias_usage_model")); assert_eq!(event.model.as_deref(), Some("gpt-5.5")); + let metadata: Value = serde_json::from_str(&event.metadata_json).unwrap(); + assert_eq!(metadata["requestedModel"], "client-alias"); + assert_eq!(metadata["upstreamModel"], "gpt-5.5"); let received = server.received_requests().await.unwrap(); let upstream_body: Value = serde_json::from_slice(&received[0].body).unwrap(); @@ -537,6 +540,20 @@ async fn responses_stream_should_record_usage_record_after_completed_stream() { assert_eq!(metadata["responseId"], "resp_stream_usage"); assert_eq!(metadata["usage"]["inputTokens"], 3); assert_eq!(metadata["usage"]["outputTokens"], 5); + assert!( + metadata["firstTokenMs"] + .as_i64() + .is_some_and(|value| value > 0), + "stream usage metadata should include first token latency: {metadata:?}", + ); + let first_token_bucket: (i64, i64) = sqlx::query_as( + "select first_token_latency_sum, first_token_latency_count from usage_time_buckets limit 1", + ) + .fetch_one(&pool) + .await + .unwrap(); + assert!(first_token_bucket.0 > 0); + assert_eq!(first_token_bucket.1, 1); assert_rate_limit_header(&metadata, "retry-after", "7"); assert_rate_limit_header(&metadata, "x-ratelimit-limit-requests", "99"); assert!(metadata.get("requestBody").is_none()); @@ -702,6 +719,10 @@ async fn responses_should_record_request_count_when_5xx_retries_are_exhausted() assert_eq!(status, StatusCode::BAD_GATEWAY); assert_eq!(authorizations.len(), 3, "requests: {authorizations:?}"); assert_eq!(failed_usage.map(|row| row.0).unwrap_or_default(), 1); + let event = latest_response_usage_record(&pool).await; + let metadata: Value = serde_json::from_str(&event.metadata_json).unwrap(); + assert_eq!(event.account_id.as_deref(), Some("acct_chat")); + assert_eq!(metadata["accountEmail"], "user@example.com"); } #[tokio::test] diff --git a/web/src/api/modules/usage.ts b/web/src/api/modules/usage.ts index f1a260ecb..b223f0811 100644 --- a/web/src/api/modules/usage.ts +++ b/web/src/api/modules/usage.ts @@ -24,9 +24,33 @@ export function getUsageRecordSummary(params?: any) { }) } -export function getUsageRecordInsights(params?: any) { +export function getUsageRecordModelDistribution(params?: any) { return request({ - url: '/api/admin/usage/records/insights', + url: '/api/admin/usage/records/insights/models', + method: 'GET', + params, + }) +} + +export function getUsageRecordEndpointDistribution(params?: any) { + return request({ + url: '/api/admin/usage/records/insights/endpoints', + method: 'GET', + params, + }) +} + +export function getUsageRecordTokenTrend(params?: any) { + return request({ + url: '/api/admin/usage/records/insights/token-trend', + method: 'GET', + params, + }) +} + +export function getUsageRecordLatencyTrend(params?: any) { + return request({ + url: '/api/admin/usage/records/insights/latency-trend', method: 'GET', params, }) diff --git a/web/src/components/base/BaseScrollbar.vue b/web/src/components/base/BaseScrollbar.vue index 4fab63271..1b0d30f80 100644 --- a/web/src/components/base/BaseScrollbar.vue +++ b/web/src/components/base/BaseScrollbar.vue @@ -7,14 +7,12 @@ const props = withDefaults( defineProps<{ viewClass?: string maxHeight?: string - forceVisible?: boolean horizontal?: boolean vertical?: boolean }>(), { viewClass: '', maxHeight: undefined, - forceVisible: false, horizontal: false, vertical: true, }, @@ -44,9 +42,7 @@ const { start: startHideTimer, stop: stopHideTimer } = useTimeoutFn(hideScrollba const canScrollY = computed(() => thumbHeight.value > 0) const canScrollX = computed(() => horizontalThumbWidth.value > 0) -const scrollbarVisible = computed( - () => props.forceVisible || dragging.value || horizontalDragging.value || visible.value, -) +const scrollbarVisible = computed(() => dragging.value || horizontalDragging.value || visible.value) const thumbStyle = computed(() => ({ height: `${thumbHeight.value}px`, transform: `translateY(${thumbTop.value}px)`, @@ -104,7 +100,7 @@ function clearHideTimer() { function scheduleHideScrollbar() { clearHideTimer() - if (props.forceVisible || dragging.value || horizontalDragging.value) { + if (dragging.value || horizontalDragging.value) { return } @@ -113,7 +109,7 @@ function scheduleHideScrollbar() { function hideScrollbar() { clearHideTimer() - if (!props.forceVisible && !dragging.value && !horizontalDragging.value) { + if (!dragging.value && !horizontalDragging.value) { visible.value = false } } diff --git a/web/src/components/base/BaseTable/index.vue b/web/src/components/base/BaseTable/index.vue index 02765d038..09edb59d4 100644 --- a/web/src/components/base/BaseTable/index.vue +++ b/web/src/components/base/BaseTable/index.vue @@ -2,7 +2,7 @@ import { useEventListener, useResizeObserver } from '@vueuse/core' import { ChevronLeft, ChevronRight } from '@lucide/vue' import { clamp } from 'es-toolkit' -import { computed, nextTick, onMounted, shallowRef, useTemplateRef } from 'vue' +import { computed, nextTick, onMounted, shallowRef, useTemplateRef, watch } from 'vue' import BaseEmpty from '../BaseEmpty.vue' import BaseScrollbar from '../BaseScrollbar.vue' @@ -72,6 +72,7 @@ const totalPages = computed(() => getTotalPages(props.pagination)) const currentPage = computed(() => getCurrentPage(props.pagination, totalPages.value)) const pageSizeOptions = computed(() => getPageSizeOptions(props.pagination)) const tableViewStyle = computed(() => resolveTableStyle(props.minWidth)) +const hasRows = computed(() => props.rows.length > 0) const pageSizeModel = computed({ get: () => String(props.pagination?.pageSize ?? ''), @@ -193,19 +194,24 @@ function scrollWrap() { return bodyScrollbarRef.value?.wrapRef ?? null } +function resetHorizontalScrollbar() { + horizontalThumbWidth.value = 0 + horizontalThumbLeft.value = 0 + horizontalScrolled.value = false + horizontalCanScrollRight.value = false +} + function updateHorizontalScrollbar() { const wrap = scrollWrap() - if (!wrap) { + if (!wrap || !hasRows.value) { + resetHorizontalScrollbar() return } const scrollRange = maxScrollLeft(wrap) const trackWidth = horizontalTrackWidth(wrap) if (scrollRange <= 0 || trackWidth <= 0) { - horizontalThumbWidth.value = 0 - horizontalThumbLeft.value = 0 - horizontalScrolled.value = false - horizontalCanScrollRight.value = false + resetHorizontalScrollbar() return } @@ -298,6 +304,13 @@ useResizeObserver( ), updateHorizontalScrollbar, ) +watch( + () => [props.rows.length, props.minWidth], + async () => { + await nextTick() + updateHorizontalScrollbar() + }, +) useEventListener(document, 'pointermove', handleHorizontalThumbPointerMove) useEventListener(document, 'pointerup', handleHorizontalThumbPointerUp) @@ -388,9 +401,9 @@ function paginationPageClass(page: number) { @@ -409,14 +422,6 @@ function paginationPageClass(page: number) { - - -
- -
- - -