diff --git a/backend/src/admin/monitoring/dashboard.rs b/backend/src/admin/monitoring/dashboard.rs index 7ce863a1b..ff82b12d2 100644 --- a/backend/src/admin/monitoring/dashboard.rs +++ b/backend/src/admin/monitoring/dashboard.rs @@ -15,7 +15,7 @@ use crate::{ admin::{ auth::session::AdminAuth, monitoring::{ - account_usage_service::{AdminUsageSummary, AdminUsageTimeBucketRecord}, + account_usage_service::AdminUsageTimeBucketRecord, diagnostics::{ fingerprint_diagnostics, AccountCapacityDiagnostics, AccountPoolDiagnostics, }, @@ -27,7 +27,7 @@ use crate::{ UsageRecordCostDetailsData, UsageRecordTokenDetailsData, }, usage_record_service::AdminUsageRecordFilter, - usage_record_store::UsageRecordAccountUsage, + usage_record_store::{UsageRecordAccountUsage, UsageRecordSummary}, }, response::{AdminEnvelope, AdminError, AdminResponse}, }, @@ -292,10 +292,10 @@ pub(crate) async fn dashboard_summary( let today_filter = today_usage_record_filter(now); let usage_summary = state .services - .usage - .summary() + .usage_records + .summary(AdminUsageRecordFilter::default()) .await - .unwrap_or_else(|_| empty_usage_summary()); + .unwrap_or_else(|_| UsageRecordSummary::default()); let account_usage_records = state .services .usage_records @@ -375,7 +375,7 @@ pub(crate) async fn dashboard_trend( fn dashboard_cards( accounts: &[Account], buckets: &[AdminUsageTimeBucketRecord], - usage_summary: &AdminUsageSummary, + usage_summary: &UsageRecordSummary, ) -> DashboardCardsData { dashboard_cards_at(accounts, buckets, usage_summary, Utc::now()) } @@ -383,7 +383,7 @@ fn dashboard_cards( fn dashboard_cards_at( accounts: &[Account], buckets: &[AdminUsageTimeBucketRecord], - usage_summary: &AdminUsageSummary, + usage_summary: &UsageRecordSummary, now: DateTime, ) -> DashboardCardsData { let today_start = china_day_start(now); @@ -391,9 +391,9 @@ fn dashboard_cards_at( let today = usage_window(buckets, today_start, now); let yesterday = usage_window(buckets, yesterday_start, today_start); let today_cost = cost_window(buckets, today_start, now).unwrap_or(0.0); - let total_requests = nonnegative_i64_to_u64(usage_summary.request_count); - let total_tokens = nonnegative_i64_to_u64(usage_summary.total_tokens); - let total_cached_tokens = nonnegative_i64_to_u64(usage_summary.cached_tokens); + let total_requests = usage_summary.total_requests; + let total_tokens = usage_summary.total_tokens; + let total_cached_tokens = usage_summary.cached_tokens; let total_hit_rate = summary_cache_hit_rate(usage_summary); let total_accounts = accounts.len() as u64; @@ -449,27 +449,10 @@ fn dashboard_cards_at( } } -fn empty_usage_summary() -> AdminUsageSummary { - AdminUsageSummary { - account_count: 0, - request_count: 0, - empty_response_count: 0, - input_tokens: 0, - output_tokens: 0, - cached_tokens: 0, - reasoning_tokens: 0, - total_tokens: 0, - image_input_tokens: 0, - image_output_tokens: 0, - image_request_count: 0, - image_request_failed_count: 0, - } -} - -fn summary_cache_hit_rate(summary: &AdminUsageSummary) -> Option { - let input_tokens = nonnegative_i64_to_u64(summary.input_tokens); +fn summary_cache_hit_rate(summary: &UsageRecordSummary) -> Option { + let input_tokens = summary.input_tokens; if input_tokens > 0 { - Some(nonnegative_i64_to_u64(summary.cached_tokens) as f64 / input_tokens as f64) + Some(summary.cached_tokens as f64 / input_tokens as f64) } else { None } diff --git a/backend/src/admin/monitoring/mod.rs b/backend/src/admin/monitoring/mod.rs index 49f3ebd0a..e4d6e0469 100644 --- a/backend/src/admin/monitoring/mod.rs +++ b/backend/src/admin/monitoring/mod.rs @@ -5,6 +5,9 @@ pub mod account_usage_store; pub(crate) mod billing; pub mod dashboard; pub mod diagnostics; +pub mod ops_error_model; +pub mod ops_error_service; +pub mod ops_error_store; pub mod usage_record_model; pub mod usage_record_routes; pub mod usage_record_service; diff --git a/backend/src/admin/monitoring/ops_error_model.rs b/backend/src/admin/monitoring/ops_error_model.rs new file mode 100644 index 000000000..06eaf9dda --- /dev/null +++ b/backend/src/admin/monitoring/ops_error_model.rs @@ -0,0 +1,75 @@ +//! 运维错误事件模型。 + +use chrono::{DateTime, Utc}; +use serde::{Deserialize, Serialize}; +use serde_json::{Map, Value}; +use uuid::Uuid; + +/// 结构化运维错误事件。 +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +#[serde(rename_all = "camelCase")] +pub struct OpsErrorLog { + /// 事件 ID。 + pub id: String, + /// 请求 ID。 + pub request_id: Option, + /// 事件类别。 + pub kind: String, + /// 账号 ID。预账号调度失败时为空。 + pub account_id: Option, + /// HTTP 路由。 + pub route: Option, + /// 模型名。 + pub model: Option, + /// 事件状态码。 + pub status_code: Option, + /// 客户端看到的状态码。 + pub client_status_code: Option, + /// 上游 HTTP 状态码。 + pub upstream_status_code: Option, + /// 上游传输方式,例如 http_sse 或 websocket。 + pub transport: Option, + /// 同一请求内的上游尝试序号。 + pub attempt_index: Option, + /// 失败分类。 + pub failure_class: Option, + /// 上游响应 ID。 + pub response_id: Option, + /// 上游请求 ID。 + pub upstream_request_id: Option, + /// 请求耗时毫秒。 + pub latency_ms: Option, + /// 展示消息。 + pub message: String, + /// 结构化元数据。 + pub metadata: Value, + /// 创建时间。 + #[serde(serialize_with = "crate::infra::time::serialize_china_rfc3339")] + pub created_at: DateTime, +} + +impl OpsErrorLog { + /// 构造运维错误事件。 + pub fn new(kind: impl Into, message: impl Into) -> Self { + Self { + id: format!("ops_error_{}", Uuid::new_v4().simple()), + request_id: None, + kind: kind.into(), + account_id: None, + route: None, + model: None, + status_code: None, + client_status_code: None, + upstream_status_code: None, + transport: None, + attempt_index: None, + failure_class: None, + response_id: None, + upstream_request_id: None, + latency_ms: None, + message: message.into(), + metadata: Value::Object(Map::default()), + created_at: Utc::now(), + } + } +} diff --git a/backend/src/admin/monitoring/ops_error_service.rs b/backend/src/admin/monitoring/ops_error_service.rs new file mode 100644 index 000000000..c005bc8e5 --- /dev/null +++ b/backend/src/admin/monitoring/ops_error_service.rs @@ -0,0 +1,68 @@ +//! 管理端运维错误事件服务。 + +use chrono::Utc; +use thiserror::Error; + +use crate::admin::monitoring::{ + ops_error_model::OpsErrorLog, ops_error_store::SqliteOpsErrorLogStore, +}; + +/// 管理端运维错误事件错误。 +#[derive(Debug, Error)] +pub enum AdminOpsErrorLogError { + /// 写入失败。 + #[error("failed to append ops error log")] + Append, + /// 保留期清理失败。 + #[error("failed to trim expired ops error logs")] + Retention, +} + +/// 管理端运维错误事件服务。 +#[derive(Clone)] +pub struct AdminOpsErrorLogService { + store: SqliteOpsErrorLogStore, + capture_body: bool, +} + +impl AdminOpsErrorLogService { + /// 构造管理端运维错误事件服务。 + pub fn new(store: SqliteOpsErrorLogStore, capture_body: bool) -> Self { + Self { + store, + capture_body, + } + } + + /// 记录运维错误事件。 + pub async fn record(&self, mut event: OpsErrorLog) -> Result<(), AdminOpsErrorLogError> { + apply_capture_body_policy(&mut event, self.capture_body); + self.store + .append(&event) + .await + .map_err(|_| AdminOpsErrorLogError::Append)?; + self.store + .trim_to_retention(Utc::now()) + .await + .map_err(|_| AdminOpsErrorLogError::Retention)?; + Ok(()) + } +} + +fn apply_capture_body_policy(event: &mut OpsErrorLog, capture_body: bool) { + if capture_body { + return; + } + let Some(metadata) = event.metadata.as_object_mut() else { + return; + }; + for key in [ + "body", + "rawBody", + "requestBody", + "responseBody", + "upstreamBody", + ] { + metadata.remove(key); + } +} diff --git a/backend/src/admin/monitoring/ops_error_store.rs b/backend/src/admin/monitoring/ops_error_store.rs new file mode 100644 index 000000000..6392e7594 --- /dev/null +++ b/backend/src/admin/monitoring/ops_error_store.rs @@ -0,0 +1,261 @@ +//! 运维错误事件存储实现(SQLite)。 + +use chrono::{DateTime, Duration, Utc}; +use serde_json::Value; +use sqlx::SqlitePool; +use thiserror::Error; + +use crate::{ + admin::monitoring::{ + ops_error_model::OpsErrorLog, + usage_record_model::{metadata_i64, metadata_service_tier, metadata_string}, + usage_record_store::USAGE_RECORD_RETENTION_DAYS, + }, + infra::time::china_quarter_hour_start, +}; + +/// SQLite 运维错误事件错误。 +#[derive(Debug, Error)] +pub enum SqliteOpsErrorLogStoreError { + /// 数据库错误。 + #[error("sqlite ops error log database error: {0}")] + Database(#[from] sqlx::Error), +} + +/// SQLite 运维错误事件结果。 +pub type SqliteOpsErrorLogStoreResult = Result; + +/// SQLite 运维错误事件存储。 +#[derive(Clone)] +pub struct SqliteOpsErrorLogStore { + pool: SqlitePool, +} + +impl SqliteOpsErrorLogStore { + /// 构造存储。 + pub fn new(pool: SqlitePool) -> Self { + Self { pool } + } + + /// 写入运维错误事件。 + pub async fn append(&self, event: &OpsErrorLog) -> SqliteOpsErrorLogStoreResult<()> { + append_event(&self.pool, event).await + } + + /// 按保留期清理运维错误事件。 + pub async fn trim_to_retention(&self, now: DateTime) -> SqliteOpsErrorLogStoreResult { + trim_to_retention(&self.pool, now).await + } + + /// 清空运维错误事件。 + pub async fn clear(&self) -> SqliteOpsErrorLogStoreResult { + let result = sqlx::query("delete from ops_error_logs") + .execute(&self.pool) + .await?; + Ok(result.rows_affected()) + } + + /// 读取错误事件数量。 + pub async fn count(&self) -> SqliteOpsErrorLogStoreResult { + let (total,): (i64,) = sqlx::query_as("select count(*) from ops_error_logs") + .fetch_one(&self.pool) + .await?; + Ok(total.max(0) as u64) + } +} + +#[derive(Debug, Clone)] +struct OpsErrorSummaryFields { + client_status_code: Option, + upstream_status_code: Option, + transport: Option, + attempt_index: Option, + failure_class: Option, + response_id: Option, + upstream_request_id: Option, +} + +fn ops_error_summary_fields(event: &OpsErrorLog) -> OpsErrorSummaryFields { + OpsErrorSummaryFields { + client_status_code: event.client_status_code.or_else(|| { + metadata_i64( + &event.metadata, + &["clientStatusCode", "client_status_code", "clientStatus"], + ) + }), + upstream_status_code: event.upstream_status_code.or_else(|| { + metadata_i64( + &event.metadata, + &[ + "upstreamStatus", + "upstreamStatusCode", + "upstream_status_code", + ], + ) + }), + transport: event + .transport + .clone() + .or_else(|| metadata_string(&event.metadata, &["transport"])), + attempt_index: event + .attempt_index + .or_else(|| metadata_i64(&event.metadata, &["attemptIndex", "attempt_index"])), + failure_class: event + .failure_class + .clone() + .or_else(|| metadata_string(&event.metadata, &["failureClass", "failure_class"])), + response_id: event + .response_id + .clone() + .or_else(|| metadata_string(&event.metadata, &["responseId", "response_id"])), + upstream_request_id: event.upstream_request_id.clone().or_else(|| { + metadata_string( + &event.metadata, + &[ + "upstreamRequestId", + "upstream_request_id", + "openaiRequestId", + ], + ) + }), + } +} + +async fn append_event(pool: &SqlitePool, event: &OpsErrorLog) -> SqliteOpsErrorLogStoreResult<()> { + let summary = ops_error_summary_fields(event); + sqlx::query( + "insert into ops_error_logs (id, request_id, kind, account_id, route, model, status_code, client_status_code, upstream_status_code, transport, attempt_index, failure_class, response_id, upstream_request_id, latency_ms, message, metadata_json, created_at) values (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", + ) + .bind(&event.id) + .bind(&event.request_id) + .bind(&event.kind) + .bind(&event.account_id) + .bind(&event.route) + .bind(&event.model) + .bind(event.status_code) + .bind(summary.client_status_code) + .bind(summary.upstream_status_code) + .bind(summary.transport) + .bind(summary.attempt_index) + .bind(summary.failure_class) + .bind(summary.response_id) + .bind(summary.upstream_request_id) + .bind(event.latency_ms) + .bind(&event.message) + .bind(event.metadata.to_string()) + .bind(event.created_at.to_rfc3339()) + .execute(pool) + .await?; + append_error_time_bucket(pool, event).await?; + Ok(()) +} + +async fn append_error_time_bucket( + pool: &SqlitePool, + event: &OpsErrorLog, +) -> SqliteOpsErrorLogStoreResult<()> { + let bucket_start = china_quarter_hour_start(event.created_at); + let input_tokens = metadata_usage_i64(&event.metadata, "inputTokens"); + let output_tokens = metadata_usage_i64(&event.metadata, "outputTokens"); + let cached_tokens = metadata_usage_i64(&event.metadata, "cachedTokens"); + let first_token_latency = metadata_first_token_latency(&event.metadata).unwrap_or(0); + let first_token_latency_count = i64::from(first_token_latency > 0); + let latency = event.latency_ms.filter(|value| *value > 0).unwrap_or(0); + let latency_count = i64::from(latency > 0); + let now = Utc::now().to_rfc3339(); + + sqlx::query( + r" +insert into usage_time_buckets ( + bucket_start, + account_id, + model, + service_tier, + request_count, + error_count, + input_tokens, + output_tokens, + cached_tokens, + first_token_latency_sum, + first_token_latency_count, + latency_sum, + latency_count, + max_latency_ms, + min_latency_ms, + updated_at +) values (?, ?, ?, ?, 1, 1, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) +on conflict(bucket_start, account_id, model, service_tier) do update set + request_count = request_count + excluded.request_count, + error_count = error_count + excluded.error_count, + input_tokens = input_tokens + excluded.input_tokens, + output_tokens = output_tokens + excluded.output_tokens, + cached_tokens = cached_tokens + excluded.cached_tokens, + first_token_latency_sum = first_token_latency_sum + excluded.first_token_latency_sum, + first_token_latency_count = first_token_latency_count + excluded.first_token_latency_count, + latency_sum = latency_sum + excluded.latency_sum, + latency_count = latency_count + excluded.latency_count, + max_latency_ms = max(max_latency_ms, excluded.max_latency_ms), + min_latency_ms = case + when min_latency_ms = 0 then excluded.min_latency_ms + when excluded.min_latency_ms = 0 then min_latency_ms + else min(min_latency_ms, excluded.min_latency_ms) + end, + updated_at = excluded.updated_at", + ) + .bind(bucket_start.to_rfc3339()) + .bind(event.account_id.as_deref().unwrap_or_default()) + .bind(event.model.as_deref().unwrap_or_default()) + .bind(metadata_service_tier(&event.metadata).unwrap_or_default()) + .bind(input_tokens) + .bind(output_tokens) + .bind(cached_tokens) + .bind(first_token_latency) + .bind(first_token_latency_count) + .bind(latency) + .bind(latency_count) + .bind(latency) + .bind(latency) + .bind(now) + .execute(pool) + .await?; + Ok(()) +} + +fn metadata_usage_i64(metadata: &Value, field: &str) -> i64 { + metadata + .get("usage") + .and_then(|usage| usage.get(field)) + .or_else(|| metadata.get(field)) + .and_then(Value::as_i64) + .filter(|value| *value > 0) + .unwrap_or(0) +} + +fn metadata_first_token_latency(metadata: &Value) -> Option { + [ + "firstTokenMs", + "first_token_ms", + "firstTokenLatencyMs", + "first_token_latency_ms", + ] + .into_iter() + .find_map(|field| { + metadata + .get(field) + .or_else(|| metadata.get("usage").and_then(|usage| usage.get(field))) + .and_then(Value::as_i64) + .filter(|value| *value > 0) + }) +} + +async fn trim_to_retention( + pool: &SqlitePool, + now: DateTime, +) -> SqliteOpsErrorLogStoreResult { + let cutoff = now - Duration::days(USAGE_RECORD_RETENTION_DAYS); + let result = sqlx::query("delete from ops_error_logs where created_at < ?") + .bind(cutoff.to_rfc3339()) + .execute(pool) + .await?; + Ok(result.rows_affected()) +} diff --git a/backend/src/admin/monitoring/usage_record_service.rs b/backend/src/admin/monitoring/usage_record_service.rs index 29afffeb2..cd13a3a86 100644 --- a/backend/src/admin/monitoring/usage_record_service.rs +++ b/backend/src/admin/monitoring/usage_record_service.rs @@ -265,7 +265,19 @@ impl AdminUsageRecordService { /// 记录使用记录。 pub async fn record(&self, event: UsageRecord) -> Result<(), AdminUsageRecordError> { let settings = *self.settings.read().await; - if !settings.enabled && event.level != UsageRecordLevel::Error { + if !settings.enabled { + return Ok(()); + } + if !is_usage_log_event(&event) { + tracing::warn!( + usage_record_id = %event.id, + request_id = event.request_id.as_deref().unwrap_or(""), + account_id = event.account_id.as_deref().unwrap_or(""), + model = event.model.as_deref().unwrap_or(""), + level = ?event.level, + status_code = ?event.status_code, + "skipped non-usage event in usage record service" + ); return Ok(()); } self.append_with_settings(event, settings).await @@ -289,6 +301,21 @@ impl AdminUsageRecordService { } } +fn is_usage_log_event(event: &UsageRecord) -> bool { + event.level != UsageRecordLevel::Error + && event.status_code.is_none_or(|status| status < 400) + && event + .account_id + .as_deref() + .map(str::trim) + .is_some_and(|account_id| !account_id.is_empty()) + && event + .model + .as_deref() + .map(str::trim) + .is_some_and(|model| !model.is_empty()) +} + fn apply_capture_body_policy(event: &mut UsageRecord, capture_body: bool) { if capture_body { return; diff --git a/backend/src/admin/monitoring/usage_record_store.rs b/backend/src/admin/monitoring/usage_record_store.rs index 3eeda327a..3edba85b8 100644 --- a/backend/src/admin/monitoring/usage_record_store.rs +++ b/backend/src/admin/monitoring/usage_record_store.rs @@ -100,6 +100,11 @@ impl SqliteUsageRecordStore { Self { pool } } + /// 暴露底层连接池,供运行时组合层复用。 + pub fn pool(&self) -> &SqlitePool { + &self.pool + } + /// 写入使用记录。 pub async fn append(&self, event: &UsageRecord) -> SqliteUsageRecordStoreResult<()> { append_event(&self.pool, event).await diff --git a/backend/src/infra/database.rs b/backend/src/infra/database.rs index 7e332d3b4..4d9691f03 100644 --- a/backend/src/infra/database.rs +++ b/backend/src/infra/database.rs @@ -4,7 +4,7 @@ use std::{fs, path::Path, str::FromStr, time::Duration}; use sqlx::{sqlite::SqliteConnectOptions, SqlitePool}; -const CURRENT_SCHEMA_VERSION: i64 = 2; +const CURRENT_SCHEMA_VERSION: i64 = 3; const MIGRATIONS: &[Migration] = &[ Migration { version: 1, @@ -16,6 +16,11 @@ const MIGRATIONS: &[Migration] = &[ name: "rename_rotation_strategies", sql: include_str!("migrations/0002_rename_rotation_strategies.sql"), }, + Migration { + version: 3, + name: "ops_error_logs", + sql: include_str!("migrations/0003_ops_error_logs.sql"), + }, ]; struct Migration { diff --git a/backend/src/infra/migrations/0003_ops_error_logs.sql b/backend/src/infra/migrations/0003_ops_error_logs.sql new file mode 100644 index 000000000..6e6ee8995 --- /dev/null +++ b/backend/src/infra/migrations/0003_ops_error_logs.sql @@ -0,0 +1,31 @@ +create table ops_error_logs ( + id text primary key, + request_id text, + kind text not null, + account_id text, + route text, + model text, + status_code integer check (status_code is null or (status_code >= 100 and status_code <= 599)), + client_status_code integer check (client_status_code is null or (client_status_code >= 100 and client_status_code <= 599)), + upstream_status_code integer check (upstream_status_code is null or (upstream_status_code >= 100 and upstream_status_code <= 599)), + transport text, + attempt_index integer check (attempt_index is null or attempt_index >= 0), + failure_class text, + response_id text, + upstream_request_id text, + latency_ms integer check (latency_ms is null or latency_ms >= 0), + message text not null, + metadata_json text not null, + created_at text not null +); + +create index idx_ops_error_logs_created_id on ops_error_logs(created_at desc, id desc); +create index idx_ops_error_logs_request_id on ops_error_logs(request_id) where request_id is not null; +create index idx_ops_error_logs_account on ops_error_logs(account_id, created_at desc) where account_id is not null; +create index idx_ops_error_logs_route_created on ops_error_logs(route, created_at desc) where route is not null; +create index idx_ops_error_logs_model_created on ops_error_logs(model, created_at desc) where model is not null; +create index idx_ops_error_logs_status_created on ops_error_logs(status_code, created_at desc) where status_code is not null; +create index idx_ops_error_logs_transport_created on ops_error_logs(transport, created_at desc) where transport is not null; +create index idx_ops_error_logs_failure_class on ops_error_logs(failure_class, created_at desc) where failure_class is not null; +create index idx_ops_error_logs_response_id on ops_error_logs(response_id) where response_id is not null; +create index idx_ops_error_logs_upstream_request_id on ops_error_logs(upstream_request_id) where upstream_request_id is not null; diff --git a/backend/src/proxy/dispatch/responses/event_recording.rs b/backend/src/proxy/dispatch/responses/event_recording.rs index dd16d0c9c..428cc9e2a 100644 --- a/backend/src/proxy/dispatch/responses/event_recording.rs +++ b/backend/src/proxy/dispatch/responses/event_recording.rs @@ -4,8 +4,9 @@ use serde_json::{json, Map, Value}; use crate::{ admin::monitoring::{ - usage_record_model::{ResponseUsageRecord, UsageRecord, UsageRecordLevel}, - usage_record_service::AdminUsageRecordService, + ops_error_model::OpsErrorLog, + ops_error_service::AdminOpsErrorLogService, + usage_record_model::{UsageRecord, UsageRecordLevel}, }, infra::time::elapsed_millis_i64, proxy::dispatch::{ @@ -17,7 +18,7 @@ use crate::{ usage_events::{ enrich_event_route_metadata, enrich_usage_record_identity, event_kind as response_event_kind, reasoning_effort_from_request, - record_dispatch_error_event, record_response_event, DispatchErrorUsageRecord, + record_dispatch_error_event, DispatchErrorLogRecord, }, }, upstream::{ @@ -39,7 +40,7 @@ use super::{ }; pub(super) struct ResponseUpstreamErrorEventRecord<'a> { - pub(super) usage_records: &'a AdminUsageRecordService, + pub(super) ops_errors: &'a AdminOpsErrorLogService, pub(super) request_id: &'a str, pub(super) account_id: &'a str, pub(super) account_email: Option<&'a str>, @@ -55,7 +56,7 @@ pub(super) struct ResponseUpstreamErrorEventRecord<'a> { } pub(super) struct ResponseStreamFailureEventRecord<'a> { - pub(super) usage_records: &'a AdminUsageRecordService, + pub(super) ops_errors: &'a AdminOpsErrorLogService, pub(super) request_id: &'a str, pub(super) account_id: &'a str, pub(super) route: &'a str, @@ -73,7 +74,7 @@ pub(super) struct ResponseStreamFailureEventRecord<'a> { } pub(super) struct ResponseDispatchErrorEventRecord<'a> { - pub(super) usage_records: &'a AdminUsageRecordService, + pub(super) ops_errors: &'a AdminOpsErrorLogService, pub(super) request_id: &'a str, pub(super) account_id: Option<&'a str>, pub(super) route: &'a str, @@ -110,8 +111,8 @@ pub(super) async fn record_response_dispatch_error_event( dispatch_error_upstream_status_code(record.error), ); } - record_dispatch_error_event(DispatchErrorUsageRecord { - usage_records: record.usage_records, + record_dispatch_error_event(DispatchErrorLogRecord { + ops_errors: record.ops_errors, request_id: record.request_id, account_id: record.account_id, route: record.route, @@ -135,11 +136,7 @@ pub(super) async fn record_response_upstream_error_event( Some(backend_transport_name(record.transport)), ); enrich_event_route_metadata(&mut metadata, record.route); - let mut event = UsageRecord::new( - "v1.response", - UsageRecordLevel::Error, - "v1 responses upstream request failed", - ); + let mut event = OpsErrorLog::new("v1.response", "v1 responses upstream request failed"); event.request_id = Some(record.request_id.to_string()); event.account_id = Some(record.account_id.to_string()); event.route = Some(record.route.to_string()); @@ -167,8 +164,17 @@ pub(super) async fn record_response_upstream_error_event( ); } } + enrich_usage_record_identity( + &mut metadata, + Some(record.request.model()), + record.model, + record.request.client_ip.as_deref(), + record.request.client_user_agent.as_deref(), + reasoning_effort_from_request(record.request), + record.request.service_tier(), + ); event.metadata = metadata; - if let Err(error) = record.usage_records.record(event).await { + if let Err(error) = record.ops_errors.record(event).await { tracing::warn!(account_id = %record.account_id, error = %error, "failed to record upstream error event"); } } @@ -204,25 +210,38 @@ pub(super) async fn record_prefetched_response_stream_failure_event( Some(record.attempt), ); } - record_response_event(ResponseUsageRecord { - usage_records: record.usage_records, - request_id: record.request_id, - account_id: record.account_id, - route: record.route, - model: record.model, - requested_model: Some(record.requested_model), - client_ip: record.request.client_ip.as_deref(), - client_user_agent: record.request.client_user_agent.as_deref(), - reasoning_effort: reasoning_effort_from_request(record.request), - service_tier: record.request.service_tier(), - started_at: record.started_at, - status_code: event_status_code, - level: UsageRecordLevel::Error, - message: "v1 responses stream failed", - metadata, - rate_limit_headers: record.rate_limit_headers, - }) - .await; + enrich_event_route_metadata(&mut metadata, record.route); + enrich_usage_record_identity( + &mut metadata, + Some(record.requested_model), + record.model, + record.request.client_ip.as_deref(), + record.request.client_user_agent.as_deref(), + reasoning_effort_from_request(record.request), + record.request.service_tier(), + ); + let mut event = OpsErrorLog::new( + response_event_kind(record.route), + "v1 responses stream failed", + ); + event.request_id = Some(record.request_id.to_string()); + event.account_id = Some(record.account_id.to_string()); + event.route = Some(record.route.to_string()); + event.model = Some(record.model.to_string()); + event.status_code = Some(event_status_code); + event.latency_ms = Some(elapsed_millis_i64(record.started_at)); + if !record.rate_limit_headers.is_empty() { + if let Some(object) = metadata.as_object_mut() { + object.insert( + "rateLimitHeaders".to_string(), + serde_json::json!(record.rate_limit_headers), + ); + } + } + event.metadata = metadata; + if let Err(error) = record.ops_errors.record(event).await { + tracing::warn!(account_id = %record.account_id, error = %error, "failed to record prefetched stream error event"); + } } fn ensure_stream_metadata_flag(metadata: &mut Value) { @@ -469,13 +488,6 @@ pub(super) async fn record_live_response_stream_event( body, ); log_live_response_stream_finalized(context, status_code, level, message, &metadata, body); - let mut event = UsageRecord::new(response_event_kind(&context.route), level, message); - 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.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), @@ -485,12 +497,38 @@ pub(super) async fn record_live_response_stream_event( reasoning_effort_from_request(&context.request), context.request.service_tier(), ); - event.metadata = metadata; - if let Err(error) = context.usage_records.record(event).await { - tracing::warn!(account_id = %context.account_id, error = %error, "failed to record live response stream event"); + + if is_success_usage_event(level, status_code) { + let mut event = UsageRecord::new(response_event_kind(&context.route), level, message); + 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.display_model.clone()); + event.status_code = Some(status_code); + event.latency_ms = Some(elapsed_millis_i64(context.started_at)); + event.metadata = metadata; + if let Err(error) = context.usage_records.record(event).await { + tracing::warn!(account_id = %context.account_id, error = %error, "failed to record live response stream event"); + } + } else { + let mut event = OpsErrorLog::new(response_event_kind(&context.route), message); + 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.display_model.clone()); + event.status_code = Some(status_code); + event.latency_ms = Some(elapsed_millis_i64(context.started_at)); + event.metadata = metadata; + if let Err(error) = context.ops_errors.record(event).await { + tracing::warn!(account_id = %context.account_id, error = %error, "failed to record live response stream error event"); + } } } +fn is_success_usage_event(level: UsageRecordLevel, status_code: i64) -> bool { + level != UsageRecordLevel::Error && status_code < 400 +} + fn log_live_response_stream_finalized( context: &LiveResponseStreamContext, status_code: i64, diff --git a/backend/src/proxy/dispatch/responses/service.rs b/backend/src/proxy/dispatch/responses/service.rs index ad0100bef..8cb6826a6 100644 --- a/backend/src/proxy/dispatch/responses/service.rs +++ b/backend/src/proxy/dispatch/responses/service.rs @@ -17,6 +17,7 @@ use tokio::sync::Mutex; use crate::{ admin::monitoring::{ + ops_error_service::AdminOpsErrorLogService, usage_record_model::{ResponseUsageRecord, UsageRecordLevel}, usage_record_service::AdminUsageRecordService, }, @@ -110,6 +111,7 @@ pub struct ResponseDispatchService { session_affinity: Arc, reasoning_replay: Arc>, usage_records: Arc, + ops_errors: Arc, token_refresh: Arc>, installation_id: Option, cloudflare: CloudflareRecovery, @@ -121,6 +123,7 @@ pub(crate) struct ResponseDispatchServiceParts { pub codex: Arc, pub session_affinity: Arc, pub usage_records: Arc, + pub ops_errors: Arc, pub token_refresh: Arc>, pub installation_id: Option, pub cloudflare: CloudflareRecovery, @@ -178,6 +181,7 @@ impl ResponseDispatchService { DEFAULT_REASONING_REPLAY_TTL_SECS, )))), usage_records: parts.usage_records, + ops_errors: parts.ops_errors, token_refresh: parts.token_refresh, installation_id: parts.installation_id, cloudflare: parts.cloudflare, @@ -834,7 +838,7 @@ impl ResponseDispatchService { } Err(error) => { record_response_upstream_error_event(ResponseUpstreamErrorEventRecord { - usage_records: &self.usage_records, + ops_errors: &self.ops_errors, request_id, account_id: &release_account_id, account_email: account.email.as_deref(), @@ -980,7 +984,7 @@ impl ResponseDispatchService { error: &ResponseDispatchError, ) { record_response_dispatch_error_event(ResponseDispatchErrorEventRecord { - usage_records: &self.usage_records, + ops_errors: &self.ops_errors, request_id, account_id: details.account_id, route, @@ -1318,7 +1322,7 @@ impl ResponseDispatchService { if let ResponseDispatchError::Upstream(upstream_error) = &error { record_response_upstream_error_event( ResponseUpstreamErrorEventRecord { - usage_records: &self.usage_records, + ops_errors: &self.ops_errors, request_id, account_id: &release_account_id, account_email: account.email.as_deref(), @@ -1423,7 +1427,7 @@ impl ResponseDispatchService { self.account_pool.release(&release_account_id).await; record_prefetched_response_stream_failure_event( ResponseStreamFailureEventRecord { - usage_records: &self.usage_records, + ops_errors: &self.ops_errors, request_id, account_id: &release_account_id, route, @@ -1449,6 +1453,7 @@ impl ResponseDispatchService { session_affinity: Arc::clone(&self.session_affinity), reasoning_replay: Arc::clone(&self.reasoning_replay), usage_records: Arc::clone(&self.usage_records), + ops_errors: Arc::clone(&self.ops_errors), cloudflare: self.cloudflare.clone(), account_id: account.id, account_plan_type: account.plan_type, @@ -1585,7 +1590,7 @@ impl ResponseDispatchService { Err(error) => { self.account_pool.release(&release_account_id).await; record_response_upstream_error_event(ResponseUpstreamErrorEventRecord { - usage_records: &self.usage_records, + ops_errors: &self.ops_errors, request_id, account_id: &release_account_id, account_email: account.email.as_deref(), @@ -1877,7 +1882,7 @@ impl ResponseDispatchService { error: &ResponseDispatchError, ) { record_response_dispatch_error_event(ResponseDispatchErrorEventRecord { - usage_records: &self.usage_records, + ops_errors: &self.ops_errors, request_id, account_id, route: "/v1/responses/compact", diff --git a/backend/src/proxy/dispatch/responses/stream_lifecycle.rs b/backend/src/proxy/dispatch/responses/stream_lifecycle.rs index b7c55e350..f5d308c63 100644 --- a/backend/src/proxy/dispatch/responses/stream_lifecycle.rs +++ b/backend/src/proxy/dispatch/responses/stream_lifecycle.rs @@ -5,7 +5,8 @@ use tokio::sync::Mutex; use crate::{ admin::monitoring::{ - usage_record_model::UsageRecordLevel, usage_record_service::AdminUsageRecordService, + ops_error_service::AdminOpsErrorLogService, usage_record_model::UsageRecordLevel, + usage_record_service::AdminUsageRecordService, }, infra::time::elapsed_millis_i64, proxy::{ @@ -49,6 +50,7 @@ pub(super) struct LiveResponseStreamContext { pub(super) session_affinity: Arc, pub(super) reasoning_replay: Arc>, pub(super) usage_records: Arc, + pub(super) ops_errors: Arc, pub(super) cloudflare: CloudflareRecovery, pub(super) account_id: String, pub(super) account_plan_type: Option, diff --git a/backend/src/proxy/dispatch/upstream.rs b/backend/src/proxy/dispatch/upstream.rs index 229fd7295..7160a3b9a 100644 --- a/backend/src/proxy/dispatch/upstream.rs +++ b/backend/src/proxy/dispatch/upstream.rs @@ -103,7 +103,10 @@ pub(crate) async fn verify_acquired_quota_if_required( .apply_quota_snapshot(&account_id, "a) .await; if quota_snapshot_limit_reached("a) { - context.account_pool.release(&account_id).await; + context + .account_pool + .release_without_request_usage(&account_id) + .await; context.excluded_account_ids.push(account_id.clone()); *context.verify_attempts += 1; let max_attempts_reached = *context.verify_attempts >= MAX_QUOTA_VERIFY_ATTEMPTS; diff --git a/backend/src/proxy/dispatch/usage_events.rs b/backend/src/proxy/dispatch/usage_events.rs index bfb58213f..f1ec60835 100644 --- a/backend/src/proxy/dispatch/usage_events.rs +++ b/backend/src/proxy/dispatch/usage_events.rs @@ -6,15 +6,16 @@ use serde_json::Value; use crate::{ admin::monitoring::{ - usage_record_model::{ResponseUsageRecord, UsageRecord, UsageRecordLevel}, - usage_record_service::AdminUsageRecordService, + ops_error_model::OpsErrorLog, + ops_error_service::AdminOpsErrorLogService, + usage_record_model::{ResponseUsageRecord, UsageRecord}, }, infra::time::elapsed_millis_i64, upstream::protocol::responses::{CodexCompactRequest, CodexResponsesRequest}, }; -pub(super) struct DispatchErrorUsageRecord<'a> { - pub usage_records: &'a AdminUsageRecordService, +pub(super) struct DispatchErrorLogRecord<'a> { + pub ops_errors: &'a AdminOpsErrorLogService, pub request_id: &'a str, pub account_id: Option<&'a str>, pub route: &'a str, @@ -25,15 +26,11 @@ pub(super) struct DispatchErrorUsageRecord<'a> { pub metadata: Value, } -pub(super) async fn record_dispatch_error_event(record: DispatchErrorUsageRecord<'_>) { +pub(super) async fn record_dispatch_error_event(record: DispatchErrorLogRecord<'_>) { let mut metadata = record.metadata; enrich_event_route_metadata(&mut metadata, record.route); - let mut event = UsageRecord::new( - event_kind(record.route), - UsageRecordLevel::Error, - record.message, - ); + let mut event = OpsErrorLog::new(event_kind(record.route), record.message); event.request_id = Some(record.request_id.to_string()); event.account_id = record.account_id.map(ToString::to_string); event.route = Some(record.route.to_string()); @@ -42,7 +39,7 @@ pub(super) async fn record_dispatch_error_event(record: DispatchErrorUsageRecord event.latency_ms = Some(elapsed_millis_i64(record.started_at)); event.metadata = metadata; - if let Err(error) = record.usage_records.record(event).await { + if let Err(error) = record.ops_errors.record(event).await { tracing::warn!( account_id = record.account_id.unwrap_or(""), error = %error, diff --git a/backend/src/runtime/services.rs b/backend/src/runtime/services.rs index d771705f8..6bed5c5d4 100644 --- a/backend/src/runtime/services.rs +++ b/backend/src/runtime/services.rs @@ -6,6 +6,8 @@ use crate::{ admin::monitoring::{ account_usage_service::AdminUsageService, account_usage_store::SqliteUsageStore, + ops_error_service::AdminOpsErrorLogService, + ops_error_store::SqliteOpsErrorLogStore, usage_record_service::AdminUsageRecordService, usage_record_store::{SqliteUsageRecordStore, DEFAULT_USAGE_RECORD_CAPTURE_BODY}, }, @@ -91,6 +93,7 @@ pub struct Services { pub refresh_policy: RuntimeRefreshPolicy, pub admin_accounts: StdArc, pub usage_records: StdArc, + pub ops_errors: StdArc, pub usage: StdArc, pub account_pool: StdArc, pub token_refresh: StdArc>, @@ -193,6 +196,10 @@ impl Services { usage_record_options.enabled, usage_record_options.capture_body, )); + let ops_errors = StdArc::new(AdminOpsErrorLogService::new( + SqliteOpsErrorLogStore::new(stores.usage_records.pool().clone()), + usage_record_options.capture_body, + )); let account_pool = StdArc::new(RuntimeAccountPoolService::new( account_store_trait.clone(), account_pool_options_from_config(config), @@ -269,6 +276,7 @@ impl Services { codex: codex.clone(), session_affinity: session_affinity.clone(), usage_records: usage_records.clone(), + ops_errors: ops_errors.clone(), token_refresh: token_refresh.clone(), installation_id: installation_id.clone(), cloudflare: cloudflare_recovery, @@ -284,6 +292,7 @@ impl Services { refresh_policy, admin_accounts, usage_records, + ops_errors, usage, account_pool, token_refresh, diff --git a/backend/src/runtime/tasks/model_refresh.rs b/backend/src/runtime/tasks/model_refresh.rs index 392451a73..c97aaf429 100644 --- a/backend/src/runtime/tasks/model_refresh.rs +++ b/backend/src/runtime/tasks/model_refresh.rs @@ -88,7 +88,9 @@ impl ModelRefreshTask { ) .await; for selected in &selected_accounts { - self.account_pool.release(&selected.account.id).await; + self.account_pool + .release_without_request_usage(&selected.account.id) + .await; } let result = result?; diff --git a/backend/src/upstream/accounts/pool.rs b/backend/src/upstream/accounts/pool.rs index d03eed75e..50dc49b52 100644 --- a/backend/src/upstream/accounts/pool.rs +++ b/backend/src/upstream/accounts/pool.rs @@ -574,6 +574,22 @@ impl AccountPool { /// 释放指定账号的一个在途槽位。 pub fn release(&mut self, account_id: &str) -> Option { + self.release_slot(account_id, true) + } + + /// 释放指定账号的一个在途槽位,不累计请求用量。 + pub fn release_without_request_usage( + &mut self, + account_id: &str, + ) -> Option { + self.release_slot(account_id, false) + } + + fn release_slot( + &mut self, + account_id: &str, + record_request_usage: bool, + ) -> Option { self.accounts.get(account_id)?; let mut remove_slots = false; @@ -585,7 +601,9 @@ impl AccountPool { if remove_slots { self.slots.remove(account_id); } - self.mark_request_usage(account_id, Utc::now()); + if record_request_usage { + self.mark_request_usage(account_id, Utc::now()); + } Some(ReleasedAccountUsage { model: slot.and_then(|slot| slot.model), }) @@ -900,10 +918,29 @@ impl RuntimeAccountPoolService { /// 释放账号在途槽位。 pub async fn release(&self, account_id: &str) { - let released = self.pool.lock().await.release(account_id); + self.release_slot(account_id, true).await; + } + + /// 释放账号在途槽位,不累计请求用量。 + pub async fn release_without_request_usage(&self, account_id: &str) { + self.release_slot(account_id, false).await; + } + + async fn release_slot(&self, account_id: &str, record_request_usage: bool) { + let released = if record_request_usage { + self.pool.lock().await.release(account_id) + } else { + self.pool + .lock() + .await + .release_without_request_usage(account_id) + }; let Some(released) = released else { return; }; + if !record_request_usage { + return; + } if let Err(error) = self.store.record_request(account_id).await { tracing::warn!( account_id, diff --git a/backend/tests/admin/monitoring/dashboard.rs b/backend/tests/admin/monitoring/dashboard.rs index 606183b84..a0345e826 100644 --- a/backend/tests/admin/monitoring/dashboard.rs +++ b/backend/tests/admin/monitoring/dashboard.rs @@ -70,7 +70,7 @@ async fn dashboard_summary_should_render_dash_when_fingerprint_updated_at_is_mis #[tokio::test] async fn dashboard_summary_should_calculate_today_traffic_and_average_first_token_latency_from_time_buckets( ) { - let (app, store, pool, _dir) = dashboard_test_app( + let (app, store, _pool, _dir) = dashboard_test_app( "dashboard-traffic-latency.sqlite", crate::support::fingerprint::test_fingerprint(), ) @@ -96,23 +96,22 @@ async fn dashboard_summary_should_calculate_today_traffic_and_average_first_toke .append(&usage_record_with_tokens(yesterday_record_at, 30)) .await .unwrap(); - seed_usage_summary_with_cached_tokens(&pool, 120, 30, 24).await; let body = dashboard_summary(app).await; assert_eq!(body["data"]["cards"]["traffic"]["todayRequests"], "2"); assert_eq!(body["data"]["cards"]["traffic"]["todayRequestsValue"], 2); - assert_eq!(body["data"]["cards"]["traffic"]["totalRequests"], "42"); + assert_eq!(body["data"]["cards"]["traffic"]["totalRequests"], "3"); assert_eq!( body["data"]["cards"]["traffic"]["yesterdayRequestsValue"], 1 ); assert_eq!(body["data"]["cards"]["tokens"]["todayTokens"], "30"); assert_eq!(body["data"]["cards"]["tokens"]["todayTokensValue"], 30); - assert_eq!(body["data"]["cards"]["tokens"]["totalTokens"], "150"); + assert_eq!(body["data"]["cards"]["tokens"]["totalTokens"], "60"); assert_eq!(body["data"]["cards"]["tokens"]["yesterdayTokensValue"], 30); - assert_eq!(body["data"]["cards"]["cache"]["totalHitRate"], "20.0%"); - assert_eq!(body["data"]["cards"]["cache"]["totalCachedTokens"], "24"); + assert_eq!(body["data"]["cards"]["cache"]["totalHitRate"], "0.0%"); + assert_eq!(body["data"]["cards"]["cache"]["totalCachedTokens"], "0"); assert_eq!( body["data"]["cards"]["cache"]["averageFirstTokenLatencyMs"], "300 ms" @@ -719,39 +718,6 @@ async fn seed_usage_summary(pool: &SqlitePool, input_tokens: u64, output_tokens: .await; } -async fn seed_usage_summary_with_cached_tokens( - pool: &SqlitePool, - input_tokens: u64, - output_tokens: u64, - cached_tokens: u64, -) { - sqlx::query( - "insert into accounts (id, email, access_token, status, added_at, updated_at) values (?, ?, ?, ?, ?, ?)", - ) - .bind("acct_usage_cached") - .bind("acct-usage-cached@example.com") - .bind("cipher") - .bind("active") - .bind("2026-06-18T00:00:00Z") - .bind("2026-06-18T00:00:00Z") - .execute(pool) - .await - .unwrap(); - sqlx::query( - "insert into account_usage (account_id, request_count, input_tokens, output_tokens, cached_tokens, total_tokens, last_used_at) values (?, ?, ?, ?, ?, ?, ?)", - ) - .bind("acct_usage_cached") - .bind(42_i64) - .bind(input_tokens as i64) - .bind(output_tokens as i64) - .bind(cached_tokens as i64) - .bind((input_tokens + output_tokens) as i64) - .bind("2026-06-18T00:00:00Z") - .execute(pool) - .await - .unwrap(); -} - async fn seed_usage_summary_with_last_used( pool: &SqlitePool, input_tokens: u64, @@ -857,6 +823,7 @@ fn usage_record_with_account_tokens( fn usage_record(created_at: chrono::DateTime, model: &str, metadata: Value) -> UsageRecord { let mut record = UsageRecord::new("v1.response", UsageRecordLevel::Info, "completed"); record.model = Some(model.to_string()); + record.status_code = Some(200); record.created_at = created_at; record.metadata = metadata; record diff --git a/backend/tests/admin/monitoring/usage_record_routes.rs b/backend/tests/admin/monitoring/usage_record_routes.rs index 199f76263..add23396d 100644 --- a/backend/tests/admin/monitoring/usage_record_routes.rs +++ b/backend/tests/admin/monitoring/usage_record_routes.rs @@ -54,10 +54,16 @@ async fn admin_usage_records_should_cursor_page_events_and_include_request_id() let now = Utc::now(); let mut older = UsageRecord::new("request", UsageRecordLevel::Info, "older"); older.id = "usage_older".to_string(); + older.account_id = Some("acct_usage".to_string()); + older.model = Some("gpt-5.5".to_string()); + older.status_code = Some(200); older.created_at = now; store.append(&older).await.unwrap(); let mut newer = UsageRecord::new("request", UsageRecordLevel::Info, "newer"); newer.id = "usage_newer".to_string(); + newer.account_id = Some("acct_usage".to_string()); + newer.model = Some("gpt-5.5".to_string()); + newer.status_code = Some(200); newer.created_at = now + Duration::seconds(1); store.append(&newer).await.unwrap(); @@ -88,8 +94,11 @@ async fn admin_usage_records_should_return_numbered_page_metadata() { ("usage_old", "older timeout", 0), ("usage_new", "newer timeout", 1), ] { - let mut event = UsageRecord::new("request", UsageRecordLevel::Error, message); + let mut event = UsageRecord::new("request", UsageRecordLevel::Info, message); event.id = id.to_string(); + event.account_id = Some("acct_usage".to_string()); + event.model = Some("gpt-5.5".to_string()); + event.status_code = Some(200); event.created_at = now + Duration::seconds(offset); store.append(&event).await.unwrap(); } @@ -98,7 +107,7 @@ async fn admin_usage_records_should_return_numbered_page_metadata() { .oneshot( Request::builder() .method("GET") - .uri("/api/admin/usage/records?page=1&pageSize=1&level=error&search=timeout") + .uri("/api/admin/usage/records?page=1&pageSize=1&search=timeout") .header("cookie", "cpr_admin_session=session_1") .header("x-request-id", "req_usage_records_numbered") .body(Body::empty()) @@ -203,12 +212,13 @@ async fn admin_usage_records_should_return_table_display_fields() { 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"); + let mut event = UsageRecord::new("v1.response", UsageRecordLevel::Info, "completed"); event.id = "usage_email_snapshot".to_string(); event.account_id = Some("acct_deleted".to_string()); + event.model = Some("gpt-5.5".to_string()); + event.status_code = Some(200); event.metadata = json!({ - "accountEmail": "deleted@example.com", - "failed": true + "accountEmail": "deleted@example.com" }); store.append(&event).await.unwrap(); @@ -237,7 +247,9 @@ async fn admin_usage_records_summary_should_aggregate_filtered_usage() { admin_usage_records_test_app("admin-usage-records-summary.sqlite").await; let mut success = UsageRecord::new("request", UsageRecordLevel::Info, "summary success"); success.id = "usage_summary_success".to_string(); + success.account_id = Some("acct_summary".to_string()); success.model = Some("gpt-5.5".to_string()); + success.status_code = Some(200); success.latency_ms = Some(1200); success.metadata = json!({ "usage": { @@ -250,6 +262,7 @@ async fn admin_usage_records_summary_should_aggregate_filtered_usage() { let mut failure = UsageRecord::new("request", UsageRecordLevel::Error, "summary failure"); failure.id = "usage_summary_failure".to_string(); + failure.account_id = Some("acct_summary".to_string()); failure.model = Some("gpt-5.5-mini".to_string()); failure.status_code = Some(429); failure.latency_ms = Some(600); @@ -291,8 +304,10 @@ async fn admin_usage_record_insight_cards_should_aggregate_filtered_usage_dimens 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.account_id = Some("acct_insight".to_string()); first.route = Some("/v1/responses".to_string()); first.model = Some("gpt-5.5".to_string()); + first.status_code = Some(200); first.latency_ms = Some(240); first.metadata = json!({ "stream": true, @@ -310,6 +325,7 @@ async fn admin_usage_record_insight_cards_should_aggregate_filtered_usage_dimens let mut second = UsageRecord::new("request", UsageRecordLevel::Error, "insight failure"); second.id = "usage_insight_failure".to_string(); + second.account_id = Some("acct_insight".to_string()); second.route = Some("/v1/responses/review".to_string()); second.model = Some("gpt-5.5-mini".to_string()); second.status_code = Some(500); @@ -451,6 +467,9 @@ async fn admin_usage_record_trends_should_bucket_by_china_calendar_day() { admin_usage_records_test_app_with_pool("admin-usage-records-china-day-trend.sqlite").await; let mut june_29 = UsageRecord::new("request", UsageRecordLevel::Info, "china-day trend"); june_29.id = "usage_china_day_29".to_string(); + june_29.account_id = Some("acct_trend".to_string()); + june_29.model = Some("gpt-5.5".to_string()); + june_29.status_code = Some(200); june_29.created_at = chrono::DateTime::parse_from_rfc3339("2026-06-29T15:30:00Z") .unwrap() .with_timezone(&Utc); @@ -465,6 +484,9 @@ async fn admin_usage_record_trends_should_bucket_by_china_calendar_day() { let mut june_30 = UsageRecord::new("request", UsageRecordLevel::Info, "china-day trend"); june_30.id = "usage_china_day_30".to_string(); + june_30.account_id = Some("acct_trend".to_string()); + june_30.model = Some("gpt-5.5".to_string()); + june_30.status_code = Some(200); june_30.created_at = chrono::DateTime::parse_from_rfc3339("2026-06-29T16:30:00Z") .unwrap() .with_timezone(&Utc); @@ -517,8 +539,10 @@ async fn admin_usage_records_insight_card_endpoints_should_return_source_specifi 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.account_id = Some("acct_card".to_string()); first.route = Some("/v1/responses".to_string()); first.model = Some("gpt-5.5".to_string()); + first.status_code = Some(200); first.latency_ms = Some(240); first.metadata = json!({ "stream": true, @@ -535,8 +559,10 @@ async fn admin_usage_records_insight_card_endpoints_should_return_source_specifi let mut second = UsageRecord::new("request", UsageRecordLevel::Info, "card mini"); second.id = "usage_card_mini".to_string(); + second.account_id = Some("acct_card".to_string()); second.route = Some("/v1/responses/compact".to_string()); second.model = Some("gpt-5.5-mini".to_string()); + second.status_code = Some(200); second.latency_ms = Some(120); second.metadata = json!({ "stream": false, @@ -662,11 +688,17 @@ async fn admin_usage_records_should_filter_summary_and_table_by_time_range() { let now = Utc::now(); let mut older = UsageRecord::new("request", UsageRecordLevel::Info, "outside range"); older.id = "usage_outside_range".to_string(); + older.account_id = Some("acct_range".to_string()); + older.model = Some("gpt-5.5".to_string()); + older.status_code = Some(200); older.created_at = now - Duration::days(3); store.append(&older).await.unwrap(); let mut current = UsageRecord::new("request", UsageRecordLevel::Info, "inside range"); current.id = "usage_inside_range".to_string(); + current.account_id = Some("acct_range".to_string()); + current.model = Some("gpt-5.5".to_string()); + current.status_code = Some(200); current.created_at = now; store.append(¤t).await.unwrap(); @@ -710,18 +742,18 @@ async fn admin_usage_records_should_filter_summary_and_table_by_time_range() { #[tokio::test] async fn admin_usage_records_should_filter_and_cursor_page_events() { let (app, store, _dir) = admin_usage_records_test_app("admin-usage-records.sqlite").await; - let mut matching = UsageRecord::new("request", UsageRecordLevel::Error, "upstream timeout"); + let mut matching = UsageRecord::new("request", UsageRecordLevel::Info, "upstream timeout"); matching.id = "usage_matching".to_string(); + matching.account_id = Some("acct_filter".to_string()); matching.route = Some("/v1/responses".to_string()); + matching.model = Some("gpt-5.5".to_string()); + matching.status_code = Some(200); store.append(&matching).await.unwrap(); - store - .append(&UsageRecord::new( - "request", - UsageRecordLevel::Info, - "upstream timeout", - )) - .await - .unwrap(); + let mut non_matching_message = UsageRecord::new("request", UsageRecordLevel::Info, "completed"); + non_matching_message.account_id = Some("acct_filter".to_string()); + non_matching_message.model = Some("gpt-5.5".to_string()); + non_matching_message.status_code = Some(200); + store.append(&non_matching_message).await.unwrap(); store .append(&UsageRecord::new( "account", @@ -735,7 +767,7 @@ async fn admin_usage_records_should_filter_and_cursor_page_events() { .oneshot( Request::builder() .method("GET") - .uri("/api/admin/usage/records?kind=request&level=error&search=timeout&limit=1") + .uri("/api/admin/usage/records?kind=request&level=info&search=timeout&limit=1") .header("cookie", "cpr_admin_session=session_1") .header("x-request-id", "req_usage_records_filter") .body(Body::empty()) diff --git a/backend/tests/infra/storage_schema/mod.rs b/backend/tests/infra/storage_schema/mod.rs index 09d409ccb..36b866d30 100644 --- a/backend/tests/infra/storage_schema/mod.rs +++ b/backend/tests/infra/storage_schema/mod.rs @@ -10,7 +10,7 @@ async fn sqlite_schema_should_create_current_tables() { let pool = connect_sqlite(&url).await.unwrap(); let rows: Vec<(String,)> = sqlx::query_as( - "select name from sqlite_master where type = 'table' and name in ('schema_migrations', 'admin_users', 'admin_sessions', 'client_api_keys', 'runtime_settings', 'accounts', 'account_refresh_leases', 'account_usage', 'account_model_usage', 'usage_time_buckets', 'account_cookies', 'fingerprints', 'fingerprint_update_history', 'usage_records', 'model_plan_snapshots', 'session_affinities') order by name", + "select name from sqlite_master where type = 'table' and name in ('schema_migrations', 'admin_users', 'admin_sessions', 'client_api_keys', 'runtime_settings', 'accounts', 'account_refresh_leases', 'account_usage', 'account_model_usage', 'usage_time_buckets', 'account_cookies', 'fingerprints', 'fingerprint_update_history', 'usage_records', 'ops_error_logs', 'model_plan_snapshots', 'session_affinities') order by name", ) .fetch_all(&pool) .await @@ -29,6 +29,7 @@ async fn sqlite_schema_should_create_current_tables() { "fingerprint_update_history", "fingerprints", "model_plan_snapshots", + "ops_error_logs", "runtime_settings", "schema_migrations", "session_affinities", @@ -56,6 +57,7 @@ async fn sqlite_schema_should_record_applied_migrations() { [ (1, "initial".to_string()), (2, "rename_rotation_strategies".to_string()), + (3, "ops_error_logs".to_string()), ] ); } @@ -161,7 +163,7 @@ async fn sqlite_schema_should_persist_diagnostic_filter_indexes() { let pool = connect_sqlite(&url).await.unwrap(); let rows: Vec<(String,)> = sqlx::query_as( - "select name from sqlite_master where type = 'index' and name in ('idx_account_cookies_expires', 'idx_usage_records_level_created', 'idx_usage_records_model_created', 'idx_usage_records_route_created', 'idx_usage_records_status_created', 'idx_usage_records_upstream_status_created', 'idx_session_affinities_active_order', 'idx_usage_time_buckets_bucket', 'idx_usage_time_buckets_model_bucket') order by name", + "select name from sqlite_master where type = 'index' and name in ('idx_account_cookies_expires', 'idx_usage_records_level_created', 'idx_usage_records_model_created', 'idx_usage_records_route_created', 'idx_usage_records_status_created', 'idx_usage_records_upstream_status_created', 'idx_ops_error_logs_created_id', 'idx_ops_error_logs_status_created', 'idx_ops_error_logs_failure_class', 'idx_session_affinities_active_order', 'idx_usage_time_buckets_bucket', 'idx_usage_time_buckets_model_bucket') order by name", ) .fetch_all(&pool) .await @@ -171,6 +173,9 @@ async fn sqlite_schema_should_persist_diagnostic_filter_indexes() { rows.into_iter().map(|row| row.0).collect::>(), [ "idx_account_cookies_expires", + "idx_ops_error_logs_created_id", + "idx_ops_error_logs_failure_class", + "idx_ops_error_logs_status_created", "idx_session_affinities_active_order", "idx_usage_records_level_created", "idx_usage_records_model_created", @@ -475,6 +480,41 @@ async fn sqlite_schema_should_persist_structured_event_diagnostic_columns() { ); } +#[tokio::test] +async fn sqlite_schema_should_persist_ops_error_log_columns() { + let dir = tempfile::tempdir().unwrap(); + let db = dir.path().join("ops-error-columns.sqlite"); + let url = format!("sqlite://{}", db.display()); + let pool = connect_sqlite(&url).await.unwrap(); + + let rows: Vec<(String,)> = sqlx::query_as( + "select name from pragma_table_info('ops_error_logs') where name in ('request_id', 'account_id', 'route', 'model', 'status_code', 'client_status_code', 'upstream_status_code', 'transport', 'attempt_index', 'failure_class', 'response_id', 'upstream_request_id', 'latency_ms', 'metadata_json') order by name", + ) + .fetch_all(&pool) + .await + .unwrap(); + + assert_eq!( + rows.into_iter().map(|row| row.0).collect::>(), + [ + "account_id", + "attempt_index", + "client_status_code", + "failure_class", + "latency_ms", + "metadata_json", + "model", + "request_id", + "response_id", + "route", + "status_code", + "transport", + "upstream_request_id", + "upstream_status_code", + ] + ); +} + #[tokio::test] async fn sqlite_schema_should_reject_invalid_account_status() { let dir = tempfile::tempdir().unwrap(); @@ -507,6 +547,22 @@ async fn sqlite_schema_should_reject_invalid_usage_record_level() { assert!(result.is_err()); } +#[tokio::test] +async fn sqlite_schema_should_reject_invalid_ops_error_status_code() { + let dir = tempfile::tempdir().unwrap(); + let db = dir.path().join("ops-error-status-check.sqlite"); + let url = format!("sqlite://{}", db.display()); + let pool = connect_sqlite(&url).await.unwrap(); + + let result = sqlx::query( + "insert into ops_error_logs (id, kind, status_code, message, metadata_json, created_at) values ('ops_bad', 'request', 99, 'bad', '{}', '2026-06-14T00:00:00Z')", + ) + .execute(&pool) + .await; + + assert!(result.is_err()); +} + #[tokio::test] async fn sqlite_schema_should_reject_non_boolean_flags() { let dir = tempfile::tempdir().unwrap(); diff --git a/backend/tests/proxy/dispatch/chat_upstream/compact_routes.rs b/backend/tests/proxy/dispatch/chat_upstream/compact_routes.rs index 929e6f2ee..acd7046bb 100644 --- a/backend/tests/proxy/dispatch/chat_upstream/compact_routes.rs +++ b/backend/tests/proxy/dispatch/chat_upstream/compact_routes.rs @@ -274,7 +274,7 @@ async fn responses_compact_should_return_rate_limit_error_when_fallback_is_exhau assert_eq!(body["error"]["code"], "rate_limit_exceeded"); assert!(message.contains("All accounts exhausted (1 rate-limited)")); assert!(message.contains("compact quota reached")); - let event = latest_response_usage_record(&pool).await; + let event = latest_response_ops_error_log(&pool).await; let metadata: Value = serde_json::from_str(&event.metadata_json).unwrap(); assert_eq!( event.request_id.as_deref(), @@ -339,7 +339,7 @@ async fn responses_compact_should_preserve_upstream_client_error_status() { assert_eq!(status, StatusCode::BAD_REQUEST); assert_eq!(body["error"]["type"], "invalid_request_error"); assert_eq!(body["error"]["code"], "codex_api_error"); - let event = latest_response_usage_record(&pool).await; + let event = latest_response_ops_error_log(&pool).await; let metadata: Value = serde_json::from_str(&event.metadata_json).unwrap(); assert_eq!( event.request_id.as_deref(), @@ -361,7 +361,7 @@ async fn responses_compact_should_preserve_upstream_client_error_status() { #[tokio::test] async fn responses_compact_should_return_responses_error_when_no_accounts_are_available() { let server = MockServer::start().await; - let (app, api_key, _dir) = test_app_without_accounts(server.uri()).await; + let (app, api_key, _pool, _dir) = test_app_without_accounts(server.uri()).await; let response = app .oneshot( diff --git a/backend/tests/proxy/dispatch/chat_upstream/mod.rs b/backend/tests/proxy/dispatch/chat_upstream/mod.rs index 135c3a0fb..e11baf8b0 100644 --- a/backend/tests/proxy/dispatch/chat_upstream/mod.rs +++ b/backend/tests/proxy/dispatch/chat_upstream/mod.rs @@ -327,7 +327,9 @@ async fn test_app_with_account_pool_and_affinity( test_app_with_account_pool_config_and_affinity(base_url, |_| {}, Some(response_id)).await } -async fn test_app_without_accounts(base_url: String) -> (axum::Router, String, tempfile::TempDir) { +async fn test_app_without_accounts( + base_url: String, +) -> (axum::Router, String, SqlitePool, tempfile::TempDir) { let dir = tempfile::tempdir().unwrap(); let db = dir.path().join("openai-responses-no-accounts.sqlite"); let url = format!("sqlite://{}", db.display()); @@ -336,7 +338,7 @@ async fn test_app_without_accounts(base_url: String) -> (axum::Router, String, t insert_model_snapshot(&pool).await; let state = test_app_state_with_pool_and_installation_id( &test_config(url, base_url), - pool, + pool.clone(), TEST_INSTALLATION_ID.to_string(), ) .await; @@ -346,7 +348,7 @@ async fn test_app_without_accounts(base_url: String) -> (axum::Router, String, t .restore_from_repository() .await .expect("empty account pool should restore"); - (router::router().with_state(state), api_key, dir) + (router::router().with_state(state), api_key, pool, dir) } async fn test_app_with_account_pool_and_logging( @@ -1445,6 +1447,40 @@ async fn latest_response_usage_record(pool: &SqlitePool) -> ResponseUsageRecordS latest_usage_record(pool, "v1.response").await } +async fn latest_response_ops_error_log(pool: &SqlitePool) -> ResponseUsageRecordSnapshot { + let row: Option<( + Option, + Option, + Option, + Option, + Option, + Option, + String, + )> = sqlx::query_as( + "select request_id, account_id, route, model, status_code, response_id, metadata_json + from ops_error_logs + where kind = ? + order by created_at desc, id desc + limit 1", + ) + .bind("v1.response") + .fetch_optional(pool) + .await + .unwrap(); + let (request_id, account_id, route, model, status_code, response_id, metadata_json) = + row.unwrap_or_else(|| panic!("expected a v1.response ops error log")); + ResponseUsageRecordSnapshot { + level: "error".to_string(), + request_id, + account_id, + route, + model, + status_code, + response_id, + metadata_json, + } +} + async fn latest_usage_record(pool: &SqlitePool, kind: &str) -> ResponseUsageRecordSnapshot { let page = SqliteUsageRecordStore::new(pool.clone()) .list( diff --git a/backend/tests/proxy/dispatch/chat_upstream/responses_recovery.rs b/backend/tests/proxy/dispatch/chat_upstream/responses_recovery.rs index 32ce9795b..ebb55035e 100644 --- a/backend/tests/proxy/dispatch/chat_upstream/responses_recovery.rs +++ b/backend/tests/proxy/dispatch/chat_upstream/responses_recovery.rs @@ -93,7 +93,7 @@ async fn responses_should_classify_sse_cyber_policy_failure_as_bad_request() { #[tokio::test] async fn responses_should_return_no_available_accounts_error_when_no_accounts_are_available() { let server = MockServer::start().await; - let (app, api_key, _dir) = test_app_without_accounts(server.uri()).await; + let (app, api_key, pool, _dir) = test_app_without_accounts(server.uri()).await; let response = app .oneshot( @@ -121,6 +121,35 @@ async fn responses_should_return_no_available_accounts_error_when_no_accounts_ar assert_eq!(status, StatusCode::SERVICE_UNAVAILABLE); assert_eq!(body["error"]["type"], "server_error"); assert_eq!(body["error"]["code"], "no_available_accounts"); + + let (usage_count,): (i64,) = sqlx::query_as("select count(*) from usage_records") + .fetch_one(&pool) + .await + .unwrap(); + let (ops_count, account_id, model, failure_class): ( + i64, + Option, + Option, + Option, + ) = sqlx::query_as( + "select count(*), max(account_id), max(model), max(failure_class) from ops_error_logs", + ) + .fetch_one(&pool) + .await + .unwrap(); + let (bucket_requests, bucket_errors): (i64, i64) = sqlx::query_as( + "select coalesce(sum(request_count), 0), coalesce(sum(error_count), 0) from usage_time_buckets", + ) + .fetch_one(&pool) + .await + .unwrap(); + + assert_eq!(usage_count, 0); + assert_eq!(ops_count, 1); + assert_eq!(account_id, None); + assert_eq!(model.as_deref(), Some("gpt-5.5")); + assert_eq!(failure_class.as_deref(), Some("no_available_accounts")); + assert_eq!((bucket_requests, bucket_errors), (1, 1)); } #[tokio::test] @@ -2128,7 +2157,7 @@ async fn responses_should_return_rate_limit_error_when_429_fallback_is_exhausted assert_eq!(body["error"]["code"], "rate_limit_exceeded"); assert_eq!(quota_state.0, 1); assert!(quota_state.1.is_some()); - let event = latest_response_usage_record(&pool).await; + let event = latest_response_ops_error_log(&pool).await; let metadata: Value = serde_json::from_str(&event.metadata_json).unwrap(); assert_eq!(event.level, "error"); assert_eq!( @@ -2212,7 +2241,7 @@ async fn responses_stream_should_return_rate_limit_error_when_429_fallback_is_ex assert_eq!(quota_state.0, 1); assert!(quota_state.1.is_some()); assert!(!response_request_id.is_empty()); - let event = latest_response_usage_record(&pool).await; + let event = latest_response_ops_error_log(&pool).await; let metadata: Value = serde_json::from_str(&event.metadata_json).unwrap(); assert_eq!(event.level, "error"); assert_eq!( diff --git a/backend/tests/proxy/dispatch/chat_upstream/usage_logging.rs b/backend/tests/proxy/dispatch/chat_upstream/usage_logging.rs index 5e2686ed4..93ba315fc 100644 --- a/backend/tests/proxy/dispatch/chat_upstream/usage_logging.rs +++ b/backend/tests/proxy/dispatch/chat_upstream/usage_logging.rs @@ -678,7 +678,7 @@ async fn responses_stream_should_record_usage_record_after_late_disconnect() { let response = response_task.await.unwrap(); close_upstream.send(()).unwrap(); let body = response_text(response).await; - let event = latest_response_usage_record(&pool).await; + let event = latest_response_ops_error_log(&pool).await; let metadata: Value = serde_json::from_str(&event.metadata_json).unwrap(); assert!(body.contains("stream_disconnected")); @@ -744,7 +744,7 @@ async fn responses_stream_should_record_failure_detail_after_transport_error() { let response = response_task.await.unwrap(); close_upstream.send(()).unwrap(); let _body = response_text(response).await; - let event = latest_response_usage_record(&pool).await; + let event = latest_response_ops_error_log(&pool).await; let metadata: Value = serde_json::from_str(&event.metadata_json).unwrap(); assert_eq!(metadata["failureSource"], "proxy"); @@ -788,7 +788,7 @@ async fn responses_stream_should_not_record_first_token_when_metadata_only_prefi let response = response_task.await.unwrap(); close_upstream.send(()).unwrap(); let body = response_text(response).await; - let event = latest_response_usage_record(&pool).await; + let event = latest_response_ops_error_log(&pool).await; let metadata: Value = serde_json::from_str(&event.metadata_json).unwrap(); let first_token_bucket: (i64, i64) = sqlx::query_as( "select coalesce(sum(first_token_latency_sum), 0), coalesce(sum(first_token_latency_count), 0) from usage_time_buckets", @@ -855,7 +855,7 @@ 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 event = latest_response_ops_error_log(&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"); @@ -906,7 +906,7 @@ async fn responses_should_record_attempt_trace_after_retrying_another_account() ) .await .unwrap(); - let event = latest_response_usage_record(&pool).await; + let event = latest_response_ops_error_log(&pool).await; let metadata: Value = serde_json::from_str(&event.metadata_json).unwrap(); assert_eq!(response.status(), StatusCode::BAD_GATEWAY); diff --git a/backend/tests/runtime/tasks/model_refresh.rs b/backend/tests/runtime/tasks/model_refresh.rs index 90c11119c..03038c911 100644 --- a/backend/tests/runtime/tasks/model_refresh.rs +++ b/backend/tests/runtime/tasks/model_refresh.rs @@ -1,6 +1,9 @@ use super::*; -use std::collections::BTreeMap; +use std::{ + collections::BTreeMap, + sync::atomic::{AtomicUsize, Ordering}, +}; use codex_proxy_rs::upstream::{ accounts::pool::{AccountAcquireRequest, AccountPoolOptions, RuntimeAccountPoolService}, @@ -50,6 +53,8 @@ async fn model_refresh_task_should_sync_model_plan_allowlist_to_account_pool() { plan_account("acct-plus", "plus"), plan_account("acct-team", "team"), ], + request_usage_records: Arc::new(AtomicUsize::new(0)), + model_request_usage_records: Arc::new(AtomicUsize::new(0)), }); let account_pool = Arc::new(RuntimeAccountPoolService::new( account_store.clone(), @@ -74,6 +79,16 @@ async fn model_refresh_task_should_sync_model_plan_allowlist_to_account_pool() { .await .expect("team model should acquire an account"); assert_eq!(acquired.account.id, "acct-team"); + assert_eq!( + account_store.request_usage_records.load(Ordering::SeqCst), + 0 + ); + assert_eq!( + account_store + .model_request_usage_records + .load(Ordering::SeqCst), + 0 + ); } fn empty_model_config() -> ModelConfig { @@ -139,6 +154,8 @@ impl CodexModelCatalogClient for FakeModelCatalogClient { struct PlanAccountStore { accounts: Vec, + request_usage_records: Arc, + model_request_usage_records: Arc, } #[async_trait] @@ -174,8 +191,11 @@ impl AccountStore for PlanAccountStore { async fn record_usage_delta( &self, _account_id: &str, - _usage: AccountUsageDelta, + usage: AccountUsageDelta, ) -> AccountStoreResult<()> { + if usage.requests > 0 { + self.request_usage_records.fetch_add(1, Ordering::SeqCst); + } Ok(()) } @@ -183,8 +203,12 @@ impl AccountStore for PlanAccountStore { &self, _account_id: &str, _model: &str, - _usage: AccountModelUsageDelta, + usage: AccountModelUsageDelta, ) -> AccountStoreResult<()> { + if usage.requests > 0 { + self.model_request_usage_records + .fetch_add(1, Ordering::SeqCst); + } Ok(()) }