Merge pull request #429 from AAEE86/rust

fix(usage): 统一用户排行榜 Token 统计口径

apps/aether-gateway/src/tests/control/admin/stats.rs

@@ -1375,7 +1375,7 @@ async fn gateway_handles_admin_stats_leaderboard_models_locally_with_trusted_adm
     assert_eq!(payload["metric"], "tokens");
     assert_eq!(payload["items"][0]["rank"], 1);
     assert_eq!(payload["items"][0]["id"], "gpt-5");
-    assert_eq!(payload["items"][0]["value"], 180);
+    assert_eq!(payload["items"][0]["value"], 160);
     assert_eq!(payload["items"][1]["id"], "claude-3-5-sonnet");
     assert_eq!(payload["items"][1]["value"], 100);
     assert_eq!(*upstream_hits.lock().expect("mutex should lock"), 0);

apps/aether-gateway/src/tests/control/admin/usage.rs

@@ -1682,7 +1682,7 @@ async fn gateway_handles_admin_usage_detail_locally_with_trusted_admin_principal
     assert_eq!(payload["provider"], "OpenAI");
     assert_eq!(payload["model"], "gpt-5");
     assert_eq!(payload["effective_input_tokens"], 115);
-    assert_eq!(payload["total_tokens"], 170);
+    assert_eq!(payload["total_tokens"], 165);
     assert_eq!(payload["cache_creation_cost"], 0.0);
     assert_eq!(payload["cache_read_cost"], 0.0);
     assert_eq!(

crates/aether-admin/src/observability/stats.rs

@@ -1,3 +1,4 @@
+use crate::observability::usage::admin_usage_total_tokens;
 use aether_data::repository::auth::StoredAuthApiKeySnapshot;
 use aether_data_contracts::repository::{
     provider_catalog::StoredProviderCatalogProvider,
@@ -1799,12 +1800,7 @@ pub fn build_user_leaderboard_items(
                     cost: 0.0,
                 });
         entry.requests = entry.requests.saturating_add(1);
-        entry.tokens = entry.tokens.saturating_add(
+        entry.tokens = entry.tokens.saturating_add(admin_usage_total_tokens(item));
-            item.input_tokens
-                .saturating_add(item.output_tokens)
-                .saturating_add(item.cache_creation_input_tokens)
-                .saturating_add(item.cache_read_input_tokens),
-        );
         entry.cost += item.total_cost_usd;
     }
 
@@ -2202,4 +2198,24 @@ mod tests {
         assert_eq!(leaderboard.len(), 1);
         assert_eq!(leaderboard[0].name, "alice");
     }
+
+    #[test]
+    fn user_leaderboard_tokens_match_dashboard_effective_token_rules() {
+        let item = StoredRequestUsageAudit {
+            input_tokens: 100,
+            output_tokens: 20,
+            total_tokens: 999,
+            cache_creation_input_tokens: 0,
+            cache_creation_ephemeral_5m_input_tokens: 12,
+            cache_creation_ephemeral_1h_input_tokens: 8,
+            cache_read_input_tokens: 80,
+            ..sample_usage(Some("legacy-default"))
+        };
+
+        let leaderboard =
+            build_user_leaderboard_items(&[item], &BTreeMap::new(), false, false, false);
+
+        assert_eq!(leaderboard.len(), 1);
+        assert_eq!(leaderboard[0].tokens, 140);
+    }
 }

crates/aether-admin/src/observability/usage.rs

