chore: bump version to 3.0.30

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

.github/workflows/release.yml

@@ -626,7 +626,7 @@ jobs:
           make_latest: ${{ needs.release-info.outputs.prerelease == 'true' && 'false' || 'true' }}
           body: |
             > [!WARNING]
-            > **Upgrading across major versions?** Database schema and configuration are only compatible within a major version, and the one-click self-update refuses cross-major targets (releases before 3.0 lack this gate — do not use their built-in update to reach a newer major). Deploy the new major fresh (new `.runtime/` data directory) and re-import or re-authorize provider accounts and client keys. See the [upgrade notes](https://github.com/${{ github.repository }}/blob/main/deploy/README.md#镜像升级与源码构建).
+            > **Upgrading across major versions?** Online upgrades across major versions are not supported. Deploy the new major fresh with a new `.runtime/` data directory, then re-import or re-authorize provider accounts and client keys. See the [upgrade notes](https://github.com/${{ github.repository }}/blob/main/deploy/README.md#镜像升级与源码构建).
 
             ## 📥 Installation
 

README.md

@@ -168,10 +168,8 @@ docker compose -f deploy/compose.yaml up -d
 > 运行状态。
 
 > [!WARNING]
-> 跨大版本不是原地升级：数据库 schema 与配置只在同一大版本内兼容，内置一键自更新
-> 会拒绝跨大版本目标（3.0 之前的旧版没有这道门禁，不要用旧版界面的一键更新跨大版本）。
-> 升级到新的大版本需要全新部署（全新 `.runtime/` 数据目录），并重新导入或重新授权
-> Provider 账号与客户端 Key。
+> 不支持跨大版本在线升级。升级到新的大版本时，请使用全新的 `.runtime/` 数据目录重新部署，
+> 并重新导入或重新授权 Provider 账号与客户端 Key。
 
 Compose 默认只绑定 `127.0.0.1`。公网接入应使用 HTTPS 反向代理，转发 WebSocket
 upgrade 与真实客户端 IP；不要暴露 PostgreSQL 或 Redis。

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

@@ -630,7 +630,8 @@ impl ProviderAccountRepository for PgProviderAccountRepository {
                      else availability_observed_at
                  end,
                  updated_at = case when enabled then greatest(now(), $6) else updated_at end
-             where id = $1 and credential_revision = $2",
+             where id = $1 and credential_revision = $2
+               and (availability_observed_at is null or availability_observed_at <= $6)",
         )
         .bind(update.account_id)
         .bind(to_i64(update.expected_revision.get())?)
@@ -675,7 +676,8 @@ impl ProviderAccountRepository for PgProviderAccountRepository {
             "update provider_accounts
              set provider_quota_json = $3, quota_observed_at = $4,
                  updated_at = greatest(now(), coalesce($4, now()))
-             where id = $1 and credential_revision = $2",
+             where id = $1 and credential_revision = $2
+               and ($4 is null or quota_observed_at is null or quota_observed_at <= $4)",
         )
         .bind(account_id)
         .bind(to_i64(expected_revision.get())?)

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

@@ -58,30 +58,6 @@ impl TestDatabase {
             .execute(&mut *migration)
             .await
             .expect("apply terminal migration");
-        sqlx::raw_sql(include_str!(
-            "../../../../migrations/0002_drop_opaque_response_id_index.sql"
-        ))
-        .execute(&mut *migration)
-        .await
-        .expect("apply opaque response ID migration");
-        sqlx::raw_sql(include_str!(
-            "../../../../migrations/0003_store_opaque_response_ids_as_bytes.sql"
-        ))
-        .execute(&mut *migration)
-        .await
-        .expect("apply opaque response storage migration");
-        sqlx::raw_sql(include_str!(
-            "../../../../migrations/0004_add_model_request_service_tier.sql"
-        ))
-        .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,
@@ -177,25 +153,19 @@ 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, 5));
+    assert_eq!((first_table_count, migration_count), (8, 1));
     assert_eq!(response_id_types, ["bytea", "bytea"]);
     assert!(!raw_response_id_index_exists);
 }
 
 #[test]
 fn migrations_should_leave_transaction_ownership_to_sqlx() {
-    let transaction_statements = [
-        include_str!("../../../../migrations/0001_initial.sql"),
-        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)
-    .map(str::trim)
-    .filter(|line| matches!(*line, "begin;" | "commit;"))
-    .count();
+    let transaction_statements = [include_str!("../../../../migrations/0001_initial.sql")]
+        .into_iter()
+        .flat_map(str::lines)
+        .map(str::trim)
+        .filter(|line| matches!(*line, "begin;" | "commit;"))
+        .count();
 
     assert_eq!(transaction_statements, 0);
 }

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

@@ -111,6 +111,97 @@ async fn core_quota_batch_reads_only_observed_accounts_in_one_contract_call() {
     database.close().await;
 }
 
