fix: 修正用量失败状态判断与 Codex 会话头

apps/aether-gateway/src/handlers/admin/observability/monitoring/usage_helpers.rs

@@ -1,9 +1,7 @@
+use aether_admin::observability::usage::admin_usage_is_failed;
 use aether_data_contracts::repository::usage::StoredRequestUsageAudit;
 
 pub(super) fn admin_monitoring_usage_is_error(item: &StoredRequestUsageAudit) -> bool {
-    item.status_code.is_some_and(|value| value >= 400)
-        || item.status.trim().eq_ignore_ascii_case("failed")
-        || item.status.trim().eq_ignore_ascii_case("error")
-        || item.error_message.is_some()
+    item.status.trim().eq_ignore_ascii_case("error") || admin_usage_is_failed(item)
         || item.error_category.is_some()
 }

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

@@ -181,6 +181,18 @@ pub fn admin_usage_matches_api_format(
         .is_some_and(|value| value.eq_ignore_ascii_case(api_format))
 }
 
+pub fn admin_usage_is_failed(item: &StoredRequestUsageAudit) -> bool {
+    let status = item.status.trim();
+    if !status.is_empty() {
+        return status.eq_ignore_ascii_case("failed");
+    }
+    item.status_code.is_some_and(|value| value >= 400)
+        || item
+            .error_message
+            .as_deref()
+            .is_some_and(|value| !value.trim().is_empty())
+}
+
 pub fn admin_usage_matches_status(item: &StoredRequestUsageAudit, status: Option<&str>) -> bool {
     let Some(status) = status.map(str::trim).filter(|value| !value.is_empty()) else {
         return true;
@@ -192,11 +204,7 @@ pub fn admin_usage_matches_status(item: &StoredRequestUsageAudit, status: Option
             item.status_code.is_some_and(|value| value >= 400) || item.error_message.is_some()
         }
         "pending" | "streaming" | "completed" | "cancelled" => item.status == status,
-        "failed" => {
-            item.status == "failed"
-                || item.status_code.is_some_and(|value| value >= 400)
-                || item.error_message.is_some()
-        }
+        "failed" => admin_usage_is_failed(item),
         "active" => matches!(item.status.as_str(), "pending" | "streaming"),
         _ => true,
     }
@@ -227,6 +235,76 @@ pub fn admin_usage_provider_key_name(
         .or_else(|| admin_usage_request_metadata_string(item, "key_name"))
 }
 
+#[cfg(test)]
+mod tests {
+    use super::{admin_usage_is_failed, admin_usage_matches_status};
+    use aether_data_contracts::repository::usage::StoredRequestUsageAudit;
+
+    fn sample_usage(
+        status: &str,
+        status_code: Option<i32>,
+        error_message: Option<&str>,
+    ) -> StoredRequestUsageAudit {
+        StoredRequestUsageAudit::new(
+            "usage-1".to_string(),
+            "req-1".to_string(),
+            Some("user-1".to_string()),
+            Some("api-key-1".to_string()),
+            Some("alice".to_string()),
+            Some("default".to_string()),
+            "OpenAI".to_string(),
+            "gpt-5".to_string(),
+            None,
+            Some("provider-1".to_string()),
+            Some("endpoint-1".to_string()),
+            Some("provider-key-1".to_string()),
+            Some("chat".to_string()),
+            Some("openai:chat".to_string()),
+            Some("openai".to_string()),
+            Some("chat".to_string()),
+            Some("openai:chat".to_string()),
+            Some("openai".to_string()),
+            Some("chat".to_string()),
+            false,
+            false,
+            10,
+            20,
+            30,
+            0.0,
+            0.0,
+            status_code,
+            error_message.map(ToOwned::to_owned),
+            None,
+            Some(120),
+            None,
+            status.to_string(),
+            "settled".to_string(),
+            100,
+            101,
+            Some(102),
+        )
+        .expect("usage should build")
+    }
+
+    #[test]
+    fn explicit_completed_status_wins_over_legacy_failure_fields() {
+        let item = sample_usage("completed", Some(429), Some("rate limited on first attempt"));
+        assert!(!admin_usage_is_failed(&item));
+        assert!(!admin_usage_matches_status(&item, Some("failed")));
+        assert!(admin_usage_matches_status(&item, Some("completed")));
+    }
+
+    #[test]
+    fn legacy_failure_signals_still_work_when_status_is_missing() {
+        let item = StoredRequestUsageAudit {
+            status: String::new(),
+            ..sample_usage("completed", Some(429), Some("rate limited"))
+        };
+        assert!(admin_usage_is_failed(&item));
+        assert!(admin_usage_matches_status(&item, Some("failed")));
+    }
+}
+
 pub fn admin_usage_record_json(
     item: &StoredRequestUsageAudit,
     users_by_id: &BTreeMap<String, StoredUserSummary>,

crates/aether-ai-pipeline/src/planner/standard/codex.rs

@@ -181,41 +181,13 @@ pub fn apply_codex_openai_cli_special_headers(
         return;
     }
 
-    if let Some(account_id) = extract_codex_account_id(decrypted_auth_config_raw) {
-        provider_request_headers.insert("chatgpt-account-id".to_string(), account_id);
-    }
-
-    if !provider_request_headers
-        .get("x-client-request-id")
-        .map(|value| !value.trim().is_empty())
-        .unwrap_or(false)
-    {
-        if let Some(request_id) = request_id.map(str::trim).filter(|value| !value.is_empty()) {
-            provider_request_headers
-                .insert("x-client-request-id".to_string(), request_id.to_string());
-        }
-    }
-
     let prompt_cache_key = provider_request_body
         .get("prompt_cache_key")
         .and_then(Value::as_str)
         .map(str::trim)
         .filter(|value| !value.is_empty());
 
-    let Some(short_id) = prompt_cache_key.and_then(build_short_codex_header_id) else {
-        return;
-    };
-
     if !header_map_has_non_empty_value(original_headers, "session_id") {
-        provider_request_headers.insert("session_id".to_string(), short_id.clone());
-    }
-
-    if !provider_api_format
-        .trim()
-        .eq_ignore_ascii_case("openai:compact")
-        && !header_map_has_non_empty_value(original_headers, "conversation_id")
-    {
-        let session_id = provider_request_headers.get("session_id").unwrap();
-        provider_request_headers.insert("conversation_id".to_string(), session_id.clone());
+        provider_request_headers.insert("session_id".to_string(), prompt_cache_key.unwrap().to_string());
     }
 }

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

@@ -415,9 +415,18 @@ DO UPDATE SET
   output_price_per_1m = CASE WHEN "usage".billing_status = 'pending' THEN COALESCE(EXCLUDED.output_price_per_1m, "usage".output_price_per_1m) ELSE "usage".output_price_per_1m END,
   total_cost_usd = CASE WHEN "usage".billing_status = 'pending' THEN COALESCE(EXCLUDED.total_cost_usd, "usage".total_cost_usd) ELSE "usage".total_cost_usd END,
   actual_total_cost_usd = CASE WHEN "usage".billing_status = 'pending' THEN COALESCE(EXCLUDED.actual_total_cost_usd, "usage".actual_total_cost_usd) ELSE "usage".actual_total_cost_usd END,
-  status_code = CASE WHEN "usage".billing_status = 'pending' THEN COALESCE(EXCLUDED.status_code, "usage".status_code) ELSE "usage".status_code END,
-  error_message = CASE WHEN "usage".billing_status = 'pending' THEN COALESCE(EXCLUDED.error_message, "usage".error_message) ELSE "usage".error_message END,
-  error_category = CASE WHEN "usage".billing_status = 'pending' THEN COALESCE(EXCLUDED.error_category, "usage".error_category) ELSE "usage".error_category END,
+  status_code = CASE WHEN "usage".billing_status = 'pending' THEN CASE
+    WHEN EXCLUDED.status IN ('pending', 'streaming', 'completed', 'cancelled') AND EXCLUDED.status_code IS NULL THEN NULL
+    ELSE COALESCE(EXCLUDED.status_code, "usage".status_code)
+  END ELSE "usage".status_code END,
+  error_message = CASE WHEN "usage".billing_status = 'pending' THEN CASE
+    WHEN EXCLUDED.status IN ('pending', 'streaming', 'completed', 'cancelled') THEN EXCLUDED.error_message
+    ELSE COALESCE(EXCLUDED.error_message, "usage".error_message)
+  END ELSE "usage".error_message END,
+  error_category = CASE WHEN "usage".billing_status = 'pending' THEN CASE
+    WHEN EXCLUDED.status IN ('pending', 'streaming', 'completed', 'cancelled') THEN EXCLUDED.error_category
+    ELSE COALESCE(EXCLUDED.error_category, "usage".error_category)
+  END ELSE "usage".error_category END,
   response_time_ms = CASE WHEN "usage".billing_status = 'pending' THEN COALESCE(EXCLUDED.response_time_ms, "usage".response_time_ms) ELSE "usage".response_time_ms END,
   first_byte_time_ms = CASE WHEN "usage".billing_status = 'pending' THEN COALESCE(EXCLUDED.first_byte_time_ms, "usage".first_byte_time_ms) ELSE "usage".first_byte_time_ms END,
   status = CASE WHEN "usage".billing_status = 'pending' THEN EXCLUDED.status ELSE "usage".status END,
@@ -1074,4 +1083,17 @@ mod tests {
         assert!(super::UPSERT_SQL.contains("WHEN $50 IS NULL THEN NULL"));
         assert!(super::UPSERT_SQL.contains("TO_TIMESTAMP($51::double precision)"));
     }
+
+    #[test]
+    fn usage_sql_clears_stale_failure_fields_for_non_failed_status_updates() {
+        assert!(super::UPSERT_SQL.contains(
+            "WHEN EXCLUDED.status IN ('pending', 'streaming', 'completed', 'cancelled') AND EXCLUDED.status_code IS NULL THEN NULL"
+        ));
+        assert!(super::UPSERT_SQL.contains(
+            "WHEN EXCLUDED.status IN ('pending', 'streaming', 'completed', 'cancelled') THEN EXCLUDED.error_message"
+        ));
+        assert!(super::UPSERT_SQL.contains(
+            "WHEN EXCLUDED.status IN ('pending', 'streaming', 'completed', 'cancelled') THEN EXCLUDED.error_category"
+        ));
+    }
 }

