chore: bump version to 3.0.24

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

backend/crates/gateway-protocol/src/openai/events.rs

@@ -392,6 +392,12 @@ pub fn rate_limits_to_header_pairs(rate_limits: &ParsedRateLimits) -> Vec<(Strin
             &format!("{prefix}-secondary"),
             details.secondary,
         );
+        if let Some(allowed) = details.allowed {
+            headers.push((format!("{prefix}-allowed"), allowed.to_string()));
+        }
+        if let Some(limit_reached) = details.limit_reached {
+            headers.push((format!("{prefix}-limit-reached"), limit_reached.to_string()));
+        }
         if let Some(limit_name) = &details.limit_name {
             headers.push((format!("{prefix}-limit-name"), limit_name.clone()));
         }

backend/crates/gateway-protocol/tests/openai/events.rs

@@ -440,6 +440,8 @@ fn parse_rate_limits_event_should_extract_websocket_metered_limit() {
         "metered_limit_name": "codex-other",
         "limit_name": "Other requests",
         "rate_limits": {
+            "allowed": false,
+            "limit_reached": true,
             "primary": {
                 "used_percent": 33,
                 "window_minutes": 60,
@@ -450,7 +452,8 @@ fn parse_rate_limits_event_should_extract_websocket_metered_limit() {
             "has_credits": true,
             "unlimited": false,
             "balance": "9.25"
-        }
+        },
+        "rate_limit_reached_type": "usage_limit_reached"
     });
 
     let parsed = parse_rate_limits_event(&event).expect("rate-limit event");
@@ -460,16 +463,22 @@ fn parse_rate_limits_event_should_extract_websocket_metered_limit() {
         (
             parsed.active_limit.as_deref(),
             details.limit_name.as_deref(),
+            details.allowed,
+            details.limit_reached,
             details.primary.expect("primary window").window_minutes,
             parsed.plan_type.as_deref(),
             parsed.credits.expect("credits").balance,
+            parsed.rate_limit_reached_type.as_deref(),
         ),
         (
             Some("codex_other"),
             Some("Other requests"),
+            Some(false),
+            Some(true),
             Some(60),
             Some("team"),
             Some("9.25".to_owned()),
+            Some("usage_limit_reached"),
         )
     );
 }
@@ -498,9 +507,15 @@ fn rate_limit_header_round_trip_should_preserve_wire_facts() {
             "300".to_owned(),
         ),
         ("x-codex-active-limit".to_owned(), "codex".to_owned()),
+        ("x-codex-allowed".to_owned(), "false".to_owned()),
+        ("x-codex-limit-reached".to_owned(), "true".to_owned()),
         ("x-codex-credits-has-credits".to_owned(), "true".to_owned()),
         ("x-codex-credits-unlimited".to_owned(), "false".to_owned()),
         ("x-codex-plan-type".to_owned(), "plus".to_owned()),
+        (
+            "x-codex-rate-limit-reached-type".to_owned(),
+            "usage_limit_reached".to_owned(),
+        ),
     ];
     let parsed = parse_rate_limit_headers(&source).expect("initial parse");
     let encoded = rate_limits_to_header_pairs(&parsed);

backend/crates/gateway-store/src/postgres/observability.rs

@@ -1992,27 +1992,35 @@ async fn provider_account_metrics(
     observed_at: DateTime<Utc>,
 ) -> StoreResult<ProviderAccountMetrics> {
     let row = sqlx::query(
-        "select count(*)::bigint as total,
+        "with normalized as (
+           select enabled,
+                  case
+                    when not enabled then 'disabled'
+                    when availability = 'banned' then 'banned'
+                    when availability = 'quota_exhausted'
+                      or (
+                        availability = 'cooldown'
+                        and availability_reason = 'usage_limit_exhausted'
+                        and (cooldown_until is null or cooldown_until > $1)
+                      )
+                      then 'quota_exhausted'
+                    when availability in ('expired', 'invalid')
+                      or (access_token_expires_at is not null and access_token_expires_at <= $1)
+                      then 'expired'
+                    else 'active'
+                  end as status
+             from provider_accounts
+         )
+         select count(*)::bigint as total,
                 count(*) filter (where enabled)::bigint as enabled,
-                count(*) filter (
-                  where not (
-                    enabled and availability = 'ready'
-                      and (access_token_expires_at is null or access_token_expires_at > $1)
-                  )
-                )::bigint as unavailable,
-                count(*) filter (
-                  where enabled and availability = 'ready'
-                    and (access_token_expires_at is null or access_token_expires_at > $1)
-                )::bigint as active,
-                count(*) filter (
-                  where availability = 'expired'
-                    or (access_token_expires_at is not null and access_token_expires_at <= $1)
-                )::bigint as expired,
-                count(*) filter (where availability = 'quota_exhausted')::bigint
-                  as quota_exhausted,
-                count(*) filter (where not enabled)::bigint as disabled,
-                count(*) filter (where availability = 'banned')::bigint as banned
-         from provider_accounts",
+                count(*) filter (where status in ('expired', 'disabled', 'banned'))::bigint
+                  as unavailable,
+                count(*) filter (where status = 'active')::bigint as active,
+                count(*) filter (where status = 'expired')::bigint as expired,
+                count(*) filter (where status = 'quota_exhausted')::bigint as quota_exhausted,
+                count(*) filter (where status = 'disabled')::bigint as disabled,
+                count(*) filter (where status = 'banned')::bigint as banned
+           from normalized",
     )
     .bind(observed_at)
     .fetch_one(pool)
@@ -2035,11 +2043,29 @@ async fn active_provider_account_ids(
     observed_at: DateTime<Utc>,
 ) -> StoreResult<Vec<String>> {
     sqlx::query(
-        "select id
-         from provider_accounts
-         where enabled and availability = 'ready'
-           and (access_token_expires_at is null or access_token_expires_at > $1)
-         order by id",
+        "with normalized as (
+           select id,
+                  case
+                    when not enabled then 'disabled'
+                    when availability = 'banned' then 'banned'
+                    when availability = 'quota_exhausted'
+                      or (
+                        availability = 'cooldown'
+                        and availability_reason = 'usage_limit_exhausted'
+                        and (cooldown_until is null or cooldown_until > $1)
+                      )
+                      then 'quota_exhausted'
+                    when availability in ('expired', 'invalid')
+                      or (access_token_expires_at is not null and access_token_expires_at <= $1)
+                      then 'expired'
+                    else 'active'
+                  end as status
+             from provider_accounts
+         )
+         select id
+           from normalized
+          where status = 'active'
+          order by id",
     )
     .bind(observed_at)
     .fetch_all(pool)

backend/crates/gateway-store/src/postgres/provider_accounts.rs

@@ -125,6 +125,26 @@ fn parse_availability(value: &str) -> StoreResult<AccountAvailability> {
     AccountAvailability::parse(value).ok_or_else(|| invalid("unknown availability value"))
 }
 