+#[tokio::test]
+async fn delayed_provider_observations_cannot_overwrite_newer_state_or_quota() {
+    let Some(database) = TestDatabase::create("provider_account_observation_fence").await else {
+        return;
+    };
+    let repository = PgProviderAccountRepository::new(database.pool.clone());
+    let account_id = ProviderAccountId::new("acct_observation_fence").expect("account id");
+    repository
+        .insert_provider_account(account(account_id.as_str(), "user-observation-fence"))
+        .await
+        .expect("insert observation fence fixture");
+    let revision = CredentialRevision::new(1).expect("revision");
+    let newer = SystemTime::now()
+        .checked_add(Duration::from_secs(60))
+        .expect("newer observation time");
+    let older = newer
+        .checked_sub(Duration::from_secs(30))
+        .expect("older observation time");
+
+    let quota = |marker: &str, observed_at| QuotaObservation {
+        account_id: account_id.clone(),
+        expected_revision: revision,
+        quota: Some(OpaqueProviderData::new(
+            json!({"marker": marker})
+                .as_object()
+                .expect("quota object")
+                .clone(),
+        )),
+        observed_at: Some(observed_at),
+    };
+    assert_eq!(
+        repository
+            .compare_and_swap_quota(quota("newer", newer))
+            .await
+            .expect("persist newer quota"),
+        QuotaWriteOutcome::Updated
+    );
+    assert_eq!(
+        repository
+            .compare_and_swap_quota(quota("older", older))
+            .await
+            .expect("reject older quota"),
+        QuotaWriteOutcome::Conflict
+    );
+
+    repository
+        .apply_state_change(AccountStateChange {
+            account_id: account_id.clone(),
+            expected_revision: revision,
+            availability: AccountAvailability::QuotaExhausted,
+            reason: Some("quota_exhausted".to_owned()),
+            cooldown_until: Some(newer),
+            observed_at: newer,
+        })
+        .await
+        .expect("persist newer account state");
+    assert!(
+        repository
+            .apply_state_change(AccountStateChange {
+                account_id: account_id.clone(),
+                expected_revision: revision,
+                availability: AccountAvailability::Ready,
+                reason: None,
+                cooldown_until: None,
+                observed_at: older,
+            })
+            .await
+            .is_err(),
+        "a delayed state observation must conflict"
+    );
+
+    let current = repository
+        .get_account(&account_id)
+        .await
+        .expect("load fenced account")
+        .expect("fenced account");
+    assert_eq!(current.availability(), AccountAvailability::QuotaExhausted);
+    let observed = repository
+        .get_quotas(std::slice::from_ref(&account_id))
+        .await
+        .expect("load fenced quota")
+        .pop()
+        .expect("fenced quota");
+    assert_eq!(
+        observed.quota.expect("quota document").expose_to_provider()["marker"],
+        "newer"
+    );
+
+    database.close().await;
+}
+
 #[tokio::test]
 async fn terminal_admin_list_filters_and_sorts_before_pagination_with_retained_usage() {
     let Some(database) = TestDatabase::create("provider_account_terminal_list").await else {

backend/crates/providers/openai/src/admin.rs

@@ -25,8 +25,9 @@ use gateway_admin::model::{MutationActor, MutationContext, Revision};
 use gateway_admin::ports::provider::{ProviderAdmin, ProviderAdminError, ProviderAdminErrorKind};
 use gateway_core::accounting::Money;
 use gateway_core::engine::credential::{
-    CredentialRevision, LoadedCredential, NewProviderAccount, OpaqueProviderData,
-    PlaintextCredential, ProviderAccount, ProviderAccountId, ProviderAccountStore,
+    AccountAvailability, CredentialRevision, LoadedCredential, NewProviderAccount,
+    OpaqueProviderData, PlaintextCredential, ProviderAccount, ProviderAccountId,
+    ProviderAccountStore,
 };
 use gateway_core::error::StoreErrorKind;
 use gateway_core::operation::{GenerateRequest, Operation, ProtocolPayload};
@@ -458,21 +459,23 @@ impl ProviderAdmin for OpenAiAdminProvider {
             refresh,
             rolling_usage: _,
         } = request;
-        self.account(&account_id).await?;
+        let mut account = self.account(&account_id).await?;
         let snapshot = if refresh {
-            Some(
+            let snapshot = Some(
                 self.quota
                     .refresh_account(&account_id)
                     .await
                     .map_err(map_quota_error)?,
-            )
+            );
+            account = self.account(&account_id).await?;
+            snapshot
         } else {
             self.quota
                 .read_account(&account_id)
                 .await
                 .map_err(map_quota_error)?
         };
-        Ok(snapshot.map_or_else(empty_quota, project_quota))
+        Ok(project_quota(snapshot, &account))
     }
 
     async fn models(
@@ -746,7 +749,18 @@ fn empty_quota() -> ProviderQuota {
     }
 }
 
-fn project_quota(snapshot: CodexAccountQuotaSnapshot) -> ProviderQuota {
+fn project_quota(
+    snapshot: Option<CodexAccountQuotaSnapshot>,
+    account: &ProviderAccount,
+) -> ProviderQuota {
+    let mut quota = snapshot.map_or_else(empty_quota, project_quota_snapshot);
+    if account.availability() == AccountAvailability::QuotaExhausted {
+        force_confirmed_exhaustion_projection(&mut quota, account.cooldown_until());
+    }
+    quota
+}
+
+fn project_quota_snapshot(snapshot: CodexAccountQuotaSnapshot) -> ProviderQuota {
     let mut provider_data = Map::new();
     provider_data.insert(
         "remaining_percent".to_owned(),
@@ -799,6 +813,68 @@ fn project_quota(snapshot: CodexAccountQuotaSnapshot) -> ProviderQuota {
     }
 }
 
+fn force_confirmed_exhaustion_projection(
+    quota: &mut ProviderQuota,
+    cooldown_until: Option<SystemTime>,
+) {
+    // 真实 429 对“当前已耗尽”的证明强于可能延迟结算的 usage 文档。这里只覆盖
+    // Admin 展示投影，原始 Provider JSON 仍作为后续判断窗口是否真正恢复的基线。
+    quota.provider_data = Some(ProviderDocument::new(OpaqueProviderData::new(
+        Map::from_iter([
+            (
+                "remaining_percent".to_owned(),
+                Value::Number(Number::from(0)),
+            ),
+            ("exhausted".to_owned(), Value::Bool(true)),
+        ]),
+    )));
+    let reset_at = cooldown_until.map(DateTime::<Utc>::from);
+    if quota.windows.is_empty() {
+        quota.windows.push(ProviderQuotaWindow {
+            key: "confirmed-quota-exhaustion".to_owned(),
+            group: "other".to_owned(),
+            label: "额度".to_owned(),
+            source: None,
+            window_seconds: None,
+            used_percent: Some(100.0),
+            reset_at,
+            local_usage: None,
+            provider_data: None,
+        });
+        return;
+    }
+
+    let matched_reset = reset_at.and_then(|reset_at| {
+        quota
+            .windows
+            .iter()
+            .enumerate()
+            .filter_map(|(index, window)| {
+                let distance = window.reset_at.map(|window_reset| {
+                    (window_reset.timestamp() - reset_at.timestamp()).unsigned_abs()
+                })?;
+                Some((index, distance))
+            })
+            .min_by_key(|(_, distance)| *distance)
+            .map(|(index, _)| index)
+    });
+    let index = matched_reset
+        .or_else(|| {
+            quota
+                .windows
+                .iter()
+                .enumerate()
+                .max_by(|(_, left), (_, right)| {
+                    left.used_percent
+                        .unwrap_or(-1.0)
+                        .total_cmp(&right.used_percent.unwrap_or(-1.0))
+                })
+                .map(|(index, _)| index)
+        })
+        .unwrap_or(0);
+    quota.windows[index].used_percent = Some(100.0);
+}
+
 const fn quota_group(kind: CodexQuotaWindowKind) -> &'static str {
     match kind {
         CodexQuotaWindowKind::Monthly => "monthly",

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

@@ -194,6 +194,19 @@ pub struct CodexQuotaSyncSummary {
     pub stale: u64,
 }
 
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+enum QuotaRecoveryEvidence {
+    SuccessfulResponse,
+    RefreshedSnapshot { confirms_recovery: bool },
+    PostFailureSnapshot,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+enum QuotaRefreshMode {
+    Reconcile,
+    AfterConfirmedFailure,
+}
+
 impl CodexQuotaSyncSummary {
     #[must_use]
     pub const fn has_operational_failures(self) -> bool {
@@ -636,9 +649,20 @@ impl CodexCredentialQuotaService {
             let observed_at = SystemTime::now();
             match self.fetch_usage_with_recovery(&client, &account).await {
                 Ok(FetchedCodexQuota { account, value }) => {
+                    let previous_snapshot = observations
+                        .as_ref()
+                        .and_then(|observations| observations.get(account.id()))
+                        .filter(|observation| observation.expected_revision == account.revision())
+                        .and_then(quota_snapshot_from_observation);
                     // 单账号解析或落库失败只影响该账号；其余账号继续同步。
                     if let Err(error) = self
-                        .apply_fetched_quota(&account, &value, observed_at, &mut summary)
+                        .apply_fetched_quota(
+                            &account,
+                            &value,
+                            observed_at,
+                            previous_snapshot.as_ref(),
+                            &mut summary,
+                        )
                         .await
                     {
                         summary.transient += 1;
@@ -753,9 +777,16 @@ impl CodexCredentialQuotaService {
         account: &ProviderAccount,
         value: &Value,
         observed_at: SystemTime,
+        previous_snapshot: Option<&CodexAccountQuotaSnapshot>,
         summary: &mut CodexQuotaSyncSummary,
     ) -> Result<(), CodexCredentialQuotaError> {
-        let fact = parse_codex_quota_usage(value)?;
+        let snapshot = parse_account_quota_snapshot(
+            account.id().clone(),
+            account.revision(),
+            observed_at,
+            value,
+        )?;
+        let fact = snapshot.fact();
         let object = value
             .as_object()
             .cloned()
@@ -793,6 +824,9 @@ impl CodexCredentialQuotaService {
             current.cooldown_until(),
             fact,
             SystemTime::now(),
+            QuotaRecoveryEvidence::RefreshedSnapshot {
+                confirms_recovery: quota_snapshot_confirms_recovery(previous_snapshot, &snapshot),
+            },
         ) {
             let _ = self
                 .repository
@@ -859,6 +893,7 @@ impl CodexCredentialQuotaService {
             current.cooldown_until(),
             fact,
             observed_at,
+            QuotaRecoveryEvidence::SuccessfulResponse,
         ) {
             self.repository
                 .apply_state(
@@ -934,6 +969,24 @@ impl CodexCredentialQuotaService {
     pub async fn refresh_account(
         &self,
         account_id: &ProviderAccountId,
+    ) -> Result<CodexAccountQuotaSnapshot, CodexCredentialQuotaError> {
+        self.refresh_account_with_mode(account_id, QuotaRefreshMode::Reconcile)
+            .await
+    }
+
+    /// 真实限额失败后的异步刷新只补齐展示快照，不允许同一轮陈旧 usage 撤销失败状态。
+    pub(crate) async fn refresh_account_after_failure(
+        &self,
+        account_id: &ProviderAccountId,
+    ) -> Result<CodexAccountQuotaSnapshot, CodexCredentialQuotaError> {
+        self.refresh_account_with_mode(account_id, QuotaRefreshMode::AfterConfirmedFailure)
+            .await
+    }
+
+    async fn refresh_account_with_mode(
+        &self,
+        account_id: &ProviderAccountId,
+        mode: QuotaRefreshMode,
     ) -> Result<CodexAccountQuotaSnapshot, CodexCredentialQuotaError> {
         let account = self
             .store
@@ -941,6 +994,7 @@ impl CodexCredentialQuotaService {
             .await?
             .filter(|account| account.provider().as_str() == "openai")
             .ok_or(CodexCredentialQuotaError::NotFound)?;
+        let previous_snapshot = self.read_snapshot_for(&account).await?;
         let observed_at = SystemTime::now();
         if !access_token_is_current(&account, observed_at) {
             return Err(CodexCredentialQuotaError::CredentialRefreshRequired);
@@ -1017,6 +1071,17 @@ impl CodexCredentialQuotaService {
             current.cooldown_until(),
             snapshot.fact(),
             SystemTime::now(),
+            match mode {
+                QuotaRefreshMode::AfterConfirmedFailure => {
+                    QuotaRecoveryEvidence::PostFailureSnapshot
+                }
+                QuotaRefreshMode::Reconcile => QuotaRecoveryEvidence::RefreshedSnapshot {
+                    confirms_recovery: quota_snapshot_confirms_recovery(
+                        previous_snapshot.as_ref(),
+                        &snapshot,
+                    ),
+                },
+            },
         ) {
             self.repository
                 .apply_state(
@@ -1239,9 +1304,13 @@ fn quota_success_state(
     cooldown_until: Option<SystemTime>,
     fact: CodexQuotaFact,
     now: SystemTime,
+    recovery_evidence: QuotaRecoveryEvidence,
 ) -> Option<(AccountAvailability, Option<SystemTime>)> {
     if fact.exhausted() {
-        let observed_cooldown = fact.resets_at().map(SystemTime::from).or(cooldown_until);
+        let observed_cooldown = match (fact.resets_at().map(SystemTime::from), cooldown_until) {
+            (Some(observed), Some(confirmed)) => Some(observed.max(confirmed)),
+            (observed, confirmed) => observed.or(confirmed),
+        };
         return (current != AccountAvailability::QuotaExhausted
             || cooldown_until != observed_cooldown)
             .then_some((AccountAvailability::QuotaExhausted, observed_cooldown));
@@ -1251,8 +1320,13 @@ fn quota_success_state(
         AccountAvailability::Invalid
         | AccountAvailability::Expired
         | AccountAvailability::Banned
-        | AccountAvailability::Unknown
-        | AccountAvailability::QuotaExhausted => Some(AccountAvailability::Ready),
+        | AccountAvailability::Unknown => Some(AccountAvailability::Ready),
+        AccountAvailability::QuotaExhausted
+            if !quota_recovery_confirmed(recovery_evidence, cooldown_until, now) =>
+        {
+            None
+        }
+        AccountAvailability::QuotaExhausted => Some(AccountAvailability::Ready),
         AccountAvailability::Ready => None,
         AccountAvailability::Cooldown
             if cooldown_until.is_some_and(|cooldown_until| cooldown_until <= now) =>
@@ -1264,6 +1338,59 @@ fn quota_success_state(
     availability.map(|availability| (availability, None))
 }
 
+fn quota_recovery_confirmed(
+    evidence: QuotaRecoveryEvidence,
+    cooldown_until: Option<SystemTime>,
+    now: SystemTime,
+) -> bool {
+    if cooldown_until.is_some_and(|cooldown_until| cooldown_until <= now) {
+        return true;
+    }
+    match evidence {
+        QuotaRecoveryEvidence::SuccessfulResponse => true,
+        QuotaRecoveryEvidence::RefreshedSnapshot { confirms_recovery } => confirms_recovery,
+        QuotaRecoveryEvidence::PostFailureSnapshot => false,
+    }
+}
+
+fn quota_snapshot_confirms_recovery(
+    previous: Option<&CodexAccountQuotaSnapshot>,
+    current: &CodexAccountQuotaSnapshot,
+) -> bool {
+    if current.fact().exhausted() {
+        return false;
+    }
+    let Some(previous) = previous else {
+        return false;
+    };
+    if previous
+        .fact()
+        .remaining_percent()
+        .zip(current.fact().remaining_percent())
+        .is_some_and(|(previous, current)| current > previous)
+    {
+        return true;
+    }
+
+    // 单窗口账号可以用同一窗口的用量回落或 reset 前移证明上游提前翻新；多窗口
+    // 无法从账号级 429 确认具体受限窗口，保守等待整体剩余额度改善或真实请求成功。
+    let ([previous], [current]) = (previous.windows(), current.windows()) else {
+        return false;
+    };
+    if previous.key() != current.key() {
+        return false;
+    }
+    let usage_recovered = previous
+        .used_percent()
+        .zip(current.used_percent())
+        .is_some_and(|(previous, current)| current < previous);
+    let window_advanced = previous
+        .reset_at()
+        .zip(current.reset_at())
+        .is_some_and(|(previous, current)| current > previous);
+    usage_recovered || window_advanced
+}
+
 fn quota_state_transition(
     error: &CodexClientError,
     observed_at: SystemTime,

backend/crates/providers/openai/src/provider.rs

@@ -84,6 +84,9 @@ const MAX_COOKIE_HEADER_BYTES: usize = 16 * 1024;
 /// 提交边界前最多保留 64 KiB 原始上游 chunk；达到阈值后结束无感换号窗口，
 /// 但不会把上游数据改写成协议失败。
 const MAX_STREAM_PREFETCH_BYTES: usize = 64 * 1024;
+/// 短暂保留 response.created 等结构事件，让随后到达的明确拒绝可以无感换号；
+/// 到期即放行，避免模型长时间思考时让客户端一直收不到首事件。
+const STREAM_REPLAY_GRACE: Duration = Duration::from_millis(1_200);
 // 额度拒绝后先给上游额度结算留出时间，再以受限时长同步 usage 快照。
 const QUOTA_FAILURE_REFRESH_DELAY: Duration = Duration::from_secs(2);
 const QUOTA_FAILURE_REFRESH_TIMEOUT: Duration = Duration::from_secs(5);
@@ -682,8 +685,6 @@ fn cold_response_stream(response: ColdResponse) -> EventStream {
             lease.escape_reason(),
             lease.account_switch(),
         );
-        let stream_commit_policy =
-            StreamCommitPolicy::for_continuation(context.continuation_attempt());
         let mut response = create_response_attempt(
             &client,
             &request,
@@ -829,12 +830,13 @@ fn cold_response_stream(response: ColdResponse) -> EventStream {
         let mut decoder = CodexCanonicalDecoder::new(upstream_model.as_str())
             .with_requested_service_tier(request.service_tier())
             .with_raw_sse_passthrough();
-        let mut pre_commit_events = PreCommitClientEvents::new(stream_commit_policy);
+        let mut pre_commit_events = PreCommitClientEvents::new();
         loop {
             let Some(stream_deadline) = remaining(context.deadline()) else {
                 Err(provider_error(ProviderErrorKind::Timeout, UpstreamSendState::Sent))?;
                 return;
             };
+            let replay_grace_deadline = pre_commit_events.replay_grace_deadline();
             let next = tokio::select! {
                 biased;
                 _ = cancellation.cancelled() => Err(MappedProviderFailure::plain(provider_error(
@@ -845,14 +847,21 @@ fn cold_response_stream(response: ColdResponse) -> EventStream {
                     ProviderErrorKind::Timeout,
                     UpstreamSendState::Sent,
                 ))),
+                _ = wait_for_replay_grace(replay_grace_deadline) => Ok(PreCommitPoll::GraceElapsed),
                 chunk = body.next() => match chunk {
-                    Some(Ok(chunk)) => Ok(Some(chunk)),
+                    Some(Ok(chunk)) => Ok(PreCommitPoll::Upstream(Some(chunk))),
                     Some(Err(error)) => Err(map_stream_error(error)),
-                    None => Ok(None),
+                    None => Ok(PreCommitPoll::Upstream(None)),
                 },
             };
             let next = match next {
-                Ok(next) => next,
+                Ok(PreCommitPoll::Upstream(next)) => next,
+                Ok(PreCommitPoll::GraceElapsed) => {
+                    for event in pre_commit_events.commit_pending() {
+                        yield event;
+                    }
+                    continue;
+                }
                 Err(failure) => {
                     let updates = take_rate_limit_updates(rate_limit_updates.as_ref()).await;
                     if !updates.is_empty() {
@@ -1801,54 +1810,37 @@ impl MappedProviderFailure {
     }
 }
 
-/// 与 2.x 一致的流提交策略：普通请求优先首事件，续写/重放优先可恢复性。
-#[derive(Clone, Copy)]
-enum StreamCommitPolicy {
-    FirstForwardableEvent,
-    UntilOutputOrTerminal,
+enum PreCommitPoll<T> {
+    Upstream(T),
+    GraceElapsed,
 }
 
-impl StreamCommitPolicy {
-    const fn for_continuation(attempt: ContinuationAttempt) -> Self {
-        match attempt {
-            ContinuationAttempt::None => Self::FirstForwardableEvent,
-            ContinuationAttempt::Native
-            | ContinuationAttempt::ReplayOwner
-            | ContinuationAttempt::ReplayAny => Self::UntilOutputOrTerminal,
-        }
-    }
-
-    fn boundary_reached(
-        self,
-        events: &[ProviderEvent],
-        timing_signals: ResponseEventSignals,
-        completed: bool,
-    ) -> bool {
-        match self {
-            Self::FirstForwardableEvent => events.iter().any(ProviderEvent::has_client_event),
-            Self::UntilOutputOrTerminal => timing_signals.semantic_output || completed,
-        }
+async fn wait_for_replay_grace(deadline: Option<Instant>) {
+    if let Some(deadline) = deadline {
+        tokio::time::sleep_until(tokio::time::Instant::from_std(deadline)).await;
+    } else {
+        std::future::pending::<()>().await;
     }
 }
 
 /// 提交边界前的上游事件预取。
 ///
-/// 原始 chunk 计数而不是重编码后的 event 大小。64 KiB 只限定无感换号窗口；达到
-/// 阈值会提交已缓存 wire，绝不能因为网关私有资源规则伪造上游协议失败。一旦提交，
-/// 后续事件不再具备无痕重放资格。
+/// 原始 chunk 计数而不是重编码后的 event 大小。时间与 64 KiB 共同限定无感换号
+/// 窗口；任一边界到达都会提交已缓存 wire，不能因网关私有资源规则伪造上游协议
+/// 失败。一旦提交，后续事件不再具备无痕重放资格。
 struct PreCommitClientEvents {
-    policy: StreamCommitPolicy,
     pending: Vec<ProviderEvent>,
     prefetched_bytes: usize,
+    replay_grace_deadline: Option<Instant>,
     committed: bool,
 }
 
 impl PreCommitClientEvents {
-    const fn new(policy: StreamCommitPolicy) -> Self {
+    const fn new() -> Self {
         Self {
-            policy,
             pending: Vec::new(),
             prefetched_bytes: 0,
+            replay_grace_deadline: None,
             committed: false,
         }
     }
@@ -1868,13 +1860,17 @@ impl PreCommitClientEvents {
         if self.committed {
             return incoming;
         }
-        let boundary_reached = self
-            .policy
-            .boundary_reached(&incoming, timing_signals, completed);
+        let starts_replay_grace = incoming.iter().any(ProviderEvent::has_client_event);
         self.pending.extend(incoming);
-        if boundary_reached || self.prefetched_bytes > MAX_STREAM_PREFETCH_BYTES {
+        if timing_signals.semantic_output
+            || completed
+            || self.prefetched_bytes > MAX_STREAM_PREFETCH_BYTES
+        {
             return self.commit_pending();
         }
+        if starts_replay_grace && self.replay_grace_deadline.is_none() {
+            self.replay_grace_deadline = Instant::now().checked_add(STREAM_REPLAY_GRACE);
+        }
         Vec::new()
     }
 
@@ -1903,9 +1899,14 @@ impl PreCommitClientEvents {
     fn take_for_failure(&mut self, incoming: Vec<ProviderEvent>) -> Vec<ProviderEvent> {
         self.pending.extend(incoming);
         self.prefetched_bytes = 0;
+        self.replay_grace_deadline = None;
         std::mem::take(&mut self.pending)
     }
 
+    const fn replay_grace_deadline(&self) -> Option<Instant> {
+        self.replay_grace_deadline
+    }
+
     const fn is_committed(&self) -> bool {
         self.committed
     }
@@ -1913,6 +1914,7 @@ impl PreCommitClientEvents {
     fn commit_pending(&mut self) -> Vec<ProviderEvent> {
         self.committed = true;
         self.prefetched_bytes = 0;
+        self.replay_grace_deadline = None;
         std::mem::take(&mut self.pending)
     }
 }
@@ -2009,15 +2011,15 @@ fn schedule_authoritative_quota_refresh_after_failure(
     account: &ProviderAccount,
 ) {
     // 限额错误可以确认账号状态，但 Responses 事件中的 used_percent 可能仍停在上一结算点。
-    // usage 快照在后台补齐，不能阻塞原始失败响应。先等待 2 秒让上游结算，
-    // 再查询；5 秒仅限制实际查询自身。
+    // usage 快照在后台补齐展示基线，不得撤销同一轮真实失败，也不能阻塞原始响应。
+    // 先等待 2 秒让上游结算，再查询；5 秒仅限制实际查询自身。
     let quota = Arc::clone(quota);
     let account_id = account.id().clone();
     drop(tokio::spawn(async move {
         tokio::time::sleep(QUOTA_FAILURE_REFRESH_DELAY).await;
         match tokio::time::timeout(
             QUOTA_FAILURE_REFRESH_TIMEOUT,
-            quota.refresh_account(&account_id),
+            quota.refresh_account_after_failure(&account_id),
         )
         .await
         {

backend/crates/providers/openai/tests/admin.rs

@@ -20,8 +20,8 @@ use gateway_admin::model::provider_credentials::{
 use gateway_admin::model::{MutationActor, MutationContext, Revision};
 use gateway_admin::ports::provider::ProviderAdminErrorKind;
 use gateway_core::engine::credential::{
-    CredentialRevision, OpaqueProviderData, ProviderAccount, ProviderAccountId,
-    ProviderAccountStore, QuotaObservation,
+    AccountAvailability as CoreAccountAvailability, AccountStateChange, CredentialRevision,
+    OpaqueProviderData, ProviderAccount, ProviderAccountId, ProviderAccountStore, QuotaObservation,
 };
 use gateway_core::operation::{GenerateRequest, Operation, ProtocolPayload};
 use gateway_core::provider_ports::{
@@ -503,6 +503,108 @@ async fn openai_admin_provider_projects_codex_additional_limit_as_the_primary_qu
     assert_eq!(monthly[0].used_percent, Some(2.0));
 }
 
+#[tokio::test]
+async fn openai_admin_projects_confirmed_quota_exhaustion_as_full_without_mutating_raw_usage() {
+    let store = Arc::new(MemoryAccountStore::default());
+    let account_id = "acct_admin_confirmed_exhaustion";
+    store
+        .seed_oauth_credential(ImportCodexOAuthCredential {
+            account_id: account_id.to_owned(),
+            name: "admin confirmed exhaustion".to_owned(),
+            secret: secret("admin-confirmed-exhaustion-access"),
+            verified_account: profile("chatgpt-admin-confirmed-exhaustion"),
+            next_refresh_at: Some(Utc::now() + chrono::Duration::minutes(30)),
+            enabled: true,
+        })
+        .await;
+    let account = store.account(account_id).expect("stored account");
+    let reset_at = 1_900_000_000_u64;
+    store
+        .compare_and_swap_quota(QuotaObservation {
+            account_id: account.id().clone(),
+            expected_revision: account.revision(),
+            quota: Some(OpaqueProviderData::new(
+                json!({
+                    "rate_limit": {
+                        "allowed": true,
+                        "limit_reached": false,
+                        "primary_window": {"used_percent": 86, "reset_at": reset_at}
+                    }
+                })
+                .as_object()
+                .expect("quota object")
+                .clone(),
+            )),
+            observed_at: Some(SystemTime::now()),
+        })
+        .await
+        .expect("persist raw quota");
+    store
+        .apply_state_change(AccountStateChange {
+            account_id: account.id().clone(),
+            expected_revision: account.revision(),
+            availability: CoreAccountAvailability::QuotaExhausted,
+            reason: Some("usage_limit_exhausted".to_owned()),
+            cooldown_until: Some(SystemTime::UNIX_EPOCH + Duration::from_secs(reset_at)),
+            observed_at: SystemTime::now(),
+        })
+        .await
+        .expect("mark account exhausted");
+    let config = valid_config();
+    let bundle = provider_openai::initialize(
+        config.config.clone(),
+        provider_ports_with(Arc::clone(&store), Arc::new(TestOAuthPending::default())),
+    )
+    .await
+    .expect("OpenAI bundle");
+
+    let exhausted = bundle
+        .admin_provider()
+        .quota(ProviderQuotaRequest {
+            account_id: account.id().clone(),
+            refresh: false,
+            rolling_usage: None,
+        })
+        .await
+        .expect("project exhausted quota");
+
+    assert_eq!(exhausted.windows.len(), 1);
+    assert_eq!(exhausted.windows[0].used_percent, Some(100.0));
+    let raw = store
+        .get_quotas(std::slice::from_ref(account.id()))
+        .await
+        .expect("read raw quota")
+        .pop()
+        .expect("raw quota");
+    assert_eq!(
+        raw.quota.expect("raw document").expose_to_provider()["rate_limit"]["primary_window"]["used_percent"],
+        86
+    );
+
+    store
+        .apply_state_change(AccountStateChange {
+            account_id: account.id().clone(),
+            expected_revision: account.revision(),
+            availability: CoreAccountAvailability::Ready,
+            reason: None,
+            cooldown_until: None,
+            observed_at: SystemTime::now(),
+        })
+        .await
+        .expect("recover account");
+    let recovered = bundle
+        .admin_provider()
+        .quota(ProviderQuotaRequest {
+            account_id: account.id().clone(),
+            refresh: false,
+            rolling_usage: None,
+        })
+        .await
+        .expect("project recovered quota");
+
+    assert_eq!(recovered.windows[0].used_percent, Some(86.0));
+}
+
 #[tokio::test]
 async fn openai_admin_provider_rejects_unprepared_mutations_before_store_commit() {
     let store = Arc::new(MemoryAccountStore::default());

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

@@ -498,6 +498,38 @@ async fn manual_quota_refresh_updates_account_state_from_provider_response() {
         AccountAvailability::QuotaExhausted
     );
 
+    server.reset().await;
+    Mock::given(method("GET"))
+        .and(path("/api/codex/usage"))
+        .and(header(
+            "authorization",
+            format!("Bearer token-{account_id}"),
+        ))
+        .respond_with(ResponseTemplate::new(200).set_body_json(json!({
+            "rate_limit": {
+                "allowed": true,
+                "limit_reached": false,
+                "primary_window": {"used_percent": 100, "reset_at": 1_900_000_000}
+            }
+        })))
+        .mount(&server)
+        .await;
+
+    let stale = service
+        .refresh_account(account.id())
+        .await
+        .expect("refresh contradictory full quota");
+
+    assert!(!stale.fact().exhausted());
+    assert_eq!(
+        store
+            .account(account_id)
+            .expect("account after contradictory quota")
+            .availability(),
+        AccountAvailability::QuotaExhausted,
+        "an unchanged usage window cannot override a confirmed quota failure"
+    );
+
     server.reset().await;
     Mock::given(method("GET"))
         .and(path("/api/codex/usage"))
@@ -1018,6 +1050,118 @@ async fn periodic_quota_synchronization_rechecks_exhausted_accounts_without_wait
     assert_eq!(repeated, CodexQuotaSyncSummary::default());
 }
 
+#[tokio::test]
+async fn periodic_quota_synchronization_requires_a_real_snapshot_improvement_to_recover() {
+    let store = Arc::new(MemoryAccountStore::default());
+    let account_id = "acct_periodic_recovery_evidence";
+    create_account(&store, account_id).await;
+    let account = store.account(account_id).expect("created account");
+    let reset_at = 1_900_000_000;
+    store
+        .compare_and_swap_quota(QuotaObservation {
+            account_id: account.id().clone(),
+            expected_revision: account.revision(),
+            quota: Some(OpaqueProviderData::new(
+                json!({
+                    "rate_limit": {
+                        "allowed": true,
+                        "limit_reached": false,
+                        "primary_window": {"used_percent": 86, "reset_at": reset_at}
+                    }
+                })
+                .as_object()
+                .expect("quota object")
+                .clone(),
+            )),
+            observed_at: Some(SystemTime::now()),
+        })
+        .await
+        .expect("seed post-failure quota snapshot");
+    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_add(Duration::from_secs(24 * 60 * 60)),
+            observed_at: SystemTime::now(),
+        })
+        .await
+        .expect("mark account exhausted");
+    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": 86, "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 unchanged = service
+        .synchronize()
+        .await
+        .expect("unchanged quota recheck");
+
+    assert_eq!(unchanged.updated, 1);
+    assert_eq!(
+        store
+            .account(account_id)
+            .expect("account after unchanged snapshot")
+            .availability(),
+        AccountAvailability::QuotaExhausted
+    );
+
+    server.reset().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": 20, "reset_at": reset_at}
+            }
+        })))
+        .expect(1)
+        .mount(&server)
+        .await;
+    // 新 worker 实例模拟下一次 30 分钟复核，避免测试等待真实节流周期。
+    let next_cycle = quota_service_with_base_url(
+        &store,
+        reqwest::Client::builder()
+            .no_proxy()
+            .build()
+            .expect("client"),
+        server.uri(),
+    );
+
+    let improved = next_cycle
+        .synchronize()
+        .await
+        .expect("improved quota recheck");
+
+    assert_eq!(improved.updated, 1);
+    assert_eq!(
+        store
+            .account(account_id)
+            .expect("account after improved snapshot")
+            .availability(),
+        AccountAvailability::Ready
+    );
+}
+
 #[tokio::test]
 async fn periodic_quota_synchronization_throttles_the_same_exhausted_account() {
     let store = Arc::new(MemoryAccountStore::default());

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

@@ -1036,15 +1036,16 @@ async fn usage_limit_failure_returns_promptly_and_refreshes_the_authoritative_qu
     Mock::given(method("GET"))
         .and(path("/api/codex/usage"))
         .and(header("authorization", format!("Bearer at-{account_id}")))
-        // 限额错误本身只能确认状态；这里必须覆盖上一帧被动观察到的 99%。
+        // 生产环境会短暂返回互相矛盾的快照：数字已到 100%，但布尔位仍声称可用。
+        // 后台同步应覆盖展示数据，不能据此撤销真实请求已经确认的额度耗尽。
         .respond_with(
             ResponseTemplate::new(200)
                 // 验证后台 usage 同步不能把原始的额度错误响应拖到查询完成之后。
                 .set_delay(Duration::from_millis(750))
                 .set_body_json(json!({
                     "rate_limit": {
-                        "allowed": false,
-                        "limit_reached": true,
+                        "allowed": true,
+                        "limit_reached": false,
                         "primary_window": {"used_percent": 100, "reset_at": 1_900_000_000}
                     }
                 })),
@@ -1142,13 +1143,21 @@ async fn usage_limit_failure_returns_promptly_and_refreshes_the_authoritative_qu
             .count(),
         1
     );
+    assert_eq!(
+        store
+            .account(account_id)
+            .expect("account after stale full usage refresh")
+            .availability(),
+        AccountAvailability::QuotaExhausted,
+        "a contradictory full usage snapshot must not unlock a confirmed exhausted account"
+    );
 }
 
 #[tokio::test]
-async fn ordinary_request_should_forward_created_before_later_failed_chunk() {
+async fn ordinary_request_should_hold_created_until_later_failure_can_rotate() {
     let store = Arc::new(MemoryAccountStore::default());
     create_account(&store, "acct_first_event_latency").await;
-    let (base_url, release, _first_chunk_sent, server) = paused_chunked_sse_server(
+    let (base_url, release, first_chunk_sent, server) = paused_chunked_sse_server(
         concat!(
             "event: response.created\n",
             "data: {\"type\":\"response.created\",\"response\":{\"id\":\"resp_first_event\",\"model\":\"gpt-5.4\",\"status\":\"in_progress\"}}\n\n"
@@ -1169,41 +1178,111 @@ async fn ordinary_request_should_forward_created_before_later_failed_chunk() {
         .await
         .expect("prepare provider stream");
 
-    let first_event = loop {
-        let next = timeout(Duration::from_secs(1), stream.next())
-            .await
-            .expect("first chunk must not wait for the withheld failure")
-            .expect("provider stream must stay open")
-            .expect("provider event");
-        if next.has_client_event() {
-            break next;
+    let mut first_chunk_sent = Box::pin(first_chunk_sent);
+    loop {
+        tokio::select! {
+            sent = &mut first_chunk_sent => {
+                sent.expect("first upstream chunk");
+                break;
+            }
+            next = stream.next() => {
+                let event = next
+                    .expect("provider stream must stay open")
+                    .expect("provider event");
+                assert!(!event.has_client_event(), "response.created must remain replayable");
+            }
         }
-    };
-    assert_eq!(
-        first_event.wire_event().and_then(|wire| wire.event_type()),
-        Some("response.created")
+    }
+
+    let exposed = timeout(Duration::from_millis(100), async {
+        loop {
+            let next = stream
+                .next()
+                .await
+                .expect("provider stream must stay open")
+                .expect("provider event");
+            if next.has_client_event() {
+                return next;
+            }
+        }
+    })
+    .await;
+    assert!(
+        exposed.is_err(),
+        "a structural event must not commit the downstream before a later 429"
     );
 
     release.send(()).expect("release second upstream chunk");
-    let mut later_wire_types = Vec::new();
-    let failure = loop {
+    let mut failure = loop {
         let next = timeout(Duration::from_secs(1), stream.next())
             .await
             .expect("later failure must arrive after release")
             .expect("provider stream must return typed failure");
         match next {
-            Ok(event) => {
-                if let Some(event_type) = event.wire_event().and_then(|wire| wire.event_type()) {
-                    later_wire_types.push(event_type.to_owned());
-                }
-            }
+            Ok(event) => assert!(
+                !event.has_client_event(),
+                "failure attempt leaked downstream"
+            ),
             Err(error) => break error,
         }
     };
 
-    assert_eq!(later_wire_types, vec!["response.failed"]);
-    assert!(!failure.replay_is_safe());
-    assert!(!failure.has_atomic_client_events());
+    assert!(failure.replay_is_safe());
+    assert_eq!(
+        failure
+            .take_atomic_client_events()
+            .iter()
+            .filter_map(|event| event.wire_event()?.event_type())
+            .collect::<Vec<_>>(),
+        vec!["response.created", "response.failed"]
+    );
+    server.await.expect("chunked SSE server");
+}
+
+#[tokio::test]
+async fn ordinary_request_should_bound_structural_event_replay_grace() {
+    let store = Arc::new(MemoryAccountStore::default());
+    create_account(&store, "acct_bounded_replay_grace").await;
+    let (base_url, release, _first_chunk_sent, server) = paused_chunked_sse_server(
+        concat!(
+            "event: response.created\n",
+            "data: {\"type\":\"response.created\",\"response\":{\"id\":\"resp_bounded_grace\",\"model\":\"gpt-5.4\",\"status\":\"in_progress\"}}\n\n"
+        )
+        .to_owned(),
+        String::new(),
+    )
+    .await;
+    let mut stream = provider_with_base_url(&store, base_url)
+        .execute(
+            planned_request("openai", http_generate_operation()),
+            context("req_bounded_replay_grace", CancellationToken::new()),
+        )
+        .await
+        .expect("prepare provider stream");
+
+    let first_event = timeout(Duration::from_secs(2), async {
+        loop {
+            let event = stream
+                .next()
+                .await
+                .expect("provider stream must stay open")
+                .expect("provider event");
+            if event.has_client_event() {
+                return event;
+            }
+        }
+    })
+    .await
+    .expect("response.created must be released after the bounded grace period");
+
+    assert_eq!(
+        first_event.wire_event().and_then(|wire| wire.event_type()),
+        Some("response.created")
+    );
+    release.send(()).expect("finish upstream response");
+    while let Some(event) = stream.next().await {
+        event.expect("clean upstream EOF");
+    }
     server.await.expect("chunked SSE server");
 }
 

backend/crates/providers/openai/tests/support.rs

@@ -36,6 +36,7 @@ struct StoredAccount {
     account: ProviderAccount,
     credential: gateway_core::engine::credential::PlaintextCredential,
     quota: Option<QuotaObservation>,
+    state_observed_at: Option<SystemTime>,
 }
 
 #[derive(Default)]
@@ -129,6 +130,7 @@ impl ProviderAccountStore for MemoryAccountStore {
                 account: input.account,
                 credential: input.credential,
                 quota: None,
+                state_observed_at: None,
             },
         );
         Ok(())
@@ -260,6 +262,15 @@ impl ProviderAccountStore for MemoryAccountStore {
         if stored.account.revision() != observation.expected_revision {
             return Ok(QuotaWriteOutcome::Conflict);
         }
+        if stored
+            .quota
+            .as_ref()
+            .and_then(|quota| quota.observed_at)
+            .zip(observation.observed_at)
+            .is_some_and(|(current, incoming)| current > incoming)
+        {
+            return Ok(QuotaWriteOutcome::Conflict);
+        }
         stored.quota = Some(observation);
         Ok(QuotaWriteOutcome::Updated)
     }
@@ -272,6 +283,12 @@ impl ProviderAccountStore for MemoryAccountStore {
         if stored.account.revision() != change.expected_revision {
             return Err(store_error(StoreErrorKind::Conflict));
         }
+        if stored
+            .state_observed_at
+            .is_some_and(|current| current > change.observed_at)
+        {
+            return Err(store_error(StoreErrorKind::Conflict));
+        }
         stored.account = rebuild_account(
             &stored.account,
             AccountRebuild {
@@ -285,6 +302,7 @@ impl ProviderAccountStore for MemoryAccountStore {
                 profile: None,
             },
         );
+        stored.state_observed_at = Some(change.observed_at);
         Ok(())
     }
 

backend/migrations/.frozen-sha256

@@ -1,5 +1 @@
-0554fed953bd1440b50bc4033dc8bad96b51e132c57c2ccf1553728a991d5cb2  0001_initial.sql
-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
+a4681a5c040e80c5234e879293fc34972080e5251dfbc7c40e1374f8e4170a4c  0001_initial.sql

backend/migrations/0001_initial.sql

@@ -166,7 +166,14 @@ create table provider_accounts (
     )
   ),
   constraint provider_accounts_cooldown_ck check (
-    (availability = 'cooldown') = (cooldown_until is not null)
+    (
+      availability != 'cooldown'
+      or cooldown_until is not null
+    )
+    and (
+      cooldown_until is null
+      or availability in ('cooldown', 'quota_exhausted')
+    )
   ),
   constraint provider_accounts_time_ck check (
     created_at <= updated_at
@@ -208,6 +215,7 @@ create table model_requests (
   requested_model_id text not null,
   provider_kind text,
   upstream_model_id text,
+  service_tier text,
   provider_account_id text,
   provider_account_ref text,
   upstream_transport text,
@@ -220,9 +228,11 @@ create table model_requests (
   outcome text not null default 'running',
   client_status_code integer,
   upstream_status_code integer,
-  client_response_id text,
+  -- Provider response IDs are opaque JSON strings. Bytes preserve embedded NUL
+  -- and avoid imposing PostgreSQL text-index limits on protocol handles.
+  client_response_id bytea,
   upstream_request_id text,
-  upstream_response_id text,
+  upstream_response_id bytea,
   error_kind text,
   provider_error_code text,
   error_message text,
@@ -321,6 +331,13 @@ create table model_requests (
     provider_observation_json is null
     or jsonb_typeof(provider_observation_json) = 'object'
   ),
+  constraint model_requests_service_tier_ck check (
+    service_tier is null
+    or (
+      octet_length(service_tier) between 1 and 64
+      and service_tier !~ '[[:cntrl:]]'
+    )
+  ),
   constraint model_requests_cost_ck check (
     cost_source in ('provider_reported', 'calculated', 'unavailable')
     and (
@@ -414,10 +431,6 @@ create index model_requests_running_deadline_idx
 create index model_requests_retention_idx
   on model_requests (completed_at)
   where outcome <> 'running';
-create unique index model_requests_client_response_uq
-  on model_requests (client_response_id)
-  where client_response_id is not null;
-
 create table ops_events (
   id text primary key,
   model_request_id text,

backend/migrations/0002_drop_opaque_response_id_index.sql

@@ -1,3 +0,0 @@
--- Provider response IDs are opaque protocol values, not relational identifiers.
--- Indexing the raw text makes PostgreSQL reject otherwise valid long handles.
-drop index model_requests_client_response_uq;

backend/migrations/0003_store_opaque_response_ids_as_bytes.sql

@@ -1,7 +0,0 @@
--- Response IDs are opaque JSON strings. PostgreSQL text rejects NUL even though JSON accepts it.
--- Store their UTF-8 bytes without an index; Rust decodes them only at the audit/API boundary.
-alter table model_requests
-  alter column client_response_id type bytea
-    using convert_to(client_response_id, 'UTF8'),
-  alter column upstream_response_id type bytea
-    using convert_to(upstream_response_id, 'UTF8');

backend/migrations/0004_add_model_request_service_tier.sql

@@ -1,9 +0,0 @@
-alter table model_requests
-  add column service_tier text,
-  add constraint model_requests_service_tier_ck check (
-    service_tier is null
-    or (
-      octet_length(service_tier) between 1 and 64
-      and service_tier !~ '[[:cntrl:]]'
-    )
-  );

backend/migrations/0005_allow_quota_exhausted_cooldown.sql

@@ -1,12 +0,0 @@
-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')
-    )
-  );

backend/migrations/README.md

@@ -6,14 +6,16 @@ sqlx 在服务启动时（`serve` 监听之前）按编号顺序执行本目录
 
 ## 冻结规则
 
-- 已合入 main 的迁移文件**永久冻结**，一个字节都不能改——包括注释和空白。
-  任何改动都会让所有已部署实例（含自更新链路上的存量用户）在下次重启时
-  因 checksum 失配而无法启动，且没有自动恢复路径。
-- schema 变更一律新增编号迁移（`0003_...`、`0004_...`），永远不回改旧文件。
+- 同一大版本内，已合入 main 的迁移文件**永久冻结**，一个字节都不能改——包括注释和
+  空白。任何改动都会让该大版本的已部署实例在下次重启时因 checksum 失配而无法启动，
+  且没有自动恢复路径。
+- 同一大版本内的 schema 变更一律新增编号迁移（`0002_...`、`0003_...`），永远不回改
+  旧文件。切换大版本时允许把已完成升级的 schema 折叠成新的 `0001` 基线，因为跨大版本
+  不支持在线升级，目标版本必须使用全新数据目录部署。
 - `.frozen-sha256` 是冻结清单（`sha256sum` 格式）。CI 会校验：清单内文件
   字节未变，且目录里每个 `*.sql` 都已入册。新增迁移时在同一提交里执行
   `sha256sum 000N_xxx.sql >> .frozen-sha256` 入册。
-- 若确需修正已冻结迁移里的错误，用新迁移做补偿性变更（`alter` / 回填），
+- 若同一大版本内确需修正已冻结迁移里的错误，用新迁移做补偿性变更（`alter` / 回填），
   不要动原文件。
 
 ## 本地测试库

deploy/README.md

@@ -130,10 +130,8 @@ Provider schema 以明文 JSON 保存于 PostgreSQL。Redis 只保存可重建
 ## 镜像升级与源码构建
 
 > [!WARNING]
-> 以下命令只适用于同一大版本内的升级。数据库 schema 与配置只在同一大版本内兼容，
-> 跨大版本直接换镜像会导致服务无法启动；内置一键自更新会拒绝跨大版本目标（3.0 之前
-> 的旧版没有这道门禁，不要用旧版界面的一键更新跨大版本）。跨大版本请全新部署（全新
-> `.runtime/` 数据目录），并重新导入或重新授权 Provider 账号与客户端 Key。
+> 以下命令只适用于同一大版本内的升级，不支持跨大版本在线升级。跨大版本请使用全新的
+> `.runtime/` 数据目录重新部署，并重新导入或重新授权 Provider 账号与客户端 Key。
 
 ```bash
 docker compose -f deploy/compose.yaml build codex-proxy-rs

docs/architecture.md

@@ -170,11 +170,11 @@ Admin API 不要求客户端提交配置 revision，也不向客户端暴露配
   客户端可见协议结果。队列不是第二份业务权威，也不做无幂等键的盲目重试。
 - retention 只删除已满足保留规则的历史事实，不改变运行中请求。
 
-业务 schema 由 `backend/migrations/` 下的编号迁移按序定义：`0001_initial.sql` 建表，`0002` 删除
-opaque response ID 唯一索引，`0003` 将 client/upstream response ID 改为无格式假设的 UTF-8 bytes，
-`0004` 增加响应观测得到的 `model_requests.service_tier`。已合入的迁移永久冻结——
-`.frozen-sha256` 是冻结清单，CI 校验清单相对 PR base 只增不改，启动时 sqlx 重校验已应用迁移的
-checksum，不一致直接拒绝启动。规则见 `backend/migrations/README.md`。
+业务 schema 由 `backend/migrations/0001_initial.sql` 直接定义当前大版本的完整新库基线，其中
+response ID 使用无格式假设的 UTF-8 bytes，`model_requests.service_tier` 保存响应观测到的服务档位，
+且额度耗尽账号可以携带冷却截止时间。同一大版本内已合入的迁移永久冻结——
+`.frozen-sha256` 是冻结清单，CI 校验清单相对 PR base 只增不改，启动时 sqlx 重校验已应用
+迁移的 checksum，不一致直接拒绝启动。规则见 `backend/migrations/README.md`。
 
 ## 9. Redis 终态
 

release/version.yaml

@@ -1 +1 @@
-version: "3.0.29"
+version: "3.0.30"

skills/codex-proxy-rs/references/repo-guide.md

@@ -40,9 +40,9 @@
 
 ## 存储
 
-- PostgreSQL 业务表只有 `0001_initial.sql` 定义的七张；后续编号迁移只做增量。当前 `0002` 删除
-  opaque response ID 索引，`0003` 把 response ID 改为 bytes，`0004` 增加响应观测的 service tier。
-- 已应用迁移按字节冻结：`backend/migrations/.frozen-sha256` 是冻结清单，CI 相对 PR base 做 append-only 校验；schema 变更一律新增编号迁移，规则见 `backend/migrations/README.md`。
+- PostgreSQL 业务表只有 `0001_initial.sql` 定义的七张；该文件直接表达当前大版本的完整新库基线，
+  包括 bytes response ID、响应观测 service tier 与额度耗尽冷却约束。
+- 同一大版本内已应用迁移按字节冻结：`backend/migrations/.frozen-sha256` 是冻结清单，CI 相对 PR base 做 append-only 校验；schema 变更一律新增编号迁移，规则见 `backend/migrations/README.md`。
 - `config_revision` 只用于会改变调度快照或安全配置的管理 mutation。
 - quota、cooldown、catalog generation、自动 refresh 不推进全局 revision。
 - refresh 只推进账号 `credential_revision`；Redis cooldown 是可丢失热缓存。