frontend/src/features/usage/components/UsageRecordsTable.vue

@@ -205,7 +205,7 @@
           <div class="flex items-center gap-1.5">
             <!-- 状态 Badge -->
             <Badge
-              v-if="record.status === 'failed' || (record.status_code && record.status_code >= 400) || record.error_message"
+              v-if="isUsageRecordFailed(record)"
               variant="destructive"
               class="whitespace-nowrap text-[10px] px-1.5 h-4 leading-4 inline-flex items-center"
             >
@@ -521,7 +521,7 @@
               传输中
             </Badge>
             <Badge
-              v-else-if="record.status === 'failed' || (record.status_code && record.status_code >= 400) || record.error_message"
+              v-else-if="isUsageRecordFailed(record)"
               variant="destructive"
               class="whitespace-nowrap"
             >
@@ -678,6 +678,7 @@ import { RefreshCcw, Search } from 'lucide-vue-next'
 import { formatTokens, formatCurrency } from '@/utils/format'
 import { formatDateTime } from '../composables'
 import { getEffectiveInputTokens } from '../token-normalization'
+import { isUsageRecordFailed } from '../utils/status'
 import { useRowClick } from '@/composables/useRowClick'
 import { formatApiFormat } from '@/api/endpoints/types/api-format'
 import type { DateRangeParams, UsageRecord } from '../types'

