diff --git a/backend/crates/gateway-admin/src/model/accounts.rs b/backend/crates/gateway-admin/src/model/accounts.rs index 14d07696e..00887e6b8 100644 --- a/backend/crates/gateway-admin/src/model/accounts.rs +++ b/backend/crates/gateway-admin/src/model/accounts.rs @@ -212,6 +212,9 @@ pub enum AccountConnectionTestEvent { message: String, provider_error_code: Option, provider_error_type: Option, + upstream_status: Option, + upstream_content_type: Option, + upstream_body: Option, account_status: AccountStatus, }, } diff --git a/backend/crates/gateway-admin/src/use_case/accounts.rs b/backend/crates/gateway-admin/src/use_case/accounts.rs index e0fae70ee..158a460ba 100644 --- a/backend/crates/gateway-admin/src/use_case/accounts.rs +++ b/backend/crates/gateway-admin/src/use_case/accounts.rs @@ -452,12 +452,33 @@ impl AccountsService for DefaultAccountsService { account_status: status, })) .collect(), - Err(error) => vec![AccountConnectionTestEvent::Failed { - message: error.client_message().to_owned(), - provider_error_code: error.client_error_code().map(ToOwned::to_owned), - provider_error_type: error.client_error_type().map(ToOwned::to_owned), - account_status: status, - }], + Err(error) => { + let upstream_status = error + .upstream_response() + .map(gateway_core::engine::probe::AccountProbeUpstreamResponse::status); + let upstream_content_type = error + .upstream_response() + .and_then(|response| response.content_type()) + .and_then(|value| std::str::from_utf8(value).ok()) + .map(ToOwned::to_owned); + let upstream_body = error + .upstream_response() + .map(|response| String::from_utf8_lossy(response.body()).into_owned()); + let message = upstream_body + .as_deref() + .filter(|body| !body.is_empty()) + .unwrap_or_else(|| error.client_message()) + .to_owned(); + vec![AccountConnectionTestEvent::Failed { + message, + provider_error_code: error.client_error_code().map(ToOwned::to_owned), + provider_error_type: error.client_error_type().map(ToOwned::to_owned), + upstream_status, + upstream_content_type, + upstream_body, + account_status: status, + }] + } } }) .flat_map(futures::stream::iter); @@ -468,22 +489,25 @@ impl AccountsService for DefaultAccountsService { fn account_status(account: &AccountRecord, now: chrono::DateTime) -> AccountStatus { if !account.enabled { AccountStatus::Disabled + } else if account.availability == crate::model::accounts::AccountAvailability::Banned { + AccountStatus::Banned + } else if account.availability == crate::model::accounts::AccountAvailability::QuotaExhausted + || (account.availability == crate::model::accounts::AccountAvailability::Cooldown + && account.availability_reason.as_deref() == Some("usage_limit_exhausted") + && account.cooldown_until.is_none_or(|until| until > now)) + { + AccountStatus::QuotaExhausted + } else if matches!( + account.availability, + crate::model::accounts::AccountAvailability::Expired + | crate::model::accounts::AccountAvailability::Invalid + ) || account + .access_token_expires_at + .is_some_and(|expires_at| expires_at <= now) + { + AccountStatus::Expired } else { - match account.availability { - crate::model::accounts::AccountAvailability::Banned => AccountStatus::Banned, - crate::model::accounts::AccountAvailability::QuotaExhausted => { - AccountStatus::QuotaExhausted - } - crate::model::accounts::AccountAvailability::Expired - | crate::model::accounts::AccountAvailability::Invalid => AccountStatus::Expired, - _ if account - .access_token_expires_at - .is_some_and(|expires_at| expires_at <= now) => - { - AccountStatus::Expired - } - _ => AccountStatus::Active, - } + AccountStatus::Active } } diff --git a/backend/crates/gateway-admin/src/use_case/openai.rs b/backend/crates/gateway-admin/src/use_case/openai.rs index af6f59dac..511c5f18c 100644 --- a/backend/crates/gateway-admin/src/use_case/openai.rs +++ b/backend/crates/gateway-admin/src/use_case/openai.rs @@ -13,8 +13,7 @@ use crate::{ CredentialDeletionResult, CredentialDetails, CredentialImportCommit, CredentialImportResult, CredentialListQuery, CredentialMutation, CredentialMutationResult, CredentialPage, ImportCredentials, PrepareCredentialImport, - PrepareCredentialRotation, PreparedAuthorizationCredential, ProviderQuotaRequest, - RotateCredential, StartAuthorization, + PrepareCredentialRotation, ProviderQuotaRequest, RotateCredential, StartAuthorization, }, }, ports::{provider::ProviderAdmin, store::AccountStore}, @@ -195,10 +194,6 @@ impl OpenAiService for DefaultOpenAiService { "OpenAI authorization", ) .await?; - let creates_account = matches!( - &prepared.credential, - PreparedAuthorizationCredential::Create(_) - ); let result = commit_authorization( self.accounts.as_ref(), prepared, @@ -210,13 +205,11 @@ impl OpenAiService for DefaultOpenAiService { .account_facts_changed(std::slice::from_ref(&result.account_id)) .await; publish_committed(self.snapshot.as_ref(), result.config_revision).await?; - if creates_account { - self.observe_initial_quotas( - std::slice::from_ref(&result.account_id), - &context.request_id, - ) - .await; - } + self.observe_initial_quotas( + std::slice::from_ref(&result.account_id), + &context.request_id, + ) + .await; Ok(result) } diff --git a/backend/crates/gateway-admin/tests/use_case/accounts.rs b/backend/crates/gateway-admin/tests/use_case/accounts.rs index af1483a7a..15b1614b1 100644 --- a/backend/crates/gateway-admin/tests/use_case/accounts.rs +++ b/backend/crates/gateway-admin/tests/use_case/accounts.rs @@ -6,7 +6,7 @@ use futures::{StreamExt as _, future::BoxFuture}; use gateway_core::{ engine::{ credential::{OpaqueProviderData, ProviderAccountId}, - probe::{AccountProbe, AccountProbeRequest, AccountProbeResult}, + probe::{AccountProbe, AccountProbeError, AccountProbeRequest, AccountProbeResult}, }, error::{ClientVisibleUpstreamError, GatewayError, GatewayErrorKind}, operation::{GenerateRequest, Operation, ProtocolPayload}, @@ -794,6 +794,7 @@ async fn connection_test_should_probe_unavailable_account() { provider_error_code: Some(code), provider_error_type: Some(error_type), account_status: gateway_admin::model::accounts::AccountStatus::QuotaExhausted, + .. }) if message == "included usage exhausted" && code == "usage_exhausted" && error_type == "invalid_request_error" @@ -891,6 +892,36 @@ async fn accounts_list_should_return_complete_directory_semantics() { ); } +#[tokio::test] +async fn accounts_list_should_prefer_usage_limit_exhaustion_over_stale_token_expiry() { + let provider = FakeProviderAdmin::new("openai", events()); + let mut account = account_record("openai"); + account.availability = AccountAvailability::Cooldown; + account.availability_reason = Some("usage_limit_exhausted".to_owned()); + account.cooldown_until = Some(Utc::now() + TimeDelta::minutes(5)); + account.access_token_expires_at = Some(Utc::now() - TimeDelta::minutes(5)); + let store = FakeAccountStore::with_account(account, events()); + + let page = accounts_service(provider, store) + .await + .accounts() + .list(AccountListQuery { + page: 1, + page_size: gateway_admin::model::PageSize::new(20).expect("page size"), + provider_kind: None, + search: None, + status: None, + sort: None, + }) + .await + .expect("usage-limited account directory"); + + assert_eq!( + page.items.first().expect("account item").status, + gateway_admin::model::accounts::AccountStatus::QuotaExhausted, + ); +} + #[tokio::test] async fn accounts_list_should_attach_local_usage_to_quota_windows() { let provider = FakeProviderAdmin::new("openai", events()); @@ -1141,7 +1172,7 @@ impl AccountProbe for SuccessfulAccountProbe { fn probe( &self, _: AccountProbeRequest, - ) -> BoxFuture<'_, Result> { + ) -> BoxFuture<'_, Result> { Box::pin(async { Ok(AccountProbeResult { text: vec!["OK".to_owned()], @@ -1156,7 +1187,7 @@ impl AccountProbe for FailingAccountProbe { fn probe( &self, _: AccountProbeRequest, - ) -> BoxFuture<'_, Result> { + ) -> BoxFuture<'_, Result> { Box::pin(async { let detail = ClientVisibleUpstreamError::new( "included usage exhausted", @@ -1167,7 +1198,8 @@ impl AccountProbe for FailingAccountProbe { GatewayErrorKind::RateLimited, "upstream capacity is temporarily unavailable", ) - .with_client_visible_upstream_error(detail)) + .with_client_visible_upstream_error(detail) + .into()) }) } } diff --git a/backend/crates/gateway-admin/tests/use_case/mod.rs b/backend/crates/gateway-admin/tests/use_case/mod.rs index 89f3872c0..abecdf741 100644 --- a/backend/crates/gateway-admin/tests/use_case/mod.rs +++ b/backend/crates/gateway-admin/tests/use_case/mod.rs @@ -60,7 +60,7 @@ use gateway_admin::{ use gateway_core::{ engine::{ credential::ProviderAccountId, - probe::{AccountProbe, AccountProbeRequest, AccountProbeResult}, + probe::{AccountProbe, AccountProbeError, AccountProbeRequest, AccountProbeResult}, }, error::{GatewayError, GatewayErrorKind}, policy::ClientApiKeyId, @@ -647,12 +647,13 @@ impl AccountProbe for UnavailableProbe { fn probe( &self, _: AccountProbeRequest, - ) -> BoxFuture<'_, Result> { + ) -> BoxFuture<'_, Result> { Box::pin(async { Err(GatewayError::new( GatewayErrorKind::Internal, "test account probe is unavailable", - )) + ) + .into()) }) } } diff --git a/backend/crates/gateway-admin/tests/use_case/openai.rs b/backend/crates/gateway-admin/tests/use_case/openai.rs index 889e04c06..7fd9cfc92 100644 --- a/backend/crates/gateway-admin/tests/use_case/openai.rs +++ b/backend/crates/gateway-admin/tests/use_case/openai.rs @@ -411,8 +411,10 @@ async fn openai_reauthorization_should_commit_after_credential_revision_advances "store.commit_authorization", "guard.finish", "provider.account_facts_changed", + "provider.quota", ] ); + assert_eq!(provider.quota_requests().len(), 1); assert_eq!(store.audit_requests(), ["oauth-complete-openai"]); } diff --git a/backend/crates/gateway-api/src/admin/accounts.rs b/backend/crates/gateway-api/src/admin/accounts.rs index 21a8a8a4a..515a6c8bb 100644 --- a/backend/crates/gateway-api/src/admin/accounts.rs +++ b/backend/crates/gateway-api/src/admin/accounts.rs @@ -531,12 +531,18 @@ impl From for AccountConnectionTestEvent { message, provider_error_code, provider_error_type, + upstream_status, + upstream_content_type, + upstream_body, account_status, } => serde_json::json!({ "type": "error", "error": message, "providerErrorCode": provider_error_code, "providerErrorType": provider_error_type, + "upstreamStatus": upstream_status, + "upstreamContentType": upstream_content_type, + "upstreamBody": upstream_body, "accountStatus": account_status_name(account_status) }), }; diff --git a/backend/crates/gateway-api/tests/admin/accounts.rs b/backend/crates/gateway-api/tests/admin/accounts.rs index 452a4b427..a74f68e2a 100644 --- a/backend/crates/gateway-api/tests/admin/accounts.rs +++ b/backend/crates/gateway-api/tests/admin/accounts.rs @@ -337,6 +337,9 @@ mod actions { message: "upstream unavailable".to_owned(), provider_error_code: Some("usage_exhausted".to_owned()), provider_error_type: Some("invalid_request_error".to_owned()), + upstream_status: Some(429), + upstream_content_type: Some("application/json".to_owned()), + upstream_body: Some(r#"{"error":{"type":"usage_limit_reached"}}"#.to_owned()), account_status: AccountStatus::Active, }, ] @@ -368,6 +371,9 @@ mod actions { "error": "upstream unavailable", "providerErrorCode": "usage_exhausted", "providerErrorType": "invalid_request_error", + "upstreamStatus": 429, + "upstreamContentType": "application/json", + "upstreamBody": r#"{"error":{"type":"usage_limit_reached"}}"#, "accountStatus": "active" }), ] diff --git a/backend/crates/gateway-api/tests/admin/mod.rs b/backend/crates/gateway-api/tests/admin/mod.rs index 14d573ae5..b35bfa5d6 100644 --- a/backend/crates/gateway-api/tests/admin/mod.rs +++ b/backend/crates/gateway-api/tests/admin/mod.rs @@ -58,9 +58,8 @@ use gateway_api::admin::AdminSessionState; use gateway_core::{ engine::{ credential::ProviderAccountId, - probe::{AccountProbe, AccountProbeRequest, AccountProbeResult}, + probe::{AccountProbe, AccountProbeError, AccountProbeRequest, AccountProbeResult}, }, - error::GatewayError, policy::{ClientApiKeyId, RateLimits}, routing::{ ConfigRevision, ProviderKind, PublicModelId, UpstreamModelId, snapshot::SnapshotControl, @@ -675,7 +674,7 @@ impl AccountProbe for NoopProbe { fn probe( &self, _: AccountProbeRequest, - ) -> BoxFuture<'_, Result> { + ) -> BoxFuture<'_, Result> { Box::pin(async { panic!("unused account probe") }) } } diff --git a/backend/crates/gateway-core/src/engine/execution.rs b/backend/crates/gateway-core/src/engine/execution.rs index 05abfc6d7..4b049e550 100644 --- a/backend/crates/gateway-core/src/engine/execution.rs +++ b/backend/crates/gateway-core/src/engine/execution.rs @@ -17,7 +17,10 @@ use crate::engine::continuation::{ ContinuationBinding, NativeContinuationPin, NativeContinuationPort, PreviousResponseId, }; use crate::engine::coordinator::ResponseExecutionSession; -use crate::engine::probe::{AccountProbe, AccountProbeRequest, AccountProbeResult}; +use crate::engine::probe::{ + AccountProbe, AccountProbeError, AccountProbeRequest, AccountProbeResult, + AccountProbeUpstreamResponse, +}; use crate::engine::provider::ProviderRegistry; use crate::engine::{ AttemptCoordinator, AttemptRecord, CancellationToken, CoordinatedEvent, EngineError, @@ -426,7 +429,7 @@ impl DefaultExecutionService { async fn probe_inner( &self, request: AccountProbeRequest, - ) -> Result { + ) -> Result { let AccountProbeRequest { account_id, provider_kind, @@ -535,14 +538,19 @@ impl DefaultExecutionService { }) } - /// 探测失败先记录完整分类事实，再仅把 Provider 显式标记为客户端可见的 - /// 结构化 message/code/type 带入协议错误。 + /// 探测失败先记录脱敏分类事实，再把请求局部的原始上游响应交给认证管理端。 async fn observe_probe_failure( &self, observed: &ProbeObservation, started_at: SystemTime, error: &EngineError, - ) -> GatewayError { + ) -> AccountProbeError { + let upstream_response = match error { + EngineError::Provider(provider_error) => provider_error + .client_visible_upstream_response() + .map(AccountProbeUpstreamResponse::from_client_response), + _ => None, + }; if let EngineError::Provider(provider_error) = error { let latency = started_at.elapsed().unwrap_or_default(); let latency_ms = u64::try_from(latency.as_millis()).unwrap_or(u64::MAX); @@ -578,7 +586,7 @@ impl DefaultExecutionService { ); } } - gateway_error_from_engine(error) + AccountProbeError::new(gateway_error_from_engine(error), upstream_response) } } @@ -685,7 +693,7 @@ impl AccountProbe for DefaultExecutionService { fn probe( &self, request: AccountProbeRequest, - ) -> BoxFuture<'_, Result> { + ) -> BoxFuture<'_, Result> { Box::pin(async move { self.probe_inner(request).await }) } } diff --git a/backend/crates/gateway-core/src/engine/probe.rs b/backend/crates/gateway-core/src/engine/probe.rs index c28f15615..61b5b6e01 100644 --- a/backend/crates/gateway-core/src/engine/probe.rs +++ b/backend/crates/gateway-core/src/engine/probe.rs @@ -1,8 +1,11 @@ //! 管理端账号连通性测试的真实执行链端口。 +use std::fmt; + +use bytes::Bytes; use futures::future::BoxFuture; -use crate::error::GatewayError; +use crate::error::{ClientVisibleUpstreamResponse, GatewayError, GatewayErrorKind}; use crate::routing::{ProviderKind, UpstreamModelId}; use crate::{engine::credential::ProviderAccountId, operation::Operation}; @@ -19,9 +22,111 @@ pub struct AccountProbeResult { pub text: Vec, } +/// 仅供当前管理端连接测试展示的原始上游失败响应。 +/// +/// 正文不进入 `Debug`、日志或持久化；Core 只在探测终态从原始 Provider 错误显式复制。 +#[derive(PartialEq, Eq)] +pub struct AccountProbeUpstreamResponse { + status: u16, + content_type: Option>, + body: Bytes, +} + +impl AccountProbeUpstreamResponse { + pub(crate) fn from_client_response(response: &ClientVisibleUpstreamResponse) -> Self { + Self { + status: response.status(), + content_type: response.content_type().map(<[u8]>::to_vec), + body: response.body().clone(), + } + } + + #[must_use] + pub const fn status(&self) -> u16 { + self.status + } + + #[must_use] + pub fn content_type(&self) -> Option<&[u8]> { + self.content_type.as_deref() + } + + #[must_use] + pub const fn body(&self) -> &Bytes { + &self.body + } +} + +impl fmt::Debug for AccountProbeUpstreamResponse { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter + .debug_struct("AccountProbeUpstreamResponse") + .field("status", &self.status) + .field( + "content_type", + &self.content_type.as_ref().map(|_| ""), + ) + .field("body", &"") + .finish() + } +} + +/// 管理端连接测试的终态错误。 +/// +/// `gateway` 保留稳定分类，`upstream_response` 只面向本次认证管理请求展示源响应。 +#[derive(Debug)] +pub struct AccountProbeError { + gateway: GatewayError, + upstream_response: Option, +} + +impl AccountProbeError { + #[must_use] + pub const fn new( + gateway: GatewayError, + upstream_response: Option, + ) -> Self { + Self { + gateway, + upstream_response, + } + } + + #[must_use] + pub const fn kind(&self) -> GatewayErrorKind { + self.gateway.kind() + } + + #[must_use] + pub fn client_message(&self) -> &str { + self.gateway.client_message() + } + + #[must_use] + pub fn client_error_type(&self) -> Option<&str> { + self.gateway.client_error_type() + } + + #[must_use] + pub fn client_error_code(&self) -> Option<&str> { + self.gateway.client_error_code() + } + + #[must_use] + pub const fn upstream_response(&self) -> Option<&AccountProbeUpstreamResponse> { + self.upstream_response.as_ref() + } +} + +impl From for AccountProbeError { + fn from(gateway: GatewayError) -> Self { + Self::new(gateway, None) + } +} + pub trait AccountProbe: Send + Sync { fn probe( &self, request: AccountProbeRequest, - ) -> BoxFuture<'_, Result>; + ) -> BoxFuture<'_, Result>; } diff --git a/backend/crates/gateway-core/src/error.rs b/backend/crates/gateway-core/src/error.rs index f62e9c079..d0f3ae1fe 100644 --- a/backend/crates/gateway-core/src/error.rs +++ b/backend/crates/gateway-core/src/error.rs @@ -256,9 +256,9 @@ impl fmt::Debug for ClientVisibleUpstreamError { /// 只供当前客户端请求使用的原始上游 HTTP 失败响应。 /// -/// 该值不属于稳定诊断事实，不能进入日志、探针或持久化。它刻意不实现 -/// [`Clone`]；[`ProviderError`] 的普通 clone 会丢弃它，只有最终失败的原对象 -/// 才能把响应交给协议 adapter。 +/// 该值不属于稳定诊断事实，不能进入日志或持久化。它刻意不实现 [`Clone`]； +/// [`ProviderError`] 的普通 clone 会丢弃它，只有最终失败的原对象才能把响应交给 +/// 原客户端协议，或由认证管理端的账号连接测试显式复制到请求局部结果。 #[derive(PartialEq, Eq)] pub struct ClientVisibleUpstreamResponse { status: u16, diff --git a/backend/crates/gateway-core/tests/engine/execution.rs b/backend/crates/gateway-core/tests/engine/execution.rs index 29eaddeb4..5dd90c82b 100644 --- a/backend/crates/gateway-core/tests/engine/execution.rs +++ b/backend/crates/gateway-core/tests/engine/execution.rs @@ -8,6 +8,7 @@ use std::{ }; use async_trait::async_trait; +use bytes::Bytes; use futures::{executor::block_on, future::BoxFuture}; use gateway_core::engine::admission::{ ClientAdmissionDecision, ClientAdmissionError, ClientAdmissionPort, ClientAdmissionRecovery, @@ -34,7 +35,8 @@ use gateway_core::engine::{ ModelRequestId, NewModelRequest, ProbeFailure, RecoveryReport, UpstreamSendState, }; use gateway_core::error::{ - GatewayErrorKind, ProviderError, ProviderErrorKind, StoreError, StoreErrorKind, + ClientVisibleUpstreamResponse, GatewayErrorKind, ProviderError, ProviderErrorKind, StoreError, + StoreErrorKind, }; use gateway_core::operation::{GenerateRequest, Operation, OperationKind, ProtocolPayload}; use gateway_core::policy::{ClientApiKeyId, ClientPolicy, PlaintextClientApiKey, RateLimits}; @@ -108,6 +110,19 @@ fn probe_failures_should_be_observable_without_a_model_request_row() { .expect_err("the provider rejects every probe"); assert_eq!(error.kind(), GatewayErrorKind::UpstreamUnavailable); + let upstream = error + .upstream_response() + .expect("probe must preserve its request-local upstream response"); + assert_eq!(upstream.status(), 502); + assert_eq!( + upstream.content_type(), + Some(b"application/json".as_slice()) + ); + assert_eq!( + upstream.body(), + &Bytes::from_static(br#"{"error":{"message":"source upstream failure"}}"#), + ); + assert!(!format!("{error:?}").contains("source upstream failure")); assert!(!store.touched.load(Ordering::SeqCst)); assert_eq!(store.probe_failures(), vec!["transport".to_owned()]); } @@ -166,10 +181,14 @@ impl Provider for FailingProvider { _request: ProviderRequest, _context: AttemptContext, ) -> Result { - Err(ProviderError::new( - ProviderErrorKind::Transport, - UpstreamSendState::NotSent, - )) + Err( + ProviderError::new(ProviderErrorKind::Transport, UpstreamSendState::NotSent) + .with_client_visible_upstream_response(ClientVisibleUpstreamResponse::new( + 502, + Some(b"application/json".to_vec()), + Bytes::from_static(br#"{"error":{"message":"source upstream failure"}}"#), + )), + ) } } diff --git a/backend/crates/gateway-protocol/src/openai/events.rs b/backend/crates/gateway-protocol/src/openai/events.rs index f185c9ccf..0bbb23a42 100644 --- a/backend/crates/gateway-protocol/src/openai/events.rs +++ b/backend/crates/gateway-protocol/src/openai/events.rs @@ -151,7 +151,7 @@ pub fn retry_after_seconds_from_value(value: &Value) -> Option { .or_else(|| { error .get("resets_in_seconds") - .and_then(Value::as_u64) + .and_then(positive_u64) .filter(|seconds| *seconds > 0) }) .or_else(|| retry_after_seconds_from_resets_at(error)) @@ -437,7 +437,7 @@ fn nested_number_field(value: &Value, path: &[&str]) -> Option { } fn retry_after_seconds_from_resets_at(error: &Value) -> Option { - let resets_at = error.get("resets_at").and_then(Value::as_u64)?; + let resets_at = error.get("resets_at").and_then(positive_u64)?; let now = std::time::SystemTime::now() .duration_since(std::time::UNIX_EPOCH) .ok()? @@ -445,6 +445,13 @@ fn retry_after_seconds_from_resets_at(error: &Value) -> Option { (resets_at > now).then_some(resets_at - now) } +fn positive_u64(value: &Value) -> Option { + value + .as_u64() + .or_else(|| value.as_str().and_then(|value| value.parse::().ok())) + .filter(|value| *value > 0) +} + fn retry_after_seconds_from_rate_limit_message(error: &Value) -> Option { let code = error .get("code") diff --git a/backend/crates/gateway-protocol/tests/openai/events.rs b/backend/crates/gateway-protocol/tests/openai/events.rs index ca307b636..9a0ec66b2 100644 --- a/backend/crates/gateway-protocol/tests/openai/events.rs +++ b/backend/crates/gateway-protocol/tests/openai/events.rs @@ -226,6 +226,32 @@ fn retry_after_should_read_structured_response_delay() { assert_eq!(retry_after_seconds_from_body(&body), Some(45)); } +#[test] +fn retry_after_should_read_string_reset_delay() { + let body = json!({ + "error": {"type": "usage_limit_reached", "resets_in_seconds": "46"} + }) + .to_string(); + + assert_eq!(retry_after_seconds_from_body(&body), Some(46)); +} + +#[test] +fn retry_after_should_read_string_reset_timestamp() { + let reset_at = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .expect("system clock") + .as_secs() + + 60; + let body = json!({ + "error": {"type": "usage_limit_reached", "resets_at": reset_at.to_string()} + }) + .to_string(); + + let retry_after = retry_after_seconds_from_body(&body).expect("retry-after"); + assert!((59..=60).contains(&retry_after)); +} + #[test] fn retry_after_should_read_nested_error_delay() { let body = json!({ diff --git a/backend/crates/gateway-store/src/postgres/provider_accounts.rs b/backend/crates/gateway-store/src/postgres/provider_accounts.rs index 78f95401c..b7b828dc2 100644 --- a/backend/crates/gateway-store/src/postgres/provider_accounts.rs +++ b/backend/crates/gateway-store/src/postgres/provider_accounts.rs @@ -1564,7 +1564,11 @@ fn admin_account_status( AdminAccountStatus::Disabled } else if account.availability == AccountAvailability::Banned { AdminAccountStatus::Banned - } else if account.availability == AccountAvailability::QuotaExhausted { + } else if account.availability == AccountAvailability::QuotaExhausted + || (account.availability == AccountAvailability::Cooldown + && account.availability_reason.as_deref() == Some("usage_limit_exhausted") + && account.cooldown_until.is_none_or(|until| until > now)) + { AdminAccountStatus::QuotaExhausted } else if account.availability == AccountAvailability::Expired || account.availability == AccountAvailability::Invalid @@ -1983,6 +1987,22 @@ async fn rotate_provider_account_in_transaction( has_refresh_token = $8, access_token_expires_at = $9, next_refresh_at = $10, + availability = case + when availability <> 'quota_exhausted' then 'ready' + else availability + end, + availability_reason = case + when availability <> 'quota_exhausted' then null + else availability_reason + end, + cooldown_until = case + when availability <> 'quota_exhausted' then null + else cooldown_until + end, + availability_observed_at = case + when availability <> 'quota_exhausted' then now() + else availability_observed_at + end, updated_at = now() where id = $1 and provider_kind = $2 and credential_revision = $3 diff --git a/backend/crates/gateway-store/tests/postgres/provider_accounts.rs b/backend/crates/gateway-store/tests/postgres/provider_accounts.rs index 0f6d212f4..9f2e00e3c 100644 --- a/backend/crates/gateway-store/tests/postgres/provider_accounts.rs +++ b/backend/crates/gateway-store/tests/postgres/provider_accounts.rs @@ -912,6 +912,16 @@ async fn provider_account_admin_mutations_are_scoped_audited_and_atomic() { )); assert_eq!(current_revision(&database.pool).await, 2); + sqlx::query( + "update provider_accounts + set availability = 'expired', availability_reason = 'credential_expired' + where id = $1", + ) + .bind("acct_admin_a") + .execute(&database.pool) + .await + .expect("seed stale credential state"); + let rotation = repository .rotate_provider_account(RotateProviderAccount { scope: scope.clone(), @@ -923,6 +933,15 @@ async fn provider_account_admin_mutations_are_scoped_audited_and_atomic() { .expect("rotate provider account"); assert_eq!(rotation.config_revision.get(), 3); assert_eq!(rotation.credential_revision.get(), 2); + let restored: (String, Option, Option>) = sqlx::query_as( + "select availability, availability_reason, cooldown_until + from provider_accounts where id = $1", + ) + .bind("acct_admin_a") + .fetch_one(&database.pool) + .await + .expect("load restored account state"); + assert_eq!(restored, ("ready".to_owned(), None, None)); let revision = repository .set_provider_account_enabled_admin(SetProviderAccountEnabled { @@ -965,6 +984,63 @@ async fn provider_account_admin_mutations_are_scoped_audited_and_atomic() { database.close().await; } +#[tokio::test] +async fn verified_credential_rotation_preserves_quota_exhaustion() { + let Some(database) = TestDatabase::create("provider_account_rotation_quota").await else { + return; + }; + let repository = PgProviderAccountRepository::new(database.pool.clone()); + let scope = ProviderAccountAdminScope { + provider_kind: "openai".to_owned(), + }; + let mut exhausted = account("acct_rotation_quota", "user-rotation-quota"); + exhausted.availability = AccountAvailability::QuotaExhausted; + exhausted.availability_reason = Some("quota_exhausted".to_owned()); + repository + .import_provider_accounts(ImportProviderAccounts { + scope: scope.clone(), + accounts: vec![exhausted], + audit: audit( + "audit_rotation_quota_import", + "import", + "acct_rotation_quota", + ), + }) + .await + .expect("import exhausted account"); + + repository + .rotate_provider_account(RotateProviderAccount { + scope, + profile: profile("acct_rotation_quota", "reauthorized account"), + credential: credential_update("acct_rotation_quota", 1, "reauthorized-secret"), + audit: audit( + "audit_rotation_quota_rotate", + "reauthorize", + "acct_rotation_quota", + ), + }) + .await + .expect("rotate exhausted account credential"); + + let availability: (String, Option) = sqlx::query_as( + "select availability, availability_reason from provider_accounts where id = $1", + ) + .bind("acct_rotation_quota") + .fetch_one(&database.pool) + .await + .expect("load exhausted account after credential rotation"); + assert_eq!( + availability, + ( + "quota_exhausted".to_owned(), + Some("quota_exhausted".to_owned()), + ), + ); + + database.close().await; +} + fn account(id: &str, upstream_user_id: &str) -> NewProviderAccount { NewProviderAccount { id: id.to_owned(), diff --git a/backend/crates/providers/openai/src/credential/selector.rs b/backend/crates/providers/openai/src/credential/selector.rs index ccf8d8d22..f2f237d96 100644 --- a/backend/crates/providers/openai/src/credential/selector.rs +++ b/backend/crates/providers/openai/src/credential/selector.rs @@ -56,6 +56,11 @@ pub enum CodexAccountFailure { Banned, /// 账号信用额度已耗尽。 QuotaExhausted, + /// 当前用量窗口已耗尽；到重置时间后可自动恢复。 + UsageLimitExhausted { + /// 上游返回的窗口重置时长。 + retry_after: Option, + }, /// 账号触发临时用量限制。 RateLimited { /// 上游明确返回的最短冷却时长。 @@ -702,9 +707,19 @@ impl CodexCredentialSelector { None, CookieRecovery::None, ), + CodexAccountFailure::UsageLimitExhausted { retry_after } => { + return self + .record_resettable_limit_failure( + account, + retry_after, + "usage_limit_exhausted", + now, + ) + .await; + } CodexAccountFailure::RateLimited { retry_after } => { return self - .record_rate_limit_failure(account, retry_after, now) + .record_resettable_limit_failure(account, retry_after, "rate_limited", now) .await; } CodexAccountFailure::CloudflareChallenge { retry_after } => { @@ -743,23 +758,20 @@ impl CodexCredentialSelector { Ok(()) } - async fn record_rate_limit_failure( + async fn record_resettable_limit_failure( &self, account: &ProviderAccount, retry_after: Option, + reason: &'static str, observed_at: SystemTime, ) -> Result<(), CredentialSelectionError> { let current = self.current_account(account.id()).await?; if current.revision() != account.revision() { return Err(CredentialRepositoryError::RevisionConflict.into()); } - if matches!( - current.availability(), - AccountAvailability::Invalid - | AccountAvailability::Expired - | AccountAvailability::Banned - | AccountAvailability::QuotaExhausted - ) { + // 一次真实 429 证明上游已经接受当前凭据，可纠正旧的认证/封禁终态。 + // 结构化额度耗尽仍拥有更高优先级，不能被通用限流降级。 + if current.availability() == AccountAvailability::QuotaExhausted { return Ok(()); } let quota_exhausted = self @@ -776,7 +788,7 @@ impl CodexCredentialSelector { } else { ( AccountAvailability::Cooldown, - Some("rate_limited".to_owned()), + Some(reason.to_owned()), observed_at.checked_add(retry_after.unwrap_or(DEFAULT_RATE_LIMIT_COOLDOWN)), ) }; @@ -847,6 +859,8 @@ impl CodexCredentialSelector { AccountAvailability::Unknown | AccountAvailability::Cooldown | AccountAvailability::Expired + | AccountAvailability::Invalid + | AccountAvailability::Banned ) { return; diff --git a/backend/crates/providers/openai/src/provider.rs b/backend/crates/providers/openai/src/provider.rs index adcc07055..cbcd59e6d 100644 --- a/backend/crates/providers/openai/src/provider.rs +++ b/backend/crates/providers/openai/src/provider.rs @@ -2191,6 +2191,7 @@ const fn provider_error_kind(category: CodexFailureCategory) -> ProviderErrorKin CodexFailureCategory::IdentityVerificationRequired | CodexFailureCategory::Banned => { ProviderErrorKind::PermissionDenied } + CodexFailureCategory::UsageLimitExhausted => ProviderErrorKind::QuotaExhausted, CodexFailureCategory::RateLimited => ProviderErrorKind::RateLimited, CodexFailureCategory::QuotaExhausted => ProviderErrorKind::QuotaExhausted, CodexFailureCategory::CloudflareChallenge @@ -2221,6 +2222,11 @@ fn account_failure( Some(CodexAccountFailure::IdentityVerificationRequired) } CodexFailureCategory::Banned => Some(CodexAccountFailure::Banned), + CodexFailureCategory::UsageLimitExhausted => { + Some(CodexAccountFailure::UsageLimitExhausted { + retry_after: retry_after_seconds.map(Duration::from_secs), + }) + } CodexFailureCategory::RateLimited => Some(CodexAccountFailure::RateLimited { retry_after: retry_after_seconds.map(Duration::from_secs), }), diff --git a/backend/crates/providers/openai/src/transport/diagnostics.rs b/backend/crates/providers/openai/src/transport/diagnostics.rs index 0cd77a09c..f0936ab96 100644 --- a/backend/crates/providers/openai/src/transport/diagnostics.rs +++ b/backend/crates/providers/openai/src/transport/diagnostics.rs @@ -90,6 +90,8 @@ pub enum CodexFailureCategory { CredentialExpired, IdentityVerificationRequired, Banned, + /// 当前用量窗口已经耗尽，但存在自动恢复的重置时间。 + UsageLimitExhausted, RateLimited, QuotaExhausted, CloudflareChallenge, @@ -217,6 +219,7 @@ impl CodexUpstreamFailure { | CodexFailureCategory::CredentialExpired | CodexFailureCategory::IdentityVerificationRequired | CodexFailureCategory::Banned + | CodexFailureCategory::UsageLimitExhausted | CodexFailureCategory::RateLimited | CodexFailureCategory::QuotaExhausted | CodexFailureCategory::CloudflareChallenge @@ -424,24 +427,36 @@ fn classify_upstream_failure( if status == Some(StatusCode::NOT_FOUND) && body.trim().is_empty() { return CodexFailureCategory::CloudflarePathBlocked; } - if [code.as_str(), error_type.as_str()] - .into_iter() - .any(is_quota_signal) - || is_quota_message(&message) - || is_quota_message(&body) + let structured_failure_signals = [code.as_str(), error_type.as_str()]; + if structured_failure_signals.into_iter().any(is_quota_signal) || status == Some(StatusCode::PAYMENT_REQUIRED) { return CodexFailureCategory::QuotaExhausted; } - if [code.as_str(), error_type.as_str()] + if structured_failure_signals + .into_iter() + .any(is_usage_limit_signal) + { + return CodexFailureCategory::UsageLimitExhausted; + } + if structured_failure_signals .into_iter() .any(is_rate_limit_signal) - || is_rate_limit_message(&message) - || is_rate_limit_message(&body) - || status == Some(StatusCode::TOO_MANY_REQUESTS) { return CodexFailureCategory::RateLimited; } + if is_usage_limit_message(&message) || is_usage_limit_message(&body) { + return CodexFailureCategory::UsageLimitExhausted; + } + if is_quota_message(&message) || is_quota_message(&body) { + return CodexFailureCategory::QuotaExhausted; + } + if is_rate_limit_message(&message) || is_rate_limit_message(&body) { + return CodexFailureCategory::RateLimited; + } + if status == Some(StatusCode::TOO_MANY_REQUESTS) { + return CodexFailureCategory::RateLimited; + } if is_upstream_overload(&code) || is_upstream_overload(&message) { return CodexFailureCategory::Unavailable; } @@ -532,18 +547,22 @@ fn is_expired_credential(value: &str) -> bool { || value.contains("invalid api key") } -fn is_rate_limit_signal(value: &str) -> bool { +fn is_usage_limit_signal(value: &str) -> bool { matches!( value, "usage_limit_reached" - | "rate_limit_exceeded" - | "rate_limit_reached" - | "rate_limit_error" | "workspace_owner_usage_limit_reached" | "workspace_member_usage_limit_reached" ) } +fn is_rate_limit_signal(value: &str) -> bool { + matches!( + value, + "rate_limit_exceeded" | "rate_limit_reached" | "rate_limit_error" + ) +} + fn is_quota_signal(value: &str) -> bool { matches!( value, @@ -557,7 +576,11 @@ fn is_quota_signal(value: &str) -> bool { } fn is_rate_limit_message(value: &str) -> bool { - value.contains("rate limit") || value.contains("usage limit") + value.contains("rate limit") +} + +fn is_usage_limit_message(value: &str) -> bool { + value.contains("usage limit") } fn is_quota_message(value: &str) -> bool { diff --git a/backend/crates/providers/openai/tests/credential/contract.rs b/backend/crates/providers/openai/tests/credential/contract.rs index 838fb52ed..9d8e5d79f 100644 --- a/backend/crates/providers/openai/tests/credential/contract.rs +++ b/backend/crates/providers/openai/tests/credential/contract.rs @@ -765,6 +765,30 @@ fn rate_limited_failure_marks_account_cooldown_with_a_cooldown_deadline() { ); } +#[test] +fn usage_limit_exhaustion_is_a_resettable_account_cooldown() { + let store = Arc::new(MemoryAccountStore::default()); + create_account(&store, "acct_primary", "at-primary"); + let account = store.account("acct_primary").expect("account"); + let selector = selector(&store, Arc::new(TestLeaseCoordinator::default())); + + block_on(selector.record_failure( + &account, + CodexAccountFailure::UsageLimitExhausted { + retry_after: Some(Duration::from_secs(30)), + }, + )) + .expect("record usage-limit exhaustion"); + + let account = store.account("acct_primary").expect("persisted account"); + assert_eq!(account.availability(), AccountAvailability::Cooldown); + assert!( + account + .cooldown_until() + .is_some_and(|until| until > SystemTime::now()) + ); +} + #[test] fn rate_limited_failure_does_not_downgrade_persisted_quota_exhaustion() { let store = Arc::new(MemoryAccountStore::default()); @@ -854,7 +878,7 @@ fn rate_limited_failure_ignores_exhaustion_after_its_reset_time() { } #[test] -fn rate_limited_failure_preserves_terminal_account_state() { +fn authenticated_rate_limit_recovers_stale_authentication_state() { let store = Arc::new(MemoryAccountStore::default()); create_account(&store, "acct_primary", "at-primary"); let account = store.account("acct_primary").expect("account"); @@ -883,10 +907,45 @@ fn rate_limited_failure_preserves_terminal_account_state() { .account("acct_primary") .expect("persisted account") .availability(), - AccountAvailability::Invalid + AccountAvailability::Cooldown ); } +#[test] +fn successful_upstream_response_recovers_non_quota_terminal_states() { + for stale in [ + AccountAvailability::Expired, + AccountAvailability::Invalid, + AccountAvailability::Banned, + ] { + let store = Arc::new(MemoryAccountStore::default()); + create_account(&store, "acct_primary", "at-primary"); + let account = store.account("acct_primary").expect("account"); + block_on(store.apply_state_change(AccountStateChange { + account_id: account.id().clone(), + expected_revision: account.revision(), + availability: stale, + reason: Some("stale_observation".to_owned()), + cooldown_until: None, + observed_at: SystemTime::now(), + })) + .expect("seed stale state"); + let current = store.account("acct_primary").expect("stale account"); + let selector = selector(&store, Arc::new(TestLeaseCoordinator::default())); + + block_on(selector.record_success(¤t, None)); + + assert_eq!( + store + .account("acct_primary") + .expect("recovered account") + .availability(), + AccountAvailability::Ready, + "stale state {stale:?}", + ); + } +} + #[test] fn rate_limited_failures_for_distinct_accounts_do_not_conflict() { let store = Arc::new(MemoryAccountStore::default()); diff --git a/backend/crates/providers/openai/tests/transport/diagnostics.rs b/backend/crates/providers/openai/tests/transport/diagnostics.rs index caa7f6852..1b95fe122 100644 --- a/backend/crates/providers/openai/tests/transport/diagnostics.rs +++ b/backend/crates/providers/openai/tests/transport/diagnostics.rs @@ -156,7 +156,93 @@ fn structured_quota_signal_takes_precedence_over_http_429() { ); } +#[test] +fn structured_usage_limit_type_is_resettable_quota_exhaustion() { + assert_eq!( + classify( + 429, + r#"{"error":{"code":null,"type":"usage_limit_reached","message":"The usage limit has been reached"}}"# + ), + CodexFailureCategory::UsageLimitExhausted + ); +} + +#[test] +fn usage_limit_message_without_a_type_is_resettable_quota_exhaustion() { + assert_eq!( + classify( + 429, + r#"{"error":{"message":"The usage limit has been reached"}}"# + ), + CodexFailureCategory::UsageLimitExhausted + ); +} + #[test] fn bare_http_429_remains_a_temporary_rate_limit() { assert_eq!(classify(429, ""), CodexFailureCategory::RateLimited); } + +#[test] +fn sub2api_openai_account_failure_matrix_is_preserved() { + let cases = [ + ( + 400, + r#"{"error":{"message":"Organization has been disabled"}}"#, + CodexFailureCategory::Banned, + ), + ( + 400, + r#"{"error":{"message":"Identity verification is required"}}"#, + CodexFailureCategory::IdentityVerificationRequired, + ), + ( + 400, + r#"{"error":{"message":"invalid request payload"}}"#, + CodexFailureCategory::InvalidRequest, + ), + ( + 401, + r#"{"error":{"code":"token_revoked","message":"token has been revoked"}}"#, + CodexFailureCategory::CredentialExpired, + ), + ( + 401, + r#"{"detail":"Unauthorized"}"#, + CodexFailureCategory::CredentialExpired, + ), + ( + 402, + r#"{"detail":{"code":"deactivated_workspace","message":"workspace disabled"}}"#, + CodexFailureCategory::Banned, + ), + ( + 402, + r#"{"error":{"message":"payment required"}}"#, + CodexFailureCategory::QuotaExhausted, + ), + ( + 403, + r#"{"error":{"message":"access forbidden"}}"#, + CodexFailureCategory::PermissionDenied, + ), + ( + 429, + r#"{"error":{"type":"usage_limit_reached","message":"quota window reached"}}"#, + CodexFailureCategory::UsageLimitExhausted, + ), + ( + 529, + r#"{"error":{"type":"server_error","message":"overloaded"}}"#, + CodexFailureCategory::Unavailable, + ), + ]; + + for (status, body, expected) in cases { + assert_eq!( + classify(status, body), + expected, + "status={status} body={body}" + ); + } +} diff --git a/frontend/src/views/accounts/composables/useAccountConnectionTest.ts b/frontend/src/views/accounts/composables/useAccountConnectionTest.ts index 9cd1d3b28..f2c3debcb 100644 --- a/frontend/src/views/accounts/composables/useAccountConnectionTest.ts +++ b/frontend/src/views/accounts/composables/useAccountConnectionTest.ts @@ -39,25 +39,12 @@ interface AccountConnectionTestOptions { onAccountStatus: (accountId: string, status: string) => void } -// providerErrorCode 到测试弹窗提示文案的映射。 -const CONNECTION_TEST_ERROR_HINTS: Record = { - account_cooling_down: '账号冷却中', - account_capacity_busy: '账号并发已占满或请求间隔未到，请稍后重试', - no_eligible_account: '该账号没有可用于所选模型的凭据', - account_selector_unavailable: '账号调度状态暂时不可读，请稍后重试', -} - function connectionTestErrorText(event: any) { - const fallback = event.error || '测试连接失败' - const hint = CONNECTION_TEST_ERROR_HINTS[event.providerErrorCode] - if (!hint) - return fallback - if (event.providerErrorCode !== 'account_cooling_down') - return hint - const seconds = /retry in (\d+)s/.exec(fallback)?.[1] - return seconds - ? `${hint}，约 ${seconds} 秒后自动恢复（上游流被中断后触发）` - : `${hint}（上游流被中断后触发）` + if (typeof event.upstreamBody === 'string' && event.upstreamBody.length > 0) + return event.upstreamBody + if (Number.isInteger(event.upstreamStatus)) + return `HTTP ${event.upstreamStatus}` + return event.error || '测试连接失败' } export function useAccountConnectionTest(options: AccountConnectionTestOptions) { @@ -293,8 +280,15 @@ export function useAccountConnectionTest(options: AccountConnectionTestOptions) if (event.type === 'error') { applyAccountStatus(event.accountStatus) connectionTestError.value = connectionTestErrorText(event) - const detail = event.providerErrorCode || event.providerErrorType - ? { code: event.providerErrorCode, type: event.providerErrorType, message: event.error } + const detail = event.upstreamStatus || event.upstreamContentType || event.upstreamBody + || event.providerErrorCode || event.providerErrorType + ? { + status: event.upstreamStatus, + contentType: event.upstreamContentType, + body: event.upstreamBody, + code: event.providerErrorCode, + type: event.providerErrorType, + } : undefined appendConnectionTestLog(connectionTestError.value, 'danger', detail) finishConnectionTest('error')