+fn validate_availability_cooldown(
+    availability: AccountAvailability,
+    has_cooldown_until: bool,
+) -> StoreResult<()> {
+    if availability == AccountAvailability::Cooldown && !has_cooldown_until {
+        return Err(invalid("cooldown availability requires cooldown_until"));
+    }
+    if has_cooldown_until
+        && !matches!(
+            availability,
+            AccountAvailability::Cooldown | AccountAvailability::QuotaExhausted
+        )
+    {
+        return Err(invalid(
+            "cooldown_until is only allowed for cooldown or quota_exhausted availability",
+        ));
+    }
+    Ok(())
+}
+
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub struct ProviderAccountSummary {
     pub id: String,
@@ -224,11 +244,7 @@ impl NewProviderAccount {
         if !self.has_refresh_token && self.next_refresh_at.is_some() {
             return Err(invalid("next_refresh_at requires a refresh token"));
         }
-        if (self.availability == AccountAvailability::Cooldown) != self.cooldown_until.is_some() {
-            return Err(invalid(
-                "cooldown_until must be present exactly for cooldown availability",
-            ));
-        }
+        validate_availability_cooldown(self.availability, self.cooldown_until.is_some())?;
         validate_object_size(
             "provider_credentials_json",
             &self.provider_credentials_json,
@@ -380,11 +396,7 @@ pub struct ProviderAccountStateUpdate {
 impl ProviderAccountStateUpdate {
     pub fn validate(&self) -> StoreResult<()> {
         require_nonempty(ENTITY, "account_id", &self.account_id)?;
-        if (self.availability == AccountAvailability::Cooldown) != self.cooldown_until.is_some() {
-            return Err(invalid(
-                "cooldown availability and cooldown_until must agree",
-            ));
-        }
+        validate_availability_cooldown(self.availability, self.cooldown_until.is_some())?;
         Ok(())
     }
 }
@@ -392,11 +404,7 @@ impl ProviderAccountStateUpdate {
 impl ProviderAccountObservation {
     pub fn validate(&self) -> StoreResult<()> {
         require_nonempty(ENTITY, "account_id", &self.account_id)?;
-        if (self.availability == AccountAvailability::Cooldown) != self.cooldown_until.is_some() {
-            return Err(invalid(
-                "cooldown availability and cooldown_until must agree",
-            ));
-        }
+        validate_availability_cooldown(self.availability, self.cooldown_until.is_some())?;
         if self.provider_quota_json.is_some() != self.quota_observed_at.is_some() {
             return Err(invalid("quota JSON and quota_observed_at must agree"));
         }

backend/crates/gateway-store/tests/postgres/mod.rs

@@ -76,6 +76,12 @@ impl TestDatabase {
         .execute(&mut *migration)
         .await
         .expect("apply model request service tier migration");
+        sqlx::raw_sql(include_str!(
+            "../../../../migrations/0005_allow_quota_exhausted_cooldown.sql"
+        ))
+        .execute(&mut *migration)
+        .await
+        .expect("apply quota exhausted cooldown migration");
         migration.commit().await.expect("commit terminal migration");
         Some(Self {
             admin,
@@ -171,7 +177,7 @@ async fn connect_and_migrate_should_apply_all_migrations_once_and_reopen_cleanly
     .expect("drop migration test database");
     admin.close().await;
 
-    assert_eq!((first_table_count, migration_count), (8, 4));
+    assert_eq!((first_table_count, migration_count), (8, 5));
     assert_eq!(response_id_types, ["bytea", "bytea"]);
     assert!(!raw_response_id_index_exists);
 }
@@ -183,6 +189,7 @@ fn migrations_should_leave_transaction_ownership_to_sqlx() {
         include_str!("../../../../migrations/0002_drop_opaque_response_id_index.sql"),
         include_str!("../../../../migrations/0003_store_opaque_response_ids_as_bytes.sql"),
         include_str!("../../../../migrations/0004_add_model_request_service_tier.sql"),
+        include_str!("../../../../migrations/0005_allow_quota_exhausted_cooldown.sql"),
     ]
     .into_iter()
     .flat_map(str::lines)

backend/crates/gateway-store/tests/postgres/observability.rs

@@ -122,16 +122,27 @@ async fn dashboard_account_metrics_should_partition_available_and_unavailable_ac
            id, provider_kind, name, upstream_user_id, authentication_kind,
            provider_credentials_json, credential_revision, has_refresh_token,
            access_token_expires_at, enabled, availability,
+           availability_reason, cooldown_until,
            availability_observed_at, created_at, updated_at
          ) values
            ('acct_available', 'openai', 'available', 'user-available', 'oauth',
-            '{}'::jsonb, 1, false, $1 + interval '1 day', true, 'ready', $1, $1, $1),
+            '{}'::jsonb, 1, false, $1 + interval '1 day', true, 'ready',
+            null, null, $1, $1, $1),
            ('acct_expired', 'openai', 'expired', 'user-expired', 'oauth',
-            '{}'::jsonb, 1, false, $1 - interval '1 day', true, 'ready', $1, $1, $1),
+            '{}'::jsonb, 1, false, $1 - interval '1 day', true, 'ready',
+            null, null, $1, $1, $1),
+           ('acct_rate_cooldown', 'openai', 'rate-cooldown', 'user-rate-cooldown', 'oauth',
+            '{}'::jsonb, 1, false, $1 + interval '1 day', true, 'cooldown',
+            'rate_limited', $1 + interval '2 hours', $1, $1, $1),
+           ('acct_usage_limit', 'openai', 'usage-limit', 'user-usage-limit', 'oauth',
+            '{}'::jsonb, 1, false, $1 + interval '1 day', true, 'cooldown',
+            'usage_limit_exhausted', $1 + interval '2 hours', $1, $1, $1),
            ('acct_banned', 'xai', 'banned', 'user-banned', 'oauth',
-            '{}'::jsonb, 1, false, $1 + interval '1 day', true, 'banned', $1, $1, $1),
+            '{}'::jsonb, 1, false, $1 + interval '1 day', true, 'banned',
+            null, null, $1, $1, $1),
            ('acct_disabled', 'xai', 'disabled', 'user-disabled', 'oauth',
-            '{}'::jsonb, 1, false, $1 + interval '1 day', false, 'ready', $1, $1, $1)",
+            '{}'::jsonb, 1, false, $1 + interval '1 day', false, 'ready',
+            null, null, $1, $1, $1)",
     )
     .bind(now)
     .execute(&database.pool)
@@ -154,10 +165,11 @@ async fn dashboard_account_metrics_should_partition_available_and_unavailable_ac
             metrics.active,
             metrics.unavailable,
             metrics.expired,
+            metrics.quota_exhausted,
             metrics.disabled,
             metrics.banned,
         ),
-        (4, 3, 1, 3, 1, 1, 1),
+        (6, 5, 2, 3, 1, 1, 1, 1),
     );
     database.close().await;
 }

backend/crates/gateway-store/tests/postgres/provider_accounts.rs

@@ -1334,7 +1334,15 @@ fn cooldown_observation_requires_expiry() {
 }
 
 #[test]