@@ -1135,7 +1135,7 @@ pub fn admin_usage_record_json(
 }
 
 pub fn admin_usage_total_tokens(item: &StoredRequestUsageAudit) -> u64 {
-    item.input_tokens
+    admin_usage_effective_input_tokens(item)
         .saturating_add(item.output_tokens)
         .saturating_add(admin_usage_cache_creation_tokens(item))
         .saturating_add(item.cache_read_input_tokens)
@@ -2292,7 +2292,7 @@ mod tests {
         admin_usage_has_fallback, admin_usage_is_failed, admin_usage_is_success,
         admin_usage_matches_search, admin_usage_matches_status, admin_usage_matches_username,
         admin_usage_record_json, admin_usage_resolve_request_capture_body,
-        admin_usage_upstream_is_stream, build_admin_usage_detail_payload,
+        admin_usage_total_tokens, admin_usage_upstream_is_stream, build_admin_usage_detail_payload,
     };
     use aether_data_contracts::repository::usage::{StoredRequestUsageAudit, UsageBodyField};
 
@@ -3054,6 +3054,22 @@ mod tests {
         assert_eq!(payload["total_tokens"], 50);
     }
 
+    #[test]
+    fn admin_usage_total_tokens_uses_effective_input_for_cached_openai_usage() {
+        let item = StoredRequestUsageAudit {
+            input_tokens: 100,
+            output_tokens: 20,
+            total_tokens: 999,
+            cache_creation_input_tokens: 0,
+            cache_creation_ephemeral_5m_input_tokens: 12,
+            cache_creation_ephemeral_1h_input_tokens: 8,
+            cache_read_input_tokens: 80,
+            ..sample_usage("completed", Some(200), None)
+        };
+
+        assert_eq!(admin_usage_total_tokens(&item), 140);
+    }
+
     #[test]
     fn detail_payload_exposes_typed_settlement_section() {
         let item = StoredRequestUsageAudit {

crates/aether-data/src/repository/usage/memory.rs

@@ -931,6 +931,13 @@ fn usage_effective_input_tokens(item: &StoredRequestUsageAudit) -> u64 {
     normalize_usage_input_tokens(api_format, input_tokens, cache_read_tokens) as u64
 }
 
+fn usage_total_tokens(item: &StoredRequestUsageAudit) -> u64 {
+    usage_effective_input_tokens(item)
+        .saturating_add(item.output_tokens)
+        .saturating_add(usage_cache_creation_tokens(item))
+        .saturating_add(item.cache_read_input_tokens)
+}
+
 fn usage_is_success(item: &StoredRequestUsageAudit) -> bool {
     matches!(
         item.status.as_str(),
@@ -1426,7 +1433,9 @@ impl UsageReadRepository for InMemoryUsageReadRepository {
                 .effective_input_tokens
                 .saturating_add(usage_effective_input_tokens(item));
             summary.output_tokens = summary.output_tokens.saturating_add(item.output_tokens);
-            summary.total_tokens = summary.total_tokens.saturating_add(item.total_tokens);
+            summary.total_tokens = summary
+                .total_tokens
+                .saturating_add(usage_total_tokens(item));
             summary.cache_creation_tokens = summary
                 .cache_creation_tokens
                 .saturating_add(usage_cache_creation_tokens(item));
@@ -2187,12 +2196,7 @@ impl UsageReadRepository for InMemoryUsageReadRepository {
                 entry.legacy_name = legacy_name;
             }
             entry.request_count = entry.request_count.saturating_add(1);
-            entry.total_tokens = entry.total_tokens.saturating_add(
+            entry.total_tokens = entry.total_tokens.saturating_add(usage_total_tokens(item));
-                item.input_tokens
-                    .saturating_add(item.output_tokens)
-                    .saturating_add(item.cache_creation_input_tokens)
-                    .saturating_add(item.cache_read_input_tokens),
-            );
             entry.total_cost_usd += item.total_cost_usd;
         }
         Ok(grouped.into_values().collect())
@@ -2989,7 +2993,8 @@ mod tests {
     };
     use aether_data_contracts::repository::usage::{
         usage_body_ref, ProviderApiKeyWindowUsageRequest, UsageAuditAggregationGroupBy,
-        UsageAuditAggregationQuery, UsageBodyField, UsageProviderPerformanceQuery,
+        UsageAuditAggregationQuery, UsageBodyField, UsageDashboardSummaryQuery,
+        UsageLeaderboardGroupBy, UsageLeaderboardQuery, UsageProviderPerformanceQuery,
         UsageTimeSeriesGranularity,
     };
     use serde_json::json;
@@ -4647,6 +4652,46 @@ mod tests {
         assert_eq!(items[0].request_id, "req-2");
     }
 
+    #[tokio::test]
+    async fn dashboard_and_leaderboard_total_tokens_use_effective_cache_aware_tokens() {
+        let mut item = sample_usage("req-cache-aware-total", 1_711_000_000);
+        item.input_tokens = 100;
+        item.output_tokens = 20;
+        item.total_tokens = 999;
+        item.cache_creation_input_tokens = 0;
+        item.cache_creation_ephemeral_5m_input_tokens = 12;
+        item.cache_creation_ephemeral_1h_input_tokens = 8;
+        item.cache_read_input_tokens = 80;
+
+        let repository = InMemoryUsageReadRepository::seed(vec![item]);
+
+        let dashboard = repository
+            .summarize_dashboard_usage(&UsageDashboardSummaryQuery {
+                created_from_unix_secs: 1_711_000_000,
+                created_until_unix_secs: 1_711_000_001,
+                user_id: None,
+            })
+            .await
+            .expect("dashboard should summarize");
+        assert_eq!(dashboard.effective_input_tokens, 20);
+        assert_eq!(dashboard.cache_creation_tokens, 20);
+        assert_eq!(dashboard.total_tokens, 140);
+
+        let leaderboard = repository
+            .summarize_usage_leaderboard(&UsageLeaderboardQuery {
+                created_from_unix_secs: 1_711_000_000,
+                created_until_unix_secs: 1_711_000_001,
+                group_by: UsageLeaderboardGroupBy::User,
+                user_id: None,
+                provider_name: None,
+                model: None,
+            })
+            .await
+            .expect("leaderboard should summarize");
+        assert_eq!(leaderboard.len(), 1);
+        assert_eq!(leaderboard[0].total_tokens, 140);
+    }
+
     #[tokio::test]
     async fn summarizes_provider_api_key_last_used_at_in_seconds() {
         let repository = InMemoryUsageReadRepository::seed(vec![

crates/aether-data/src/repository/usage/mysql.rs

@@ -1104,7 +1104,7 @@ mod tests {
             .expect("dashboard summary should load");
         assert_eq!(summary.total_requests, 2);
         assert_eq!(summary.error_requests, 1);
-        assert_eq!(summary.total_tokens, 14);
+        assert_eq!(summary.total_tokens, 10);
     }
 
     async fn seed_stats_targets(

crates/aether-data/src/repository/usage/postgres/mod.rs

@@ -6019,9 +6019,30 @@ SELECT
   MAX({legacy_name_expr}) AS legacy_name,
   COUNT(*)::BIGINT AS request_count,
   COALESCE(SUM(
-    GREATEST(COALESCE("usage".input_tokens, 0), 0)
+    CASE
+      WHEN GREATEST(COALESCE("usage".input_tokens, 0), 0) <= 0 THEN 0
+      WHEN GREATEST(COALESCE("usage".cache_read_input_tokens, 0), 0) <= 0
+      THEN GREATEST(COALESCE("usage".input_tokens, 0), 0)
+      WHEN split_part(lower(COALESCE(COALESCE("usage".endpoint_api_format, "usage".api_format), '')), ':', 1)
+           IN ('openai', 'gemini', 'google')
+      THEN GREATEST(
+        GREATEST(COALESCE("usage".input_tokens, 0), 0)
+          - GREATEST(COALESCE("usage".cache_read_input_tokens, 0), 0),
+        0
+      )
+      ELSE GREATEST(COALESCE("usage".input_tokens, 0), 0)
+    END
     + GREATEST(COALESCE("usage".output_tokens, 0), 0)
-    + GREATEST(COALESCE("usage".cache_creation_input_tokens, 0), 0)
+    + CASE
+        WHEN COALESCE("usage".cache_creation_input_tokens, 0) = 0
+             AND (
+               COALESCE("usage".cache_creation_input_tokens_5m, 0)
+               + COALESCE("usage".cache_creation_input_tokens_1h, 0)
+             ) > 0
+        THEN COALESCE("usage".cache_creation_input_tokens_5m, 0)
+           + COALESCE("usage".cache_creation_input_tokens_1h, 0)
+        ELSE COALESCE("usage".cache_creation_input_tokens, 0)
+      END
     + GREATEST(COALESCE("usage".cache_read_input_tokens, 0), 0)
   ), 0)::BIGINT AS total_tokens,
   COALESCE(SUM(COALESCE(CAST("usage".total_cost_usd AS DOUBLE PRECISION), 0)), 0)
@@ -6169,42 +6190,20 @@ WHERE date >=
                 fetch_usage_leaderboard_query(builder.build(), &self.pool).await?
             }
             UsageLeaderboardGroupBy::User => {
-                let mut builder = if query.provider_name.is_some() && query.model.is_some() {
+                if query.provider_name.is_some() && query.model.is_some() {
+                    return Ok(None);
+                }
+                let mut builder = if let Some(provider_name) = query.provider_name.as_deref() {
                     let mut builder = QueryBuilder::<Postgres>::new(
                         r#"
 SELECT
   user_id AS group_key,
   MAX(NULLIF(BTRIM(username), '')) AS legacy_name,
   COALESCE(SUM(total_requests), 0)::BIGINT AS request_count,
-  COALESCE(SUM(total_tokens), 0)::BIGINT AS total_tokens,
+  COALESCE(
-  CAST(COALESCE(SUM(total_cost), 0) AS DOUBLE PRECISION) AS total_cost_usd
+    SUM(effective_input_tokens + output_tokens + cache_creation_tokens + cache_read_tokens),
-FROM stats_user_daily_model_provider
+    0
-WHERE date >=
+  )::BIGINT AS total_tokens,
-"#,
-                    );
-                    builder
-                        .push_bind(start_day_utc)
-                        .push(" AND date < ")
-                        .push_bind(end_day_utc)
-                        .push(" AND provider_name = ")
-                        .push_bind(query.provider_name.as_deref().unwrap().to_string())
-                        .push(" AND model = ")
-                        .push_bind(query.model.as_deref().unwrap().to_string());
-                    if let Some(user_id) = query.user_id.as_deref() {
-                        builder
-                            .push(" AND user_id = ")
-                            .push_bind(user_id.to_string());
-                    }
-                    builder.push(" GROUP BY user_id ORDER BY user_id ASC");
-                    builder
-                } else if let Some(provider_name) = query.provider_name.as_deref() {
-                    let mut builder = QueryBuilder::<Postgres>::new(
-                        r#"
-SELECT
-  user_id AS group_key,
-  MAX(NULLIF(BTRIM(username), '')) AS legacy_name,
-  COALESCE(SUM(total_requests), 0)::BIGINT AS request_count,
-  COALESCE(SUM(total_tokens), 0)::BIGINT AS total_tokens,
   CAST(COALESCE(SUM(total_cost), 0) AS DOUBLE PRECISION) AS total_cost_usd
 FROM stats_user_daily_provider
 WHERE date >=
@@ -6230,7 +6229,10 @@ SELECT
   user_id AS group_key,
   MAX(NULLIF(BTRIM(username), '')) AS legacy_name,
   COALESCE(SUM(total_requests), 0)::BIGINT AS request_count,
-  COALESCE(SUM(total_tokens), 0)::BIGINT AS total_tokens,
+  COALESCE(
+    SUM(effective_input_tokens + output_tokens + cache_creation_tokens + cache_read_tokens),
+    0
+  )::BIGINT AS total_tokens,
   CAST(COALESCE(SUM(total_cost), 0) AS DOUBLE PRECISION) AS total_cost_usd
 FROM stats_user_daily_model
 WHERE date >=
@@ -6257,7 +6259,7 @@ SELECT
   MAX(NULLIF(BTRIM(username), '')) AS legacy_name,
   COALESCE(SUM(total_requests), 0)::BIGINT AS request_count,
   COALESCE(
-    SUM(input_tokens + output_tokens + cache_creation_tokens + cache_read_tokens),
+    SUM(effective_input_tokens + output_tokens + cache_creation_tokens + cache_read_tokens),
     0
   )::BIGINT AS total_tokens,
   CAST(COALESCE(SUM(total_cost), 0) AS DOUBLE PRECISION) AS total_cost_usd

crates/aether-data/src/repository/usage/sqlite.rs

@@ -1170,7 +1170,7 @@ ORDER BY request_id
             .expect("dashboard summary should load");
         assert_eq!(summary.total_requests, 2);
         assert_eq!(summary.error_requests, 1);
-        assert_eq!(summary.total_tokens, 14);
+        assert_eq!(summary.total_tokens, 10);
     }
 
     async fn seed_stats_targets(pool: &sqlx::SqlitePool) {