chore: bump version to 2.3.1

Co-authored-by: Codex <noreply@openai.com>

backend/src/api/admin/dashboard_routes.rs

@@ -205,12 +205,12 @@ pub(crate) async fn dashboard_summary(
     let capacity = state.services.account_pool.capacity_summary_now().await;
     let now = Utc::now();
     let today_filter = today_usage_record_filter(now);
-    let lifetime_usage = state
+    let retained_usage = state
         .services
         .usage
-        .summary()
+        .retained_totals()
         .await
-        .map_err(|error| dashboard_data_error("account usage summary", &error))?;
+        .map_err(|error| dashboard_data_error("retained usage summary", &error))?;
     let account_usage_records = state
         .services
         .usage_records
@@ -256,7 +256,7 @@ pub(crate) async fn dashboard_summary(
             cards: dashboard_cards(
                 dashboard_account_counts(&accounts),
                 &time_buckets,
-                &lifetime_usage,
+                &retained_usage,
             ),
             trend,
             health_timeline: dashboard_health_timeline_data(&time_buckets),

backend/src/telemetry/account_usage/query.rs

@@ -6,7 +6,7 @@ use thiserror::Error;
 
 use crate::infra::format::nonnegative_i64_to_u64;
 use crate::telemetry::{
-    account_usage::store::{PgAccountUsageStore, UsageListRecord, UsageSummary},
+    account_usage::store::{PgAccountUsageStore, UsageListRecord},
     billing,
     buckets::query::{
         ModelBucketUsage, ModelUsageWindow, PgRequestBucketQuery, UsageBucketTotals,
@@ -42,13 +42,12 @@ impl AccountUsageQueryService {
             .map_err(|_| AccountUsageQueryError::List)
     }
 
-    /// 汇总账号用量。
-    pub async fn summary(&self) -> Result<AccountUsageSummary, AccountUsageQueryError> {
-        self.store
-            .usage_summary()
+    /// 汇总请求时间桶保留期内的全局用量。
+    pub async fn retained_totals(&self) -> Result<UsageBucketTotals, AccountUsageQueryError> {
+        self.buckets
+            .retained_totals()
             .await
-            .map(AccountUsageSummary::from)
-            .map_err(|_| AccountUsageQueryError::Summary)
+            .map_err(|_| AccountUsageQueryError::RetainedTotals)
     }
 
     /// 列出指定时间范围内的聚合时间桶。
@@ -95,8 +94,8 @@ impl AccountUsageQueryService {
 pub enum AccountUsageQueryError {
     #[error("failed to list account usage")]
     List,
-    #[error("failed to summarize account usage")]
-    Summary,
+    #[error("failed to summarize retained usage")]
+    RetainedTotals,
     #[error("failed to list usage time buckets")]
     TimeBuckets,
 }
@@ -129,23 +128,6 @@ pub struct AccountUsageRecord {
     pub last_used_at: Option<DateTime<Utc>>,
 }
 
-/// 管理端账号用量汇总。
-#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
-pub struct AccountUsageSummary {
-    pub account_count: i64,
-    pub request_count: i64,
-    pub empty_response_count: i64,
-    pub input_tokens: i64,
-    pub output_tokens: i64,
-    pub cached_tokens: i64,
-    pub reasoning_tokens: i64,
-    pub total_tokens: i64,
-    pub image_input_tokens: i64,
-    pub image_output_tokens: i64,
-    pub image_request_count: i64,
-    pub image_request_failed_count: i64,
-}
-
 /// 管理端时间桶用量记录。
 #[derive(Debug, Clone, PartialEq)]
 pub struct AccountUsageTimeBucket {
@@ -166,25 +148,6 @@ pub struct AccountUsageTimeBucket {
     pub billing_amount_usd: Option<f64>,
 }
 
-impl From<UsageSummary> for AccountUsageSummary {
-    fn from(s: UsageSummary) -> Self {
-        Self {
-            account_count: s.account_count,
-            request_count: s.request_count,
-            empty_response_count: s.empty_response_count,
-            input_tokens: s.input_tokens,
-            output_tokens: s.output_tokens,
-            cached_tokens: s.cached_tokens,
-            reasoning_tokens: s.reasoning_tokens,
-            total_tokens: s.total_tokens,
-            image_input_tokens: s.image_input_tokens,
-            image_output_tokens: s.image_output_tokens,
-            image_request_count: s.image_request_count,
-            image_request_failed_count: s.image_request_failed_count,
-        }
-    }
-}
-
 impl From<UsageTimeBucketRecord> for AccountUsageTimeBucket {
     fn from(record: UsageTimeBucketRecord) -> Self {
         let billing_amount_usd = time_bucket_billing_amount_usd(&record);

backend/src/telemetry/account_usage/store.rs

@@ -129,22 +129,6 @@ select
 from account_usage au
 left join accounts a on a.id = au.account_id";
 
-const USAGE_SUMMARY_SQL: &str = r"
-select
-  count(*) as account_count,
-  coalesce(sum(request_count), 0)::bigint as request_count,
-  coalesce(sum(empty_response_count), 0)::bigint as empty_response_count,
-  coalesce(sum(input_tokens), 0)::bigint as input_tokens,
-  coalesce(sum(output_tokens), 0)::bigint as output_tokens,
-  coalesce(sum(cached_tokens), 0)::bigint as cached_tokens,
-  coalesce(sum(reasoning_tokens), 0)::bigint as reasoning_tokens,
-  coalesce(sum(total_tokens), 0)::bigint as total_tokens,
-  coalesce(sum(image_input_tokens), 0)::bigint as image_input_tokens,
-  coalesce(sum(image_output_tokens), 0)::bigint as image_output_tokens,
-  coalesce(sum(image_request_count), 0)::bigint as image_request_count,
-  coalesce(sum(image_request_failed_count), 0)::bigint as image_request_failed_count
-from account_usage";
-
 /// PostgreSQL 用量存储错误。
 #[derive(Debug, Error)]
 pub enum PgAccountUsageStoreError {
@@ -312,35 +296,6 @@ pub struct UsageListRecord {
     pub last_used_at: Option<DateTime<Utc>>,
 }
 
-/// 账号用量汇总。
-#[derive(Debug, Clone)]
-pub struct UsageSummary {
-    /// 账号数。
-    pub account_count: i64,
-    /// 总请求数。
-    pub request_count: i64,
-    /// 总空响应数。
-    pub empty_response_count: i64,
-    /// 总输入 token。
-    pub input_tokens: i64,
-    /// 总输出 token。
-    pub output_tokens: i64,
-    /// 总缓存 token。
-    pub cached_tokens: i64,
-    /// 总 reasoning token。
-    pub reasoning_tokens: i64,
-    /// 总 token。
-    pub total_tokens: i64,
-    /// 总图片输入 token。
-    pub image_input_tokens: i64,
-    /// 总图片输出 token。
-    pub image_output_tokens: i64,
-    /// 总图片请求成功数。
-    pub image_request_count: i64,
-    /// 总图片请求失败数。
-    pub image_request_failed_count: i64,
-}
-
 /// PostgreSQL 用量存储。
 #[derive(Clone)]
 pub struct PgAccountUsageStore {
@@ -492,25 +447,6 @@ where account_id in (",
         let rows = builder.build().fetch_all(&self.pool).await?;
         rows.iter().map(usage_list_from_row).collect()
     }
-
-    /// 汇总账号用量。
-    pub async fn usage_summary(&self) -> PgAccountUsageStoreResult<UsageSummary> {
-        let row = sqlx::query(USAGE_SUMMARY_SQL).fetch_one(&self.pool).await?;
-        Ok(UsageSummary {
-            account_count: row.get("account_count"),
-            request_count: row.get("request_count"),
-            empty_response_count: row.get("empty_response_count"),
-            input_tokens: row.get("input_tokens"),
-            output_tokens: row.get("output_tokens"),
-            cached_tokens: row.get("cached_tokens"),
-            reasoning_tokens: row.get("reasoning_tokens"),
-            total_tokens: row.get("total_tokens"),
-            image_input_tokens: row.get("image_input_tokens"),
-            image_output_tokens: row.get("image_output_tokens"),
-            image_request_count: row.get("image_request_count"),
-            image_request_failed_count: row.get("image_request_failed_count"),
-        })
-    }
 }
 
 #[async_trait]

backend/src/telemetry/buckets/query.rs

@@ -29,6 +29,14 @@ where bucket_start >= $1 and bucket_start <= $2
 group by bucket_start, model, service_tier
 order by bucket_start asc, model asc, service_tier asc";
 
+const RETAINED_USAGE_TOTALS_SQL: &str = r"
+select
+  coalesce(sum(success_count + error_count), 0)::bigint as request_count,
+  coalesce(sum(input_tokens), 0)::bigint as input_tokens,
+  coalesce(sum(output_tokens), 0)::bigint as output_tokens,
+  coalesce(sum(cached_tokens), 0)::bigint as cached_tokens
+from request_time_buckets";
+
 /// 请求时间桶查询错误。
 #[derive(Debug, Error)]
 pub enum PgRequestBucketQueryError {
@@ -127,6 +135,19 @@ impl PgRequestBucketQuery {
         Ok(rows.iter().map(usage_time_bucket_from_row).collect())
     }
 
+    /// 汇总请求时间桶保留期内的全局用量，不受当前账号集合影响。
+    pub async fn retained_totals(&self) -> Result<UsageBucketTotals, PgRequestBucketQueryError> {
+        let row = sqlx::query(RETAINED_USAGE_TOTALS_SQL)
+            .fetch_one(&self.pool)
+            .await?;
+        Ok(UsageBucketTotals {
+            request_count: row.get("request_count"),
+            input_tokens: row.get("input_tokens"),
+            output_tokens: row.get("output_tokens"),
+            cached_tokens: row.get("cached_tokens"),
+        })
+    }
+
     pub async fn usage_by_windows(
         &self,
         windows: &[UsageBucketWindow],

backend/src/telemetry/dashboard.rs

@@ -11,7 +11,7 @@ use crate::infra::{
     time::{china_day_start, china_quarter_hour_start},
 };
 
-use super::account_usage::query::{AccountUsageSummary, AccountUsageTimeBucket};
+use super::{account_usage::query::AccountUsageTimeBucket, buckets::query::UsageBucketTotals};
 
 const HEALTH_TIMELINE_SLOT_MINUTES: i64 = 15;
 const HEALTH_TIMELINE_SLOTS: i64 = 24 * 4;
@@ -174,15 +174,15 @@ impl UsageWindow {
 pub fn dashboard_cards(
     accounts: DashboardAccountCounts,
     buckets: &[AccountUsageTimeBucket],
-    lifetime_usage: &AccountUsageSummary,
+    retained_usage: &UsageBucketTotals,
 ) -> DashboardCardsData {
-    dashboard_cards_at(accounts, buckets, lifetime_usage, Utc::now())
+    dashboard_cards_at(accounts, buckets, retained_usage, Utc::now())
 }
 
 pub fn dashboard_cards_at(
     accounts: DashboardAccountCounts,
     buckets: &[AccountUsageTimeBucket],
-    lifetime_usage: &AccountUsageSummary,
+    retained_usage: &UsageBucketTotals,
     now: DateTime<Utc>,
 ) -> DashboardCardsData {
     let today_start = china_day_start(now);
@@ -190,9 +190,10 @@ pub fn dashboard_cards_at(
     let today = usage_window(buckets, today_start, now);
     let yesterday = usage_window(buckets, yesterday_start, today_start);
     let today_billing_amount = billing_window(buckets, today_start, now).unwrap_or(0.0);
-    let total_requests = nonnegative_i64_to_u64(lifetime_usage.request_count);
-    let total_tokens = nonnegative_i64_to_u64(lifetime_usage.total_tokens);
-    let total_cached_tokens = nonnegative_i64_to_u64(lifetime_usage.cached_tokens);
+    let total_requests = nonnegative_i64_to_u64(retained_usage.request_count);
+    let total_tokens = nonnegative_i64_to_u64(retained_usage.input_tokens)
+        .saturating_add(nonnegative_i64_to_u64(retained_usage.output_tokens));
+    let total_cached_tokens = nonnegative_i64_to_u64(retained_usage.cached_tokens);
 
     DashboardCardsData {
         accounts: DashboardAccountsCardData {
@@ -220,7 +221,7 @@ pub fn dashboard_cards_at(
             today_hit_rate: format_rate(today.cache_hit_rate()),
             today_hit_rate_value: today.cache_hit_rate(),
             yesterday_hit_rate_value: yesterday.cache_hit_rate(),
-            total_hit_rate: format_rate(summary_cache_hit_rate(lifetime_usage)),
+            total_hit_rate: format_rate(summary_cache_hit_rate(retained_usage)),
             total_cached_tokens: format_tokens(total_cached_tokens),
             average_first_token_latency_ms: format_optional_duration_ms(
                 today.avg_first_token_latency(),
@@ -540,7 +541,7 @@ fn health_tone(bucket: UsageWindow, success_rate: f64, is_future: bool) -> char
     }
 }
 
-fn summary_cache_hit_rate(summary: &AccountUsageSummary) -> Option<f64> {
+fn summary_cache_hit_rate(summary: &UsageBucketTotals) -> Option<f64> {
     let input_tokens = nonnegative_i64_to_u64(summary.input_tokens);
     (input_tokens > 0)
         .then(|| nonnegative_i64_to_u64(summary.cached_tokens) as f64 / input_tokens as f64)

backend/src/telemetry/usage/insights.rs

@@ -162,12 +162,27 @@ impl UsageDiagnosticsDimension {
 
     fn expression(self) -> &'static str {
         match self {
-            Self::Model => "model_name",
-            Self::Account => "coalesce(nullif(account_id, ''), '未知账号')",
-            Self::ApiKey => "coalesce(nullif(client_api_key_id, ''), '未知 API Key')",
-            Self::Provider => "coalesce(nullif(provider, ''), '未知 Provider')",
-            Self::Transport => "coalesce(nullif(transport, ''), '未知传输')",
-            Self::FailureClass => "failure_class",
+            Self::Model => "terminal_requests.model_name",
+            Self::Account => {
+                "coalesce(nullif(concat_ws(' → ', nullif(diagnostic_accounts.email, ''), nullif(terminal_requests.account_id, '')), ''), '未知账号')"
+            }
+            Self::ApiKey => {
+                "coalesce(nullif(terminal_requests.client_api_key_id, ''), '未知 API Key')"
+            }
+            Self::Provider => "coalesce(nullif(terminal_requests.provider, ''), '未知 Provider')",
+            Self::Transport => "coalesce(nullif(terminal_requests.transport, ''), '未知传输')",
+            Self::FailureClass => "terminal_requests.failure_class",
+        }
+    }
+
+    fn source(self) -> &'static str {
+        match self {
+            Self::Account => {
+                "terminal_requests left join accounts diagnostic_accounts on diagnostic_accounts.id = terminal_requests.account_id"
+            }
+            Self::Model | Self::ApiKey | Self::Provider | Self::Transport | Self::FailureClass => {
+                "terminal_requests"
+            }
         }
     }
 }
@@ -307,8 +322,9 @@ pub async fn diagnostics(
     dimension: UsageDiagnosticsDimension,
 ) -> PgUsageRecordStoreResult<UsageDiagnosticsInsights> {
     let expression = dimension.expression();
+    let source = dimension.source();
     let failure_filter = if dimension == UsageDiagnosticsDimension::FailureClass {
-        "where not is_success"
+        "where not terminal_requests.is_success"
     } else {
         ""
     };
@@ -321,7 +337,7 @@ pub async fn diagnostics(
            percentile_cont(0.95) within group (order by latency_ms)\n\
              filter (where is_success and latency_ms is not null) as latency_p95_ms,\n\
            (select count(*) from terminal_requests)::bigint as total_requests\n\
-         from terminal_requests {failure_filter}\n\
+         from {source} {failure_filter}\n\
          group by name\n\
          order by request_count desc, name asc\n\
          limit 8"
@@ -610,6 +626,7 @@ async fn load_diagnostic_costs(
     dimension: UsageDiagnosticsDimension,
 ) -> PgUsageRecordStoreResult<HashMap<String, f64>> {
     let expression = dimension.expression();
+    let source = dimension.source();
     let sql = format!(
         "with {TERMINAL_REQUESTS_CTE}\n\
          select {expression} as name, billing_model, service_tier,\n\
@@ -622,8 +639,8 @@ async fn load_diagnostic_costs(
            coalesce(sum(input_tokens), 0)::bigint as input_tokens,\n\
            coalesce(sum(output_tokens), 0)::bigint as output_tokens,\n\
            coalesce(sum(least(coalesce(cached_tokens, 0), coalesce(input_tokens, 0))), 0)::bigint as cached_tokens\n\
-         from terminal_requests\n\
-         where is_success\n\
+         from {source}\n\
+         where terminal_requests.is_success\n\
          group by name, billing_model, service_tier, is_long"
     );
     // SAFETY: the grouping expression comes from the closed diagnostics dimension enum.

backend/src/upstream/openai/transport/websocket_pool.rs

@@ -84,7 +84,7 @@ pub struct CodexWebSocketPoolConfig {
     pub maintenance_interval: Option<Duration>,
     /// idle socket 探活 ping 间隔；`None` 表示维护时只做过期清理。
     pub ping_interval: Option<Duration>,
-    /// 发送 ping 后等待上游响应的超时时间。
+    /// 发送 ping 后等待上游响应的超时时间；零值表示不校验 Pong deadline。
     pub ping_timeout: Duration,
     /// idle socket 无活动多久后视为失活。
     pub liveness_timeout: Option<Duration>,
@@ -110,10 +110,11 @@ impl Default for CodexWebSocketPoolConfig {
 }
 
 impl CodexWebSocketPoolConfig {
-    /// pump 后台任务的保活策略：从连接池配置派生出 ping 间隔与 liveness 超时。
+    /// pump 后台任务的保活策略：从连接池配置派生出 ping/pong 与 liveness 策略。
     pub(crate) fn keepalive(&self) -> PumpKeepalive {
         PumpKeepalive {
             ping_interval: self.ping_interval,
+            ping_timeout: (!self.ping_timeout.is_zero()).then_some(self.ping_timeout),
             liveness_timeout: self.liveness_timeout,
         }
     }

backend/src/upstream/openai/transport/websocket_pump.rs

@@ -2,7 +2,7 @@
 //!
 //! 每条上游 WebSocket 连接都由一个后台 pump 任务独占：
 //!   - 持续读取 socket：一旦观察到 `Close` / EOF / 传输错误，立即把连接标记为 `closed`。
-//!   - 自动回应上游 `Ping`（保活），并按 `ping_interval` 主动 `Ping`（穿透 NAT/中间盒空闲计时器）。
+//!   - 自动回应上游 `Ping`，并按 `ping_interval` 主动 `Ping`；匹配的 `Pong` 必须在 deadline 内返回。
 //!   - 可选 `liveness_timeout`：长时间无任何入站活动时判定连接失活并退出。
 //!
 //! 因此空闲连接的“死没死”在后台被实时感知；复用方只需零成本读取 [`PumpedWebSocket::is_closed`]，
@@ -11,7 +11,7 @@
 //! 收发都通过 channel 与 pump 任务交互：
 //!   - 发送：`send` 走 command channel，等待 pump 回执。
 //!   - 接收：`next` 从 message channel 取出 pump 转发的入站帧（`Ping`/`Pong` 已被 pump 吞掉）。
-//!   - 入站缓冲满时暂停读取 socket，并继续处理关闭命令与保活；消费恢复后按原顺序继续转发。
+//!   - 入站缓冲满时暂停读取 socket 和主动探活，仅继续处理发送/关闭命令；消费恢复后按原顺序继续转发。
 
 use std::sync::{
     Arc, Mutex,
@@ -39,6 +39,8 @@ const PUMP_COMMAND_BUFFER: usize = 32;
 pub(crate) struct PumpKeepalive {
     /// 主动 `Ping` 间隔；`None` 表示 pump 不主动 ping（仅被动读取 + 回应上游 ping）。
     pub(crate) ping_interval: Option<Duration>,
+    /// 主动 `Ping` 发出后等待匹配 `Pong` 的 deadline；`None` 表示不校验响应。
+    pub(crate) ping_timeout: Option<Duration>,
     /// 无入站活动多久后判定失活；`None` 表示只靠显式 close/传输错误发现死亡。
     pub(crate) liveness_timeout: Option<Duration>,
 }
@@ -48,6 +50,7 @@ impl PumpKeepalive {
     pub(crate) fn disabled() -> Self {
         Self {
             ping_interval: None,
+            ping_timeout: None,
             liveness_timeout: None,
         }
     }
@@ -70,6 +73,7 @@ pub(crate) enum PumpExitReason {
     UpstreamCloseFrame { frame: Option<String> },
     UpstreamEof,
     InboundTransportError { message: String },
+    PongTimeout { timeout: Duration },
     LivenessTimeout { timeout: Duration },
     KeepaliveTransportError { message: String },
     MessageReceiverClosed,
@@ -84,6 +88,7 @@ impl PumpExitReason {
             Self::UpstreamCloseFrame { .. } => "upstream_close_frame",
             Self::UpstreamEof => "upstream_eof",
             Self::InboundTransportError { .. } => "inbound_transport_error",
+            Self::PongTimeout { .. } => "pong_timeout",
             Self::LivenessTimeout { .. } => "liveness_timeout",
             Self::KeepaliveTransportError { .. } => "keepalive_transport_error",
             Self::MessageReceiverClosed => "message_receiver_closed",
@@ -96,7 +101,9 @@ impl PumpExitReason {
             | Self::InboundTransportError { message }
             | Self::KeepaliveTransportError { message } => Some(message.clone()),
             Self::UpstreamCloseFrame { frame } => frame.clone(),
-            Self::LivenessTimeout { timeout } => Some(format!("{timeout:?}")),
+            Self::PongTimeout { timeout } | Self::LivenessTimeout { timeout } => {
+                Some(format!("{timeout:?}"))
+            }
             Self::CommandChannelClosed
             | Self::LocalClose
             | Self::UpstreamEof
@@ -110,6 +117,7 @@ impl PumpExitReason {
             Self::OutboundTransportError { .. }
                 | Self::UpstreamEof
                 | Self::InboundTransportError { .. }
+                | Self::PongTimeout { .. }
                 | Self::LivenessTimeout { .. }
                 | Self::KeepaliveTransportError { .. }
         )
@@ -120,6 +128,7 @@ impl PumpExitReason {
             self,
             Self::CommandChannelClosed
                 | Self::UpstreamCloseFrame { .. }
+                | Self::PongTimeout { .. }
                 | Self::LivenessTimeout { .. }
                 | Self::MessageReceiverClosed
         )
@@ -131,6 +140,12 @@ struct PendingInbound {
     exit_reason: Option<PumpExitReason>,
 }
 
+#[derive(Debug, Clone, Copy)]
+struct PendingPing {
+    payload: [u8; 8],
+    deadline: Instant,
+}
+
 /// 由后台 pump 任务托管的 WebSocket 连接句柄。
 pub(crate) struct PumpedWebSocket {
     connection_id: Uuid,
@@ -248,16 +263,9 @@ async fn pump_loop(
 ) {
     let mut last_activity = Instant::now();
     let ping_interval = keepalive.ping_interval.filter(|d| !d.is_zero());
+    let ping_timeout = keepalive.ping_timeout.filter(|d| !d.is_zero());
     let liveness_timeout = keepalive.liveness_timeout.filter(|d| !d.is_zero());
-    // ticker 由 ping 间隔或 liveness 检查间隔中较小者驱动：即使不主动 ping，
-    // 也要周期性醒来检查 liveness。两者都未配置时不启动 ticker。
-    let tick_interval = match (ping_interval, liveness_timeout) {
-        (Some(ping), Some(live)) => Some(ping.min(live)),
-        (Some(ping), None) => Some(ping),
-        (None, Some(live)) => Some(live),
-        (None, None) => None,
-    };
-    let mut ticker = tick_interval.map(|d| {
+    let mut ping_ticker = ping_interval.map(|d| {
         // 首个 tick 推迟一整个间隔：tokio::time::interval 默认会让首个 tick 立即就绪，
         // 否则连接一建立就会立刻发一帧 Ping，与首个请求 Text 抢跑、打乱帧序。
         let mut ticker = tokio::time::interval_at(tokio::time::Instant::now() + d, d);
@@ -266,6 +274,8 @@ async fn pump_loop(
     });
 
     let mut pending_inbound: Option<PendingInbound> = None;
+    let mut pending_ping: Option<PendingPing> = None;
+    let mut ping_sequence = 0_u64;
     let mut backpressure_events = 0_u64;
 
     let reason = 'pump: loop {
@@ -282,6 +292,8 @@ async fn pump_loop(
                     if let Some(reason) = pending.exit_reason {
                         break 'pump reason;
                     }
+                    // 背压期间无法读取 socket，不能把这段本地下游阻塞误算成上游静默。
+                    last_activity = Instant::now();
                 }
                 command = rx_command.recv() => {
                     let Some(command) = command else {
@@ -291,21 +303,12 @@ async fn pump_loop(
                         break 'pump reason;
                     }
                 }
-                _ = tick(&mut ticker) => {
-                    if let Some(reason) = handle_tick(
-                        &mut inner,
-                        last_activity,
-                        ping_interval,
-                        liveness_timeout,
-                        false,
-                    ).await {
-                        break 'pump reason;
-                    }
-                }
             }
             continue;
         }
 
+        let pong_deadline = pending_ping.map(|ping| ping.deadline);
+        let liveness_deadline = liveness_timeout.map(|timeout| last_activity + timeout);
         tokio::select! {
             command = rx_command.recv() => {
                 let Some(command) = command else {
@@ -318,36 +321,38 @@ async fn pump_loop(
             message = inner.next() => {
                 match message {
                     None => break 'pump PumpExitReason::UpstreamEof,
-                    Some(Ok(Message::Ping(payload))) => {
-                        last_activity = Instant::now();
-                        if let Err(error) = inner.send(Message::Pong(payload)).await {
-                            break 'pump PumpExitReason::KeepaliveTransportError {
-                                message: error.to_string(),
-                            };
-                        }
-                    }
-                    Some(Ok(Message::Pong(_))) => {
-                        last_activity = Instant::now();
-                    }
                     Some(Ok(message)) => {
                         last_activity = Instant::now();
-                        let terminal_reason = match &message {
-                            Message::Close(frame) => Some(PumpExitReason::UpstreamCloseFrame {
-                                frame: frame.as_ref().map(|frame| format!("{frame:?}")),
-                            }),
-                            _ => None,
-                        };
-                        if let Some(reason) = enqueue_inbound(
-                            &tx_message,
-                            &mut pending_inbound,
-                            connection_id,
-                            &mut backpressure_events,
-                            PendingInbound {
-                                item: Ok(message),
-                                exit_reason: terminal_reason,
-                            },
-                        ) {
-                            break 'pump reason;
+                        observe_inbound(connection_id, &mut pending_ping, &message);
+                        match message {
+                            Message::Ping(payload) => {
+                                if let Err(error) = inner.send(Message::Pong(payload)).await {
+                                    break 'pump PumpExitReason::KeepaliveTransportError {
+                                        message: error.to_string(),
+                                    };
+                                }
+                            }
+                            Message::Pong(_) => {}
+                            message => {
+                                let terminal_reason = match &message {
+                                    Message::Close(frame) => Some(PumpExitReason::UpstreamCloseFrame {
+                                        frame: frame.as_ref().map(|frame| format!("{frame:?}")),
+                                    }),
+                                    _ => None,
+                                };
+                                if let Some(reason) = enqueue_inbound(
+                                    &tx_message,
+                                    &mut pending_inbound,
+                                    connection_id,
+                                    &mut backpressure_events,
+                                    PendingInbound {
+                                        item: Ok(message),
+                                        exit_reason: terminal_reason,
+                                    },
+                                ) {
+                                    break 'pump reason;
+                                }
+                            }
                         }
                     }
                     Some(Err(err)) => {
@@ -369,17 +374,28 @@ async fn pump_loop(
                     }
                 }
             }
-            _ = tick(&mut ticker) => {
-                if let Some(reason) = handle_tick(
-                    &mut inner,
-                    last_activity,
-                    ping_interval,
-                    liveness_timeout,
-                    true,
-                ).await {
-                    break 'pump reason;
+            _ = tick(&mut ping_ticker) => {
+                if pending_ping.is_none() {
+                    match send_keepalive_ping(
+                        &mut inner,
+                        &mut ping_sequence,
+                        ping_timeout,
+                    ).await {
+                        Ok(pending) => pending_ping = pending,
+                        Err(reason) => break 'pump reason,
+                    }
                 }
             }
+            _ = wait_until(pong_deadline) => {
+                break 'pump PumpExitReason::PongTimeout {
+                    timeout: ping_timeout.expect("pending ping must have a timeout"),
+                };
+            }
+            _ = wait_until(liveness_deadline) => {
+                break 'pump PumpExitReason::LivenessTimeout {
+                    timeout: liveness_timeout.expect("liveness deadline must have a timeout"),
+                };
+            }
         }
     };
 
@@ -433,27 +449,39 @@ async fn handle_command(inner: &mut RawWsStream, command: PumpCommand) -> Option
     }
 }
 
-async fn handle_tick(
+fn observe_inbound(connection_id: Uuid, pending_ping: &mut Option<PendingPing>, message: &Message) {
+    let Some(pending) = pending_ping else {
+        return;
+    };
+    if let Message::Pong(payload) = message
+        && payload.as_ref() != pending.payload
+    {
+        tracing::debug!(
+            websocket_connection_id = %connection_id,
+            "Responses WebSocket pump received a non-matching Pong"
+        );
+    }
+    // 匹配 Pong 是探针的直接回执；其他入站帧同样证明上游链路仍可达。
+    *pending_ping = None;
+}
+
+async fn send_keepalive_ping(
     inner: &mut RawWsStream,
-    last_activity: Instant,
-    ping_interval: Option<Duration>,
-    liveness_timeout: Option<Duration>,
-    check_liveness: bool,
-) -> Option<PumpExitReason> {
-    if check_liveness
-        && let Some(timeout) = liveness_timeout
-        && last_activity.elapsed() >= timeout
-    {
-        return Some(PumpExitReason::LivenessTimeout { timeout });
-    }
-    if ping_interval.is_some()
-        && let Err(error) = inner.send(Message::Ping(Vec::new().into())).await
-    {
-        return Some(PumpExitReason::KeepaliveTransportError {
+    sequence: &mut u64,
+    ping_timeout: Option<Duration>,
+) -> Result<Option<PendingPing>, PumpExitReason> {
+    let payload = sequence.to_be_bytes();
+    *sequence = sequence.wrapping_add(1);
+    inner
+        .send(Message::Ping(payload.to_vec().into()))
+        .await
+        .map_err(|error| PumpExitReason::KeepaliveTransportError {
             message: error.to_string(),
-        });
-    }
-    None
+        })?;
+    Ok(ping_timeout.map(|timeout| PendingPing {
+        payload,
+        deadline: Instant::now() + timeout,
+    }))
 }
 
 fn log_pump_exit(connection_id: Uuid, reason: &PumpExitReason, backpressure_events: u64) {
@@ -494,3 +522,11 @@ async fn tick(ticker: &mut Option<tokio::time::Interval>) {
         None => std::future::pending::<()>().await,
     }
 }
+
+/// 等待可选 deadline；`None` 时永远挂起，让 `select!` 分支实际禁用。
+async fn wait_until(deadline: Option<Instant>) {
+    match deadline {
+        Some(deadline) => tokio::time::sleep_until(deadline).await,
+        None => std::future::pending::<()>().await,
+    }
+}

backend/tests/api/admin/dashboard_routes.rs

@@ -117,7 +117,7 @@ async fn dashboard_summary_should_calculate_today_traffic_and_average_first_toke
         .append(&usage_record_with_tokens(yesterday_record_at, 30))
         .await
         .unwrap();
-    seed_lifetime_usage_summary(&pool, 7, 400, 500, 100).await;
+    seed_retained_usage_bucket(&pool, today_start - Duration::days(8), 4, 340, 500, 100).await;
 
     let body = dashboard_summary(app).await;
 
@@ -441,23 +441,6 @@ async fn dashboard_summary_should_match_versioned_model_names_for_billing() {
     assert_eq!(billing_amount, "$1.75");
 }
 
-#[tokio::test]
-async fn dashboard_summary_should_not_price_usage_summary_without_model_dimension() {
-    let (app, _store, pool, _dir) = dashboard_test_app(
-        "dashboard-billing-summary-fallback",
-        crate::support::fingerprint::test_fingerprint(),
-    )
-    .await;
-    seed_usage_summary(&pool, 100_000, 100_000).await;
-
-    let body = dashboard_summary(app).await;
-
-    assert_eq!(
-        body["data"]["cards"]["tokens"]["todayBillingAmountUsd"],
-        "$0.00"
-    );
-}
-
 #[tokio::test]
 async fn dashboard_summary_should_return_backend_formatted_time_fields() {
     let (app, store, _pool, _dir) = dashboard_test_app(
@@ -812,62 +795,30 @@ async fn dashboard_trend(app: axum::Router, kind: &str) -> Value {
     response_json(response).await
 }
 
-async fn seed_usage_summary(pool: &PgPool, input_tokens: u64, output_tokens: u64) {
-    seed_usage_summary_with_last_used(pool, input_tokens, output_tokens, "2026-06-18T00:00:00Z")
-        .await;
-}
-
-async fn seed_lifetime_usage_summary(
+async fn seed_retained_usage_bucket(
     pool: &PgPool,
+    bucket_start: chrono::DateTime<Utc>,
     request_count: u64,
     input_tokens: u64,
     output_tokens: u64,
     cached_tokens: u64,
 ) {
-    seed_dashboard_account(pool, "acct_lifetime", "lifetime@example.com").await;
     sqlx::query(
-        "insert into account_usage
-         (account_id, request_count, input_tokens, output_tokens, cached_tokens, total_tokens)
-         values ($1, $2, $3, $4, $5, $6)",
+        "insert into request_time_buckets
+         (bucket_start, provider, account_id, model, service_tier, success_count,
+          input_tokens, output_tokens, cached_tokens, updated_at)
+         values ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10)",
     )
-    .bind("acct_lifetime")
+    .bind(china_quarter_hour_start(bucket_start))
+    .bind("openai")
+    .bind("acct_retained")
+    .bind("gpt-5.5")
+    .bind("__unknown__")
     .bind(request_count as i64)
     .bind(input_tokens as i64)
     .bind(output_tokens as i64)
     .bind(cached_tokens as i64)
-    .bind((input_tokens + output_tokens) as i64)
-    .execute(pool)
-    .await
-    .unwrap();
-}
-
-async fn seed_usage_summary_with_last_used(
-    pool: &PgPool,
-    input_tokens: u64,
-    output_tokens: u64,
-    last_used_at: &str,
-) {
-    sqlx::query(
-        "insert into accounts (id, email, access_token, status, added_at, updated_at) values ($1, $2, $3, $4, $5, $6)",
-    )
-    .bind("acct_usage")
-    .bind("acct-usage@example.com")
-    .bind("cipher")
-    .bind("active")
-    .bind(crate::support::storage::timestamp("2026-06-18T00:00:00Z"))
-    .bind(crate::support::storage::timestamp("2026-06-18T00:00:00Z"))
-    .execute(pool)
-    .await
-    .unwrap();
-    sqlx::query(
-        "insert into account_usage (account_id, request_count, input_tokens, output_tokens, total_tokens, last_used_at) values ($1, $2, $3, $4, $5, $6)",
-    )
-    .bind("acct_usage")
-    .bind(1_i64)
-    .bind(input_tokens as i64)
-    .bind(output_tokens as i64)
-    .bind((input_tokens + output_tokens) as i64)
-    .bind(crate::support::storage::timestamp(last_used_at))
+    .bind(Utc::now())
     .execute(pool)
     .await
     .unwrap();

backend/tests/api/admin/usage_routes.rs

@@ -124,6 +124,20 @@ async fn usage_summary_and_trend_expose_only_real_success_metrics() {
 async fn usage_insights_should_collapse_retry_errors_to_terminal_requests() {
     let (app, usage, ops, _guard) = monitoring_test_app("monitoring-insights", true).await;
     let now = chrono::Utc::now();
+    sqlx::query(
+        "insert into accounts
+         (id, email, access_token, status, added_at, updated_at)
+         values ($1, $2, $3, $4, $5, $6)",
+    )
+    .bind("acct_usage")
+    .bind("usage-account@example.com")
+    .bind("cipher")
+    .bind("active")
+    .bind(now)
+    .bind(now)
+    .execute(usage.pool())
+    .await
+    .unwrap();
 
     let mut success = success_record("eventually succeeded");
     success.id = "usage_insights_success".to_string();
@@ -184,6 +198,19 @@ async fn usage_insights_should_collapse_retry_errors_to_terminal_requests() {
             .unwrap()
             .is_empty()
     );
+
+    let account_diagnostics = admin_get(
+        &app,
+        "/api/admin/usage/records/insights/diagnostics?dimension=account",
+    )
+    .await;
+    assert!(
+        account_diagnostics["data"]["items"]
+            .as_array()
+            .unwrap()
+            .iter()
+            .any(|item| item["name"] == "usage-account@example.com → acct_usage")
+    );
 }
 
 #[tokio::test]

backend/tests/upstream/openai/transport/websocket.rs

@@ -1327,21 +1327,33 @@ async fn codex_backend_client_stream_should_preserve_burst_during_downstream_bac
 
     let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
     let addr = listener.local_addr().unwrap();
+    let (terminal_drained_tx, terminal_drained_rx) = tokio::sync::oneshot::channel();
     let server = tokio::spawn(async move {
         let (stream, _) = listener.accept().await.unwrap();
         let mut websocket = accept_codex_test_websocket(stream).await;
         let _message = websocket.next().await.unwrap().unwrap();
-        websocket
-            .send(Message::Text(
-                json!({
-                    "type": "response.output_text.delta",
-                    "delta": "initial;"
-                })
-                .to_string()
-                .into(),
-            ))
-            .await
-            .unwrap();
+        let (mut sink, mut source) = websocket.split();
+        // 服务端持续读取控制帧，让 tungstenite 正常回 Pong。若客户端在本地入站背压
+        // 期间仍执行 Pong deadline，它仍会因为暂时读不到已返回的 Pong 而误杀连接。
+        let control_frames = tokio::spawn(async move {
+            while let Some(message) = source.next().await {
+                match message.unwrap() {
+                    Message::Ping(_) | Message::Pong(_) => {}
+                    Message::Close(_) => break,
+                    other => panic!("unexpected client frame during response: {other:?}"),
+                }
+            }
+        });
+        sink.send(Message::Text(
+            json!({
+                "type": "response.output_text.delta",
+                "delta": "initial;"
+            })
+            .to_string()
+            .into(),
+        ))
+        .await
+        .unwrap();
 
         for index in 0..BURST_FRAMES {
             let frame = json!({
@@ -1349,22 +1361,34 @@ async fn codex_backend_client_stream_should_preserve_burst_during_downstream_bac
                 "delta": format!("chunk-{index};")
             })
             .to_string();
-            if websocket.send(Message::Text(frame.into())).await.is_err() {
+            if sink.send(Message::Text(frame.into())).await.is_err() {
+                control_frames.abort();
                 return;
             }
         }
-        let _ = websocket
+        let _ = sink
             .send(Message::Text(
                 completed_websocket_response("resp_backpressure", 3, 257).into(),
             ))
             .await;
-        let _ = websocket.close(None).await;
+        terminal_drained_rx
+            .await
+            .expect("client should drain the terminal response before server close");
+        let _ = sink.send(Message::Close(None)).await;
+        control_frames.abort();
     });
+    let pool = Arc::new(CodexWebSocketPool::with_config(CodexWebSocketPoolConfig {
+        ping_interval: Some(Duration::from_millis(5)),
+        ping_timeout: Duration::from_millis(20),
+        liveness_timeout: None,
+        ..websocket_pool_config_for_tests(None, None, None)
+    }));
     let backend = CodexBackendClient::new(
         reqwest::Client::builder().no_proxy().build().unwrap(),
         format!("http://{addr}"),
         crate::support::fingerprint::runtime_test_fingerprint(),
-    );
+    )
+    .with_websocket_pool(pool);
     let request = pooled_websocket_request("conversation-backpressure");
 
     let mut response = backend
@@ -1386,6 +1410,7 @@ async fn codex_backend_client_stream_should_preserve_burst_during_downstream_bac
     })
     .await
     .expect("backpressured websocket stream should finish");
+    terminal_drained_tx.send(()).unwrap();
     server.await.unwrap();
 
     assert_eq!(

backend/tests/upstream/openai/transport/websocket_pool.rs

@@ -714,29 +714,48 @@ async fn websocket_pool_should_use_http_for_new_keys_after_account_cap() {
 }
 
 #[tokio::test]
-async fn websocket_pool_should_evict_idle_connection_when_liveness_lapses_despite_pings() {
-    // 不变量：pump 自己发出的 keepalive ping 不算“入站活动”，因此一个只收 ping、
-    // 从不回 pong / 从不发帧的静默连接，仍会被 liveness watchdog 判定失活并驱逐；
-    // 复用前 acquire 读到 is_closed 即开新连接。
+async fn websocket_pool_should_replace_idle_connection_after_pong_deadline() {
+    // 服务端读取 Ping 后暂不继续 poll，避免 tungstenite 自动回 Pong；pump 必须在独立
+    // deadline 到期时主动关闭连接，acquire 随后只读 closed 状态并直接新建连接。
     let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
     let addr = listener.local_addr().unwrap();
     let accepted_connections = Arc::new(AtomicUsize::new(0));
     let accepted_connections_for_server = Arc::clone(&accepted_connections);
-    let (release_first_tx, release_first_rx) = tokio::sync::oneshot::channel();
+    let (ping_seen_tx, ping_seen_rx) = tokio::sync::oneshot::channel();
+    let (inspect_close_tx, inspect_close_rx) = tokio::sync::oneshot::channel();
+    let (pong_timeout_closed_tx, pong_timeout_closed_rx) = tokio::sync::oneshot::channel();
     let server = tokio::spawn(async move {
         let (first_stream, _) = listener.accept().await.unwrap();
         accepted_connections_for_server.fetch_add(1, Ordering::SeqCst);
-        let first_connection = tokio::spawn(async move {
-            let mut first_websocket = accept_codex_test_websocket(first_stream).await;
-            let _first_message = first_websocket.next().await.unwrap().unwrap();
-            first_websocket
-                .send(Message::Text(
-                    completed_websocket_response("resp_no_pong_first", 2, 1).into(),
-                ))
-                .await
-                .unwrap();
-            let _ = release_first_rx.await;
-        });
+        let mut first_websocket = accept_codex_test_websocket(first_stream).await;
+        let _first_message = first_websocket.next().await.unwrap().unwrap();
+        first_websocket
+            .send(Message::Text(
+                completed_websocket_response("resp_no_pong_first", 2, 1).into(),
+            ))
+            .await
+            .unwrap();
+        let ping = timeout(Duration::from_secs(1), first_websocket.next())
+            .await
+            .expect("pump should send a keepalive ping")
+            .expect("keepalive ping should be present")
+            .expect("keepalive ping should be valid");
+        let Message::Ping(payload) = ping else {
+            panic!("expected keepalive ping, got {ping:?}");
+        };
+        assert_eq!(payload.len(), 8);
+        ping_seen_tx.send(()).unwrap();
+
+        inspect_close_rx
+            .await
+            .expect("test should release the server after the Pong deadline");
+        let close = timeout(Duration::from_secs(1), first_websocket.next())
+            .await
+            .expect("Pong deadline should close the idle websocket")
+            .expect("Pong deadline should send a close frame")
+            .expect("close frame should be valid");
+        std::assert_matches!(close, Message::Close(_));
+        pong_timeout_closed_tx.send(()).unwrap();
 
         let (second_stream, _) = listener.accept().await.unwrap();
         accepted_connections_for_server.fetch_add(1, Ordering::SeqCst);
@@ -749,11 +768,11 @@ async fn websocket_pool_should_evict_idle_connection_when_liveness_lapses_despit
             .await
             .unwrap();
         second_websocket.close(None).await.unwrap();
-        first_connection.await.unwrap();
     });
     let pool = Arc::new(CodexWebSocketPool::with_config(CodexWebSocketPoolConfig {
-        ping_interval: Some(Duration::from_millis(1)),
-        liveness_timeout: Some(Duration::from_millis(20)),
+        ping_interval: Some(Duration::from_millis(10)),
+        ping_timeout: Duration::from_millis(30),
+        liveness_timeout: None,
         maintenance_interval: None,
         ..websocket_pool_config_for_tests(None, None, None)
     }));
@@ -772,12 +791,16 @@ async fn websocket_pool_should_evict_idle_connection_when_liveness_lapses_despit
         )
         .await
         .expect("first websocket response should succeed");
-    tokio::time::pause();
-    tokio::time::advance(Duration::from_millis(25)).await;
-    tokio::task::yield_now().await;
-    // 只加速首条 idle 连接的失活窗口；恢复实时时钟，避免新连接在发送
-    // response.create 前被测试时钟自动推进到下一次 liveness timeout。
-    tokio::time::resume();
+    timeout(Duration::from_secs(1), ping_seen_rx)
+        .await
+        .expect("server should observe the keepalive ping")
+        .expect("server should report the keepalive ping");
+    tokio::time::sleep(Duration::from_millis(80)).await;
+    inspect_close_tx.send(()).unwrap();
+    timeout(Duration::from_secs(1), pong_timeout_closed_rx)
+        .await
+        .expect("server should observe the Pong-timeout close")
+        .expect("server should report the Pong-timeout close");
     let second = backend
         .create_response(
             &request,
@@ -785,7 +808,6 @@ async fn websocket_pool_should_evict_idle_connection_when_liveness_lapses_despit
         )
         .await
         .expect("second websocket response should use a fresh connection");
-    release_first_tx.send(()).unwrap();
     server.await.unwrap();
 
     assert!(first.body.contains("resp_no_pong_first"));
@@ -901,6 +923,7 @@ async fn codex_backend_client_should_keep_idle_pooled_websocket_alive_across_rep
 
         // pump 会在 idle 期间反复发送 keepalive ping；服务端计数并回 pong，
         // 直到下一个业务请求（response.create）到达为止。
+        let mut previous_ping = None;
         loop {
             let message = timeout(Duration::from_secs(1), websocket.next())
                 .await
@@ -909,6 +932,11 @@ async fn codex_backend_client_should_keep_idle_pooled_websocket_alive_across_rep
                 .expect("frame should be valid");
             match message {
                 Message::Ping(payload) => {
+                    assert_eq!(payload.len(), 8);
+                    if let Some(previous) = &previous_ping {
+                        assert_ne!(previous, &payload, "keepalive ping sequence must be unique");
+                    }
+                    previous_ping = Some(payload.clone());
                     ping_count_for_server.fetch_add(1, Ordering::SeqCst);
                     websocket.send(Message::Pong(payload)).await.unwrap();
                 }
@@ -978,7 +1006,66 @@ async fn codex_backend_client_should_keep_idle_pooled_websocket_alive_across_rep
 
     assert!(first.body.contains("resp_pool_background_first"));
     assert!(second.body.contains("resp_pool_background_second"));
-    assert!(ping_count.load(Ordering::SeqCst) >= 1);
+    assert!(ping_count.load(Ordering::SeqCst) >= 2);
+}
+
+#[tokio::test]
+async fn codex_backend_client_should_treat_active_business_frames_as_ping_liveness() {
+    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let addr = listener.local_addr().unwrap();
+    let server = tokio::spawn(async move {
+        let (stream, _) = listener.accept().await.unwrap();
+        let mut websocket = accept_codex_test_websocket(stream).await;
+        let _request = websocket.next().await.unwrap().unwrap();
+
+        for index in 0..8 {
+            tokio::time::sleep(Duration::from_millis(20)).await;
+            websocket
+                .send(Message::Text(
+                    json!({
+                        "type": "response.output_text.delta",
+                        "delta": format!("active-{index};")
+                    })
+                    .to_string()
+                    .into(),
+                ))
+                .await
+                .unwrap();
+        }
+        websocket
+            .send(Message::Text(
+                completed_websocket_response("resp_active_without_pong", 3, 8).into(),
+            ))
+            .await
+            .unwrap();
+        websocket.close(None).await.unwrap();
+    });
+    let pool = Arc::new(CodexWebSocketPool::with_config(CodexWebSocketPoolConfig {
+        ping_interval: Some(Duration::from_millis(10)),
+        ping_timeout: Duration::from_millis(60),
+        liveness_timeout: None,
+        ..websocket_pool_config_for_tests(None, None, None)
+    }));
+    let backend = CodexBackendClient::new(
+        reqwest::Client::builder().no_proxy().build().unwrap(),
+        format!("http://{addr}"),
+        crate::support::fingerprint::runtime_test_fingerprint(),
+    )
+    .with_websocket_pool(Arc::clone(&pool));
+    let request = pooled_websocket_request("conversation-active-without-pong");
+
+    let response = backend
+        .create_response(
+            &request,
+            request_context("req_active_without_pong", Some("chatgpt-account")),
+        )
+        .await
+        .expect("active business frames should satisfy the Ping liveness probe");
+    server.await.unwrap();
+
+    assert!(response.body.contains("active-0;"));
+    assert!(response.body.contains("active-7;"));
+    assert!(response.body.contains("resp_active_without_pong"));
 }
 
 #[tokio::test]

docs/architecture.md

@@ -277,7 +277,7 @@ Pool key 为 `(base_url, account_id, local_conversation_id)`，slot 状态为：
 
 每条连接记录最新 completed response ID。Exact continuation 只有 ID 完全匹配时才能租用；socket 缺失、busy、失活或 ID 不匹配时不新建连接碰运气，由 HistoryController 决定 full replay 或失败。
 
-池化 socket 默认 25 秒 ping、5 秒探活、55 分钟最大寿命；后台 pump 处理 ping/pong、EOF、close 和 liveness，复用前只读取连接状态。
+池化 socket 默认每 25 秒发送带唯一序列的 Ping，匹配 Pong 的 deadline 为 5 秒，最大寿命为 55 分钟。任意业务入站帧同样证明链路存活；下游背压导致 pump 无法读取 socket 时暂停主动探活。后台 pump 统一处理 ping/pong、EOF、close 和 liveness，复用前只读取原子连接状态，不增加网络往返。
 
 ### Origin breaker
 

frontend/src/views/accounts/components/AccountIdentityCell.vue

@@ -61,9 +61,9 @@ const avatarClass = computed(() => {
     >
       {{ initial }}
     </span>
-    <div class="min-w-0">
+    <div class="min-w-0 flex-1">
       <div class="flex min-w-0 items-center gap-2">
-        <span class="truncate text-[14px] font-[760] text-(--cp-text-primary)">
+        <span class="min-w-0 flex-1 truncate text-[14px] font-[760] text-(--cp-text-primary)">
           {{ displayTitle }}
         </span>
         <AccountPlanBadge v-if="showPlan" :plan-type="account.planType" size="sm" />

frontend/src/views/accounts/components/AccountPlanBadge.vue

@@ -23,8 +23,8 @@ const label = computed(() => props.planType?.trim() || 'Free')
 
 const sizeClass = computed(() =>
   props.size === 'sm'
-    ? 'h-5 max-w-24 rounded-full px-1.75 text-[11px] font-[720]'
-    : 'h-5.5 max-w-full rounded-full px-2 text-[11px] font-[760]',
+    ? 'h-5 rounded-full px-1.75 text-[11px] font-[720]'
+    : 'h-5.5 rounded-full px-2 text-[11px] font-[760]',
 )
 
 const paletteClass = computed(() => {
@@ -39,9 +39,9 @@ const paletteClass = computed(() => {
 
 <template>
   <span
-    class="inline-flex min-w-0 items-center justify-center leading-none capitalize"
+    class="inline-flex shrink-0 items-center justify-center whitespace-nowrap leading-none capitalize"
     :class="[sizeClass, paletteClass]"
   >
-    <span class="min-w-0 truncate">{{ label }}</span>
+    <span>{{ label }}</span>
   </span>
 </template>

frontend/src/views/usage/components/UsageDiagnosticCard.vue

@@ -80,7 +80,7 @@ const tableRows = computed(() => (isStale.value ? [] : displayItems.value))
 
 function diagnosticNameDisplay(name: string) {
   const full = name.trim() || '未知'
-  if (dimension.value !== 'model') {
+  if (dimension.value !== 'model' && dimension.value !== 'account') {
     return { primary: full, secondary: '', full }
   }
 

frontend/src/views/usage/composables/useUsageRecordsTable.ts

@@ -1,5 +1,6 @@
 import { clamp } from 'es-toolkit'
 import { onMounted, shallowRef, watch } from 'vue'
+import type { Ref } from 'vue'
 
 import {
   getUsageRecordInsightsDiagnostics,
@@ -10,7 +11,20 @@ import {
 import { toast } from '@/components/base/BaseToast'
 import { withMinimumDuration } from '@/utils/async'
 
-export function useUsageRecordsTable(options: any) {
+import type { UsageTimeRangeParams } from './useUsageTimeRange'
+
+interface UseUsageRecordsTableOptions {
+  page: Ref<number>
+  pageSize: Ref<number>
+  searchQuery: Ref<string>
+  timeRangeParams: Readonly<Ref<UsageTimeRangeParams>>
+  totalRecords: Ref<number>
+  refreshTimeRangeEnd: () => void
+}
+
+type UsageLoadScope = 'all' | 'table'
+
+export function useUsageRecordsTable(options: UseUsageRecordsTableOptions) {
   const loading = shallowRef(true)
   const analyticsLoading = shallowRef(true)
   const records = shallowRef<any[]>([])
@@ -26,7 +40,7 @@ export function useUsageRecordsTable(options: any) {
     search: options.searchQuery.value || undefined,
   })
 
-  async function loadUsageRecords(scope = 'all') {
+  async function loadUsageRecords(scope: UsageLoadScope = 'all') {
     const requestId = ++loadRequestId
     try {
       loading.value = true
@@ -135,7 +149,8 @@ export function useUsageRecordsTable(options: any) {
     if (refreshingList.value || loading.value) return
     refreshingList.value = true
     try {
-      await withMinimumDuration(() => loadUsageRecords('table'))
+      options.refreshTimeRangeEnd()
+      await withMinimumDuration(() => loadUsageRecords('all'))
     } finally {
       refreshingList.value = false
     }

frontend/src/views/usage/composables/useUsageTimeRange.ts

@@ -0,0 +1,40 @@
+import dayjs, { type Dayjs } from 'dayjs'
+import { computed, shallowRef } from 'vue'
+
+export type UsageTimeRange = 'today' | '7d' | '30d'
+
+export interface UsageTimeRangeParams extends Record<string, string> {
+  startTime: string
+  endTime: string
+}
+
+export function useUsageTimeRange(initialRange: UsageTimeRange = 'today') {
+  const timeRange = shallowRef<UsageTimeRange>(initialRange)
+  const rangeEnd = shallowRef(dayjs())
+  const timeRangeParams = computed(() => buildTimeRange(timeRange.value, rangeEnd.value))
+
+  function refreshTimeRangeEnd() {
+    rangeEnd.value = dayjs()
+  }
+
+  return {
+    timeRange,
+    timeRangeParams,
+    refreshTimeRangeEnd,
+  }
+}
+
+function buildTimeRange(range: UsageTimeRange, end: Dayjs): UsageTimeRangeParams {
+  if (range === 'today') {
+    return {
+      startTime: end.startOf('day').toISOString(),
+      endTime: end.toISOString(),
+    }
+  }
+
+  const days = range === '30d' ? 29 : 6
+  return {
+    startTime: end.subtract(days, 'day').startOf('day').toISOString(),
+    endTime: end.toISOString(),
+  }
+}

frontend/src/views/usage/index.vue

@@ -1,7 +1,6 @@
 <script setup lang="ts">
 import { CalendarDays, Eye, Minimize2 } from '@lucide/vue'
-import dayjs from 'dayjs'
-import { computed, shallowRef, watch } from 'vue'
+import { shallowRef, watch } from 'vue'
 
 import BaseButton from '@/components/base/BaseButton.vue'
 import BaseCard from '@/components/base/BaseCard.vue'
@@ -19,6 +18,7 @@ import {
 import { useUsageRecordDetail } from './composables/useUsageRecordDetail'
 import { useUsageFilters } from './composables/useUsageFilters'
 import { useUsageRecordsTable } from './composables/useUsageRecordsTable'
+import { useUsageTimeRange } from './composables/useUsageTimeRange'
 import UsageClientIpCell from './components/UsageClientIpCell.vue'
 import UsageBillingCell from './components/UsageBillingCell.vue'
 import UsageFilters from './components/UsageFilters.vue'
@@ -31,34 +31,12 @@ import UsageSummaryCards from './components/UsageSummaryCards.vue'
 import UsageTokenCell from './components/UsageTokenCell.vue'
 
 const totalRecords = shallowRef(0)
-const timeRange = shallowRef('today')
 const recordView = shallowRef('success')
 const recordViewOptions = [
   { label: '成功记录', value: 'success' },
   { label: '错误排查', value: 'errors' },
 ]
-const timeRangeParams = computed(() => {
-  const now = dayjs()
-
-  if (timeRange.value === 'today') {
-    return {
-      startTime: now.startOf('day').toISOString(),
-      endTime: now.toISOString(),
-    }
-  }
-
-  if (timeRange.value === '30d') {
-    return {
-      startTime: now.subtract(29, 'day').startOf('day').toISOString(),
-      endTime: now.toISOString(),
-    }
-  }
-
-  return {
-    startTime: now.subtract(6, 'day').startOf('day').toISOString(),
-    endTime: now.toISOString(),
-  }
-})
+const { timeRange, timeRangeParams, refreshTimeRangeEnd } = useUsageTimeRange()
 
 const {
   page,
@@ -87,6 +65,7 @@ const {
   searchQuery,
   timeRangeParams,
   totalRecords,
+  refreshTimeRangeEnd,
 })
 
 const { showDetailModal, selectedUsageRecord, handleViewDetail } = useUsageRecordDetail({
@@ -96,6 +75,7 @@ const { showDetailModal, selectedUsageRecord, handleViewDetail } = useUsageRecor
 bindUsageRecordLoader(loadUsageRecords)
 
 watch(timeRange, () => {
+  refreshTimeRangeEnd()
   page.value = 1
   void loadUsageRecords('all')
 })

release/version.yaml

@@ -1 +1 @@
-version: "2.3.0"
+version: "2.3.1"