From a4f18438767438126cd51f6ab5df668f784e2b0b Mon Sep 17 00:00:00 2001 From: zyycn Date: Wed, 15 Jul 2026 10:15:10 +0800 Subject: [PATCH] fix: unify account failures and usage diagnostics Co-authored-by: Codex --- backend/src/bootstrap/tasks/coordinator.rs | 2 +- backend/src/bootstrap/tasks/quota_refresh.rs | 8 +- .../dispatch/controllers/account_failure.rs | 216 +++--------- .../src/dispatch/controllers/account_state.rs | 38 --- .../src/dispatch/controllers/cloudflare.rs | 2 +- .../dispatch/controllers/cyber_policy/mod.rs | 6 +- backend/src/dispatch/controllers/history.rs | 2 +- backend/src/dispatch/controllers/mod.rs | 7 +- backend/src/dispatch/controllers/quota.rs | 148 ++------ .../dispatch/controllers/telemetry/events.rs | 16 + backend/src/dispatch/lifecycle/attempt.rs | 2 +- backend/src/dispatch/lifecycle/contract.rs | 6 +- backend/src/dispatch/transport/mod.rs | 3 +- backend/src/fleet/account_failure.rs | 317 ++++++++++++++++++ backend/src/fleet/manage/probe.rs | 57 +--- backend/src/fleet/manage/quota.rs | 14 +- backend/src/fleet/manage/types.rs | 9 +- backend/src/fleet/mod.rs | 1 + backend/src/fleet/quota/runtime.rs | 27 +- backend/src/telemetry/usage/insights.rs | 17 +- .../openai/failure.rs} | 11 +- backend/src/upstream/openai/mod.rs | 1 + .../src/upstream/openai/transport/client.rs | 33 -- backend/src/upstream/openai/transport/mod.rs | 1 - .../api/admin/accounts_routes/quota_routes.rs | 57 ++++ backend/tests/api/admin/usage_routes.rs | 39 ++- .../tests/dispatch/service/usage_logging.rs | 1 + backend/tests/fleet/account_failure.rs | 79 +++++ backend/tests/fleet/mod.rs | 1 + backend/tests/fleet/quota.rs | 74 +++- backend/tests/support/storage.rs | 30 +- .../tests/upstream/openai/transport/client.rs | 18 +- docs/architecture.md | 25 +- .../usage/components/UsageDiagnosticCard.vue | 10 +- 34 files changed, 766 insertions(+), 512 deletions(-) delete mode 100644 backend/src/dispatch/controllers/account_state.rs create mode 100644 backend/src/fleet/account_failure.rs rename backend/src/{dispatch/transport/observation.rs => upstream/openai/failure.rs} (95%) create mode 100644 backend/tests/fleet/account_failure.rs diff --git a/backend/src/bootstrap/tasks/coordinator.rs b/backend/src/bootstrap/tasks/coordinator.rs index fe3c5be98..021b9155f 100644 --- a/backend/src/bootstrap/tasks/coordinator.rs +++ b/backend/src/bootstrap/tasks/coordinator.rs @@ -82,6 +82,7 @@ impl TaskCoordinator { stores.accounts.clone(), stores.account_usage.clone(), services.codex.clone(), + services.account_pool.clone(), services.account_pseudonymizer.clone(), config .quota @@ -91,7 +92,6 @@ impl TaskCoordinator { 30 * 60, ) .with_cookie_store(stores.cookies.clone()) - .with_account_pool(services.account_pool.clone()) .start(), ); coordinator.push("fingerprint_update", { diff --git a/backend/src/bootstrap/tasks/quota_refresh.rs b/backend/src/bootstrap/tasks/quota_refresh.rs index 8f89d83f2..3af81c476 100644 --- a/backend/src/bootstrap/tasks/quota_refresh.rs +++ b/backend/src/bootstrap/tasks/quota_refresh.rs @@ -35,6 +35,7 @@ impl QuotaRefreshTask { store: PgAccountStore, usage_store: PgAccountUsageStore, codex: Arc, + account_pool: Arc, account_pseudonymizer: Arc, interval_secs: u64, min_refresh_interval_secs: u64, @@ -44,6 +45,7 @@ impl QuotaRefreshTask { store, usage_store, codex, + account_pool, account_pseudonymizer, min_refresh_interval_secs, ), @@ -58,12 +60,6 @@ impl QuotaRefreshTask { self } - /// 设置运行时账号池，用于刷新后同步内存调度状态。 - pub fn with_account_pool(mut self, account_pool: Arc) -> Self { - self.service = self.service.with_account_pool(account_pool); - self - } - /// 启动后台刷新任务。 pub fn start(self) -> SchedulerHandle { let config = PeriodicTaskConfig::new( diff --git a/backend/src/dispatch/controllers/account_failure.rs b/backend/src/dispatch/controllers/account_failure.rs index 598da988b..33e52ebb8 100644 --- a/backend/src/dispatch/controllers/account_failure.rs +++ b/backend/src/dispatch/controllers/account_failure.rs @@ -2,19 +2,25 @@ use crate::{ dispatch::{ - controllers::{ - ControllerFailureFact, - account_state::{AccountStateEffect, AccountStateEffects}, - }, + controllers::ControllerFailureFact, errors::ClientFailure, failure::exhaustion::{AccountExhaustionRecord, ExhaustedAccountKind}, lifecycle::contract::{ AttemptDecision, AttemptObservation, AttemptObservationKind, CompleteResponseFacts, }, - transport::observation::{UpstreamFailureFacts, UpstreamFailureKind}, }, - fleet::{account::AccountStatus, pool::AccountPoolService}, - upstream::openai::{protocol::responses::ResponsesSseFailure, transport::CodexBackendClient}, + fleet::{ + account_failure::{ + AccountFailureKind, AccountStateEffect, apply_account_state_effect_immediately, + classify_response_failure, classify_upstream_failure, + }, + pool::AccountPoolService, + }, + upstream::openai::{ + failure::{UpstreamFailureFacts, UpstreamFailureKind}, + protocol::responses::ResponsesSseFailure, + transport::CodexBackendClient, + }, }; pub(super) struct AccountFailureController; @@ -46,7 +52,7 @@ impl AccountFailureController { return; }; if let Some(effect) = &classified.effect { - AccountStateEffects::apply(account_pool, codex, account_id, effect).await; + apply_account_state_effect_immediately(account_pool, codex, account_id, effect).await; } } @@ -110,15 +116,7 @@ impl AccountFailureController { .as_deref() .unwrap_or_default() .to_ascii_lowercase(); - let error_type = failure - .upstream_type - .as_deref() - .unwrap_or_default() - .to_ascii_lowercase(); - let message = failure.message.to_ascii_lowercase(); - if is_model_unsupported(&code) - || is_model_unsupported(&message) - || code.contains("invalid_request") + if code.contains("invalid_request") || code.contains("not_found") || code.contains("context_window") || code.contains("invalid_prompt") @@ -126,17 +124,18 @@ impl AccountFailureController { { return Some(ClientFailure::new(failure.clone(), 400, false)); } - if is_auth_failure(&code, &error_type, &message) { - return Some(ClientFailure::new(failure.clone(), 401, false)); - } - if code.contains("forbidden") || code.contains("banned") || error_type == "permission_error" - { - return Some(ClientFailure::new(failure.clone(), 403, false)); - } if code.contains("server_overloaded") { return Some(ClientFailure::new(failure.clone(), 503, true)); } - None + classify_response_failure(failure).map(|classified| { + let status = match classified.kind { + AccountFailureKind::ModelUnsupported => 400, + AccountFailureKind::Expired => 401, + AccountFailureKind::Disabled | AccountFailureKind::Banned => 403, + AccountFailureKind::QuotaExhausted | AccountFailureKind::RateLimited => 429, + }; + ClientFailure::new(failure.clone(), status, false) + }) } } @@ -168,167 +167,42 @@ fn classify_failure( } fn classify_upstream(account_id: &str, facts: &UpstreamFailureFacts) -> Option { - let status = facts.status_code; - let body = facts.body.as_str(); - let code = facts.code.as_deref().unwrap_or_default(); - let error_type = facts.error_type.as_deref().unwrap_or_default(); - let message = facts.message.as_str(); - - if is_model_unsupported(code) || is_model_unsupported(message) || is_model_unsupported(body) { - return Some(model_unsupported_failure(account_id, body)); - } - if let Some(account_status) = explicit_account_status(code, error_type, body) { - return Some(account_status_failure( - account_id, - account_status, - body, - status, - )); - } - if status == Some(403) && !is_html(body) { - return Some(account_status_failure( - account_id, - AccountStatus::Banned, - body, - status, - )); - } - if status == Some(401) { - return Some(account_status_failure( - account_id, - AccountStatus::Expired, - body, - status, - )); - } - None + classified_failure( + account_id, + &facts.body, + facts.status_code, + classify_upstream_failure(facts)?, + ) } fn classify_sse(account_id: &str, failure: &ResponsesSseFailure) -> Option { - let code = failure.upstream_code.as_deref().unwrap_or_default(); - let error_type = failure.upstream_type.as_deref().unwrap_or_default(); - if is_model_unsupported(code) || is_model_unsupported(&failure.message) { - return Some(model_unsupported_failure( - account_id, - &crate::dispatch::failure::sse::sse_failure_error_body(failure), - )); - } - let account_status = - explicit_account_status(code, error_type, &failure.message).or_else(|| { - (code.eq_ignore_ascii_case("forbidden") || error_type == "permission_error") - .then_some(AccountStatus::Banned) - })?; - Some(account_status_failure( + classified_failure( account_id, - account_status, &crate::dispatch::failure::sse::sse_failure_error_body(failure), failure.explicit_status_code, - )) + classify_response_failure(failure)?, + ) } -fn model_unsupported_failure(account_id: &str, message: &str) -> ClassifiedFailure { - ClassifiedFailure { - exhaustion: AccountExhaustionRecord::new( - account_id, - ExhaustedAccountKind::ModelUnsupported, - message, - ), - effect: None, - } -} - -fn account_status_failure( +fn classified_failure( account_id: &str, - status: AccountStatus, message: &str, status_code: Option, -) -> ClassifiedFailure { - let kind = match status { - AccountStatus::Expired => ExhaustedAccountKind::Expired, - AccountStatus::Disabled => ExhaustedAccountKind::Disabled, - AccountStatus::Banned => ExhaustedAccountKind::Banned, - AccountStatus::Active | AccountStatus::QuotaExhausted => { - unreachable!("account failure controller only invalidates unavailable statuses") - } + classified: crate::fleet::account_failure::ClassifiedAccountFailure, +) -> Option { + let kind = match classified.kind { + AccountFailureKind::ModelUnsupported => ExhaustedAccountKind::ModelUnsupported, + AccountFailureKind::Expired => ExhaustedAccountKind::Expired, + AccountFailureKind::Disabled => ExhaustedAccountKind::Disabled, + AccountFailureKind::Banned => ExhaustedAccountKind::Banned, + AccountFailureKind::QuotaExhausted | AccountFailureKind::RateLimited => return None, }; let mut exhaustion = AccountExhaustionRecord::new(account_id, kind, message); if let Some(status_code) = status_code { exhaustion = exhaustion.with_status_code(status_code); } - ClassifiedFailure { + Some(ClassifiedFailure { exhaustion, - effect: Some(AccountStateEffect::SetStatus(status)), - } -} - -fn explicit_account_status(code: &str, error_type: &str, message: &str) -> Option { - let code = code.trim().to_ascii_lowercase(); - let error_type = error_type.trim().to_ascii_lowercase(); - let message = message.to_ascii_lowercase(); - if matches!( - code.as_str(), - "identity_verification_required" | "verification_required" - ) || message.contains("identity verification is required") - { - return Some(AccountStatus::Disabled); - } - if matches!( - code.as_str(), - "account_banned" - | "account_deactivated" - | "account_disabled" - | "account_suspended" - | "deactivated_workspace" - | "organization_disabled" - | "workspace_deactivated" - ) || message.contains("account is banned") - || message.contains("account has been banned") - || message.contains("account deactivated") - || message.contains("account has been deactivated") - || message.contains("account disabled") - || message.contains("account has been disabled") - || message.contains("account suspended") - || message.contains("organization has been disabled") - || message.contains("workspace has been deactivated") - || message.contains("deactivated_workspace") - { - return Some(AccountStatus::Banned); - } - is_auth_failure(&code, &error_type, &message).then_some(AccountStatus::Expired) -} - -fn is_auth_failure(code: &str, error_type: &str, message: &str) -> bool { - matches!( - code, - "token_invalid" - | "token_invalidated" - | "token_expired" - | "token_revoked" - | "refresh_token_invalidated" - | "unauthorized" - | "invalid_api_key" - | "authentication_error" - ) || error_type == "authentication_error" - || message.contains("token revoked") - || message.contains("token invalidated") - || message.contains("token invalid") - || message.contains("token expired") - || message.contains("unauthorized") - || message.contains("invalid api key") -} - -fn is_model_unsupported(value: &str) -> bool { - let value = value.to_ascii_lowercase(); - value.contains("model_not_supported") - || value.contains("model_not_available") - || (value.contains("model") - && (value.contains("not supported") - || value.contains("not available") - || value.contains("not_supported") - || value.contains("not_available"))) -} - -fn is_html(value: &str) -> bool { - let value = value.trim_start().to_ascii_lowercase(); - value.starts_with("), -} - -pub(super) struct AccountStateEffects; - -impl AccountStateEffects { - pub(super) async fn apply( - account_pool: &AccountPoolService, - codex: &CodexBackendClient, - account_id: &str, - effect: &AccountStateEffect, - ) { - match effect { - AccountStateEffect::SetStatus(status) => { - account_pool - .set_status_immediately(account_id, *status) - .await; - } - AccountStateEffect::MarkQuotaLimitedUntil(until) => { - account_pool - .mark_quota_limited_until_immediately(account_id, *until) - .await; - } - } - codex.evict_websocket_account(account_id).await; - } -} diff --git a/backend/src/dispatch/controllers/cloudflare.rs b/backend/src/dispatch/controllers/cloudflare.rs index b7a4e82f8..f58732c9b 100644 --- a/backend/src/dispatch/controllers/cloudflare.rs +++ b/backend/src/dispatch/controllers/cloudflare.rs @@ -9,13 +9,13 @@ use crate::{ AttemptDecision, AttemptObservation, AttemptObservationKind, AttemptReturnKind, }, lifecycle::stream::StreamTerminal, - transport::observation::UpstreamFailureFacts, }, fleet::{ account::AccountStatus, cookies::{CloudflareChallengeCooldownTracker, CloudflarePathBlockTracker, PgCookieStore}, pool::AccountPoolService, }, + upstream::openai::failure::UpstreamFailureFacts, }; pub(super) struct CloudflareController; diff --git a/backend/src/dispatch/controllers/cyber_policy/mod.rs b/backend/src/dispatch/controllers/cyber_policy/mod.rs index 5951019c8..00b12efe2 100644 --- a/backend/src/dispatch/controllers/cyber_policy/mod.rs +++ b/backend/src/dispatch/controllers/cyber_policy/mod.rs @@ -15,9 +15,11 @@ use crate::{ controllers::ControllerFailureFact, errors::ClientFailure, lifecycle::contract::{AttemptDecision, AttemptObservation, AttemptReturnKind}, - transport::observation::UpstreamFailureFacts, }, - upstream::openai::protocol::responses::{CodexResponsesRequest, ResponsesSseFailure}, + upstream::openai::{ + failure::UpstreamFailureFacts, + protocol::responses::{CodexResponsesRequest, ResponsesSseFailure}, + }, }; use self::{store::Store, types::SessionState}; diff --git a/backend/src/dispatch/controllers/history.rs b/backend/src/dispatch/controllers/history.rs index 1d427a3ee..dd200ff11 100644 --- a/backend/src/dispatch/controllers/history.rs +++ b/backend/src/dispatch/controllers/history.rs @@ -13,9 +13,9 @@ use crate::{ AttemptDecision, AttemptObservation, AttemptObservationKind, AttemptReturnKind, AttemptRoutingFacts, CompleteResponseFacts, PinnedCandidateAcquireFailureKind, }, - transport::observation::{UpstreamFailureFacts, UpstreamFailureKind}, }, upstream::openai::{ + failure::{UpstreamFailureFacts, UpstreamFailureKind}, protocol::responses::{CodexResponsesRequest, PreviousResponseScope, StreamCommitPolicy}, transport::CodexBackendTransport, transport::websocket::PreviousResponseUnavailableReason, diff --git a/backend/src/dispatch/controllers/mod.rs b/backend/src/dispatch/controllers/mod.rs index 110abc2a1..6a5cc1ede 100644 --- a/backend/src/dispatch/controllers/mod.rs +++ b/backend/src/dispatch/controllers/mod.rs @@ -49,14 +49,15 @@ use crate::{ }, stream::StreamTerminal, }, - transport::observation::UpstreamFailureFacts, }, fleet::pool::AccountPoolService, - upstream::openai::protocol::responses::{CodexResponsesRequest, ResponsesSseFailure}, + upstream::openai::{ + failure::UpstreamFailureFacts, + protocol::responses::{CodexResponsesRequest, ResponsesSseFailure}, + }, }; mod account_failure; -mod account_state; mod affinity; pub(crate) mod cloudflare; pub(crate) mod cyber_policy; diff --git a/backend/src/dispatch/controllers/quota.rs b/backend/src/dispatch/controllers/quota.rs index 7d6aa87c1..07ceb8286 100644 --- a/backend/src/dispatch/controllers/quota.rs +++ b/backend/src/dispatch/controllers/quota.rs @@ -1,23 +1,22 @@ //! 请求前配额验证规则 owner。 -use chrono::{Duration, Utc}; use serde_json::Value; use crate::{ dispatch::{ affinity::AccountIdentityService, - controllers::{ - ControllerFailureFact, - account_state::{AccountStateEffect, AccountStateEffects}, - }, + controllers::ControllerFailureFact, errors::ClientFailure, failure::exhaustion::{AccountExhaustionRecord, ExhaustedAccountKind}, lifecycle::contract::{ AttemptDecision, AttemptObservation, AttemptObservationKind, AttemptReturnKind, }, - transport::observation::UpstreamFailureFacts, }, fleet::{ + account_failure::{ + AccountFailureKind, AccountStateEffect, apply_account_state_effect_immediately, + classify_response_failure, classify_upstream_failure, + }, pool::{AccountLease, AccountPoolService}, quota::{ quota_from_usage, quota_snapshot_limit_reached, quota_snapshot_limit_window_seconds, @@ -25,13 +24,13 @@ use crate::{ }, }, upstream::openai::{ + failure::UpstreamFailureFacts, protocol::responses::ResponsesSseFailure, transport::{CodexBackendClient, CodexRequestContext}, }, }; const LIMIT_REACHED_MESSAGE: &str = "Upstream usage quota still reports limit_reached"; -const DEFAULT_RATE_LIMIT_RETRY_SECONDS: u64 = 60; pub(super) enum QuotaEnterOutcome { Ready(Box), @@ -167,7 +166,7 @@ impl QuotaController { return; }; if let Some(effect) = &classified.effect { - AccountStateEffects::apply(account_pool, codex, account_id, effect).await; + apply_account_state_effect_immediately(account_pool, codex, account_id, effect).await; } } @@ -201,131 +200,34 @@ fn classify_upstream( account_id: &str, facts: &UpstreamFailureFacts, ) -> Option { - let signaled_kind = facts - .code - .as_deref() - .and_then(classify_quota_signal) - .or_else(|| facts.error_type.as_deref().and_then(classify_quota_signal)) - .or_else(|| classify_quota_message(&facts.message)); - match signaled_kind.or(match facts.status_code { - Some(429) => Some(StreamQuotaKind::RateLimited), - Some(402) => Some(StreamQuotaKind::QuotaExhausted), - _ => None, - })? { - StreamQuotaKind::RateLimited => Some(rate_limited_failure( - account_id, - facts.body.clone(), - facts.retry_after_seconds, - )), - StreamQuotaKind::QuotaExhausted => { - Some(quota_exhausted_failure(account_id, facts.body.clone())) - } - } + classified_quota_failure(account_id, &facts.body, classify_upstream_failure(facts)?) } fn classify_stream( account_id: &str, failure: &ResponsesSseFailure, ) -> Option { - let kind = classify_stream_signal(failure).or(match failure.explicit_status_code { - Some(429) => Some(StreamQuotaKind::RateLimited), - Some(402) => Some(StreamQuotaKind::QuotaExhausted), - _ => None, - })?; let body = crate::dispatch::failure::sse::sse_failure_error_body(failure); - Some(match kind { - StreamQuotaKind::RateLimited => { - rate_limited_failure(account_id, body, failure.retry_after_seconds) - } - StreamQuotaKind::QuotaExhausted => quota_exhausted_failure(account_id, body), - }) + classified_quota_failure(account_id, &body, classify_response_failure(failure)?) } -#[derive(Clone, Copy)] -enum StreamQuotaKind { - RateLimited, - QuotaExhausted, -} - -fn classify_stream_signal(failure: &ResponsesSseFailure) -> Option { - failure - .upstream_code - .as_deref() - .and_then(classify_quota_signal) - .or_else(|| { - failure - .upstream_type - .as_deref() - .and_then(classify_quota_signal) - }) - .or_else(|| classify_quota_message(&failure.message)) -} - -fn classify_quota_signal(signal: &str) -> Option { - let signal = signal.trim().to_ascii_lowercase(); - match signal.as_str() { - "usage_limit_reached" - | "rate_limit_exceeded" - | "rate_limit_reached" - | "rate_limit_error" - | "workspace_owner_usage_limit_reached" - | "workspace_member_usage_limit_reached" => Some(StreamQuotaKind::RateLimited), - "quota_exhausted" - | "quota_exceeded" - | "payment_required" - | "insufficient_quota" - | "workspace_owner_credits_depleted" - | "workspace_member_credits_depleted" => Some(StreamQuotaKind::QuotaExhausted), - signal if signal.starts_with("billing_limit") => Some(StreamQuotaKind::QuotaExhausted), - _ => None, - } -} - -fn classify_quota_message(message: &str) -> Option { - let message = message.to_ascii_lowercase(); - if message.contains("rate limit") || message.contains("usage limit") { - return Some(StreamQuotaKind::RateLimited); - } - if message.contains("quota") - || message.contains("payment required") - || message.contains("billing limit") - { - return Some(StreamQuotaKind::QuotaExhausted); - } - None -} - -fn rate_limited_failure( +fn classified_quota_failure( account_id: &str, - message: String, - retry_after_seconds: Option, -) -> ClassifiedQuotaFailure { - let seconds = retry_after_seconds - .unwrap_or(DEFAULT_RATE_LIMIT_RETRY_SECONDS) - .min(i64::MAX as u64) as i64; - ClassifiedQuotaFailure { - exhaustion: AccountExhaustionRecord::new( - account_id, - ExhaustedAccountKind::RateLimited, - message, - ), - effect: Some(AccountStateEffect::MarkQuotaLimitedUntil( - Utc::now() + Duration::seconds(seconds), - )), - } -} - -fn quota_exhausted_failure(account_id: &str, message: String) -> ClassifiedQuotaFailure { - ClassifiedQuotaFailure { - exhaustion: AccountExhaustionRecord::new( - account_id, - ExhaustedAccountKind::QuotaExhausted, - message, - ), - effect: Some(AccountStateEffect::SetStatus( - crate::fleet::account::AccountStatus::QuotaExhausted, - )), - } + message: &str, + classified: crate::fleet::account_failure::ClassifiedAccountFailure, +) -> Option { + let kind = match classified.kind { + AccountFailureKind::RateLimited => ExhaustedAccountKind::RateLimited, + AccountFailureKind::QuotaExhausted => ExhaustedAccountKind::QuotaExhausted, + AccountFailureKind::ModelUnsupported + | AccountFailureKind::Expired + | AccountFailureKind::Disabled + | AccountFailureKind::Banned => return None, + }; + Some(ClassifiedQuotaFailure { + exhaustion: AccountExhaustionRecord::new(account_id, kind, message), + effect: classified.effect, + }) } fn acquired_with_verified_quota(mut acquired: AccountLease, quota: &Value) -> AccountLease { diff --git a/backend/src/dispatch/controllers/telemetry/events.rs b/backend/src/dispatch/controllers/telemetry/events.rs index f5fd89e68..516f4a832 100644 --- a/backend/src/dispatch/controllers/telemetry/events.rs +++ b/backend/src/dispatch/controllers/telemetry/events.rs @@ -546,6 +546,9 @@ pub(super) async fn record_live_response_stream_event( status_code, body, ); + if !is_success_usage_event(level, status_code) { + ensure_stream_failure_class(&mut metadata); + } log_live_response_stream_finalized(context, status_code, level, message, &metadata); enrich_usage_record_identity( &mut metadata, @@ -601,6 +604,19 @@ fn is_success_usage_event(level: UsageRecordLevel, status_code: i64) -> bool { level != UsageRecordLevel::Error && status_code < 400 } +fn ensure_stream_failure_class(metadata: &mut Value) { + let failure_class = ["failureClass", "upstreamCode", "terminal"] + .into_iter() + .find_map(|field| metadata_string_field(metadata, field)) + .map(str::trim) + .filter(|value| !value.is_empty()) + .map(ToString::to_string); + let (Some(failure_class), Some(object)) = (failure_class, metadata.as_object_mut()) else { + return; + }; + object.insert("failureClass".to_string(), Value::String(failure_class)); +} + fn log_live_response_stream_finalized( context: &StreamContext<'_>, status_code: i64, diff --git a/backend/src/dispatch/lifecycle/attempt.rs b/backend/src/dispatch/lifecycle/attempt.rs index 5cb4d3f36..a733db143 100644 --- a/backend/src/dispatch/lifecycle/attempt.rs +++ b/backend/src/dispatch/lifecycle/attempt.rs @@ -21,7 +21,6 @@ use crate::{ create_response_with_account, prepare_response_transport_with_account, }, canonical::{CanonicalStreamDecoder, normalize_complete_response}, - observation::upstream_failure_facts, prefetch::{PrefetchedStream, StreamPrefetchError, prefetch_until_commit}, }, }, @@ -30,6 +29,7 @@ use crate::{ pool::{AccountCandidateLease, AccountPoolService}, }, upstream::openai::{ + failure::upstream_failure_facts, protocol::responses::{ CodexResponsesRequest, CollectedResponse, TransportRequirement, transport_requirement, }, diff --git a/backend/src/dispatch/lifecycle/contract.rs b/backend/src/dispatch/lifecycle/contract.rs index 38840b6f4..87ceb0bfc 100644 --- a/backend/src/dispatch/lifecycle/contract.rs +++ b/backend/src/dispatch/lifecycle/contract.rs @@ -6,13 +6,11 @@ use crate::{ dispatch::{ errors::{ClientFailure, ResponseDispatchError}, failure::exhaustion::{AccountExhaustionRecord, ExhaustedAccount}, - transport::{ - canonical::{CanonicalStreamBatch, CanonicalStreamDecoder}, - observation::UpstreamFailureFacts, - }, + transport::canonical::{CanonicalStreamBatch, CanonicalStreamDecoder}, }, fleet::{account::Account, pool::AccountLease}, upstream::openai::{ + failure::UpstreamFailureFacts, protocol::{ responses::{CodexResponsesRequest, CollectedResponse, ResponsesSseFailure}, sse::SseError, diff --git a/backend/src/dispatch/transport/mod.rs b/backend/src/dispatch/transport/mod.rs index d178e8e92..78b8e0223 100644 --- a/backend/src/dispatch/transport/mod.rs +++ b/backend/src/dispatch/transport/mod.rs @@ -1,6 +1,5 @@ -//! Attempt 阶段的账号作用域上游调用与事实规范化。 +//! Attempt 阶段的账号作用域上游调用与流规范化。 pub(in crate::dispatch) mod account; pub(crate) mod canonical; -pub(in crate::dispatch) mod observation; pub(in crate::dispatch) mod prefetch; diff --git a/backend/src/fleet/account_failure.rs b/backend/src/fleet/account_failure.rs new file mode 100644 index 000000000..200801955 --- /dev/null +++ b/backend/src/fleet/account_failure.rs @@ -0,0 +1,317 @@ +//! 上游失败到账号状态 effect 的唯一领域分类。 + +use chrono::{Duration, Utc}; + +use crate::{ + fleet::{account::AccountStatus, pool::AccountPoolService}, + upstream::openai::{ + failure::{UpstreamFailureFacts, upstream_failure_facts}, + protocol::responses::ResponsesSseFailure, + transport::{CodexBackendClient, CodexClientError}, + }, +}; + +const DEFAULT_RATE_LIMIT_RETRY_SECONDS: u64 = 60; + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum AccountFailureKind { + ModelUnsupported, + Expired, + Disabled, + Banned, + QuotaExhausted, + RateLimited, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum AccountStateEffect { + SetStatus(AccountStatus), + MarkQuotaLimitedUntil(chrono::DateTime), +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct ClassifiedAccountFailure { + pub kind: AccountFailureKind, + pub effect: Option, +} + +impl ClassifiedAccountFailure { + pub fn account_status(&self) -> Option { + match self.effect.as_ref()? { + AccountStateEffect::SetStatus(status) => Some(*status), + AccountStateEffect::MarkQuotaLimitedUntil(_) => None, + } + } +} + +pub fn classify_client_failure(error: &CodexClientError) -> Option { + classify_upstream_failure(&upstream_failure_facts(error)) +} + +pub fn classify_upstream_failure(facts: &UpstreamFailureFacts) -> Option { + classify_failure(FailureFields { + status_code: facts.status_code, + code: facts.code.as_deref(), + error_type: facts.error_type.as_deref(), + message: &facts.message, + body: &facts.body, + retry_after_seconds: facts.retry_after_seconds, + }) +} + +pub fn classify_response_failure( + failure: &ResponsesSseFailure, +) -> Option { + classify_failure(FailureFields { + status_code: failure.explicit_status_code, + code: failure.upstream_code.as_deref(), + error_type: failure.upstream_type.as_deref(), + message: &failure.message, + body: &failure.message, + retry_after_seconds: failure.retry_after_seconds, + }) +} + +pub async fn apply_account_state_effect_immediately( + account_pool: &AccountPoolService, + codex: &CodexBackendClient, + account_id: &str, + effect: &AccountStateEffect, +) { + match effect { + AccountStateEffect::SetStatus(status) => { + account_pool + .set_status_immediately(account_id, *status) + .await; + } + AccountStateEffect::MarkQuotaLimitedUntil(until) => { + account_pool + .mark_quota_limited_until_immediately(account_id, *until) + .await; + } + } + codex.evict_websocket_account(account_id).await; +} + +pub async fn apply_account_state_effect( + account_pool: &AccountPoolService, + codex: &CodexBackendClient, + account_id: &str, + effect: &AccountStateEffect, +) -> bool { + let updated = match effect { + AccountStateEffect::SetStatus(status) => account_pool.set_status(account_id, *status).await, + AccountStateEffect::MarkQuotaLimitedUntil(until) => { + account_pool + .mark_quota_limited_until(account_id, *until) + .await + } + }; + codex.evict_websocket_account(account_id).await; + updated +} + +pub async fn apply_client_failure_effect( + account_pool: &AccountPoolService, + codex: &CodexBackendClient, + account_id: &str, + error: &CodexClientError, +) -> Option { + let classified = classify_client_failure(error)?; + if let Some(effect) = &classified.effect { + apply_account_state_effect(account_pool, codex, account_id, effect).await; + } + Some(classified) +} + +struct FailureFields<'a> { + status_code: Option, + code: Option<&'a str>, + error_type: Option<&'a str>, + message: &'a str, + body: &'a str, + retry_after_seconds: Option, +} + +fn classify_failure(fields: FailureFields<'_>) -> Option { + let code = fields.code.unwrap_or_default(); + let error_type = fields.error_type.unwrap_or_default(); + + if [code, fields.message, fields.body] + .into_iter() + .any(is_model_unsupported) + { + return Some(classified(AccountFailureKind::ModelUnsupported, None)); + } + if let Some(status) = explicit_account_status(code, error_type, fields.message) + .or_else(|| explicit_account_status(code, error_type, fields.body)) + { + return Some(status_failure(status)); + } + if fields.status_code == Some(403) && !is_html(fields.body) { + return Some(status_failure(AccountStatus::Banned)); + } + if fields.status_code == Some(401) { + return Some(status_failure(AccountStatus::Expired)); + } + + let quota_kind = [code, error_type] + .into_iter() + .find_map(classify_quota_signal) + .or_else(|| classify_quota_message(fields.message)) + .or_else(|| classify_quota_message(fields.body)) + .or(match fields.status_code { + Some(429) => Some(AccountFailureKind::RateLimited), + Some(402) => Some(AccountFailureKind::QuotaExhausted), + _ => None, + })?; + Some(match quota_kind { + AccountFailureKind::RateLimited => { + let seconds = fields + .retry_after_seconds + .unwrap_or(DEFAULT_RATE_LIMIT_RETRY_SECONDS) + .min(i64::MAX as u64) as i64; + classified( + AccountFailureKind::RateLimited, + Some(AccountStateEffect::MarkQuotaLimitedUntil( + Utc::now() + Duration::seconds(seconds), + )), + ) + } + AccountFailureKind::QuotaExhausted => classified( + AccountFailureKind::QuotaExhausted, + Some(AccountStateEffect::SetStatus(AccountStatus::QuotaExhausted)), + ), + AccountFailureKind::ModelUnsupported + | AccountFailureKind::Expired + | AccountFailureKind::Disabled + | AccountFailureKind::Banned => unreachable!("quota classifier returned account status"), + }) +} + +fn status_failure(status: AccountStatus) -> ClassifiedAccountFailure { + let kind = match status { + AccountStatus::Expired => AccountFailureKind::Expired, + AccountStatus::Disabled => AccountFailureKind::Disabled, + AccountStatus::Banned => AccountFailureKind::Banned, + AccountStatus::Active | AccountStatus::QuotaExhausted => { + unreachable!("explicit account failure must make the account unavailable") + } + }; + classified(kind, Some(AccountStateEffect::SetStatus(status))) +} + +fn classified( + kind: AccountFailureKind, + effect: Option, +) -> ClassifiedAccountFailure { + ClassifiedAccountFailure { kind, effect } +} + +fn explicit_account_status(code: &str, error_type: &str, message: &str) -> Option { + let code = code.trim().to_ascii_lowercase(); + let error_type = error_type.trim().to_ascii_lowercase(); + let message = message.to_ascii_lowercase(); + if matches!( + code.as_str(), + "identity_verification_required" | "verification_required" + ) || message.contains("identity verification is required") + { + return Some(AccountStatus::Disabled); + } + if matches!( + code.as_str(), + "account_banned" + | "account_deactivated" + | "account_disabled" + | "account_suspended" + | "deactivated_workspace" + | "organization_disabled" + | "workspace_deactivated" + ) || message.contains("account is banned") + || message.contains("account has been banned") + || message.contains("account deactivated") + || message.contains("account has been deactivated") + || message.contains("account disabled") + || message.contains("account has been disabled") + || message.contains("account suspended") + || message.contains("organization has been disabled") + || message.contains("workspace has been deactivated") + || message.contains("deactivated_workspace") + || code == "forbidden" + || error_type == "permission_error" + { + return Some(AccountStatus::Banned); + } + is_auth_failure(&code, &error_type, &message).then_some(AccountStatus::Expired) +} + +fn is_auth_failure(code: &str, error_type: &str, message: &str) -> bool { + matches!( + code, + "token_invalid" + | "token_invalidated" + | "token_expired" + | "token_revoked" + | "refresh_token_invalidated" + | "unauthorized" + | "invalid_api_key" + | "authentication_error" + ) || error_type == "authentication_error" + || message.contains("token revoked") + || message.contains("token invalidated") + || message.contains("token invalid") + || message.contains("token expired") + || message.contains("unauthorized") + || message.contains("invalid api key") +} + +fn classify_quota_signal(signal: &str) -> Option { + let signal = signal.trim().to_ascii_lowercase(); + match signal.as_str() { + "usage_limit_reached" + | "rate_limit_exceeded" + | "rate_limit_reached" + | "rate_limit_error" + | "workspace_owner_usage_limit_reached" + | "workspace_member_usage_limit_reached" => Some(AccountFailureKind::RateLimited), + "quota_exhausted" + | "quota_exceeded" + | "payment_required" + | "insufficient_quota" + | "workspace_owner_credits_depleted" + | "workspace_member_credits_depleted" => Some(AccountFailureKind::QuotaExhausted), + signal if signal.starts_with("billing_limit") => Some(AccountFailureKind::QuotaExhausted), + _ => None, + } +} + +fn classify_quota_message(message: &str) -> Option { + let message = message.to_ascii_lowercase(); + if message.contains("rate limit") || message.contains("usage limit") { + return Some(AccountFailureKind::RateLimited); + } + if message.contains("quota") + || message.contains("payment required") + || message.contains("billing limit") + { + return Some(AccountFailureKind::QuotaExhausted); + } + None +} + +fn is_model_unsupported(value: &str) -> bool { + let value = value.to_ascii_lowercase(); + value.contains("model_not_supported") + || value.contains("model_not_available") + || (value.contains("model") + && (value.contains("not supported") + || value.contains("not available") + || value.contains("not_supported") + || value.contains("not_available"))) +} + +fn is_html(value: &str) -> bool { + let value = value.trim_start().to_ascii_lowercase(); + value.starts_with(" Option { - if is_banned_upstream_error(error) { - Some(AccountStatus::Banned) - } else { - match error { - CodexClientError::Upstream { status, .. } if status.as_u16() == 402 => { - Some(AccountStatus::QuotaExhausted) - } - CodexClientError::Upstream { status, body, .. } if status.as_u16() == 401 => { - Some(if is_banned_auth_signal(body) { - AccountStatus::Banned - } else { - AccountStatus::Expired - }) - } - _ => None, - } - } + classify_client_failure(error)?.account_status() } fn sse_failure_account_status(failure: &ResponsesSseFailure) -> Option { - let code = failure - .upstream_code - .as_deref() - .unwrap_or_default() - .to_ascii_lowercase(); - let message = failure.message.to_ascii_lowercase(); - if matches!(code.as_str(), "quota_exceeded" | "insufficient_quota") || message.contains("quota") - { - return Some(AccountStatus::QuotaExhausted); - } - let authentication_failed = matches!( - code.as_str(), - "token_invalid" - | "token_expired" - | "token_revoked" - | "account_deactivated" - | "unauthorized" - | "invalid_api_key" - ) || message.contains("token revoked") - || message.contains("token invalid") - || message.contains("token expired"); - authentication_failed.then(|| { - if is_banned_auth_signal(&failure.message) { - AccountStatus::Banned - } else { - AccountStatus::Expired - } - }) + classify_response_failure(failure)?.account_status() } fn take_sse_frame(buffer: &mut String) -> Option { diff --git a/backend/src/fleet/manage/quota.rs b/backend/src/fleet/manage/quota.rs index dd29e8570..d93eeb413 100644 --- a/backend/src/fleet/manage/quota.rs +++ b/backend/src/fleet/manage/quota.rs @@ -1,6 +1,7 @@ use secrecy::ExposeSecret; use super::{AccountManageService, types::AccountManageError}; +use crate::fleet::account_failure::apply_client_failure_effect; impl AccountManageService { async fn usage_cookie_header(&self, account_id: &str) -> Option { @@ -53,11 +54,14 @@ impl AccountManageService { client_request_id: None, turn_id: None, }; - let raw = self - .codex - .fetch_usage(context) - .await - .map_err(|e| AccountManageError::FetchQuota(e.to_string()))?; + let raw = match self.codex.fetch_usage(context).await { + Ok(raw) => raw, + Err(error) => { + apply_client_failure_effect(&self.account_pool, &self.codex, account_id, &error) + .await; + return Err(AccountManageError::FetchQuota(error.to_string())); + } + }; let normalized = crate::fleet::quota::quota_from_usage(&raw); self.account_pool .apply_quota_snapshot(account_id, &normalized) diff --git a/backend/src/fleet/manage/types.rs b/backend/src/fleet/manage/types.rs index 29d9431de..75629b4d9 100644 --- a/backend/src/fleet/manage/types.rs +++ b/backend/src/fleet/manage/types.rs @@ -7,6 +7,7 @@ use thiserror::Error; use crate::{ fleet::{ account::{AccountStatus, AccountStatus as AcctStatus}, + account_failure::classify_client_failure, store::StoredAccountMetadata, }, upstream::openai::{token_client::RefreshFailure, transport::CodexClientError}, @@ -323,11 +324,9 @@ fn normalized_plan_type(value: &str) -> Option { } pub(super) fn import_status_from_usage_error(error: &CodexClientError) -> Option { - if crate::upstream::openai::transport::is_banned_upstream_error(error) { - Some(AccountStatus::Banned) - } else { - None - } + classify_client_failure(error)? + .account_status() + .filter(|status| *status == AccountStatus::Banned) } pub(super) fn parse_account_status(status: &str) -> Result { diff --git a/backend/src/fleet/mod.rs b/backend/src/fleet/mod.rs index 79b535847..0a1f9bc73 100644 --- a/backend/src/fleet/mod.rs +++ b/backend/src/fleet/mod.rs @@ -1,6 +1,7 @@ //! 账号、调度、配额与刷新领域。 pub mod account; +pub mod account_failure; pub mod cookies; pub mod import; pub mod manage; diff --git a/backend/src/fleet/quota/runtime.rs b/backend/src/fleet/quota/runtime.rs index 6970308c5..39a9a1c7d 100644 --- a/backend/src/fleet/quota/runtime.rs +++ b/backend/src/fleet/quota/runtime.rs @@ -13,6 +13,7 @@ use tracing::warn; use crate::fleet::{ account::{Account, AccountStatus}, + account_failure::apply_client_failure_effect, pool::AccountPoolService, quota::{ quota_from_usage, quota_snapshot_limit_reached, quota_snapshot_limit_window_seconds, @@ -41,7 +42,7 @@ pub struct QuotaRefreshService { request_spacing: Duration, account_pseudonymizer: Arc, cookie_store: Option, - account_pool: Option>, + account_pool: Arc, } impl QuotaRefreshService { @@ -50,6 +51,7 @@ impl QuotaRefreshService { store: PgAccountStore, usage_store: PgAccountUsageStore, codex: Arc, + account_pool: Arc, account_pseudonymizer: Arc, ) -> Self { Self { @@ -60,7 +62,7 @@ impl QuotaRefreshService { request_spacing: Self::default_request_spacing(), account_pseudonymizer, cookie_store: None, - account_pool: None, + account_pool, } } @@ -73,6 +75,7 @@ impl QuotaRefreshService { store: PgAccountStore, usage_store: PgAccountUsageStore, codex: Arc, + account_pool: Arc, account_pseudonymizer: Arc, min_refresh_interval_secs: u64, ) -> Self { @@ -84,7 +87,7 @@ impl QuotaRefreshService { request_spacing: Self::default_request_spacing(), account_pseudonymizer, cookie_store: None, - account_pool: None, + account_pool, } } @@ -100,12 +103,6 @@ impl QuotaRefreshService { self } - /// 设置运行时账号池，用于刷新后同步内存调度状态。 - pub fn with_account_pool(mut self, account_pool: Arc) -> Self { - self.account_pool = Some(account_pool); - self - } - /// 执行一次配额锁定或待验证账号刷新，复用调用方传入的最近刷新记录。 pub async fn refresh_locked_accounts( &self, @@ -211,6 +208,13 @@ impl QuotaRefreshService { self.sync_runtime_pool_account(&account.id).await; } Err(error) => { + apply_client_failure_effect( + &self.account_pool, + &self.codex, + &account.id, + &error, + ) + .await; warn!( account_id = %account.id, error = %error, @@ -238,10 +242,7 @@ impl QuotaRefreshService { } async fn sync_runtime_pool_account(&self, account_id: &str) { - let Some(account_pool) = &self.account_pool else { - return; - }; - if let Err(error) = account_pool.sync_account_from_store(account_id).await { + if let Err(error) = self.account_pool.sync_account_from_store(account_id).await { warn!( account_id = %account_id, error = %error, diff --git a/backend/src/telemetry/usage/insights.rs b/backend/src/telemetry/usage/insights.rs index 55e4b6873..456e30156 100644 --- a/backend/src/telemetry/usage/insights.rs +++ b/backend/src/telemetry/usage/insights.rs @@ -49,7 +49,13 @@ error_terminals as ( case when request_id is null then 'event:' || id else 'request:' || request_id end as request_key, id, request_id, created_at as terminal_at, client_api_key_id, provider, account_id, transport, coalesce(nullif(model, ''), '未知模型') as model_name, - coalesce(nullif(failure_class, ''), '未分类错误') as failure_class, + coalesce( + nullif(failure_class, ''), + nullif(metadata_json->>'failureClass', ''), + nullif(metadata_json->>'upstreamCode', ''), + nullif(metadata_json->>'terminal', ''), + '未分类错误' + ) as failure_class, latency_ms from ops_error_logs where created_at >= $1 and created_at < $2 @@ -166,9 +172,7 @@ impl UsageDiagnosticsDimension { 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::ApiKey => "coalesce(nullif(diagnostic_api_keys.name, ''), '未知密钥')", Self::Provider => "coalesce(nullif(terminal_requests.provider, ''), '未知 Provider')", Self::Transport => "coalesce(nullif(terminal_requests.transport, ''), '未知传输')", Self::FailureClass => "terminal_requests.failure_class", @@ -180,7 +184,10 @@ impl UsageDiagnosticsDimension { 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 => { + Self::ApiKey => { + "terminal_requests left join client_api_keys diagnostic_api_keys on diagnostic_api_keys.id = terminal_requests.client_api_key_id" + } + Self::Model | Self::Provider | Self::Transport | Self::FailureClass => { "terminal_requests" } } diff --git a/backend/src/dispatch/transport/observation.rs b/backend/src/upstream/openai/failure.rs similarity index 95% rename from backend/src/dispatch/transport/observation.rs rename to backend/src/upstream/openai/failure.rs index da96c3868..b87f9b39b 100644 --- a/backend/src/dispatch/transport/observation.rs +++ b/backend/src/upstream/openai/failure.rs @@ -1,4 +1,4 @@ -//! 上游原始错误到 attempt 通用事实的单次规范化。 +//! Codex 上游原始错误到稳定失败事实的规范化。 use serde_json::Value; @@ -8,7 +8,7 @@ use crate::upstream::openai::transport::{ }; #[derive(Clone, Copy, Debug, PartialEq, Eq)] -pub(in crate::dispatch) enum UpstreamFailureKind { +pub enum UpstreamFailureKind { HttpStatus, HttpConnect, HttpTimeout, @@ -25,7 +25,7 @@ pub(in crate::dispatch) enum UpstreamFailureKind { } #[derive(Clone, Debug, PartialEq, Eq)] -pub(in crate::dispatch) struct UpstreamFailureFacts { +pub struct UpstreamFailureFacts { pub kind: UpstreamFailureKind, pub status_code: Option, pub code: Option, @@ -35,9 +35,7 @@ pub(in crate::dispatch) struct UpstreamFailureFacts { pub retry_after_seconds: Option, } -pub(in crate::dispatch) fn upstream_failure_facts( - error: &CodexClientError, -) -> UpstreamFailureFacts { +pub fn upstream_failure_facts(error: &CodexClientError) -> UpstreamFailureFacts { match error { CodexClientError::Http(error) => UpstreamFailureFacts { kind: if error.is_connect() { @@ -156,6 +154,7 @@ fn error_fields(body: &str) -> (Option, Option, Option) let error = value .pointer("/response/error") .or_else(|| value.get("error")) + .or_else(|| value.get("detail")) .unwrap_or(&value); ( error diff --git a/backend/src/upstream/openai/mod.rs b/backend/src/upstream/openai/mod.rs index 209c19ea6..2739aec09 100644 --- a/backend/src/upstream/openai/mod.rs +++ b/backend/src/upstream/openai/mod.rs @@ -1,5 +1,6 @@ //! OpenAI/Codex 上游客户端。 +pub mod failure; pub mod fingerprint; pub mod protocol; pub mod token_client; diff --git a/backend/src/upstream/openai/transport/client.rs b/backend/src/upstream/openai/transport/client.rs index 9837930f0..754e11afe 100644 --- a/backend/src/upstream/openai/transport/client.rs +++ b/backend/src/upstream/openai/transport/client.rs @@ -166,39 +166,6 @@ impl CodexClientError { } } -/// 判断上游错误正文是否表示账号已封禁或停用。 -pub fn is_banned_auth_signal(value: &str) -> bool { - let value = value.to_ascii_lowercase(); - value.contains("account_deactivated") - || value.contains("account deactivated") - || value.contains("account has been deactivated") - || value.contains("deactivated") - || value.contains("banned") -} - -/// 判断 402 错误正文是否带有 OpenAI 工作区停用标记。 -pub fn is_deactivated_workspace_error_body(value: &str) -> bool { - serde_json::from_str::(value).is_ok_and(|value| { - value.pointer("/detail/code").and_then(Value::as_str) == Some("deactivated_workspace") - }) -} - -/// 判断 Codex 上游错误是否表示账号已封禁或停用。 -pub fn is_banned_upstream_error(error: &CodexClientError) -> bool { - matches!( - error, - CodexClientError::Upstream { status, body, .. } - if (status.as_u16() == 403 - && !is_html_error_body(body)) - || (status.as_u16() == 402 && is_deactivated_workspace_error_body(body)) - ) -} - -fn is_html_error_body(value: &str) -> bool { - let value = value.trim_start().to_ascii_lowercase(); - value.starts_with(" = Result; diff --git a/backend/src/upstream/openai/transport/mod.rs b/backend/src/upstream/openai/transport/mod.rs index 3ecf75382..17d0946ef 100644 --- a/backend/src/upstream/openai/transport/mod.rs +++ b/backend/src/upstream/openai/transport/mod.rs @@ -20,7 +20,6 @@ pub use self::{ CodexModelCatalogClient, CodexModelCatalogClientError, CodexModelCatalogRequest, CodexRateLimitHeaderUpdates, CodexRequestContext, CodexResponseMetadata, CodexTransportDecision, CodexTransportMetrics, CodexTurnStateUpdate, build_reqwest_client, - is_banned_auth_signal, is_banned_upstream_error, is_deactivated_workspace_error_body, }, diagnostics::CodexUpstreamDiagnostics, endpoints::{ diff --git a/backend/tests/api/admin/accounts_routes/quota_routes.rs b/backend/tests/api/admin/accounts_routes/quota_routes.rs index 23b200966..92008f21f 100644 --- a/backend/tests/api/admin/accounts_routes/quota_routes.rs +++ b/backend/tests/api/admin/accounts_routes/quota_routes.rs @@ -195,3 +195,60 @@ async fn admin_account_quota_should_send_usage_cookie() { "2K" ); } + +#[tokio::test] +async fn admin_account_quota_should_ban_deactivated_workspace() { + let server = MockServer::start().await; + Mock::given(method("GET")) + .and(path("/backend-api/wham/usage")) + .respond_with( + ResponseTemplate::new(402) + .set_body_json(json!({ "detail": { "code": "deactivated_workspace" } })), + ) + .expect(1) + .mount(&server) + .await; + let (app, _state, pool, _dir) = admin_accounts_test_app_with_api_base_url( + "admin-account-quota-deactivated", + 89, + format!("{}/backend-api", server.uri()), + ) + .await; + seed_account( + &pool, + NewAccount { + id: "acct_quota_deactivated".to_string(), + email: Some("quota-deactivated@example.com".to_string()), + account_id: Some("chatgpt-quota-deactivated".to_string()), + user_id: None, + label: None, + plan_type: Some("plus".to_string()), + access_token: SecretString::new("access-quota-deactivated".to_string().into()), + refresh_token: None, + access_token_expires_at: None, + status: AccountStatus::Active, + added_at: None, + }, + ) + .await; + + let response = app + .oneshot( + Request::builder() + .method("GET") + .uri("/api/admin/accounts/quota?id=acct_quota_deactivated") + .header("cookie", "cpr_admin_session=session_1") + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + let status: String = sqlx::query_scalar("select status from accounts where id = $1") + .bind("acct_quota_deactivated") + .fetch_one(&pool) + .await + .unwrap(); + + assert_eq!(response.status(), StatusCode::BAD_GATEWAY); + assert_eq!(status, "banned"); +} diff --git a/backend/tests/api/admin/usage_routes.rs b/backend/tests/api/admin/usage_routes.rs index bce142617..f4b43da41 100644 --- a/backend/tests/api/admin/usage_routes.rs +++ b/backend/tests/api/admin/usage_routes.rs @@ -138,10 +138,24 @@ async fn usage_insights_should_collapse_retry_errors_to_terminal_requests() { .execute(usage.pool()) .await .unwrap(); + sqlx::query( + "insert into client_api_keys + (id, name, prefix, key, enabled, created_at, last_used_at) + values ($1, $2, $3, $4, true, $5, null)", + ) + .bind("key_diagnostic") + .bind("生产密钥") + .bind("sk-diagnostic") + .bind("sk-diagnostic-secret") + .bind(now) + .execute(usage.pool()) + .await + .unwrap(); let mut success = success_record("eventually succeeded"); success.id = "usage_insights_success".to_string(); success.request_id = Some("req_retried".to_string()); + success.client_api_key_id = Some("key_diagnostic".to_string()); success.input_tokens = Some(100); success.output_tokens = Some(20); success.cached_tokens = Some(20); @@ -157,14 +171,18 @@ async fn usage_insights_should_collapse_retry_errors_to_terminal_requests() { retried_error.created_at = now - chrono::Duration::seconds(2); ops.append(&retried_error).await.unwrap(); - for (id, failure_class, seconds) in [ - ("ops_insights_failed_older", "upstream", 2), - ("ops_insights_failed_latest", "response_failed", 1), + for (id, failure_class, seconds, persist_class) in [ + ("ops_insights_failed_older", "upstream", 2, true), + ("ops_insights_failed_latest", "response_failed", 1, false), ] { let mut error = OpsErrorLog::new("upstream", failure_class); error.id = id.to_string(); error.request_id = Some("req_failed".to_string()); - error.failure_class = Some(failure_class.to_string()); + if persist_class { + error.failure_class = Some(failure_class.to_string()); + } else { + error.metadata = json!({ "upstreamCode": failure_class }); + } error.created_at = now - chrono::Duration::seconds(seconds); ops.append(&error).await.unwrap(); } @@ -211,6 +229,19 @@ async fn usage_insights_should_collapse_retry_errors_to_terminal_requests() { .iter() .any(|item| item["name"] == "usage-account@example.com → acct_usage") ); + + let api_key_diagnostics = admin_get( + &app, + "/api/admin/usage/records/insights/diagnostics?dimension=apiKey", + ) + .await; + assert!( + api_key_diagnostics["data"]["items"] + .as_array() + .unwrap() + .iter() + .any(|item| item["name"] == "生产密钥") + ); } #[tokio::test] diff --git a/backend/tests/dispatch/service/usage_logging.rs b/backend/tests/dispatch/service/usage_logging.rs index 20a5192b0..ed84b4726 100644 --- a/backend/tests/dispatch/service/usage_logging.rs +++ b/backend/tests/dispatch/service/usage_logging.rs @@ -796,6 +796,7 @@ async fn responses_stream_should_record_usage_record_after_late_disconnect() { assert_eq!(metadata["stream"], true); assert_eq!(metadata["failed"], true); assert_eq!(metadata["upstreamCode"], "stream_disconnected"); + assert_eq!(event.failure_class.as_deref(), Some("stream_disconnected")); assert_eq!(metadata["failureSource"], "proxy"); assert_eq!(metadata["synthetic"], true); assert_rate_limit_header(&metadata, "retry-after", "11"); diff --git a/backend/tests/fleet/account_failure.rs b/backend/tests/fleet/account_failure.rs new file mode 100644 index 000000000..73dcb6e19 --- /dev/null +++ b/backend/tests/fleet/account_failure.rs @@ -0,0 +1,79 @@ +use codex_proxy_rs::{ + fleet::{ + account::AccountStatus, + account_failure::{AccountFailureKind, AccountStateEffect, classify_client_failure}, + }, + upstream::openai::transport::{ + CodexBackendTransport, CodexClientError, CodexUpstreamDiagnostics, + }, +}; +use reqwest::StatusCode; + +#[test] +fn deactivated_workspace_should_be_banned_before_generic_payment_classification() { + let failure = classify(402, r#"{"detail":{"code":"deactivated_workspace"}}"#); + + assert_eq!(failure.kind, AccountFailureKind::Banned); + assert_eq!( + failure.effect, + Some(AccountStateEffect::SetStatus(AccountStatus::Banned)) + ); +} + +#[test] +fn generic_payment_required_should_exhaust_quota() { + let failure = classify(402, r#"{"detail":{"code":"payment_required"}}"#); + + assert_eq!(failure.kind, AccountFailureKind::QuotaExhausted); + assert_eq!( + failure.effect, + Some(AccountStateEffect::SetStatus(AccountStatus::QuotaExhausted)) + ); +} + +#[test] +fn unauthorized_should_expire_account() { + let failure = classify(401, r#"{"error":{"message":"unauthorized"}}"#); + + assert_eq!(failure.kind, AccountFailureKind::Expired); + assert_eq!( + failure.effect, + Some(AccountStateEffect::SetStatus(AccountStatus::Expired)) + ); +} + +#[test] +fn rate_limit_should_create_temporary_quota_effect() { + let failure = classify(429, r#"{"error":{"code":"rate_limit_exceeded"}}"#); + + assert_eq!(failure.kind, AccountFailureKind::RateLimited); + assert!(matches!( + failure.effect, + Some(AccountStateEffect::MarkQuotaLimitedUntil(_)) + )); +} + +#[test] +fn html_forbidden_page_should_not_ban_account() { + let error = upstream_error(403, "temporary edge failure"); + + assert!(classify_client_failure(&error).is_none()); +} + +fn classify( + status: u16, + body: &str, +) -> codex_proxy_rs::fleet::account_failure::ClassifiedAccountFailure { + classify_client_failure(&upstream_error(status, body)).expect("failure should be classified") +} + +fn upstream_error(status: u16, body: &str) -> CodexClientError { + CodexClientError::Upstream { + status: StatusCode::from_u16(status).unwrap(), + body: body.to_string(), + retry_after_seconds: None, + diagnostics: CodexUpstreamDiagnostics::with_status(status), + set_cookie_headers: Vec::new(), + transport: CodexBackendTransport::HttpSse, + } +} diff --git a/backend/tests/fleet/mod.rs b/backend/tests/fleet/mod.rs index a4fac2f80..f0adb646b 100644 --- a/backend/tests/fleet/mod.rs +++ b/backend/tests/fleet/mod.rs @@ -1,3 +1,4 @@ +pub(crate) mod account_failure; pub(crate) mod cookies; pub(crate) mod pool; pub(crate) mod quota; diff --git a/backend/tests/fleet/quota.rs b/backend/tests/fleet/quota.rs index a47d3f8c5..bcf7ddec0 100644 --- a/backend/tests/fleet/quota.rs +++ b/backend/tests/fleet/quota.rs @@ -1,13 +1,16 @@ use std::{collections::HashMap, sync::Arc, time::Duration as StdDuration}; use chrono::{Duration, Utc}; -use codex_proxy_rs::fleet::account::AccountStatus; -use codex_proxy_rs::fleet::cookies::PgCookieStore; -use codex_proxy_rs::fleet::quota::{ - QuotaRefreshService, quota_from_usage, quota_snapshot_limit_reached, - quota_snapshot_limit_window_seconds, quota_snapshot_reset_at, +use codex_proxy_rs::fleet::{ + account::AccountStatus, + cookies::PgCookieStore, + pool::{AccountPoolOptions, AccountPoolService}, + quota::{ + QuotaRefreshService, quota_from_usage, quota_snapshot_limit_reached, + quota_snapshot_limit_window_seconds, quota_snapshot_reset_at, + }, + store::{AccountStore, NewAccount, PgAccountStore}, }; -use codex_proxy_rs::fleet::store::{NewAccount, PgAccountStore}; use codex_proxy_rs::infra::identity::AccountPseudonymizer; use codex_proxy_rs::telemetry::account_usage::store::PgAccountUsageStore; use codex_proxy_rs::upstream::openai::transport::CodexBackendClient; @@ -24,6 +27,15 @@ fn test_account_pseudonymizer() -> Arc { Arc::new(AccountPseudonymizer::new([7; 32])) } +fn test_account_pool(pool: &sqlx::PgPool) -> Arc { + Arc::new(AccountPoolService::new( + Arc::new(PgAccountStore::new(pool.clone())) as Arc, + Arc::new(PgAccountUsageStore::new(pool.clone())), + AccountPoolOptions::default(), + 0, + )) +} + #[test] fn quota_from_usage_should_preserve_spend_control_individual_limit() { let usage = json!({ @@ -321,6 +333,7 @@ async fn quota_refresh_service_should_send_usage_cookie_when_cookie_store_is_con store, PgAccountUsageStore::new(pool.clone()), Arc::new(codex), + test_account_pool(&pool), test_account_pseudonymizer(), ) .with_cookie_store(cookies); @@ -403,6 +416,7 @@ async fn quota_refresh_service_should_fetch_usage_for_quota_locked_accounts_and_ store.clone(), PgAccountUsageStore::new(pool.clone()), Arc::new(codex), + test_account_pool(&pool), test_account_pseudonymizer(), ); @@ -435,6 +449,49 @@ async fn quota_refresh_service_should_fetch_usage_for_quota_locked_accounts_and_ ); } +#[tokio::test] +async fn quota_refresh_service_should_ban_deactivated_workspace() { + let server = MockServer::start().await; + Mock::given(method("GET")) + .and(path("/api/codex/usage")) + .respond_with( + ResponseTemplate::new(402) + .set_body_json(json!({ "detail": { "code": "deactivated_workspace" } })), + ) + .expect(1) + .mount(&server) + .await; + + let (pool, _guard) = init_test_db("quota-refresh-deactivated-workspace").await; + let store = PgAccountStore::new(pool.clone()); + insert_quota_locked_account(&store, &pool, "acct-deactivated", "access-token").await; + let codex = Arc::new(CodexBackendClient::new( + reqwest::Client::new(), + server.uri(), + crate::support::fingerprint::runtime_test_fingerprint(), + )); + let service = QuotaRefreshService::new( + store, + PgAccountUsageStore::new(pool.clone()), + codex, + test_account_pool(&pool), + test_account_pseudonymizer(), + ); + + let summary = service + .refresh_locked_accounts(&mut HashMap::new()) + .await + .expect("quota refresh should complete"); + let status: String = sqlx::query_scalar("select status from accounts where id = $1") + .bind("acct-deactivated") + .fetch_one(&pool) + .await + .unwrap(); + + assert_eq!(summary.failed, 1); + assert_eq!(status, "banned"); +} + #[tokio::test] async fn quota_refresh_service_should_refresh_quota_exhausted_accounts() { let server = MockServer::start().await; @@ -479,6 +536,7 @@ async fn quota_refresh_service_should_refresh_quota_exhausted_accounts() { store, PgAccountUsageStore::new(pool.clone()), Arc::new(codex), + test_account_pool(&pool), test_account_pseudonymizer(), ); @@ -515,6 +573,7 @@ async fn quota_refresh_service_should_skip_recent_locked_account_before_cooldown store, PgAccountUsageStore::new(pool.clone()), Arc::new(codex), + test_account_pool(&pool), test_account_pseudonymizer(), 1_800, ); @@ -555,6 +614,7 @@ async fn quota_refresh_service_should_skip_recent_locked_account_inside_cooldown store, PgAccountUsageStore::new(pool.clone()), Arc::new(codex), + test_account_pool(&pool), test_account_pseudonymizer(), 1_800, ); @@ -612,6 +672,7 @@ async fn quota_refresh_service_should_bypass_recent_skip_after_cooldown_grace() store, PgAccountUsageStore::new(pool.clone()), Arc::new(codex), + test_account_pool(&pool), test_account_pseudonymizer(), 1_800, ); @@ -667,6 +728,7 @@ async fn quota_refresh_service_should_stagger_multiple_locked_account_requests() store, PgAccountUsageStore::new(pool.clone()), Arc::new(codex), + test_account_pool(&pool), test_account_pseudonymizer(), ) .with_request_spacing(StdDuration::from_millis(200)); diff --git a/backend/tests/support/storage.rs b/backend/tests/support/storage.rs index 3e92236e6..2c6207e55 100644 --- a/backend/tests/support/storage.rs +++ b/backend/tests/support/storage.rs @@ -3,7 +3,7 @@ use std::{env, sync::OnceLock, thread}; use chrono::{DateTime, Utc}; use codex_proxy_rs::{ auth::store::{PgAdminUserStore, RedisAdminSessionStore}, - bootstrap::services::BackgroundTaskStores, + bootstrap::{config::BootstrapConfig, services::BackgroundTaskStores}, dispatch::affinity::RedisSessionAffinityStore, fleet::{cookies::PgCookieStore, refresh::RedisRefreshLeaseStore, store::PgAccountStore}, infra::{database::migrate, redis::RedisConnection}, @@ -24,6 +24,7 @@ const DEFAULT_TEST_REDIS_URL: &str = "redis://127.0.0.1:6379"; static DATABASE_INIT_LIMIT: tokio::sync::Semaphore = tokio::sync::Semaphore::const_new(2); static REDIS_TEST_CLEANUP: tokio::sync::OnceCell<()> = tokio::sync::OnceCell::const_new(); static REDIS_TEST_RUN_ID: OnceLock = OnceLock::new(); +static LOCAL_BOOTSTRAP_CONFIG: OnceLock> = OnceLock::new(); pub(crate) struct TestDatabaseGuard { database_name: Option, @@ -186,11 +187,34 @@ pub(crate) fn background_task_stores(pool: PgPool, redis: RedisConnection) -> Ba } pub(crate) fn test_database_url() -> String { - env::var("CPR_TEST_DATABASE_URL").unwrap_or_else(|_| DEFAULT_TEST_DATABASE_URL.to_string()) + // 显式测试拓扑优先；本地开发默认跟随 Compose 配置；无本地配置的 CI 使用固定测试值。 + env::var("CPR_TEST_DATABASE_URL") + .ok() + .or_else(|| local_bootstrap_config().map(|config| config.database_url().to_string())) + .unwrap_or_else(|| DEFAULT_TEST_DATABASE_URL.to_string()) } pub(crate) fn test_redis_url() -> String { - env::var("CPR_TEST_REDIS_URL").unwrap_or_else(|_| DEFAULT_TEST_REDIS_URL.to_string()) + env::var("CPR_TEST_REDIS_URL") + .ok() + .or_else(|| local_bootstrap_config().map(|config| config.redis_url().to_string())) + .unwrap_or_else(|| DEFAULT_TEST_REDIS_URL.to_string()) +} + +fn local_bootstrap_config() -> Option<&'static BootstrapConfig> { + LOCAL_BOOTSTRAP_CONFIG + .get_or_init(|| { + let path = + std::path::Path::new(env!("CARGO_MANIFEST_DIR")).join("../deploy/config.yaml"); + if !path.is_file() { + return None; + } + Some( + BootstrapConfig::load_from_path(path) + .unwrap_or_else(|error| panic!("load local test service config: {error}")), + ) + }) + .as_ref() } pub(crate) fn timestamp(value: &str) -> DateTime { diff --git a/backend/tests/upstream/openai/transport/client.rs b/backend/tests/upstream/openai/transport/client.rs index 79d0e512b..3b1c9588e 100644 --- a/backend/tests/upstream/openai/transport/client.rs +++ b/backend/tests/upstream/openai/transport/client.rs @@ -5,27 +5,11 @@ use std::sync::{ use super::{accept_codex_test_websocket, completed_websocket_response, request_context}; use codex_proxy_rs::upstream::openai::protocol::responses::CodexResponsesRequest; -use codex_proxy_rs::upstream::openai::transport::{ - CodexBackendClient, CodexWebSocketPool, is_deactivated_workspace_error_body, -}; +use codex_proxy_rs::upstream::openai::transport::{CodexBackendClient, CodexWebSocketPool}; use futures::{SinkExt, StreamExt}; use tokio::net::TcpListener; use tokio_tungstenite::tungstenite::Message; -#[test] -fn deactivated_workspace_error_body_should_match_detail_code() { - let body = r#"{"detail":{"code":"deactivated_workspace"}}"#; - - assert!(is_deactivated_workspace_error_body(body)); -} - -#[test] -fn deactivated_workspace_error_body_should_not_match_other_json_paths() { - let body = r#"{"error":{"code":"deactivated_workspace"}}"#; - - assert!(!is_deactivated_workspace_error_body(body)); -} - #[tokio::test] async fn codex_backend_client_should_apply_configured_websocket_pool() { let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); diff --git a/docs/architecture.md b/docs/architecture.md index 6820d6a21..5262199a0 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -28,11 +28,11 @@ PostgreSQL 和 Redis 都是启动必需依赖。运行中任一健康检查失 ## 架构不变量 1. `v1/*` 使用 Request → Attempt → Stream 三层静态洋葱生命周期；管理端、健康检查和静态资源不进入业务洋葱。 -2. 一个功能只有一个规则 owner。阈值、分类、状态机和 decision 只能存在于对应 controller。 +2. 一个功能只有一个规则 owner。跨入口的账号失败语义属于 `fleet/account_failure.rs`；`v1/*` 的 retry、exhaustion 和 commit decision 属于对应 controller。 3. API 只解析和编码；lifecycle 只编排顺序、retry、commit 和 finalize；transport 只建立连接并产出 typed facts。 4. 上游响应一旦可能收到 payload，transport 或账号都不得自动重放同一请求。 5. 账号认证身份按 attempt 重建；客户端会话拓扑按原值透传；两者不能共用一个“身份”抽象。 -6. 账号明确失效先在线性化的内存路径生效并驱逐 WebSocket，再异步持久化 PostgreSQL。 +6. `v1/*` 热路径中的账号明确失效先在线性化的内存路径生效并驱逐 WebSocket，再异步持久化 PostgreSQL；管理端和后台任务可等待同一状态 effect 持久化完成。 7. 每项生产能力只有一条执行路径，不设置兼容 wrapper、双状态机或运行时切换的新旧实现。 8. `backend/src/` 只放生产代码；所有测试和 test-only helper 位于 `backend/tests/`。 @@ -104,6 +104,7 @@ backend/src/ ```text upstream/openai/ ├── token_client.rs +├── failure.rs ├── protocol/ │ ├── responses.rs │ ├── events.rs @@ -127,6 +128,7 @@ upstream/openai/ ``` - `protocol` 定义 Responses body、canonical SSE/WS 事实和序列化，不包含账号策略。 +- `failure.rs` 将各 transport 错误规范化为稳定的 `UpstreamFailureFacts`，不解释账号状态。 - `transport` 拥有 HTTP/SSE、WebSocket、TLS、header、连接池、熔断和 transport metrics。 - `fingerprint` 拥有 Codex Desktop 指纹的存储、运行时快照和更新。 - `token_client.rs` 只实现 OAuth token 刷新协议。 @@ -137,6 +139,7 @@ upstream/openai/ - `store`：账号与 quota 的 PostgreSQL 映射和事务写入。 - `pool`：进程内账号快照、并发 slot、请求间隔和候选 lease。 +- `account_failure.rs`：HTTP、SSE、手动额度查询和后台额度刷新共享的账号失败分类与状态 effect。 - `scheduler`：候选过滤与排序的唯一 owner，支持 `smart`、`quota_reset_priority`、`round_robin`、`sticky`。 - `scheduler/feedback`：进程内 EWMA 错误率和首字延迟。 - `quota`：配额查询、窗口与运行时限制。 @@ -225,15 +228,16 @@ Stream terminal 包括 completed、incomplete、failed、上游断开、下游 | --- | --- | | previous response 所有权、scope、full replay | `controllers/history.rs` | | cyber policy 会话排除和 CAS 清理 | `controllers/cyber_policy/` | -| quota 与 rate limit | `controllers/quota.rs` | -| 封禁、过期、模型不支持等账号事实 | `controllers/account_failure.rs` | -| 内存失效、异步持久化和 WS 驱逐 effect | `controllers/account_state.rs` | +| v1 quota 与 rate limit attempt decision | `controllers/quota.rs` | +| v1 封禁、过期、模型不支持的 exhaustion decision | `controllers/account_failure.rs` | +| transport 错误到稳定上游事实 | `upstream/openai/failure.rs` | +| 跨入口账号失败分类、状态 effect 与 WS 驱逐 | `fleet/account_failure.rs` | | Cloudflare Cookie 与冷却 | `controllers/cloudflare.rs` | | 会话亲和写入 | `controllers/affinity.rs` | | usage 结算 | `controllers/usage.rs` | | 代理事件与 transport metrics | `controllers/telemetry/` | -Controller 之间不直接调用。跨功能优先级只在 `ControllerSet` 声明；API、transport、store 和 lifecycle 不重复识别功能语义。 +Controller 之间不直接调用。跨功能优先级只在 `ControllerSet` 声明；API、transport、store 和 lifecycle 不重复识别功能语义。Controller 只把共享账号分类翻译为 v1 生命周期 decision；手动额度查询和后台额度刷新执行同一状态 effect，但各自保留自己的返回与调度职责。 ## 上游双热传输 @@ -327,10 +331,11 @@ Scheduler 先过滤不可用账号，再按运行时 `rotation_strategy` 排序 明确上游事实的处理顺序： 1. transport 提取 status、code、type、message、retry-after 等事实。 -2. controller 分类为 quota、rate limit、expired、disabled、banned、model unsupported、Cloudflare 或 cyber policy。 -3. 对账号全局事实，账号池 mutex 内立即失效 runtime state 并移除 slot。 -4. 驱逐该账号的 idle/busy/connecting WebSocket entry。 -5. lifecycle 从冻结候选中选择下一账号。 +2. `fleet/account_failure.rs` 将跨入口账号事实统一分类为 quota、rate limit、expired、disabled、banned 或 model unsupported；Cloudflare 与 cyber policy 保持各自 owner。 +3. v1 controller 把共享分类翻译为 retry/exhaustion decision。 +4. 对账号全局事实，账号池 mutex 内立即失效 runtime state 并移除 slot。 +5. 驱逐该账号的 idle/busy/connecting WebSocket entry。 +6. lifecycle 从冻结候选中选择下一账号。 6. PostgreSQL 状态写入后台有序队列。 数据库延迟不会延长当前换号路径，也不会使并发请求重新租到已失效账号。 diff --git a/frontend/src/views/usage/components/UsageDiagnosticCard.vue b/frontend/src/views/usage/components/UsageDiagnosticCard.vue index 2497de8b0..060353573 100644 --- a/frontend/src/views/usage/components/UsageDiagnosticCard.vue +++ b/frontend/src/views/usage/components/UsageDiagnosticCard.vue @@ -79,7 +79,9 @@ const displayItems = computed(() => const tableRows = computed(() => (isStale.value ? [] : displayItems.value)) function diagnosticNameDisplay(name: string) { - const full = name.trim() || '未知' + const raw = name.trim() || '未知' + const full = + dimension.value === 'transport' ? ({ websocket: 'WS', http_sse: 'SSE' }[raw] ?? raw) : raw if (dimension.value !== 'model' && dimension.value !== 'account') { return { primary: full, secondary: '', full } } @@ -141,7 +143,11 @@ function diagnosticNameDisplay(name: string) { v-if="row.nameDisplay.secondary" class="flex min-w-0 items-center gap-1.25 text-(--cp-text-secondary)" > - + {{ row.nameDisplay.secondary }}