-fn imported_account_rejects_mismatched_cooldown_runtime_facts() {
+fn quota_exhausted_account_accepts_cooldown_runtime_facts() {
+    let mut imported = account("acct_quota_cooldown", "user-quota-cooldown");
+    imported.availability = AccountAvailability::QuotaExhausted;
+    imported.cooldown_until = Some(Utc::now() + TimeDelta::minutes(5));
+    assert!(imported.validate().is_ok());
+}
+
+#[test]
+fn imported_account_rejects_cooldown_until_for_ready_runtime_facts() {
     let mut imported = account("acct_invalid_cooldown", "user-invalid-cooldown");
     imported.cooldown_until = Some(Utc::now() + TimeDelta::minutes(5));
     assert!(imported.validate().is_err());

backend/crates/providers/openai/src/credential/quota.rs

@@ -30,6 +30,7 @@ const MAX_RATE_LIMIT_COOLDOWN: Duration = Duration::from_secs(60 * 60);
 const QUOTA_SCHEDULING_TTL: Duration = Duration::from_secs(10 * 60);
 const QUOTA_HYDRATION_FAILURE_TTL: Duration = Duration::from_secs(5);
 const QUOTA_PERIODIC_REFRESH_MIN_INTERVAL: Duration = Duration::from_secs(30 * 60);
+const QUOTA_RESET_RECHECK_GRACE: Duration = Duration::from_secs(30);
 // ready 账号主要依赖真实请求的被动观察；这里只提供低频、定额的状态兜底。
 const QUOTA_READY_REFRESH_MAX_AGE: Duration = Duration::from_secs(2 * 60 * 60);
 const QUOTA_READY_REFRESH_BATCH_SIZE: usize = 8;
@@ -307,7 +308,7 @@ enum CodexQuotaFetchError {
 enum CodexQuotaStateTransition {
     Banned { reason: &'static str },
     Exhausted,
-    Cooldown { until: Option<SystemTime> },
+    ResettableLimit { until: Option<SystemTime> },
 }
 
 impl CodexQuotaStateTransition {
@@ -315,7 +316,7 @@ impl CodexQuotaStateTransition {
         match self {
             Self::Banned { .. } => AccountAvailability::Banned,
             Self::Exhausted => AccountAvailability::QuotaExhausted,
-            Self::Cooldown { .. } => AccountAvailability::Cooldown,
+            Self::ResettableLimit { .. } => AccountAvailability::QuotaExhausted,
         }
     }
 
@@ -323,13 +324,13 @@ impl CodexQuotaStateTransition {
         match self {
             Self::Banned { reason } => reason,
             Self::Exhausted => "quota_exhausted",
-            Self::Cooldown { .. } => "quota_rate_limited",
+            Self::ResettableLimit { .. } => "quota_rate_limited",
         }
     }
 
     const fn cooldown_until(self) -> Option<SystemTime> {
         match self {
-            Self::Cooldown { until } => until,
+            Self::ResettableLimit { until } => until,
             Self::Banned { .. } | Self::Exhausted => None,
         }
     }
@@ -547,7 +548,9 @@ fn quota_refresh_candidate(
     let snapshot = quota_snapshot_for_account(&account, observation);
     let kind = match account.availability() {
         AccountAvailability::QuotaExhausted => CodexQuotaRefreshKind::Exhausted {
-            reset_at: quota_reset_at_from_snapshot(snapshot.as_ref()),
+            reset_at: account
+                .cooldown_until()
+                .or_else(|| quota_reset_at_from_snapshot(snapshot.as_ref())),
         },
         AccountAvailability::Cooldown
             if snapshot
@@ -618,22 +621,24 @@ fn reserve_exhausted_quota_refresh(
     now: SystemTime,
     refreshed_at: Instant,
 ) -> bool {
-    let reset_recheck_due = reset_at.is_some_and(|reset_at| {
-        reset_at <= now
-            && state
-                .last_exhausted_reset_recheck
-                .get(account.id())
-                .is_none_or(|marker| {
-                    marker.revision != account.revision() || marker.reset_at != reset_at
-                })
-    });
-    if !reset_recheck_due && !periodic_quota_refresh_due(state, account.id(), refreshed_at) {
-        return false;
-    }
-    state
-        .last_periodic_refresh_at
-        .insert(account.id().clone(), refreshed_at);
-    if let Some(reset_at) = reset_at.filter(|reset_at| *reset_at <= now) {
+    if let Some(reset_at) = reset_at {
+        if !quota_reset_recheck_ready(reset_at, now) {
+            return false;
+        }
+        let same_reset_already_checked = state
+            .last_exhausted_reset_recheck
+            .get(account.id())
+            .is_some_and(|marker| {
+                marker.revision == account.revision() && marker.reset_at == reset_at
+            });
+        if same_reset_already_checked
+            && !periodic_quota_refresh_due(state, account.id(), refreshed_at)
+        {
+            return false;
+        }
+        state
+            .last_periodic_refresh_at
+            .insert(account.id().clone(), refreshed_at);
         state.last_exhausted_reset_recheck.insert(
             account.id().clone(),
             CodexQuotaResetMarker {
@@ -641,10 +646,23 @@ fn reserve_exhausted_quota_refresh(
                 reset_at,
             },
         );
+        return true;
     }
+    if !periodic_quota_refresh_due(state, account.id(), refreshed_at) {
+        return false;
+    }
+    state
+        .last_periodic_refresh_at
+        .insert(account.id().clone(), refreshed_at);
     true
 }
 
+fn quota_reset_recheck_ready(reset_at: SystemTime, now: SystemTime) -> bool {
+    reset_at
+        .checked_add(QUOTA_RESET_RECHECK_GRACE)
+        .is_some_and(|ready_at| ready_at <= now)
+}
+
 fn periodic_quota_refresh_due(
     state: &CodexQuotaProjectionState,
     account_id: &ProviderAccountId,
@@ -802,16 +820,12 @@ impl CodexCredentialQuotaService {
                         summary.transient += 1;
                         continue;
                     };
-                    let preserve_exhaustion = account.availability()
-                        == AccountAvailability::QuotaExhausted
-                        && transition.availability() == AccountAvailability::Cooldown;
                     match transition {
                         CodexQuotaStateTransition::Banned { .. } => summary.banned += 1,
-                        CodexQuotaStateTransition::Exhausted => summary.exhausted += 1,
-                        CodexQuotaStateTransition::Cooldown { .. } => summary.cooldown += 1,
-                    }
-                    if preserve_exhaustion {
-                        continue;
+                        CodexQuotaStateTransition::Exhausted
+                        | CodexQuotaStateTransition::ResettableLimit { .. } => {
+                            summary.exhausted += 1;
+                        }
                     }
                     let _ = self
                         .repository
@@ -856,19 +870,23 @@ impl CodexCredentialQuotaService {
             {
                 continue;
             }
-            if !matches!(
-                self.has_active_exhaustion(&current, observed_at).await,
-                Ok(true)
-            ) {
-                continue;
-            }
+            let snapshot = match self.read_snapshot_for(&current).await {
+                Ok(Some(snapshot))
+                    if quota_fact_is_active_exhaustion(snapshot.fact(), observed_at) =>
+                {
+                    snapshot
+                }
+                _ => continue,
+            };
+            let cooldown_until =
+                quota_reset_at_from_snapshot(Some(&snapshot)).or_else(|| current.cooldown_until());
             if let Err(error) = self
                 .repository
                 .apply_state(
                     &current,
                     AccountAvailability::QuotaExhausted,
                     Some("quota_exhausted".to_owned()),
-                    None,
+                    cooldown_until,
                     observed_at,
                 )
                 .await
@@ -881,8 +899,11 @@ impl CodexCredentialQuotaService {
                 continue;
             }
             let enabled = current.enabled();
-            *account =
-                current.with_runtime_state(enabled, AccountAvailability::QuotaExhausted, None);
+            *account = current.with_runtime_state(
+                enabled,
+                AccountAvailability::QuotaExhausted,
+                cooldown_until,
+            );
         }
     }
 
@@ -927,10 +948,10 @@ impl CodexCredentialQuotaService {
         } else {
             summary.updated += 1;
         }
-        if let Some(availability) = quota_success_availability(
+        if let Some((availability, cooldown_until)) = quota_success_state(
             current.availability(),
             current.cooldown_until(),
-            fact.exhausted(),
+            fact,
             SystemTime::now(),
         ) {
             let _ = self
@@ -939,7 +960,7 @@ impl CodexCredentialQuotaService {
                     &current,
                     availability,
                     fact.exhausted().then_some("quota_exhausted".to_owned()),
-                    None,
+                    cooldown_until,
                     observed_at,
                 )
                 .await;
@@ -993,10 +1014,10 @@ impl CodexCredentialQuotaService {
         }
         self.scheduling
             .observe(current.id().clone(), current.revision(), observed_at, fact);
-        if let Some(availability) = quota_success_availability(
+        if let Some((availability, cooldown_until)) = quota_success_state(
             current.availability(),
             current.cooldown_until(),
-            fact.exhausted(),
+            fact,
             observed_at,
         ) {
             self.repository
@@ -1004,7 +1025,7 @@ impl CodexCredentialQuotaService {
                     &current,
                     availability,
                     fact.exhausted().then_some("quota_exhausted".to_owned()),
-                    None,
+                    cooldown_until,
                     observed_at,
                 )
                 .await?;
@@ -1026,19 +1047,6 @@ impl CodexCredentialQuotaService {
         self.read_snapshot_for(&account).await
     }
 
-    pub(super) async fn has_active_exhaustion(
-        &self,
-        account: &ProviderAccount,
-        now: SystemTime,
-    ) -> Result<bool, CodexCredentialQuotaError> {
-        match self.read_snapshot_for(account).await {
-            Ok(snapshot) => Ok(snapshot
-                .is_some_and(|snapshot| quota_fact_is_active_exhaustion(snapshot.fact(), now))),
-            Err(CodexCredentialQuotaError::RevisionConflict) => Ok(false),
-            Err(error) => Err(error),
-        }
-    }
-
     async fn read_snapshot_for(
         &self,
         account: &ProviderAccount,
@@ -1164,10 +1172,10 @@ impl CodexCredentialQuotaService {
             snapshot.observed_at(),
             snapshot.fact(),
         );
-        if let Some(availability) = quota_success_availability(
+        if let Some((availability, cooldown_until)) = quota_success_state(
             current.availability(),
             current.cooldown_until(),
-            snapshot.fact().exhausted(),
+            snapshot.fact(),
             SystemTime::now(),
         ) {
             self.repository
@@ -1178,7 +1186,7 @@ impl CodexCredentialQuotaService {
                         .fact()
                         .exhausted()
                         .then_some("quota_exhausted".to_owned()),
-                    None,
+                    cooldown_until,
                     observed_at,
                 )
                 .await?;
@@ -1386,32 +1394,34 @@ fn passive_rate_limit_window(window: RateLimitWindow) -> Map<String, Value> {
     snapshot
 }
 
-fn quota_success_availability(
+fn quota_success_state(
     current: AccountAvailability,
     cooldown_until: Option<SystemTime>,
-    exhausted: bool,
+    fact: CodexQuotaFact,
     now: SystemTime,
-) -> Option<AccountAvailability> {
-    let observed = if exhausted {
-        AccountAvailability::QuotaExhausted
-    } else {
-        AccountAvailability::Ready
-    };
-    match current {
+) -> Option<(AccountAvailability, Option<SystemTime>)> {
+    if fact.exhausted() {
+        let observed_cooldown = fact.resets_at().map(SystemTime::from).or(cooldown_until);
+        return (current != AccountAvailability::QuotaExhausted
+            || cooldown_until != observed_cooldown)
+            .then_some((AccountAvailability::QuotaExhausted, observed_cooldown));
+    }
+
+    let availability = match current {
         AccountAvailability::Invalid
         | AccountAvailability::Expired
         | AccountAvailability::Banned
-        | AccountAvailability::Unknown => Some(observed),
-        AccountAvailability::QuotaExhausted => (!exhausted).then_some(AccountAvailability::Ready),
-        AccountAvailability::Ready => exhausted.then_some(AccountAvailability::QuotaExhausted),
-        AccountAvailability::Cooldown if exhausted => Some(AccountAvailability::QuotaExhausted),
+        | AccountAvailability::Unknown
+        | AccountAvailability::QuotaExhausted => Some(AccountAvailability::Ready),
+        AccountAvailability::Ready => None,
         AccountAvailability::Cooldown
             if cooldown_until.is_some_and(|cooldown_until| cooldown_until <= now) =>
         {
             Some(AccountAvailability::Ready)
         }
         AccountAvailability::Cooldown => None,
-    }
+    };
+    availability.map(|availability| (availability, None))
 }
 
 fn quota_state_transition(
@@ -1439,7 +1449,7 @@ fn quota_state_transition(
                 retry_after_seconds.unwrap_or(DEFAULT_RATE_LIMIT_COOLDOWN.as_secs()),
             )
             .min(MAX_RATE_LIMIT_COOLDOWN);
-            Some(CodexQuotaStateTransition::Cooldown {
+            Some(CodexQuotaStateTransition::ResettableLimit {
                 until: observed_at.checked_add(duration),
             })
         }
@@ -1877,7 +1887,6 @@ impl QuotaAggregate {
                 self.remaining_percent
                     .map_or(remaining, |current| current.min(remaining)),
             );
-            self.exhausted |= used >= 100.0;
         }
         if let Some(reset) = object.get("reset_at") {
             let seconds = reset

backend/crates/providers/openai/src/credential/refresh.rs

@@ -256,8 +256,6 @@ impl CodexCredentialRefreshService {
                         | AccountAvailability::Banned
                         | AccountAvailability::Invalid
                 )
-                && (account.availability() != AccountAvailability::Cooldown
-                    || account.cooldown_until().is_some_and(|until| until <= now))
                 && refresh_due_at(account, now).is_some()
         });
         accounts.sort_by_key(|account| (refresh_due_at(account, now), account.id().clone()));

backend/crates/providers/openai/src/credential/selector.rs

@@ -769,31 +769,31 @@ impl CodexCredentialSelector {
         if current.revision() != account.revision() {
             return Err(CredentialRepositoryError::RevisionConflict.into());
         }
-        // 一次真实 429 证明上游已经接受当前凭据，可纠正旧的认证/封禁终态。
-        // 结构化额度耗尽仍拥有更高优先级，不能被通用限流降级。
-        if current.availability() == AccountAvailability::QuotaExhausted {
-            return Ok(());
-        }
-        let quota_exhausted = self
-            .quota
-            .has_active_exhaustion(&current, observed_at)
-            .await
-            .map_err(|_| CredentialSelectionError::Store)?;
-        let (availability, reason, cooldown_until) = if quota_exhausted {
-            (
-                AccountAvailability::QuotaExhausted,
-                Some("quota_exhausted".to_owned()),
-                None,
-            )
-        } else {
-            (
-                AccountAvailability::Cooldown,
-                Some(reason.to_owned()),
-                observed_at.checked_add(retry_after.unwrap_or(DEFAULT_RATE_LIMIT_COOLDOWN)),
-            )
+        let cooldown_until =
+            observed_at.checked_add(retry_after.unwrap_or(DEFAULT_RATE_LIMIT_COOLDOWN));
+        let cooldown_until = match (current.availability(), current.cooldown_until()) {
+            (AccountAvailability::QuotaExhausted, Some(existing)) => {
+                cooldown_until.map_or(Some(existing), |until| Some(existing.max(until)))
+            }
+            _ => cooldown_until,
         };
+        let reason = if current.availability() == AccountAvailability::QuotaExhausted
+            && current.cooldown_until().is_none()
+        {
+            Some("quota_exhausted".to_owned())
+        } else {
+            Some(reason.to_owned())
+        };
+        // 一次真实 429 证明上游已经接受当前凭据，可纠正旧的认证/封禁终态。
+        // resettable 用量限制沿用 v2 的 quota lock 语义，并保留/延长复核时间。
         self.repository
-            .apply_state(&current, availability, reason, cooldown_until, observed_at)
+            .apply_state(
+                &current,
+                AccountAvailability::QuotaExhausted,
+                reason,
+                cooldown_until,
+                observed_at,
+            )
             .await?;
         Ok(())
     }

backend/crates/providers/openai/src/transport/client_sse.rs

@@ -3,7 +3,8 @@ use std::{sync::Arc, time::Instant};
 use futures::{StreamExt, TryStreamExt};
 use gateway_protocol::openai::{
     X_OPENAI_INTERNAL_CODEX_RESPONSES_LITE_HEADER, X_OPENAI_MEMGEN_REQUEST_HEADER,
-    events::retry_after_seconds_from_body,
+    events::{self, retry_after_seconds_from_body},
+    sse::{SseEventDecoder, SseFrame},
 };
 use reqwest::{
     Client, Response as ReqwestResponse,
@@ -130,13 +131,14 @@ impl CodexBackendClient {
             });
         }
 
+        let rate_limit_header_updates = Arc::new(tokio::sync::Mutex::new(Vec::new()));
         Ok(CodexBackendStreamingResponse {
-            body: http_sse_stream(response),
+            body: http_sse_stream(response, Arc::clone(&rate_limit_header_updates)),
             transport: CodexBackendTransport::HttpSse,
             turn_state,
             set_cookie_headers,
             rate_limit_headers,
-            rate_limit_header_updates: None,
+            rate_limit_header_updates: Some(rate_limit_header_updates),
             turn_state_update: None,
             websocket_pool_decision: None,
             diagnostics,
@@ -594,20 +596,74 @@ fn websocket_connection_profile(headers: &HeaderMap) -> String {
         .join("\0")
 }
 
-fn http_sse_stream(response: ReqwestResponse) -> CodexBackendSseStream {
+fn http_sse_stream(
+    response: ReqwestResponse,
+    rate_limit_updates: CodexRateLimitHeaderUpdates,
+) -> CodexBackendSseStream {
     let stream: CodexBackendSseStream =
         Box::pin(response.bytes_stream().map_err(CodexClientError::Http));
-    Box::pin(futures::stream::unfold(Some(stream), |stream| async move {
-        let mut stream = stream?;
-        match tokio::time::timeout(UPSTREAM_STREAM_IDLE_TIMEOUT, stream.next()).await {
-            Ok(Some(chunk)) => Some((chunk, Some(stream))),
-            Ok(None) => None,
-            Err(_) => Some((
-                Err(CodexClientError::StreamIdleTimeout {
-                    timeout: UPSTREAM_STREAM_IDLE_TIMEOUT,
-                }),
-                None,
-            )),
+    let stream: CodexBackendSseStream =
+        Box::pin(futures::stream::unfold(Some(stream), |stream| async move {
+            let mut stream = stream?;
+            match tokio::time::timeout(UPSTREAM_STREAM_IDLE_TIMEOUT, stream.next()).await {
+                Ok(Some(chunk)) => Some((chunk, Some(stream))),
+                Ok(None) => None,
+                Err(_) => Some((
+                    Err(CodexClientError::StreamIdleTimeout {
+                        timeout: UPSTREAM_STREAM_IDLE_TIMEOUT,
+                    }),
+                    None,
+                )),
+            }
+        }));
+    observe_http_sse_rate_limits(stream, rate_limit_updates)
+}
+
+fn observe_http_sse_rate_limits(
+    stream: CodexBackendSseStream,
+    updates: CodexRateLimitHeaderUpdates,
+) -> CodexBackendSseStream {
+    Box::pin(futures::stream::unfold(
+        (stream, SseEventDecoder::default(), updates),
+        |(mut stream, mut decoder, updates)| async move {
+            match stream.next().await {
+                Some(chunk) => {
+                    if let Ok(bytes) = &chunk {
+                        append_http_sse_rate_limit_updates(decoder.push_frames(bytes), &updates)
+                            .await;
+                    }
+                    Some((chunk, (stream, decoder, updates)))
+                }
+                None => {
+                    append_http_sse_rate_limit_updates(decoder.finish_frames(), &updates).await;
+                    None
+                }
+            }
+        },
+    ))
+}
+
+async fn append_http_sse_rate_limit_updates(
+    frames: Vec<SseFrame>,
+    updates: &CodexRateLimitHeaderUpdates,
+) {
+    let mut headers = Vec::new();
+    for frame in frames {
+        for event in frame.events() {
+            if event
+                .event
+                .as_deref()
+                .is_some_and(|event| event != "codex.rate_limits")
+            {
+                continue;
+            }
+            let Some(rate_limits) = events::parse_rate_limits_event_raw(&event.data) else {
+                continue;
+            };
+            headers.extend(events::rate_limits_to_header_pairs(&rate_limits));
         }
-    }))
+    }
+    if !headers.is_empty() {
+        updates.lock().await.extend(headers);
+    }
 }

backend/crates/providers/openai/tests/credential/contract.rs

@@ -3,12 +3,10 @@ use std::num::NonZeroU32;
 use std::sync::Arc;
 use std::time::{Duration, SystemTime};
 
-use chrono::Utc;
 use futures::executor::block_on;
 use gateway_core::engine::credential::{
     AccountAttemptFeedback, AccountAvailability, AccountFeedbackStats, AccountSelectionPolicy,
-    AccountStateChange, OpaqueProviderData, ProviderAccountId, ProviderAccountStore as _,
-    QuotaObservation, RotationStrategy,
+    AccountStateChange, ProviderAccountId, ProviderAccountStore as _, RotationStrategy,
 };
 use gateway_core::engine::{
     AccountAttemptContext, AttemptContext, CancellationToken, ModelRequestId, RequestAttemptContext,
@@ -24,7 +22,6 @@ use provider_openai::credential::{
 };
 use provider_openai::transport::profile::{CodexWireProfile, CodexWireProfileState};
 use secrecy::ExposeSecret;
-use serde_json::json;
 use url::Url;
 
 use crate::support::{
@@ -121,7 +118,7 @@ fn selector_with_runtime(
         os_version: "6.8".to_owned(),
         arch: "x86_64".to_owned(),
         terminal: "selector-contract".to_owned(),
-        verified_at: Utc::now(),
+        verified_at: chrono::Utc::now(),
     });
     let http = reqwest::Client::builder().build().expect("HTTP client");
     let agent_identity = agent_identity_service(store);
@@ -731,7 +728,7 @@ fn credential_expired_failure_marks_unified_account_expired() {
 }
 
 #[test]
-fn rate_limited_failure_marks_account_cooldown_with_a_cooldown_deadline() {
+fn rate_limited_failure_marks_quota_exhausted_with_a_reset_deadline() {
     let store = Arc::new(MemoryAccountStore::default());
     create_account(&store, "acct_primary", "at-primary");
     let selector = selector(&store, Arc::new(TestLeaseCoordinator::default()));
@@ -757,7 +754,7 @@ fn rate_limited_failure_marks_account_cooldown_with_a_cooldown_deadline() {
     .expect("record rate-limit failure");
 
     let account = store.account("acct_primary").expect("account");
-    assert_eq!(account.availability(), AccountAvailability::Cooldown);
+    assert_eq!(account.availability(), AccountAvailability::QuotaExhausted);
     assert!(
         account
             .cooldown_until()
@@ -765,28 +762,31 @@ fn rate_limited_failure_marks_account_cooldown_with_a_cooldown_deadline() {
     );
 }
 
-#[test]
-fn usage_limit_exhaustion_is_a_resettable_account_cooldown() {
+#[tokio::test]
+async fn usage_limit_exhaustion_marks_quota_exhausted_without_usage_probe() {
     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");
+    selector
+        .record_failure(
+            &account,
+            CodexAccountFailure::UsageLimitExhausted {
+                retry_after: Some(Duration::from_secs(30)),
+            },
+        )
+        .await
+        .expect("record usage-limit exhaustion");
 
     let account = store.account("acct_primary").expect("persisted account");
-    assert_eq!(account.availability(), AccountAvailability::Cooldown);
+    assert_eq!(account.availability(), AccountAvailability::QuotaExhausted);
     assert!(
         account
             .cooldown_until()
             .is_some_and(|until| until > SystemTime::now())
     );
+    assert_eq!(store.quota_reads(), 0);
 }
 
 #[test]
@@ -794,26 +794,15 @@ fn rate_limited_failure_does_not_downgrade_persisted_quota_exhaustion() {
     let store = Arc::new(MemoryAccountStore::default());
     create_account(&store, "acct_primary", "at-primary");
     let account = store.account("acct_primary").expect("account");
-    let reset_at = Utc::now().timestamp() + 24 * 60 * 60;
-    block_on(
-        store.compare_and_swap_quota(QuotaObservation {
-            account_id: account.id().clone(),
-            expected_revision: account.revision(),
-            quota: Some(OpaqueProviderData::new(
-                json!({
-                    "rate_limit": {
-                        "allowed": false,
-                        "primary_window": {"used_percent": 100, "reset_at": reset_at}
-                    }
-                })
-                .as_object()
-                .expect("quota object")
-                .clone(),
-            )),
-            observed_at: Some(SystemTime::now()),
-        }),
-    )
-    .expect("persist exhausted quota");
+    block_on(store.apply_state_change(AccountStateChange {
+        account_id: account.id().clone(),
+        expected_revision: account.revision(),
+        availability: AccountAvailability::QuotaExhausted,
+        reason: Some("quota_exhausted".to_owned()),
+        cooldown_until: None,
+        observed_at: SystemTime::now(),
+    }))
+    .expect("mark exhausted");
     let selector = selector(&store, Arc::new(TestLeaseCoordinator::default()));
 
     block_on(selector.record_failure(
@@ -831,33 +820,54 @@ fn rate_limited_failure_does_not_downgrade_persisted_quota_exhaustion() {
             .availability(),
         AccountAvailability::QuotaExhausted
     );
+    assert!(
+        store
+            .account("acct_primary")
+            .expect("persisted account")
+            .cooldown_until()
+            .is_some_and(|until| until > SystemTime::now())
+    );
 }
 
 #[test]
-fn rate_limited_failure_ignores_exhaustion_after_its_reset_time() {
+fn rate_limited_failure_keeps_longer_existing_quota_deadline() {
+    let store = Arc::new(MemoryAccountStore::default());
+    create_account(&store, "acct_primary", "at-primary");
+    let account = store.account("acct_primary").expect("account");
+    let existing_deadline = SystemTime::now() + Duration::from_secs(300);
+    block_on(store.apply_state_change(AccountStateChange {
+        account_id: account.id().clone(),
+        expected_revision: account.revision(),
+        availability: AccountAvailability::QuotaExhausted,
+        reason: Some("usage_limit_exhausted".to_owned()),
+        cooldown_until: Some(existing_deadline),
+        observed_at: SystemTime::now(),
+    }))
+    .expect("mark exhausted");
+    let selector = selector(&store, Arc::new(TestLeaseCoordinator::default()));
+
+    block_on(selector.record_failure(
+        &account,
+        CodexAccountFailure::RateLimited {
+            retry_after: Some(Duration::from_secs(30)),
+        },
+    ))
+    .expect("record shorter rate-limit failure");
+
+    assert_eq!(
+        store
+            .account("acct_primary")
+            .expect("persisted account")
+            .cooldown_until(),
+        Some(existing_deadline)
+    );
+}
+
+#[test]
+fn rate_limited_failure_does_not_consult_stale_quota_snapshot() {
     let store = Arc::new(MemoryAccountStore::default());
     create_account(&store, "acct_primary", "at-primary");
     let account = store.account("acct_primary").expect("account");
-    let reset_at = Utc::now().timestamp() - 1;
-    block_on(
-        store.compare_and_swap_quota(QuotaObservation {
-            account_id: account.id().clone(),
-            expected_revision: account.revision(),
-            quota: Some(OpaqueProviderData::new(
-                json!({
-                    "rate_limit": {
-                        "allowed": false,
-                        "primary_window": {"used_percent": 100, "reset_at": reset_at}
-                    }
-                })
-                .as_object()
-                .expect("quota object")
-                .clone(),
-            )),
-            observed_at: Some(SystemTime::now()),
-        }),
-    )
-    .expect("persist expired quota exhaustion");
     let selector = selector(&store, Arc::new(TestLeaseCoordinator::default()));
 
     block_on(selector.record_failure(
@@ -873,8 +883,9 @@ fn rate_limited_failure_ignores_exhaustion_after_its_reset_time() {
             .account("acct_primary")
             .expect("persisted account")
             .availability(),
-        AccountAvailability::Cooldown
+        AccountAvailability::QuotaExhausted
     );
+    assert_eq!(store.quota_reads(), 0);
 }
 
 #[test]
@@ -907,7 +918,7 @@ fn authenticated_rate_limit_recovers_stale_authentication_state() {
             .account("acct_primary")
             .expect("persisted account")
             .availability(),
-        AccountAvailability::Cooldown
+        AccountAvailability::QuotaExhausted
     );
 }
 
@@ -985,7 +996,7 @@ fn rate_limited_failures_for_distinct_accounts_do_not_conflict() {
                 .expect("persisted second account")
                 .availability(),
         ],
-        [AccountAvailability::Cooldown; 2]
+        [AccountAvailability::QuotaExhausted; 2]
     );
 }
 

backend/crates/providers/openai/tests/credential/quota.rs

@@ -55,6 +55,36 @@ fn parser_treats_any_confirmed_provider_limit_as_exhausted() {
     assert!(fact.exhausted());
 }
 
+#[test]
+fn parser_keeps_full_percent_as_display_without_provider_exhaustion_signal() {
+    let fact = parse_codex_quota_usage(&json!({
+        "rate_limit": {
+            "allowed": true,
+            "limit_reached": false,
+            "primary_window": {"used_percent": 100, "reset_at": 1_900_000_000}
+        }
+    }))
+    .expect("valid full-percent quota");
+
+    assert_eq!(fact.remaining_percent(), Some(0));
+    assert!(!fact.exhausted());
+}
+
+#[test]
+fn parser_treats_confirmed_limit_reached_as_exhausted_even_below_full_percent() {
+    let fact = parse_codex_quota_usage(&json!({
+        "rate_limit": {
+            "allowed": false,
+            "limit_reached": true,
+            "primary_window": {"used_percent": 98, "reset_at": 1_900_000_000}
+        }
+    }))
+    .expect("valid provider-confirmed quota");
+
+    assert_eq!(fact.remaining_percent(), Some(2));
+    assert!(fact.exhausted());
+}
+
 #[test]
 fn parser_does_not_infer_exhaustion_from_unknown_credit_fields() {
     let fact = parse_codex_quota_usage(&json!({
@@ -369,6 +399,49 @@ async fn passive_rate_limit_headers_update_quota_and_account_state_with_revision
     );
 }
 
+#[tokio::test]
+async fn passive_full_percent_with_explicit_allowance_keeps_account_ready() {
+    let store = Arc::new(MemoryAccountStore::default());
+    create_account(&store, "acct_passive_full_allowed").await;
+    let account = store.account("acct_passive_full_allowed").expect("account");
+    let service = quota_service(&store);
+    let headers = vec![
+        ("x-codex-active-limit".to_owned(), "codex".to_owned()),
+        ("x-codex-primary-used-percent".to_owned(), "100".to_owned()),
+        (
+            "x-codex-primary-reset-at".to_owned(),
+            "1900000000".to_owned(),
+        ),
+        ("x-codex-allowed".to_owned(), "true".to_owned()),
+        ("x-codex-limit-reached".to_owned(), "false".to_owned()),
+    ];
+
+    assert!(
+        service
+            .synchronize_passive_headers(&account, &headers)
+            .await
+            .expect("passive quota")
+    );
+    let current = store
+        .account("acct_passive_full_allowed")
+        .expect("current account");
+    let snapshot = service
+        .read_account(account.id())
+        .await
+        .expect("read quota")
+        .expect("quota snapshot");
+
+    assert_eq!(
+        (
+            current.availability(),
+            current.cooldown_until(),
+            snapshot.fact().remaining_percent(),
+            snapshot.fact().exhausted(),
+        ),
+        (AccountAvailability::Ready, None, Some(0), false)
+    );
+}
+
 #[tokio::test]
 async fn manual_quota_refresh_updates_account_state_from_provider_response() {
     let store = Arc::new(MemoryAccountStore::default());
@@ -771,12 +844,18 @@ async fn periodic_quota_synchronization_repairs_cooldown_with_active_exhausted_q
 
     service.synchronize().await.expect("quota synchronization");
 
+    let reconciled = store
+        .account("acct_periodic_stuck_cooldown")
+        .expect("reconciled account");
     assert_eq!(
-        store
-            .account("acct_periodic_stuck_cooldown")
-            .expect("reconciled account")
-            .availability(),
-        AccountAvailability::QuotaExhausted
+        (reconciled.availability(), reconciled.cooldown_until()),
+        (
+            AccountAvailability::QuotaExhausted,
+            Some(
+                SystemTime::UNIX_EPOCH
+                    + Duration::from_secs(u64::try_from(reset_at).expect("positive reset")),
+            ),
+        )
     );
 }
 
@@ -815,12 +894,14 @@ async fn periodic_quota_synchronization_does_not_downgrade_exhaustion_on_rate_li
 
     service.synchronize().await.expect("quota synchronization");
 
-    assert_eq!(
-        store
-            .account("acct_periodic_exhausted")
-            .expect("persisted account")
-            .availability(),
-        AccountAvailability::QuotaExhausted
+    let account = store
+        .account("acct_periodic_exhausted")
+        .expect("persisted account");
+    assert_eq!(account.availability(), AccountAvailability::QuotaExhausted);
+    assert!(
+        account
+            .cooldown_until()
+            .is_some_and(|until| until > SystemTime::now())
     );
 }
 
@@ -951,6 +1032,60 @@ async fn periodic_quota_synchronization_attempts_quota_exhausted_accounts() {
     assert_eq!(summary.transient, 1);
 }
 
+#[tokio::test]
+async fn periodic_quota_synchronization_waits_for_quota_cooldown_deadline_with_grace() {
+    let store = Arc::new(MemoryAccountStore::default());
+    create_account(&store, "acct_periodic_exhausted_cooldown").await;
+    let account = store
+        .account("acct_periodic_exhausted_cooldown")
+        .expect("created account");
+    store
+        .apply_state_change(AccountStateChange {
+            account_id: account.id().clone(),
+            expected_revision: account.revision(),
+            availability: AccountAvailability::QuotaExhausted,
+            reason: Some("usage_limit_exhausted".to_owned()),
+            cooldown_until: SystemTime::now().checked_sub(Duration::from_secs(29)),
+            observed_at: SystemTime::now(),
+        })
+        .await
+        .expect("mark account exhausted with recent cooldown");
+    let service = blocked_network_quota_service(&store);
+
+    let waiting = service
+        .synchronize()
+        .await
+        .expect("quota synchronization before grace");
+
+    assert_eq!(waiting, CodexQuotaSyncSummary::default());
+    let current = store
+        .account("acct_periodic_exhausted_cooldown")
+        .expect("current account");
+    store
+        .apply_state_change(AccountStateChange {
+            account_id: current.id().clone(),
+            expected_revision: current.revision(),
+            availability: AccountAvailability::QuotaExhausted,
+            reason: Some("usage_limit_exhausted".to_owned()),
+            cooldown_until: SystemTime::now().checked_sub(Duration::from_secs(31)),
+            observed_at: SystemTime::now(),
+        })
+        .await
+        .expect("mark account exhausted past grace");
+
+    let due = service
+        .synchronize()
+        .await
+        .expect("quota synchronization after grace");
+    let repeated = service
+        .synchronize()
+        .await
+        .expect("quota synchronization repeated");
+
+    assert_eq!(due.transient, 1);
+    assert_eq!(repeated, CodexQuotaSyncSummary::default());
+}
+
 #[tokio::test]
 async fn periodic_quota_synchronization_throttles_the_same_exhausted_account() {
     let store = Arc::new(MemoryAccountStore::default());
@@ -1029,8 +1164,8 @@ async fn periodic_quota_synchronization_rechecks_an_exhausted_account_once_when_
         server.uri(),
     );
 
-    let first = service.synchronize().await.expect("initial refresh");
-    assert_eq!(first.transient, 1);
+    let waiting = service.synchronize().await.expect("initial cooldown wait");
+    assert_eq!(waiting, CodexQuotaSyncSummary::default());
     store
         .compare_and_swap_quota(QuotaObservation {
             account_id: account.id().clone(),
@@ -1061,6 +1196,77 @@ async fn periodic_quota_synchronization_rechecks_an_exhausted_account_once_when_
     assert_eq!(repeated, CodexQuotaSyncSummary::default());
 }
 
+#[tokio::test]
+async fn periodic_quota_synchronization_reschedules_a_confirmed_new_reset() {
+    let store = Arc::new(MemoryAccountStore::default());
+    let account_id = "acct_periodic_new_reset";
+    create_account(&store, account_id).await;
+    let account = store.account(account_id).expect("created account");
+    let initial_reset = SystemTime::UNIX_EPOCH + Duration::from_secs(1);
+    let confirmed_reset = SystemTime::UNIX_EPOCH + Duration::from_secs(2);
+    store
+        .apply_state_change(AccountStateChange {
+            account_id: account.id().clone(),
+            expected_revision: account.revision(),
+            availability: AccountAvailability::QuotaExhausted,
+            reason: Some("usage_limit_exhausted".to_owned()),
+            cooldown_until: Some(initial_reset),
+            observed_at: SystemTime::now(),
+        })
+        .await
+        .expect("mark account exhausted with due reset");
+    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": true,
+                "primary_window": {"used_percent": 100, "reset_at": 2}
+            }
+        })))
+        .mount(&server)
+        .await;
+    let service = quota_service_with_base_url(
+        &store,
+        reqwest::Client::builder()
+            .no_proxy()
+            .build()
+            .expect("client"),
+        server.uri(),
+    );
+
+    let first = service.synchronize().await.expect("first reset recheck");
+    let persisted_reset = store
+        .account(account_id)
+        .expect("account after first recheck")
+        .cooldown_until();
+    let second = service.synchronize().await.expect("new reset recheck");
+    let repeated = service.synchronize().await.expect("same reset throttled");
+    let request_count = server
+        .received_requests()
+        .await
+        .expect("received requests")
+        .len();
+
+    assert_eq!(
+        (
+            first.exhausted,
+            persisted_reset,
+            second.exhausted,
+            repeated,
+            request_count,
+        ),
+        (
+            1,
+            Some(confirmed_reset),
+            1,
+            CodexQuotaSyncSummary::default(),
+            2,
+        )
+    );
+}
+
 #[tokio::test]
 async fn invalid_quota_json_for_one_account_does_not_abort_the_batch() {
     let store = Arc::new(MemoryAccountStore::default());

backend/crates/providers/openai/tests/credential/refresh.rs

@@ -665,6 +665,74 @@ async fn expired_access_token_refreshes_even_when_next_refresh_is_deferred() {
     );
 }
 
+#[tokio::test]
+async fn expired_access_token_refreshes_during_active_cooldown() {
+    let store = Arc::new(MemoryAccountStore::default());
+    seed_due_account(&store).await;
+    let original = store.account("acct_refresh").expect("seeded account");
+    let mut expired_profile = profile("chatgpt-acct_refresh");
+    expired_profile.access_token_expires_at =
+        Some(chrono::Utc::now() - chrono::Duration::minutes(5));
+    store
+        .repository()
+        .rotate_oauth_secret(RotateCodexCredential {
+            account_id: "acct_refresh".to_owned(),
+            expected_credential_revision: original.revision().get(),
+            secret: secret("old-access"),
+            verified_account: expired_profile,
+            next_refresh_at: Some(chrono::Utc::now() + chrono::Duration::hours(12)),
+        })
+        .await
+        .expect("seed expired access token with deferred refresh");
+    let deferred = store.account("acct_refresh").expect("deferred account");
+    let cooldown_until = SystemTime::now() + Duration::from_secs(12 * 60 * 60);
+    store
+        .repository()
+        .apply_state(
+            &deferred,
+            AccountAvailability::Cooldown,
+            Some("rate_limited".to_owned()),
+            Some(cooldown_until),
+            SystemTime::now(),
+        )
+        .await
+        .expect("mark account cooling down");
+    let refresher = Arc::new(Refresher::new(Ok(success_tokens())));
+    let service = CodexCredentialRefreshService::new(
+        store.repository(),
+        refresher.clone(),
+        Arc::new(VerifiedIdentity),
+        Arc::new(RefreshLeases {
+            available: true,
+            requests: Mutex::new(Vec::new()),
+        }),
+        Arc::new(CountingCredentialState::default()),
+        runtime_policy(),
+    );
+
+    let outcomes = service.refresh_due().await.expect("refresh due");
+
+    assert!(matches!(
+        outcomes.as_slice(),
+        [CodexCredentialRefreshOutcome::Refreshed {
+            account_id,
+            credential_revision: 3,
+        }] if account_id == "acct_refresh"
+    ));
+    assert_eq!(
+        refresher.seen.lock().expect("seen tokens lock").as_slice(),
+        ["rt-old-access"]
+    );
+    let refreshed = store.account("acct_refresh").expect("refreshed account");
+    assert_eq!(refreshed.availability(), AccountAvailability::Cooldown);
+    assert_eq!(refreshed.cooldown_until(), Some(cooldown_until));
+    assert!(
+        refreshed
+            .access_token_expires_at()
+            .is_some_and(|expires_at| expires_at > SystemTime::now())
+    );
+}
+
 #[tokio::test]
 async fn unavailable_refresh_lease_prevents_duplicate_token_exchange() {
     let (store, refresher, service) = setup(Ok(success_tokens()), false).await;

backend/crates/providers/openai/tests/provider/contract.rs

@@ -36,7 +36,7 @@ use tokio::{
     sync::oneshot,
     time::timeout,
 };
-use wiremock::matchers::{method, path};
+use wiremock::matchers::{header, method, path};
 use wiremock::{Mock, MockServer, ResponseTemplate};
 
 use crate::support::{
@@ -861,7 +861,7 @@ async fn completed_response_persists_session_affinity_before_stream_consumer_sto
 }
 
 #[tokio::test]
-async fn response_failed_before_semantic_output_is_atomic_and_persists_cooldown() {
+async fn response_failed_before_semantic_output_is_atomic_and_persists_quota_lock() {
     let store = Arc::new(MemoryAccountStore::default());
     let account_id = "acct_atomic_failure";
     create_account(&store, account_id).await;
@@ -906,7 +906,7 @@ async fn response_failed_before_semantic_output_is_atomic_and_persists_cooldown(
         vec!["response.created", "response.failed"]
     );
     let account = store.account(account_id).expect("rate-limited account");
-    assert_eq!(account.availability(), AccountAvailability::Cooldown);
+    assert_eq!(account.availability(), AccountAvailability::QuotaExhausted);
     assert!(
         account
             .cooldown_until()
@@ -917,6 +917,68 @@ async fn response_failed_before_semantic_output_is_atomic_and_persists_cooldown(
     let _ = server.await;
 }
 
+#[tokio::test]
+async fn usage_limit_failure_marks_quota_exhausted_without_usage_probe() {
+    let store = Arc::new(MemoryAccountStore::default());
+    let account_id = "acct_usage_limit_request_path";
+    create_account(&store, account_id).await;
+    let server = MockServer::start().await;
+    Mock::given(method("GET"))
+        .and(path("/api/codex/usage"))
+        .and(header("authorization", format!("Bearer at-{account_id}")))
+        .respond_with(ResponseTemplate::new(500))
+        .expect(0)
+        .mount(&server)
+        .await;
+    Mock::given(method("POST"))
+        .and(path("/codex/responses"))
+        .respond_with(
+            ResponseTemplate::new(200)
+                .insert_header("content-type", "text/event-stream")
+                .set_body_string(concat!(
+                    "event: response.created\n",
+                    "data: {\"type\":\"response.created\",\"response\":{\"id\":\"resp_usage_limit_confirm\",\"model\":\"gpt-5.4\",\"status\":\"in_progress\"}}\n\n",
+                    "event: response.failed\n",
+                    "data: {\"type\":\"response.failed\",\"status_code\":429,\"retry_after_seconds\":86400,\"response\":{\"id\":\"resp_usage_limit_confirm\",\"status\":\"failed\",\"error\":{\"code\":\"usage_limit_reached\",\"message\":\"usage limit reached\"}}}\n\n"
+                )),
+        )
+        .mount(&server)
+        .await;
+
+    let mut stream = provider_with_base_url(&store, server.uri())
+        .execute(
+            planned_request("openai", http_generate_operation()),
+            context("req_usage_limit_confirm", CancellationToken::new()),
+        )
+        .await
+        .expect("prepare provider stream");
+    let failure = loop {
+        match stream.next().await {
+            Some(Ok(_)) => {}
+            Some(Err(error)) => break error,
+            None => panic!("usage-limit failure must surface a typed failure"),
+        }
+    };
+
+    assert_eq!(failure.kind(), ProviderErrorKind::QuotaExhausted);
+    let account = store.account(account_id).expect("usage-limit account");
+    assert_eq!(account.availability(), AccountAvailability::QuotaExhausted);
+    assert!(
+        account
+            .cooldown_until()
+            .is_some_and(|until| until > SystemTime::now())
+    );
+    assert!(!store.has_quota(account_id));
+    let requests = server.received_requests().await.expect("received requests");
+    assert_eq!(
+        requests
+            .iter()
+            .filter(|request| request.url.path() == "/api/codex/usage")
+            .count(),
+        0
+    );
+}
+
 #[tokio::test]
 async fn ordinary_request_should_forward_created_before_later_failed_chunk() {
     let store = Arc::new(MemoryAccountStore::default());
@@ -1338,6 +1400,76 @@ async fn successful_response_preserves_quota_exhaustion_from_rate_limit_headers(
     assert!(store.has_quota(account_id));
 }
 
+#[tokio::test]
+async fn successful_http_sse_rate_limit_event_persists_structured_exhaustion() {
+    let store = Arc::new(MemoryAccountStore::default());
+    let account_id = "acct_http_sse_exhausted";
+    create_account(&store, account_id).await;
+    let reset_at = 1_900_000_000_u64;
+    let rate_limit_event = json!({
+        "type": "codex.rate_limits",
+        "rate_limits": {
+            "allowed": false,
+            "limit_reached": true,
+            "primary": {
+                "used_percent": 42,
+                "window_minutes": 300,
+                "reset_at": reset_at,
+            },
+        },
+    });
+    let completed_event = json!({
+        "type": "response.completed",
+        "response": {
+            "id": "resp_http_sse_exhausted",
+            "model": "gpt-5.4",
+            "status": "completed",
+            "output": [],
+            "usage": {"input_tokens": 1, "output_tokens": 1, "total_tokens": 2},
+        },
+    });
+    let body = format!(
+        "event: codex.rate_limits\ndata: {rate_limit_event}\n\nevent: response.completed\ndata: {completed_event}\n\n"
+    );
+    let server = MockServer::start().await;
+    Mock::given(method("POST"))
+        .and(path("/codex/responses"))
+        .respond_with(
+            ResponseTemplate::new(200)
+                .insert_header("content-type", "text/event-stream")
+                .set_body_string(body),
+        )
+        .mount(&server)
+        .await;
+
+    let mut stream = provider_with_base_url(&store, server.uri())
+        .execute(
+            planned_request("openai", http_generate_operation()),
+            context("req_http_sse_exhausted", CancellationToken::new()),
+        )
+        .await
+        .expect("prepare provider stream");
+    while let Some(event) = stream.next().await {
+        event.expect("successful upstream response");
+    }
+
+    let account = store
+        .account(account_id)
+        .expect("account after HTTP SSE response");
+    assert_eq!(
+        (
+            account.availability(),
+            account.cooldown_until(),
+            store.has_quota(account_id),
+        ),
+        (
+            AccountAvailability::QuotaExhausted,
+            Some(SystemTime::UNIX_EPOCH + Duration::from_secs(reset_at)),
+            true,
+        )
+    );
+}
+
 #[test]
 fn request_observation_reads_openai_metadata_without_changing_the_operation() {
     let store = Arc::new(MemoryAccountStore::default());

backend/crates/providers/openai/tests/transport/http_client.rs

@@ -269,6 +269,54 @@ async fn codex_backend_client_should_parse_retry_after_from_rate_limit_error_bod
     assert_eq!(retry_after_seconds, Some(12));
 }
 
+#[tokio::test]
+async fn codex_backend_http_sse_should_capture_structured_rate_limit_event_updates() {
+    let server = wiremock::MockServer::start().await;
+    wiremock::Mock::given(wiremock::matchers::method("POST"))
+        .and(wiremock::matchers::path("/codex/responses"))
+        .respond_with(
+            wiremock::ResponseTemplate::new(200)
+                .insert_header("content-type", "text/event-stream")
+                .set_body_string(concat!(
+                    "event: codex.rate_limits\n",
+                    "data: {\"type\":\"codex.rate_limits\",\"rate_limits\":{\"allowed\":true,\"limit_reached\":false,\"primary\":{\"used_percent\":100,\"window_minutes\":300,\"reset_at\":1893456300}}}\n\n",
+                    "event: response.completed\n",
+                    "data: {\"type\":\"response.completed\",\"response\":{\"id\":\"resp_http_rate_limits\",\"output\":[],\"usage\":{\"input_tokens\":1,\"output_tokens\":1,\"total_tokens\":2}}}\n\n"
+                )),
+        )
+        .mount(&server)
+        .await;
+    let client = CodexBackendClient::new(
+        reqwest::Client::builder().no_proxy().build().unwrap(),
+        server.uri(),
+        test_wire_profile(),
+    );
+    let mut request = codex_request("gpt-5.5", "", Vec::new());
+    request.force_http_sse = true;
+
+    let response = client
+        .create_response(
+            &request,
+            request_context("req_http_rate_limit_event", Some("acct")),
+        )
+        .await
+        .expect("HTTP SSE response");
+    let structured = response
+        .rate_limit_headers
+        .iter()
+        .filter(|(name, _)| matches!(name.as_str(), "x-codex-allowed" | "x-codex-limit-reached"))
+        .cloned()
+        .collect::<Vec<_>>();
+
+    assert_eq!(
+        structured,
+        vec![
+            ("x-codex-allowed".to_owned(), "true".to_owned()),
+            ("x-codex-limit-reached".to_owned(), "false".to_owned()),
+        ]
+    );
+}
+
 #[tokio::test(start_paused = true)]
 async fn codex_backend_http_sse_should_fail_after_five_minutes_without_stream_data() {
     let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();

backend/migrations/.frozen-sha256

@@ -2,3 +2,4 @@
 0ae3945c02461347ab8dcfc64821027cd2366505778e0450d71bd82772c4e99d  0002_drop_opaque_response_id_index.sql
 e608ed89991f9d6317491714447a00a61ede90caca4fb030eb30695654afb664  0003_store_opaque_response_ids_as_bytes.sql
 df4e02cf2335733326c12429ced4cbd0b35b8a22b3fc24981c3c0549477d7c43  0004_add_model_request_service_tier.sql
+fe8f36913c6db7053148db450f05e008dab6cdb9ca8cdcfcb22d97ad9a87bf49  0005_allow_quota_exhausted_cooldown.sql

backend/migrations/0005_allow_quota_exhausted_cooldown.sql

@@ -0,0 +1,12 @@
+alter table provider_accounts
+  drop constraint provider_accounts_cooldown_ck,
+  add constraint provider_accounts_cooldown_ck check (
+    (
+      availability != 'cooldown'
+      or cooldown_until is not null
+    )
+    and (
+      cooldown_until is null
+      or availability in ('cooldown', 'quota_exhausted')
+    )
+  );

release/version.yaml

@@ -1 +1 @@
-version: "3.0.23"
+version: "3.0.24"