diff --git a/backend/crates/providers/openai/src/credential/quota/mod.rs b/backend/crates/providers/openai/src/credential/quota/mod.rs index 0a5129c13..bac8a6c0d 100644 --- a/backend/crates/providers/openai/src/credential/quota/mod.rs +++ b/backend/crates/providers/openai/src/credential/quota/mod.rs @@ -36,7 +36,10 @@ use crate::transport::{CodexBackendClient, CodexClientError, CodexRequestContext use super::agent_identity::{CodexAgentIdentityTaskService, PreparedCodexRuntimeCredential}; use super::repository::{CodexCredentialRepository, CredentialRepositoryError}; -use recovery::{quota_state_reason, quota_state_transition, quota_success_state}; +use recovery::{ + authoritative_quota_snapshot_state, quota_state_reason, quota_state_transition, + quota_success_state, +}; use snapshot::{ parse_account_quota_snapshot, quota_projection_ttl, quota_snapshot_from_observation, scheduling_signals_from_snapshot, @@ -693,7 +696,7 @@ impl CodexCredentialQuotaService { } else { summary.updated += 1; } - if let Some(availability) = quota_success_state( + if let Some(availability) = authoritative_quota_snapshot_state( current.availability(), &refreshed_snapshot, previous_reset_at, @@ -1021,7 +1024,7 @@ impl CodexCredentialQuotaService { return Err(CodexCredentialQuotaError::RevisionConflict); } self.scheduling.observe(&snapshot); - if let Some(availability) = quota_success_state( + if let Some(availability) = authoritative_quota_snapshot_state( current.availability(), &refreshed_snapshot, previous_reset_at, diff --git a/backend/crates/providers/openai/src/credential/quota/recovery.rs b/backend/crates/providers/openai/src/credential/quota/recovery.rs index 023c81d95..e3865d0d2 100644 --- a/backend/crates/providers/openai/src/credential/quota/recovery.rs +++ b/backend/crates/providers/openai/src/credential/quota/recovery.rs @@ -1,6 +1,7 @@ -//! 确认额度耗尽后的恢复证据与 availability 转换。 +//! 权威额度快照的 availability 转换。 //! -//! `QuotaExhausted -> Ready` 只能由同一 credential revision 的权威 usage 快照证明: +//! 已触顶的 core primary 会立即把可调度账号标为 `QuotaExhausted`; +//! `QuotaExhausted -> Ready` 则只能由同一 credential revision 的权威 usage 快照证明: //! 新快照未耗尽、core primary reset 已推进到下一窗口,且该窗口用量低于 10%。 use std::time::SystemTime; @@ -14,6 +15,31 @@ use crate::transport::CodexClientError; const RESET_RECOVERY_MAX_USED_PERCENT: f64 = 10.0; +/// 基于成功的权威 usage 快照收敛可调度账号的额度状态。 +/// +/// 只接受上游明确拒绝使用且 core primary 触顶的证据;secondary、附加桶、 +/// credits 与 spend control 都不能单独把整个账号标成额度耗尽。 +pub(crate) fn authoritative_quota_snapshot_state( + current: AccountAvailability, + snapshot: &CodexAccountQuotaSnapshot, + previous_reset_at: Option, +) -> Option { + match current { + AccountAvailability::Ready | AccountAvailability::Unknown + if snapshot.authoritative_core_primary_exhausted() => + { + Some(AccountAvailability::QuotaExhausted) + } + AccountAvailability::QuotaExhausted => { + quota_success_state(current, snapshot, previous_reset_at) + } + AccountAvailability::Ready | AccountAvailability::Unknown => None, + AccountAvailability::Expired + | AccountAvailability::Banned + | AccountAvailability::Invalid => None, + } +} + /// 只处理 `QuotaExhausted` 的恢复。`allowed=true`、成功响应和百分比回落都不能单独覆盖 /// 一次已确认的额度拒绝;它们在 reset 未推进时可能只是上游滞后的观测。 /// diff --git a/backend/crates/providers/openai/src/credential/quota/snapshot.rs b/backend/crates/providers/openai/src/credential/quota/snapshot.rs index 5dfb9af54..7ce271234 100644 --- a/backend/crates/providers/openai/src/credential/quota/snapshot.rs +++ b/backend/crates/providers/openai/src/credential/quota/snapshot.rs @@ -105,6 +105,7 @@ pub struct CodexAccountQuotaSnapshot { credential_revision: CredentialRevision, observed_at: SystemTime, fact: CodexQuotaFact, + authoritative_core_primary_exhausted: bool, windows: Vec, } @@ -141,6 +142,12 @@ impl CodexAccountQuotaSnapshot { .find(|window| window.key() == CORE_PRIMARY_WINDOW_KEY) } + /// 上游成功 usage 明确拒绝请求,且 canonical core primary 已确认触顶。 + #[must_use] + pub(crate) const fn authoritative_core_primary_exhausted(&self) -> bool { + self.authoritative_core_primary_exhausted + } + /// 投影滚动已过期窗口:`reset_at` 已过的窗口 `used_percent=0`、 /// `limit_reached=false`、`reset_at` 滚到下一个周期(reset 到期自动归零)。 /// 只作用于展示/调度投影,不写回持久化 raw JSON。 @@ -316,6 +323,7 @@ pub(crate) fn parse_account_quota_snapshot( let object = usage .as_object() .ok_or(CodexCredentialQuotaError::InvalidCredentialData)?; + let authoritative_core_primary_exhausted = authoritative_core_primary_exhausted(object)?; let mut windows = Vec::new(); for limit in canonical_rate_limits(object)? { parse_rate_limit_windows( @@ -332,10 +340,55 @@ pub(crate) fn parse_account_quota_snapshot( credential_revision, observed_at, fact, + authoritative_core_primary_exhausted, windows, }) } +/// 账号状态只能由顶层 canonical Codex 额度桶的明确拒绝来改变。 +/// +/// `fact.exhausted()` 和窗口 `limit_reached` 都会聚合或继承 secondary、附加桶等事实, +/// 因此不能单独用来将整个账号标为 `QuotaExhausted`。 +fn authoritative_core_primary_exhausted( + usage: &Map, +) -> Result { + let Some(rate_limit) = usage.get("rate_limit") else { + return Ok(false); + }; + if rate_limit.is_null() { + return Ok(false); + } + let rate_limit = rate_limit + .as_object() + .ok_or(CodexCredentialQuotaError::InvalidCredentialData)?; + if optional_bool(rate_limit, "allowed")? != Some(false) { + return Ok(false); + } + let Some(primary_window) = rate_limit.get("primary_window") else { + return Ok(false); + }; + if primary_window.is_null() { + return Ok(false); + } + let primary_window = primary_window + .as_object() + .ok_or(CodexCredentialQuotaError::InvalidCredentialData)?; + let primary_window_reached = optional_bool(rate_limit, "limit_reached")?.unwrap_or(false) + || optional_bool(primary_window, "limit_reached")?.unwrap_or(false) + || primary_window + .get("used_percent") + .map(|value| { + value + .as_f64() + .filter(|value| value.is_finite() && *value >= 0.0) + .map(|value| value >= 100.0) + .ok_or(CodexCredentialQuotaError::InvalidCredentialData) + }) + .transpose()? + .unwrap_or(false); + Ok(primary_window_reached) +} + /// 一个规范化后的限流桶:`key` 用于稳定区分窗口,`source` 用于展示分组。 struct CanonicalRateLimit<'a> { key: String, diff --git a/backend/crates/providers/openai/tests/credential/quota/mod.rs b/backend/crates/providers/openai/tests/credential/quota/mod.rs index 389f7e073..a835223a0 100644 --- a/backend/crates/providers/openai/tests/credential/quota/mod.rs +++ b/backend/crates/providers/openai/tests/credential/quota/mod.rs @@ -487,7 +487,7 @@ async fn passive_rate_limit_headers_replace_stale_secondary_window() { } #[tokio::test] -async fn manual_quota_refresh_drops_empty_secondary_window() { +async fn manual_quota_refresh_marks_ready_account_exhausted_and_drops_empty_secondary_window() { let store = Arc::new(MemoryAccountStore::default()); let account_id = "acct_manual_secondary_placeholder"; create_account(&store, account_id).await; @@ -533,6 +533,13 @@ async fn manual_quota_refresh_drops_empty_secondary_window() { assert_eq!(snapshot.windows().len(), 1); assert_eq!(snapshot.windows()[0].used_percent(), Some(100.0)); + assert_eq!( + store + .account(account_id) + .expect("account after refresh") + .availability(), + AccountAvailability::QuotaExhausted + ); } #[tokio::test] @@ -824,7 +831,7 @@ async fn periodic_quota_synchronization_does_not_scan_stale_ready_accounts() { } #[tokio::test] -async fn periodic_quota_synchronization_preserves_availability_when_usage_peaks() { +async fn periodic_quota_synchronization_marks_ready_account_exhausted_when_primary_is_exhausted() { let store = Arc::new(MemoryAccountStore::default()); create_account(&store, "acct_periodic_stuck_cooldown").await; let account = store @@ -871,6 +878,50 @@ async fn periodic_quota_synchronization_preserves_availability_when_usage_peaks( "primary_window": {"used_percent": 100, "reset_at": reset_at} } }))) + .expect(1) + .mount(&server) + .await; + let service = quota_service_with_base_url( + &store, + reqwest::Client::builder() + .no_proxy() + .build() + .expect("client"), + server.uri(), + ); + service + .prepare_scheduling(std::slice::from_ref(&account)) + .await; + + let summary = service.synchronize().await.expect("quota synchronization"); + + assert_eq!(summary.exhausted, 1); + let reconciled = store + .account("acct_periodic_stuck_cooldown") + .expect("reconciled account"); + assert_eq!( + reconciled.availability(), + AccountAvailability::QuotaExhausted + ); +} + +#[tokio::test] +async fn periodic_quota_synchronization_keeps_ready_when_only_secondary_window_is_exhausted() { + let store = Arc::new(MemoryAccountStore::default()); + create_account(&store, "acct_periodic_secondary_only").await; + let reset_at = Utc::now().timestamp() + 24 * 60 * 60; + let server = MockServer::start().await; + Mock::given(method("GET")) + .and(path("/api/codex/usage")) + .respond_with(ResponseTemplate::new(200).set_body_json(json!({ + "rate_limit": { + "allowed": true, + "limit_reached": false, + "primary_window": {"used_percent": 5, "reset_at": reset_at}, + "secondary_window": {"used_percent": 100, "reset_at": reset_at} + } + }))) + .expect(1) .mount(&server) .await; let service = quota_service_with_base_url( @@ -882,13 +933,99 @@ async fn periodic_quota_synchronization_preserves_availability_when_usage_peaks( server.uri(), ); - service.synchronize().await.expect("quota synchronization"); + let summary = service.synchronize().await.expect("quota synchronization"); - // 429/触顶不再写入 availability(限流不改变账号可用性),worker 只更新 quota 数据。 - let reconciled = store - .account("acct_periodic_stuck_cooldown") - .expect("reconciled account"); - assert_eq!(reconciled.availability(), AccountAvailability::Ready); + assert_eq!(summary.exhausted, 1); + assert_eq!( + store + .account("acct_periodic_secondary_only") + .expect("reconciled account") + .availability(), + AccountAvailability::Ready + ); +} + +#[tokio::test] +async fn periodic_quota_synchronization_keeps_ready_when_primary_is_full_but_requests_are_allowed() +{ + let store = Arc::new(MemoryAccountStore::default()); + create_account(&store, "acct_periodic_primary_full_but_allowed").await; + let reset_at = Utc::now().timestamp() + 24 * 60 * 60; + let server = MockServer::start().await; + Mock::given(method("GET")) + .and(path("/api/codex/usage")) + .respond_with(ResponseTemplate::new(200).set_body_json(json!({ + "rate_limit": { + "allowed": true, + "limit_reached": false, + "primary_window": {"used_percent": 100, "reset_at": reset_at} + } + }))) + .expect(1) + .mount(&server) + .await; + let service = quota_service_with_base_url( + &store, + reqwest::Client::builder() + .no_proxy() + .build() + .expect("client"), + server.uri(), + ); + + let summary = service.synchronize().await.expect("quota synchronization"); + + assert_eq!(summary.exhausted, 1); + assert_eq!( + store + .account("acct_periodic_primary_full_but_allowed") + .expect("reconciled account") + .availability(), + AccountAvailability::Ready + ); +} + +#[tokio::test] +async fn periodic_quota_synchronization_keeps_ready_without_primary_exhaustion_evidence() { + let store = Arc::new(MemoryAccountStore::default()); + create_account( + &store, + "acct_periodic_disallowed_without_primary_exhaustion", + ) + .await; + let reset_at = Utc::now().timestamp() + 24 * 60 * 60; + let server = MockServer::start().await; + Mock::given(method("GET")) + .and(path("/api/codex/usage")) + .respond_with(ResponseTemplate::new(200).set_body_json(json!({ + "rate_limit": { + "allowed": false, + "limit_reached": false, + "primary_window": {"used_percent": 5, "reset_at": reset_at} + } + }))) + .expect(1) + .mount(&server) + .await; + let service = quota_service_with_base_url( + &store, + reqwest::Client::builder() + .no_proxy() + .build() + .expect("client"), + server.uri(), + ); + + let summary = service.synchronize().await.expect("quota synchronization"); + + assert_eq!(summary.exhausted, 1); + assert_eq!( + store + .account("acct_periodic_disallowed_without_primary_exhaustion") + .expect("reconciled account") + .availability(), + AccountAvailability::Ready + ); } #[tokio::test]