frontend/src/features/usage/composables/useUsageFilters.ts

@@ -1,5 +1,6 @@
 import { ref, computed, type Ref } from 'vue'
 import type { UsageRecord, FilterStatusValue } from '../types'
+import { isUsageRecordFailed } from '../utils/status'
 
 export interface UseUsageFiltersOptions {
   /** 所有记录的响应式引用 */
@@ -64,22 +65,18 @@ export function useUsageFilters(options: UseUsageFiltersOptions) {
     if (filterStatus.value !== '__all__') {
       if (filterStatus.value === 'stream') {
         records = records.filter(record =>
-          record.is_stream && !record.error_message && (!record.status_code || record.status_code === 200)
+          record.is_stream && !isUsageRecordFailed(record)
         )
       } else if (filterStatus.value === 'standard') {
         records = records.filter(record =>
-          !record.is_stream && !record.error_message && (!record.status_code || record.status_code === 200)
+          !record.is_stream && !isUsageRecordFailed(record)
         )
       } else if (filterStatus.value === 'active') {
         records = records.filter(record =>
           record.status === 'pending' || record.status === 'streaming'
         )
       } else if (filterStatus.value === 'failed') {
-        records = records.filter(record =>
-          record.status === 'failed' ||
-          (record.status_code && record.status_code >= 400) ||
-          record.error_message
-        )
+        records = records.filter(record => isUsageRecordFailed(record))
       } else if (filterStatus.value === 'cancelled') {
         records = records.filter(record => record.status === 'cancelled')
       }

frontend/src/features/usage/utils/__tests__/status.spec.ts

@@ -0,0 +1,43 @@
+import { describe, expect, it } from 'vitest'
+
+import { isUsageRecordFailed, isUsageRecordSuccessful } from '../status'
+import type { UsageRecord } from '../../types'
+
+function buildUsageRecord(overrides: Partial<UsageRecord> = {}): UsageRecord {
+  return {
+    id: 'usage-1',
+    model: 'gpt-5',
+    input_tokens: 10,
+    output_tokens: 20,
+    total_tokens: 30,
+    cost: 0,
+    is_stream: false,
+    created_at: '2026-04-10T00:00:00Z',
+    status: 'completed',
+    ...overrides
+  }
+}
+
+describe('usage status helpers', () => {
+  it('treats explicit completed status as authoritative over stale legacy failure fields', () => {
+    const record = buildUsageRecord({
+      status: 'completed',
+      status_code: 429,
+      error_message: 'rate limited on first attempt'
+    })
+
+    expect(isUsageRecordFailed(record)).toBe(false)
+    expect(isUsageRecordSuccessful(record)).toBe(true)
+  })
+
+  it('falls back to legacy failure signals when status is missing', () => {
+    const record = buildUsageRecord({
+      status: undefined,
+      status_code: 429,
+      error_message: 'rate limited'
+    })
+
+    expect(isUsageRecordFailed(record)).toBe(true)
+    expect(isUsageRecordSuccessful(record)).toBe(false)
+  })
+})

frontend/src/features/usage/utils/status.ts

@@ -0,0 +1,28 @@
+import type { UsageRecord } from '../types'
+
+function hasLegacyFailureSignal(
+  record: Pick<UsageRecord, 'status_code' | 'error_message'>
+): boolean {
+  return (typeof record.status_code === 'number' && record.status_code >= 400) ||
+    (typeof record.error_message === 'string' && record.error_message.trim().length > 0)
+}
+
+export function isUsageRecordFailed(
+  record: Pick<UsageRecord, 'status' | 'status_code' | 'error_message'>
+): boolean {
+  const status = typeof record.status === 'string' ? record.status.trim().toLowerCase() : ''
+  if (status) {
+    return status === 'failed'
+  }
+  return hasLegacyFailureSignal(record)
+}
+
+export function isUsageRecordSuccessful(
+  record: Pick<UsageRecord, 'status' | 'status_code' | 'error_message'>
+): boolean {
+  const status = typeof record.status === 'string' ? record.status.trim().toLowerCase() : ''
+  if (status) {
+    return status === 'completed'
+  }
+  return !hasLegacyFailureSignal(record)
+}

frontend/src/views/shared/Usage.vue

@@ -144,6 +144,7 @@ import {
   getDateRangeFromPeriod
 } from '@/features/usage/composables'
 import { reconcileActiveRequestDiscovery } from '@/features/usage/utils/activeRequestDiscovery'
+import { isUsageRecordFailed } from '@/features/usage/utils/status'
 import type { DateRangeParams, FilterStatusValue } from '@/features/usage/types'
 import type { UserOption } from '@/features/usage/components/UsageRecordsTable.vue'
 import { log } from '@/utils/logger'
@@ -294,25 +295,18 @@ const filteredRecords = computed(() => {
     if (filterStatus.value !== '__all__') {
       if (filterStatus.value === 'stream') {
         records = records.filter(record =>
-          record.is_stream && !record.error_message && (!record.status_code || record.status_code === 200)
+          record.is_stream && !isUsageRecordFailed(record)
         )
       } else if (filterStatus.value === 'standard') {
         records = records.filter(record =>
-          !record.is_stream && !record.error_message && (!record.status_code || record.status_code === 200)
+          !record.is_stream && !isUsageRecordFailed(record)
         )
       } else if (filterStatus.value === 'active') {
         records = records.filter(record =>
           record.status === 'pending' || record.status === 'streaming'
         )
       } else if (filterStatus.value === 'failed') {
-        // 失败请求需要同时考虑新旧两种判断方式：
-        // 1. 新方式：status = "failed"
-        // 2. 旧方式：status_code >= 400 或 error_message 不为空
-        records = records.filter(record =>
-          record.status === 'failed' ||
-          (record.status_code && record.status_code >= 400) ||
-          record.error_message
-        )
+        records = records.filter(record => isUsageRecordFailed(record))
       } else if (filterStatus.value === 'cancelled') {
         records = records.filter(record => record.status === 'cancelled')
       }
@@ -530,10 +524,10 @@ function stopActiveDiscovery() {
   }
 }
 
-// 监听活跃请求状态，自动启动/停止刷新
-// 活跃请求的 1 秒轮询受“自动刷新”开关控制
+// 监听活跃请求状态，已显示的活跃行始终自刷新到终态。
+// “自动刷新”开关只控制全局 3 秒刷新和新活跃请求发现。
 watch(hasActiveRequests, (hasActive) => {
-  if (globalAutoRefresh.value && hasActive && isPageVisible.value) {
+  if (hasActive && isPageVisible.value) {
     startAutoRefresh()
   } else {
     stopAutoRefresh()
@@ -568,7 +562,6 @@ function handleAutoRefreshChange(value: boolean) {
     }
     startGlobalAutoRefresh()
   } else {
-    stopAutoRefresh()
     stopActiveDiscovery()
     stopGlobalAutoRefresh()
   }
@@ -582,11 +575,11 @@ function handleVisibilityChange() {
     stopGlobalAutoRefresh()
     return
   }
+  if (hasActiveRequests.value) {
+    startAutoRefresh()
+  }
   if (globalAutoRefresh.value) {
     startActiveDiscovery()
-    if (hasActiveRequests.value) {
-      startAutoRefresh()
-    }
     refreshData()
     startGlobalAutoRefresh()
   }