test(openai): move provider mapping tests to integration suite

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

backend/crates/gateway-api/src/openai/error.rs

@@ -7,8 +7,8 @@ use axum::{
     response::{IntoResponse, Response},
 };
 use gateway_core::{
-    engine::{EngineError, UpstreamSendState},
-    error::{GatewayError, GatewayErrorKind, ProviderErrorKind},
+    engine::EngineError,
+    error::{GatewayError, GatewayErrorKind},
 };
 use serde_json::{Value, json};
 
@@ -72,19 +72,6 @@ pub fn model_not_found_response() -> (StatusCode, Json<Value>) {
 #[must_use]
 pub fn gateway_error_from_engine(error: &EngineError) -> GatewayError {
     match error {
-        EngineError::Provider(error)
-            if error.kind() == ProviderErrorKind::Unavailable
-                && error.send_state() == UpstreamSendState::NotSent =>
-        {
-            let gateway = GatewayError::new(
-                GatewayErrorKind::NoAvailableProvider,
-                "no upstream provider is currently available for this request",
-            );
-            match error.client_visible_upstream_error() {
-                Some(upstream) => gateway.with_client_visible_upstream_error(upstream.clone()),
-                None => gateway,
-            }
-        }
         EngineError::Provider(error) => GatewayError::from_provider(error),
         EngineError::Cancelled => {
             GatewayError::new(GatewayErrorKind::Cancelled, "request was cancelled")
@@ -185,6 +172,16 @@ pub const fn gateway_error_contract(
             "server_error",
             "no_available_provider",
         ),
+        GatewayErrorKind::AccountCapacityUnavailable => (
+            StatusCode::SERVICE_UNAVAILABLE,
+            "server_error",
+            "account_capacity_unavailable",
+        ),
+        GatewayErrorKind::ProviderInfrastructureUnavailable => (
+            StatusCode::SERVICE_UNAVAILABLE,
+            "server_error",
+            "provider_infrastructure_unavailable",
+        ),
         GatewayErrorKind::RateLimited => (
             StatusCode::TOO_MANY_REQUESTS,
             "rate_limit_error",

backend/crates/gateway-api/tests/openai/error.rs

@@ -58,6 +58,30 @@ fn no_available_provider_error_should_map_to_service_unavailable() {
     );
 }
 
+#[test]
+fn account_capacity_error_should_map_to_service_unavailable() {
+    assert_eq!(
+        gateway_error_contract(GatewayErrorKind::AccountCapacityUnavailable),
+        (
+            StatusCode::SERVICE_UNAVAILABLE,
+            "server_error",
+            "account_capacity_unavailable",
+        )
+    );
+}
+
+#[test]
+fn provider_infrastructure_error_should_map_to_service_unavailable() {
+    assert_eq!(
+        gateway_error_contract(GatewayErrorKind::ProviderInfrastructureUnavailable),
+        (
+            StatusCode::SERVICE_UNAVAILABLE,
+            "server_error",
+            "provider_infrastructure_unavailable",
+        )
+    );
+}
+
 #[test]
 fn rate_limited_error_should_map_to_openai_retryable_status() {
     assert_eq!(
@@ -122,29 +146,32 @@ fn engine_provider_error_should_preserve_retry_classification() {
 #[test]
 fn local_provider_capacity_exhaustion_should_map_to_service_unavailable() {
     let error = EngineError::Provider(ProviderError::new(
-        ProviderErrorKind::Unavailable,
+        ProviderErrorKind::AccountCapacityUnavailable,
         UpstreamSendState::NotSent,
     ));
 
     assert_eq!(
         gateway_error_from_engine(&error),
         GatewayError::new(
-            GatewayErrorKind::NoAvailableProvider,
-            "no upstream provider is currently available for this request"
+            GatewayErrorKind::AccountCapacityUnavailable,
+            "all eligible upstream accounts are temporarily busy"
         )
     );
 }
 
 #[test]
-fn local_provider_capacity_exhaustion_should_preserve_safe_provider_detail() {
+fn no_eligible_provider_account_should_preserve_safe_provider_detail() {
     let detail = ClientVisibleUpstreamError::new(
         "no account is eligible for the requested model",
         Some("no_eligible_account".to_owned()),
         Some("account_unavailable_error".to_owned()),
     );
     let error = EngineError::Provider(
-        ProviderError::new(ProviderErrorKind::Unavailable, UpstreamSendState::NotSent)
-            .with_client_visible_upstream_error(detail),
+        ProviderError::new(
+            ProviderErrorKind::NoEligibleAccount,
+            UpstreamSendState::NotSent,
+        )
+        .with_client_visible_upstream_error(detail),
     );
     let gateway = gateway_error_from_engine(&error);
 
@@ -164,6 +191,22 @@ fn local_provider_capacity_exhaustion_should_preserve_safe_provider_detail() {
     );
 }
 
+#[test]
+fn provider_infrastructure_failure_should_have_a_distinct_client_contract() {
+    let error = EngineError::Provider(ProviderError::new(
+        ProviderErrorKind::ProviderInfrastructureUnavailable,
+        UpstreamSendState::NotSent,
+    ));
+
+    assert_eq!(
+        gateway_error_from_engine(&error),
+        GatewayError::new(
+            GatewayErrorKind::ProviderInfrastructureUnavailable,
+            "provider account infrastructure is temporarily unavailable"
+        )
+    );
+}
+
 #[test]
 fn no_eligible_provider_account_should_map_to_service_unavailable() {
     let error = EngineError::Provider(ProviderError::new(
@@ -181,16 +224,18 @@ fn no_eligible_provider_account_should_map_to_service_unavailable() {
 }
 
 #[test]
-fn sent_provider_unavailability_should_remain_bad_gateway() {
-    let error = EngineError::Provider(ProviderError::new(
-        ProviderErrorKind::Unavailable,
-        UpstreamSendState::Sent,
-    ));
+fn provider_unavailability_should_remain_bad_gateway_regardless_of_send_state() {
+    for send_state in [UpstreamSendState::NotSent, UpstreamSendState::Sent] {
+        let error = EngineError::Provider(ProviderError::new(
+            ProviderErrorKind::Unavailable,
+            send_state,
+        ));
 
-    assert_eq!(
-        gateway_error_from_engine(&error).kind(),
-        GatewayErrorKind::UpstreamUnavailable
-    );
+        assert_eq!(
+            gateway_error_from_engine(&error).kind(),
+            GatewayErrorKind::UpstreamUnavailable
+        );
+    }
 }
 
 #[test]

backend/crates/gateway-core/src/engine/coordinator.rs

@@ -165,8 +165,8 @@ where
             timings: ModelRequestTimings::default(),
             client_response_id: None,
             upstream_response_id: None,
-            last_account_exhaustion: None,
-            last_account_exhaustion_events: Vec::new(),
+            last_retryable_failure: None,
+            last_retryable_failure_events: Vec::new(),
             provider_attempt_outcomes: Vec::new(),
             pending_terminal_failure: None,
         };
@@ -245,9 +245,10 @@ pub struct ResponseExecutionSession<S: ?Sized> {
     timings: ModelRequestTimings,
     client_response_id: Option<String>,
     upstream_response_id: Option<String>,
-    last_account_exhaustion: Option<ProviderError>,
-    /// 最近一次账号耗尽在重试时被丢弃的原始失败批次；只用于后续空选路终态。
-    last_account_exhaustion_events: Vec<ProviderEvent>,
+    /// 最近一次为无感恢复而被丢弃的原始上游错误；只在后续空选路时成为终态。
+    last_retryable_failure: Option<ProviderError>,
+    /// 与 `last_retryable_failure` 同属一个 attempt 的原始失败批次。
+    last_retryable_failure_events: Vec<ProviderEvent>,
     provider_attempt_outcomes: Vec<ProviderAttemptOutcome>,
     /// 原子失败批次已交给协议层、但尚待下游提交后收敛的原 Provider 错误。
     pending_terminal_failure: Option<PendingTerminalFailure>,
@@ -659,25 +660,30 @@ where
             ProviderBoundary::Result(result) => match *result {
                 Ok(stream) => stream,
                 Err(error) => {
-                    if error.kind() == ProviderErrorKind::NoEligibleAccount
-                        && let Some(exhaustion) = self.last_account_exhaustion.take()
+                    if matches!(
+                        error.kind(),
+                        ProviderErrorKind::AccountCapacityUnavailable
+                            | ProviderErrorKind::NoEligibleAccount
+                    ) && let Some(last_failure) = self.last_retryable_failure.take()
                     {
                         let send_state = self.current_send_state();
-                        let mut events = std::mem::take(&mut self.last_account_exhaustion_events);
+                        let mut events = std::mem::take(&mut self.last_retryable_failure_events);
                         if !events.is_empty() {
                             self.observe_atomic_terminal_events(&mut events);
                             return Ok(Some(PullOutcome::TerminalFailure {
                                 events,
-                                error: exhaustion,
+                                error: last_failure,
                                 send_state,
                             }));
                         }
-                        self.finish_provider_error(&exhaustion).await?;
-                        return Err(provider_engine_error(exhaustion));
+                        self.finish_provider_error(&last_failure).await?;
+                        return Err(provider_engine_error(last_failure));
                     }
                     if !(matches!(
                         error.kind(),
-                        ProviderErrorKind::Unavailable | ProviderErrorKind::NoEligibleAccount
+                        ProviderErrorKind::AccountCapacityUnavailable
+                            | ProviderErrorKind::NoEligibleAccount
+                            | ProviderErrorKind::ProviderInfrastructureUnavailable
                     ) && error.send_state() == UpstreamSendState::NotSent)
                     {
                         self.record_provider_failure(candidate.provider().clone(), error.kind());
@@ -909,14 +915,10 @@ where
         &mut self,
         mut error: ProviderError,
     ) -> Result<StreamErrorOutcome, EngineError> {
-        // 原始 wire 只能活在本次决策栈内；后续 clone、attempt 记录与终态对象均只
-        // 接触已剥离的稳定错误字段。
+        // 原始 wire 只活在 request-local 决策状态；clone、attempt 记录与持久化终态
+        // 均只接触已剥离的稳定错误字段。
         let mut atomic_client_events = error.take_atomic_client_events();
         let current = self.current.take().ok_or(EngineError::NoActiveAttempt)?;
-        let account_exhausted = matches!(
-            error.kind(),
-            ProviderErrorKind::RateLimited | ProviderErrorKind::QuotaExhausted
-        );
         self.record_provider_failure(current.metadata.provider().clone(), error.kind());
         // attempt_send_state 是本 attempt 自身的发送事实，驱动重试门；
         // 持久化与终态用请求级水位，二者不可混用（水位会把早先 attempt 的
@@ -976,8 +978,8 @@ where
             continuation_retry || same_account_retry || ordinary_retry || pre_delivery_retry;
 
         if retryable {
-            // 普通 clone 只保留稳定事实；原始失败响应只能随最后一次客户端可见
-            // 的账号耗尽错误继续，不能进入中间失败记录。
+            // 普通 clone 只保留稳定事实；原始 wire/HTTP response 由 request-local
+            // 所有权保留到下一次 attempt 成功，或最终空选路时返回客户端。
             let persistence_error = self.request_persisted.then(|| error.clone());
             if same_account_retry {
                 if let Some(account) = current.metadata.provider_account_id() {
@@ -992,10 +994,8 @@ where
             {
                 self.excluded_accounts.insert(account.clone());
             }
-            if account_exhausted {
-                self.last_account_exhaustion_events = atomic_client_events;
-                self.last_account_exhaustion = Some(error);
-            }
+            self.last_retryable_failure_events = atomic_client_events;
+            self.last_retryable_failure = Some(error);
             if let Some(error) = persistence_error {
                 best_effort_store_write(
                     "record_intermediate_failure",

backend/crates/gateway-core/src/error.rs

@@ -129,8 +129,12 @@ pub enum ProviderErrorKind {
     RateLimited,
     /// Credential 配额耗尽。
     QuotaExhausted,
+    /// 仍有符合条件的账号，但它们暂时没有可调度容量。
+    AccountCapacityUnavailable,
     /// Provider 已确认当前请求无法选出可用账号。
     NoEligibleAccount,
+    /// Provider 的账号存储、租约协调或本地凭据数据不可用。
+    ProviderInfrastructureUnavailable,
     /// 请求超时。
     Timeout,
     /// 网络或 transport 失败。
@@ -156,7 +160,9 @@ impl ProviderErrorKind {
             Self::PermissionDenied => "permission_denied",
             Self::RateLimited => "rate_limited",
             Self::QuotaExhausted => "quota_exhausted",
+            Self::AccountCapacityUnavailable => "account_capacity_unavailable",
             Self::NoEligibleAccount => "no_eligible_account",
+            Self::ProviderInfrastructureUnavailable => "provider_infrastructure_unavailable",
             Self::Timeout => "timeout",
             Self::Transport => "transport",
             Self::Protocol => "protocol",
@@ -711,6 +717,10 @@ pub enum GatewayErrorKind {
     ModelNotFound,
     /// 当前没有可用 target。
     NoAvailableProvider,
+    /// 符合条件的上游账号暂时没有调度容量。
+    AccountCapacityUnavailable,
+    /// Provider 的本地账号基础设施不可用。
+    ProviderInfrastructureUnavailable,
     /// 上游限流。
     RateLimited,
     /// 上游暂不可用。
@@ -734,6 +744,8 @@ impl GatewayErrorKind {
             Self::PolicyDenied => "policy_denied",
             Self::ModelNotFound => "model_not_found",
             Self::NoAvailableProvider => "no_available_provider",
+            Self::AccountCapacityUnavailable => "account_capacity_unavailable",
+            Self::ProviderInfrastructureUnavailable => "provider_infrastructure_unavailable",
             Self::RateLimited => "rate_limited",
             Self::UpstreamUnavailable => "upstream_unavailable",
             Self::Timeout => "timeout",
@@ -782,10 +794,18 @@ impl GatewayError {
                 GatewayErrorKind::RateLimited,
                 "upstream capacity is temporarily unavailable",
             ),
+            ProviderErrorKind::AccountCapacityUnavailable => Self::new(
+                GatewayErrorKind::AccountCapacityUnavailable,
+                "all eligible upstream accounts are temporarily busy",
+            ),
             ProviderErrorKind::NoEligibleAccount => Self::new(
                 GatewayErrorKind::NoAvailableProvider,
                 "no upstream provider is currently available for this request",
             ),
+            ProviderErrorKind::ProviderInfrastructureUnavailable => Self::new(
+                GatewayErrorKind::ProviderInfrastructureUnavailable,
+                "provider account infrastructure is temporarily unavailable",
+            ),
             ProviderErrorKind::Timeout => {
                 Self::new(GatewayErrorKind::Timeout, "upstream request timed out")
             }

backend/crates/gateway-core/tests/engine/coordinator.rs

@@ -29,8 +29,9 @@ use gateway_core::engine::{
     ProviderAttemptOutcome, RecoveryReport, UpstreamSendState,
 };
 use gateway_core::error::{
-    ClientVisibleUpstreamResponse, ContinuationFailure, GatewayErrorKind, OpaqueUpstreamValue,
-    ProviderError, ProviderErrorKind, StoreError, StoreErrorKind,
+    ClientVisibleUpstreamError, ClientVisibleUpstreamResponse, ContinuationFailure,
+    GatewayErrorKind, OpaqueUpstreamValue, ProviderError, ProviderErrorKind, StoreError,
+    StoreErrorKind,
 };
 use gateway_core::event::{
     ContentItem, ContentKind, GatewayEvent, ProtocolWireEvent, ProviderEvent,
@@ -2554,6 +2555,128 @@ fn pre_delivery_transport_failure_rotates_account_for_non_idempotent_generation(
     assert_eq!(state.finalizations[0].outcome, ExecutionOutcome::Succeeded);
 }
 
+#[test]
+fn final_capacity_exhaustion_returns_the_last_retryable_upstream_failure() {
+    let operation = operation(RetrySafety::NonIdempotent);
+    let route_plan = plan(&operation);
+    let raw_body = Bytes::from_static(b"{\"error\":\"last upstream body\"}\x00");
+    let upstream_error =
+        ProviderError::new(ProviderErrorKind::Transport, UpstreamSendState::Ambiguous)
+            .with_pre_delivery_retry()
+            .with_client_visible_upstream_error(ClientVisibleUpstreamError::new(
+                "message too big",
+                Some("1009".to_owned()),
+                Some("websocket_close_error".to_owned()),
+            ))
+            .with_client_visible_upstream_response(ClientVisibleUpstreamResponse::new(
+                502,
+                Some(b"application/json".to_vec()),
+                raw_body.clone(),
+            ));
+    let (coordinator, store, provider) = coordinator(vec![
+        Script::Stream {
+            account_id: "acct_first",
+            items: vec![Err(upstream_error)],
+        },
+        Script::Error(ProviderError::new(
+            ProviderErrorKind::AccountCapacityUnavailable,
+            UpstreamSendState::NotSent,
+        )),
+    ]);
+
+    let mut session = block_on(coordinator.start(
+        model_request(&operation, SystemTime::now() + Duration::from_secs(30)),
+        operation,
+        route_plan,
+        None,
+        None,
+        CancellationToken::new(),
+    ))
+    .expect("start execution");
+    let error = block_on(session.collect_uncommitted())
+        .expect_err("empty selection must return the retained upstream failure");
+    let EngineError::Provider(error) = error else {
+        panic!("expected retained provider error");
+    };
+
+    assert_eq!(error.kind(), ProviderErrorKind::Transport);
+    let detail = error
+        .client_visible_upstream_error()
+        .expect("structured upstream detail");
+    assert_eq!(detail.message(), "message too big");
+    assert_eq!(detail.code(), Some("1009"));
+    assert_eq!(detail.error_type(), Some("websocket_close_error"));
+    let response = error
+        .client_visible_upstream_response()
+        .expect("original upstream response");
+    assert_eq!(response.status(), 502);
+    assert_eq!(response.body(), &raw_body);
+
+    assert_eq!(provider.contexts.lock().expect("contexts lock").len(), 2);
+    let state = store.state.lock().expect("store lock");
+    assert_eq!(state.intermediate_failures, 1);
+    assert_eq!(state.finalizations[0].outcome, ExecutionOutcome::Failed);
+    assert_eq!(
+        state.finalizations[0].send_state,
+        UpstreamSendState::Ambiguous
+    );
+}
+
+#[test]
+fn provider_infrastructure_failure_does_not_restore_the_last_upstream_failure() {
+    let operation = operation(RetrySafety::NonIdempotent);
+    let route_plan = plan(&operation);
+    let upstream_error =
+        ProviderError::new(ProviderErrorKind::Transport, UpstreamSendState::Ambiguous)
+            .with_pre_delivery_retry()
+            .with_client_visible_upstream_error(ClientVisibleUpstreamError::new(
+                "upstream marker",
+                Some("upstream_marker".to_owned()),
+                Some("upstream_error".to_owned()),
+            ))
+            .with_client_visible_upstream_response(ClientVisibleUpstreamResponse::new(
+                502,
+                Some(b"application/json".to_vec()),
+                Bytes::from_static(b"{\"error\":\"upstream marker\"}"),
+            ));
+    let (coordinator, store, provider) = coordinator(vec![
+        Script::Stream {
+            account_id: "acct_first",
+            items: vec![Err(upstream_error)],
+        },
+        Script::Error(ProviderError::new(
+            ProviderErrorKind::ProviderInfrastructureUnavailable,
+            UpstreamSendState::NotSent,
+        )),
+    ]);
+
+    let mut session = block_on(coordinator.start(
+        model_request(&operation, SystemTime::now() + Duration::from_secs(30)),
+        operation,
+        route_plan,
+        None,
+        None,
+        CancellationToken::new(),
+    ))
+    .expect("start execution");
+    let error = block_on(session.collect_uncommitted())
+        .expect_err("provider infrastructure failure must remain terminal");
+    let EngineError::Provider(error) = error else {
+        panic!("expected provider error");
+    };
+
+    assert_eq!(
+        error.kind(),
+        ProviderErrorKind::ProviderInfrastructureUnavailable
+    );
+    assert!(error.client_visible_upstream_error().is_none());
+    assert!(error.client_visible_upstream_response().is_none());
+    assert_eq!(provider.contexts.lock().expect("contexts lock").len(), 2);
+    let state = store.state.lock().expect("store lock");
+    assert_eq!(state.intermediate_failures, 1);
+    assert_eq!(state.finalizations[0].outcome, ExecutionOutcome::Failed);
+}
+
 #[test]
 fn pre_delivery_retry_marker_is_ignored_after_downstream_commit() {
     let operation = operation(RetrySafety::NonIdempotent);

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

@@ -197,14 +197,7 @@ pub struct CodexQuotaSyncSummary {
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 enum QuotaRecoveryEvidence {
     SuccessfulResponse,
-    RefreshedSnapshot { confirms_recovery: bool },
-    PostFailureSnapshot,
-}
-
-#[derive(Debug, Clone, Copy, PartialEq, Eq)]
-enum QuotaRefreshMode {
-    Reconcile,
-    AfterConfirmedFailure,
+    RefreshedSnapshot,
 }
 
 impl CodexQuotaSyncSummary {
@@ -649,20 +642,9 @@ 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,
-                            previous_snapshot.as_ref(),
-                            &mut summary,
-                        )
+                        .apply_fetched_quota(&account, &value, observed_at, &mut summary)
                         .await
                     {
                         summary.transient += 1;
@@ -777,7 +759,6 @@ impl CodexCredentialQuotaService {
         account: &ProviderAccount,
         value: &Value,
         observed_at: SystemTime,
-        previous_snapshot: Option<&CodexAccountQuotaSnapshot>,
         summary: &mut CodexQuotaSyncSummary,
     ) -> Result<(), CodexCredentialQuotaError> {
         let snapshot = parse_account_quota_snapshot(
@@ -824,9 +805,7 @@ impl CodexCredentialQuotaService {
             current.cooldown_until(),
             fact,
             SystemTime::now(),
-            QuotaRecoveryEvidence::RefreshedSnapshot {
-                confirms_recovery: quota_snapshot_confirms_recovery(previous_snapshot, &snapshot),
-            },
+            QuotaRecoveryEvidence::RefreshedSnapshot,
         ) {
             let _ = self
                 .repository
@@ -970,23 +949,20 @@ impl CodexCredentialQuotaService {
         &self,
         account_id: &ProviderAccountId,
     ) -> Result<CodexAccountQuotaSnapshot, CodexCredentialQuotaError> {
-        self.refresh_account_with_mode(account_id, QuotaRefreshMode::Reconcile)
-            .await
+        self.refresh_account_snapshot(account_id).await
     }
 
-    /// 真实限额失败后的异步刷新只补齐展示快照，不允许同一轮陈旧 usage 撤销失败状态。
+    /// 真实限额失败后的异步刷新只补齐展示快照，不允许 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
+        self.refresh_account_snapshot(account_id).await
     }
 
-    async fn refresh_account_with_mode(
+    async fn refresh_account_snapshot(
         &self,
         account_id: &ProviderAccountId,
-        mode: QuotaRefreshMode,
     ) -> Result<CodexAccountQuotaSnapshot, CodexCredentialQuotaError> {
         let account = self
             .store
@@ -994,7 +970,6 @@ 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);
@@ -1071,17 +1046,7 @@ 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,
-                    ),
-                },
-            },
+            QuotaRecoveryEvidence::RefreshedSnapshot,
         ) {
             self.repository
                 .apply_state(
@@ -1348,49 +1313,12 @@ fn quota_recovery_confirmed(
     }
     match evidence {
         QuotaRecoveryEvidence::SuccessfulResponse => true,
-        QuotaRecoveryEvidence::RefreshedSnapshot { confirms_recovery } => confirms_recovery,
-        QuotaRecoveryEvidence::PostFailureSnapshot => false,
+        // used_percent 是可滞后、会取整的观测值；确认 429 后的 100% 也是
+        // 管理端投影。它们的回落不能覆盖仍未到期的上游 reset 时间。
+        QuotaRecoveryEvidence::RefreshedSnapshot => 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

@@ -64,8 +64,7 @@ use crate::transport::protocol::responses::{
     CodexResponsesRequest, PreviousResponseScope, ResponseEventSignals,
 };
 use crate::transport::request::{
-    CodexRequestEncodeError, encode_generate_request, sanitize_cross_account_item,
-    scope_request_to_account,
+    CodexRequestEncodeError, encode_generate_request, scope_request_to_account,
 };
 use crate::transport::session::CodexSessionIdentity;
 use crate::transport::usage::normalize_service_tier;
@@ -177,17 +176,7 @@ impl Provider for CodexProvider {
             return ProviderRequestObservation::default();
         };
         let (semantics, reasoning_effort) = encode_generate_request(request, "observability")
-            .map(|mut encoded| {
-                if let Some(previous) = decode_openai_session_state(request) {
-                    let mut input = previous
-                        .transcript
-                        .iter()
-                        .map(OpenAiReplayItem::value)
-                        .cloned()
-                        .collect::<Vec<_>>();
-                    input.extend(encoded.input().iter().cloned());
-                    encoded.set_input(input);
-                }
+            .map(|encoded| {
                 let semantics = encoded.semantics();
                 let reasoning_effort = semantics.reasoning_effort.clone();
                 (semantics, reasoning_effort)
@@ -267,7 +256,6 @@ impl Provider for CodexProvider {
         let session_affinity_key = derive_codex_session_affinity_key(&upstream_request);
         let cyber_policy_session_key =
             derive_codex_cyber_policy_session_key(&upstream_request, context.client_api_key_ref());
-        let request_input = upstream_request.input().to_vec();
         let transport = selected_transport(&upstream_request);
         apply_transport(&mut upstream_request, transport);
 
@@ -299,11 +287,8 @@ impl Provider for CodexProvider {
         ) || previous_session.as_ref().is_some_and(|state| {
             state.continuation_scope == OpenAiContinuationScope::ReplayRequired
         });
-        if replay_previous_response && let Some(state) = previous_session.as_ref() {
-            let mut input = replay_input_for_account(state, lease.account_id().as_str());
-            input.reserve(request_input.len());
-            input.extend(request_input.iter().cloned());
-            upstream_request.set_input(input);
+        if replay_previous_response && previous_session.is_some() {
+            // Codex 客户端 input 是完整历史；恢复时只解除上游续链绑定，避免重复叠加代理副本。
             upstream_request.set_previous_response_id(None);
             upstream_request.previous_response_scope = None;
             upstream_request.turn_state = None;
@@ -369,8 +354,6 @@ impl Provider for CodexProvider {
         let response_store = upstream_request.store();
         let session_capture =
             (!continuation_requested || previous_session.is_some()).then(|| OpenAiSessionCapture {
-                previous: previous_session,
-                request_input,
                 account_id: lease.account_id().as_str().to_owned(),
                 conversation_id: upstream_request.local_conversation_id.clone(),
                 turn_state: upstream_request.turn_state.clone(),
@@ -425,7 +408,6 @@ struct OpenAiSessionState {
     #[serde(default)]
     turn_state: Option<String>,
     continuation_scope: OpenAiContinuationScope,
-    transcript: Vec<OpenAiReplayItem>,
 }
 
 #[derive(Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
@@ -436,27 +418,7 @@ enum OpenAiContinuationScope {
     ReplayRequired,
 }
 
-#[derive(Clone, Serialize, Deserialize)]
-#[serde(rename_all = "snake_case")]
-enum OpenAiReplayItem {
-    ClientInput(Value),
-    SanitizedOutput(Value),
-    AccountOutput { account_id: String, item: Value },
-}
-
-impl OpenAiReplayItem {
-    fn value(&self) -> &Value {
-        match self {
-            Self::ClientInput(value)
-            | Self::SanitizedOutput(value)
-            | Self::AccountOutput { item: value, .. } => value,
-        }
-    }
-}
-
 struct OpenAiSessionCapture {
-    previous: Option<OpenAiSessionState>,
-    request_input: Vec<Value>,
     account_id: String,
     conversation_id: Option<String>,
     turn_state: Option<String>,
@@ -489,33 +451,15 @@ fn attach_openai_session_update(
     events: &mut [ProviderEvent],
     capture: &mut Option<OpenAiSessionCapture>,
 ) {
-    let Some((terminal_index, output)) = events.iter().enumerate().find_map(|(index, event)| {
-        terminal_response_output(event).map(|output| (index, output.to_vec()))
-    }) else {
+    let Some(terminal_index) = events
+        .iter()
+        .position(|event| terminal_response_output(event).is_some())
+    else {
         return;
     };
     let Some(capture) = capture.take() else {
         return;
     };
-    let mut transcript = capture
-        .previous
-        .map(|state| state.transcript)
-        .unwrap_or_default();
-    project_transcript_to_account(&mut transcript, &capture.account_id);
-    transcript.extend(
-        capture
-            .request_input
-            .into_iter()
-            .map(OpenAiReplayItem::ClientInput),
-    );
-    transcript.extend(
-        output
-            .into_iter()
-            .map(|item| OpenAiReplayItem::AccountOutput {
-                account_id: capture.account_id.clone(),
-                item,
-            }),
-    );
     let Some(continuation_scope) = capture.continuation_scope else {
         return;
     };
@@ -524,7 +468,6 @@ fn attach_openai_session_update(
         conversation_id: capture.conversation_id,
         turn_state: capture.turn_state,
         continuation_scope,
-        transcript,
     }) else {
         return;
     };
@@ -552,47 +495,6 @@ fn terminal_response_output(event: &ProviderEvent) -> Option<&[Value]> {
     .flatten()
 }
 
-fn project_transcript_to_account(transcript: &mut Vec<OpenAiReplayItem>, account_id: &str) {
-    *transcript = transcript
-        .drain(..)
-        .filter_map(|item| match item {
-            OpenAiReplayItem::AccountOutput {
-                account_id: owner,
-                item,
-            } if owner != account_id => {
-                sanitize_cross_account_item(item).map(OpenAiReplayItem::SanitizedOutput)
-            }
-            item => Some(item),
-        })
-        .collect();
-}
-
-fn replay_input_for_account(state: &OpenAiSessionState, account_id: &str) -> Vec<Value> {
-    state
-        .transcript
-        .iter()
-        .filter_map(|item| match item {
-            OpenAiReplayItem::ClientInput(value) | OpenAiReplayItem::SanitizedOutput(value) => {
-                Some(value.clone())
-            }
-            OpenAiReplayItem::AccountOutput {
-                account_id: owner,
-                item,
-            } if owner == account_id => Some(without_output_id(item.clone())),
-            OpenAiReplayItem::AccountOutput { item, .. } => {
-                sanitize_cross_account_item(item.clone())
-            }
-        })
-        .collect()
-}
-
-fn without_output_id(mut item: Value) -> Value {
-    if let Value::Object(object) = &mut item {
-        object.remove("id");
-    }
-    item
-}
-
 enum CodexHandshakeAttemptError {
     Client(CodexClientError),
     Cancelled,
@@ -1767,7 +1669,10 @@ fn map_request_error(error: CodexRequestEncodeError) -> ProviderError {
 fn map_selection_error(error: CredentialSelectionError) -> ProviderError {
     match error {
         CredentialSelectionError::CapacityUnavailable { retry_after } => {
-            let error = provider_error(ProviderErrorKind::Unavailable, UpstreamSendState::NotSent);
+            let error = provider_error(
+                ProviderErrorKind::AccountCapacityUnavailable,
+                UpstreamSendState::NotSent,
+            );
             match retry_after {
                 Some(retry) => error.with_retry_after(retry),
                 None => error,
@@ -1780,9 +1685,10 @@ fn map_selection_error(error: CredentialSelectionError) -> ProviderError {
         CredentialSelectionError::InvalidCredential
         | CredentialSelectionError::Store
         | CredentialSelectionError::Coordinator
-        | CredentialSelectionError::CookiePolicy => {
-            provider_error(ProviderErrorKind::Unavailable, UpstreamSendState::NotSent)
-        }
+        | CredentialSelectionError::CookiePolicy => provider_error(
+            ProviderErrorKind::ProviderInfrastructureUnavailable,
+            UpstreamSendState::NotSent,
+        ),
     }
 }
 
@@ -2063,7 +1969,7 @@ fn stream_transport_allows_pre_delivery_retry(error: &CodexClientError) -> bool
             CodexWebSocketExchangeError::Transport(_)
                 | CodexWebSocketExchangeError::PostSendAmbiguous { .. }
                 | CodexWebSocketExchangeError::SendTimeout { .. }
-                | CodexWebSocketExchangeError::ClosedBeforeTerminal
+                | CodexWebSocketExchangeError::ClosedBeforeTerminal(_)
                 | CodexWebSocketExchangeError::ReceiveIdleTimeout { .. }
                 | CodexWebSocketExchangeError::ReusedConnectionDiedBeforeFirstEvent { .. }
                 | CodexWebSocketExchangeError::InitialEventTimeout { .. }
@@ -2151,10 +2057,19 @@ fn map_client_error(
                 send_state,
             ))
         }
-        CodexClientError::WebSocket(error) => MappedProviderFailure::plain(provider_error(
-            websocket_error_kind(&error),
-            websocket_send_state(&error),
-        )),
+        CodexClientError::WebSocket(error) => {
+            let client_visible_error = websocket_client_visible_error(&error);
+            let mut failure = MappedProviderFailure::plain(provider_error(
+                websocket_error_kind(&error),
+                websocket_send_state(&error),
+            ));
+            if let Some(client_visible_error) = client_visible_error {
+                failure.error = failure
+                    .error
+                    .with_client_visible_upstream_error(client_visible_error);
+            }
+            failure
+        }
     };
     if let Some(continuation_failure) = continuation_failure {
         failure.error = failure
@@ -2165,6 +2080,21 @@ fn map_client_error(
     failure
 }
 
+fn websocket_client_visible_error(
+    error: &CodexWebSocketExchangeError,
+) -> Option<ClientVisibleUpstreamError> {
+    let close = error.close_before_terminal()?;
+    let message = close
+        .reason()
+        .filter(|reason| !reason.is_empty())
+        .map_or_else(|| close.to_string(), str::to_owned);
+    Some(ClientVisibleUpstreamError::new(
+        message,
+        close.code().map(|code| code.to_string()),
+        Some("websocket_close_error".to_owned()),
+    ))
+}
+
 fn map_upstream_failure(
     mut failure: CodexUpstreamFailure,
     observation: Option<ProviderResponseObservation>,
@@ -2333,7 +2263,7 @@ const fn websocket_send_state(error: &CodexWebSocketExchangeError) -> UpstreamSe
         CodexWebSocketExchangeError::Transport(_)
         | CodexWebSocketExchangeError::PostSendAmbiguous { .. }
         | CodexWebSocketExchangeError::SendTimeout { .. }
-        | CodexWebSocketExchangeError::ClosedBeforeTerminal
+        | CodexWebSocketExchangeError::ClosedBeforeTerminal(_)
         | CodexWebSocketExchangeError::ReceiveIdleTimeout { .. }
         | CodexWebSocketExchangeError::ReusedConnectionDiedBeforeFirstEvent { .. }
         | CodexWebSocketExchangeError::InitialEventTimeout { .. } => UpstreamSendState::Ambiguous,
@@ -2360,7 +2290,7 @@ const fn websocket_error_kind(error: &CodexWebSocketExchangeError) -> ProviderEr
         CodexWebSocketExchangeError::Transport(_)
         | CodexWebSocketExchangeError::Connect(_)
         | CodexWebSocketExchangeError::PostSendAmbiguous { .. }
-        | CodexWebSocketExchangeError::ClosedBeforeTerminal
+        | CodexWebSocketExchangeError::ClosedBeforeTerminal(_)
         | CodexWebSocketExchangeError::ReusedConnectionDiedBeforeFirstEvent { .. } => {
             ProviderErrorKind::Transport
         }

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

@@ -477,6 +477,7 @@ pub enum CodexTransportDecision {
     Http2BreakerOpen,
     Http2PoolUnavailable,
     Http2PreSendFailure,
+    Http2PreDeliveryFailure,
 }
 
 impl CodexTransportDecision {
@@ -491,6 +492,7 @@ impl CodexTransportDecision {
             Self::Http2BreakerOpen => "http2_breaker_open",
             Self::Http2PoolUnavailable => "http2_pool_unavailable",
             Self::Http2PreSendFailure => "http2_ws_pre_send_failure",
+            Self::Http2PreDeliveryFailure => "http2_ws_pre_delivery_failure",
         }
     }
 }

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

@@ -32,10 +32,11 @@ use crate::transport::{
     },
     response_meta,
     websocket::{
-        CodexWebSocketConnection, CodexWebSocketPool, CodexWebSocketPoolKey,
-        DEFAULT_INITIAL_EVENT_TIMEOUT, WEBSOCKET_FAST_PATH_BUDGET, WebSocketOriginBreaker,
-        WebSocketPoolDecision, execute_prepared_response_create_request_stream,
-        post_send_ambiguous, prepare_response_create_request_with_pool, websocket_audit_dir,
+        CodexWebSocketConnection, CodexWebSocketExchangeError, CodexWebSocketPool,
+        CodexWebSocketPoolKey, CodexWebSocketStreamingExchange, DEFAULT_INITIAL_EVENT_TIMEOUT,
+        WEBSOCKET_FAST_PATH_BUDGET, WebSocketOriginBreaker, WebSocketPoolDecision,
+        execute_prepared_response_create_request_stream, post_send_ambiguous,
+        prepare_response_create_request_with_pool, websocket_audit_dir,
         write_websocket_audit_artifact_from_env,
     },
 };
@@ -308,7 +309,7 @@ impl CodexBackendClient {
             route,
             metrics,
         } = prepared;
-        let result = match route {
+        match route {
             PreparedResponseRoute::Http => self
                 .create_response_stream_http_sse(request, context)
                 .await
@@ -317,50 +318,124 @@ impl CodexBackendClient {
                     response
                 }),
             PreparedResponseRoute::WebSocket(route) => {
-                let PreparedWebSocketRoute { request, prepared } = *route;
-                execute_prepared_response_create_request_stream(&request, prepared)
-                    .await
-                    .map_err(websocket_exchange_error_to_client_error)
-                    .map(|exchange| {
-                        tracing::info!(
-                            request_id = %context.request_id,
-                            websocket_connection_id = %exchange.websocket_connection_id,
-                            ws_pool = exchange.pool_decision.map_or("unpooled", WebSocketPoolDecision::kind),
-                            "WebSocket response stream established"
-                        );
-                        CodexBackendStreamingResponse {
-                            body: Box::pin(
-                                exchange
-                                    .body
-                                    .map_err(post_send_ambiguous)
-                                    .map_err(websocket_exchange_error_to_client_error),
-                            ),
-                            transport: CodexBackendTransport::WebSocket,
-                            turn_state: exchange.turn_state,
-                            set_cookie_headers: exchange.set_cookie_headers,
-                            rate_limit_headers: exchange.rate_limit_headers,
-                            rate_limit_header_updates: Some(exchange.rate_limit_header_updates),
-                            turn_state_update: Some(exchange.turn_state_update),
-                            websocket_pool_decision: exchange.pool_decision,
-                            diagnostics: exchange.diagnostics,
-                            response_metadata: exchange.response_metadata,
-                            transport_metrics: metrics,
-                            connection_local_continuation: exchange.connection_local_continuation,
-                        }
-                    })
+                let PreparedWebSocketRoute {
+                    request: websocket_request,
+                    prepared,
+                } = *route;
+                let delivery_wait_started_at = Instant::now();
+                let mut exchange = match execute_prepared_response_create_request_stream(
+                    &websocket_request,
+                    prepared,
+                )
+                .await
+                {
+                    Ok(exchange) => exchange,
+                    Err(error)
+                        if requirement.allows_pre_delivery_http_fallback()
+                            && error.allows_pre_delivery_http_fallback() =>
+                    {
+                        return self
+                            .fallback_to_http_before_websocket_delivery(
+                                request,
+                                context,
+                                requirement,
+                                metrics,
+                                delivery_wait_started_at,
+                                error,
+                            )
+                            .await;
+                    }
+                    Err(error) => return Err(websocket_exchange_error_to_client_error(error)),
+                };
+                if requirement.allows_pre_delivery_http_fallback()
+                    && let Err(error) = await_websocket_delivery_boundary(&mut exchange).await
+                {
+                    if error.allows_pre_delivery_http_fallback() {
+                        return self
+                            .fallback_to_http_before_websocket_delivery(
+                                request,
+                                context,
+                                requirement,
+                                metrics,
+                                delivery_wait_started_at,
+                                error,
+                            )
+                            .await;
+                    }
+                    return Err(websocket_exchange_error_to_client_error(
+                        post_send_ambiguous(error),
+                    ));
+                }
+                tracing::info!(
+                    request_id = %context.request_id,
+                    websocket_connection_id = %exchange.websocket_connection_id,
+                    ws_pool = exchange.pool_decision.map_or("unpooled", WebSocketPoolDecision::kind),
+                    "WebSocket response stream established"
+                );
+                Ok(CodexBackendStreamingResponse {
+                    body: Box::pin(
+                        exchange
+                            .body
+                            .map_err(post_send_ambiguous)
+                            .map_err(websocket_exchange_error_to_client_error),
+                    ),
+                    transport: CodexBackendTransport::WebSocket,
+                    turn_state: exchange.turn_state,
+                    set_cookie_headers: exchange.set_cookie_headers,
+                    rate_limit_headers: exchange.rate_limit_headers,
+                    rate_limit_header_updates: Some(exchange.rate_limit_header_updates),
+                    turn_state_update: Some(exchange.turn_state_update),
+                    websocket_pool_decision: exchange.pool_decision,
+                    diagnostics: exchange.diagnostics,
+                    response_metadata: exchange.response_metadata,
+                    transport_metrics: metrics,
+                    connection_local_continuation: exchange.connection_local_continuation,
+                })
             }
-        };
-        result.inspect_err(|error| {
+        }
+        .inspect_err(|error| {
             tracing::warn!(
                 request_id = %context.request_id,
                 transport_requirement = requirement.as_str(),
                 failure_phase = "post_send_or_explicit_response",
                 error = %error,
-                "Responses stream transport failed after preparation; automatic fallback is disabled"
+                "Responses stream transport failed after preparation"
             );
         })
     }
 
+    async fn fallback_to_http_before_websocket_delivery(
+        &self,
+        request: &CodexResponsesRequest,
+        context: CodexRequestContext<'_>,
+        requirement: TransportRequirement,
+        mut metrics: CodexTransportMetrics,
+        delivery_wait_started_at: Instant,
+        error: CodexWebSocketExchangeError,
+    ) -> CodexClientResult<CodexBackendStreamingResponse> {
+        let decision_wait_ms = metrics
+            .transport_decision_wait_ms
+            .unwrap_or_default()
+            .saturating_add(elapsed_duration_millis(delivery_wait_started_at.elapsed()));
+        metrics.decision = Some(CodexTransportDecision::Http2PreDeliveryFailure);
+        metrics.transport_decision_wait_ms = Some(decision_wait_ms);
+        tracing::warn!(
+            request_id = %context.request_id,
+            account_id = context.account_id.unwrap_or_default(),
+            transport_requirement = requirement.as_str(),
+            transport_decision = CodexTransportDecision::Http2PreDeliveryFailure.as_str(),
+            transport_decision_wait_ms = decision_wait_ms,
+            error = %error,
+            "WebSocket failed before first deliverable event; using same-account HTTP"
+        );
+        self.create_response_stream_http_sse(request, context)
+            .await
+            .map(|mut response| {
+                merge_preparation_metrics(&mut response.transport_metrics, metrics);
+                response
+            })
+    }
+
     fn websocket_pool_key(
         &self,
         request: &CodexResponsesRequest,
@@ -563,6 +638,36 @@ impl CodexBackendClient {
     }
 }
 
+async fn await_websocket_delivery_boundary(
+    exchange: &mut CodexWebSocketStreamingExchange,
+) -> Result<(), CodexWebSocketExchangeError> {
+    let mut prelude = Vec::new();
+    loop {
+        match exchange.body.next().await {
+            Some(Ok(frame)) if is_websocket_lifecycle_prelude(&frame) => prelude.push(frame),
+            Some(Ok(frame)) => {
+                prelude.push(frame);
+                let remaining =
+                    std::mem::replace(&mut exchange.body, Box::pin(futures::stream::empty()));
+                exchange.body =
+                    Box::pin(futures::stream::iter(prelude.into_iter().map(Ok)).chain(remaining));
+                return Ok(());
+            }
+            Some(Err(error)) => return Err(error),
+            None => {
+                return Err(CodexWebSocketExchangeError::closed_before_terminal(
+                    None, None,
+                ));
+            }
+        }
+    }
+}
+
+fn is_websocket_lifecycle_prelude(frame: &[u8]) -> bool {
+    frame.starts_with(b"event: response.created\n")
+        || frame.starts_with(b"event: response.in_progress\n")
+}
+
 async fn read_model_catalog_body(response: ReqwestResponse) -> CodexClientResult<Vec<u8>> {
     if response
         .content_length()

backend/crates/providers/openai/src/transport/protocol/responses.rs

@@ -139,6 +139,11 @@ impl TransportRequirement {
         )
     }
 
+    /// WebSocket 已发送 payload、但尚未交付业务事件时，是否允许同账号 HTTP 重放。
+    pub fn allows_pre_delivery_http_fallback(self) -> bool {
+        matches!(self, Self::NewChain)
+    }
+
     /// 用于审计与遥测的稳定名称。
     pub fn as_str(self) -> &'static str {
         match self {

backend/crates/providers/openai/src/transport/websocket/error.rs

@@ -71,8 +71,8 @@ pub enum CodexWebSocketExchangeError {
         reason: PreviousResponseUnavailableReason,
     },
     /// 上游在 terminal 事件前关闭。
-    #[error("websocket closed before terminal event")]
-    ClosedBeforeTerminal,
+    #[error("{0}")]
+    ClosedBeforeTerminal(CodexWebSocketCloseError),
     /// 上游在指定时间内没有发送任何事件。
     #[error("websocket receive idle timeout after {timeout:?}")]
     ReceiveIdleTimeout {
@@ -87,6 +87,9 @@ pub enum CodexWebSocketExchangeError {
     ReusedConnectionDiedBeforeFirstEvent {
         /// 底层失效原因。
         message: String,
+        /// 原始 typed transport failure。
+        #[source]
+        source: Option<Box<CodexWebSocketExchangeError>>,
     },
     /// 建连并发送后，上游在配置时间内没有产生任何事件。
     #[error("websocket first upstream event not received within {timeout:?}")]
@@ -96,6 +99,55 @@ pub enum CodexWebSocketExchangeError {
     },
 }
 
+/// 上游在 terminal 事件前发送的 WebSocket close 信息。
+///
+/// close reason 只供原请求的客户端错误响应使用；`Debug` 与 `Display` 均不会输出它。
+#[derive(Clone, PartialEq, Eq)]
+pub struct CodexWebSocketCloseError {
+    code: Option<u16>,
+    reason: Option<String>,
+}
+
+impl CodexWebSocketCloseError {
+    pub(crate) fn new(code: Option<u16>, reason: Option<String>) -> Self {
+        Self { code, reason }
+    }
+
+    /// 返回上游 close code；没有 close frame 时为 `None`。
+    #[must_use]
+    pub const fn code(&self) -> Option<u16> {
+        self.code
+    }
+
+    /// 返回上游 close reason；仅可用于当前请求的客户端协议响应。
+    #[must_use]
+    pub fn reason(&self) -> Option<&str> {
+        self.reason.as_deref()
+    }
+}
+
+impl fmt::Debug for CodexWebSocketCloseError {
+    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+        formatter
+            .debug_struct("CodexWebSocketCloseError")
+            .field("code", &self.code)
+            .field("reason", &self.reason.as_ref().map(|_| "<redacted>"))
+            .finish()
+    }
+}
+
+impl fmt::Display for CodexWebSocketCloseError {
+    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+        match self.code {
+            Some(code) => write!(
+                formatter,
+                "websocket closed before terminal event (code {code})"
+            ),
+            None => formatter.write_str("websocket closed before terminal event"),
+        }
+    }
+}
+
 /// WebSocket 上游错误帧载荷。
 #[derive(Clone, PartialEq, Eq)]
 pub struct CodexWebSocketUpstreamError {
@@ -140,6 +192,25 @@ impl fmt::Display for CodexWebSocketUpstreamError {
 }
 
 impl CodexWebSocketExchangeError {
+    pub(crate) fn closed_before_terminal(code: Option<u16>, reason: Option<String>) -> Self {
+        Self::ClosedBeforeTerminal(CodexWebSocketCloseError::new(code, reason))
+    }
+
+    pub(crate) fn close_before_terminal(&self) -> Option<&CodexWebSocketCloseError> {
+        match self {
+            Self::ClosedBeforeTerminal(close) => Some(close),
+            Self::PostSendAmbiguous {
+                source: Some(source),
+                ..
+            }
+            | Self::ReusedConnectionDiedBeforeFirstEvent {
+                source: Some(source),
+                ..
+            } => source.close_before_terminal(),
+            _ => None,
+        }
+    }
+
     pub(super) fn upstream(
         status_code: u16,
         retry_after_seconds: Option<u64>,
@@ -173,4 +244,16 @@ impl CodexWebSocketExchangeError {
                 | Self::ContinuationUnavailable { .. }
         )
     }
+
+    /// 首个业务事件交付前，普通 WebSocket 失败可切到同账号 HTTP。
+    pub(in crate::transport) fn allows_pre_delivery_http_fallback(&self) -> bool {
+        match self {
+            Self::Upstream(_) => false,
+            Self::PostSendAmbiguous {
+                source: Some(source),
+                ..
+            } => source.allows_pre_delivery_http_fallback(),
+            _ => true,
+        }
+    }
 }

backend/crates/providers/openai/src/transport/websocket/exchange/io.rs

@@ -40,19 +40,20 @@ pub(super) fn reused_stream_receive_error(
     error: CodexWebSocketExchangeError,
 ) -> CodexWebSocketExchangeError {
     match error {
-        CodexWebSocketExchangeError::ClosedBeforeTerminal
+        CodexWebSocketExchangeError::ClosedBeforeTerminal(_)
         | CodexWebSocketExchangeError::ReceiveIdleTimeout { .. }
         | CodexWebSocketExchangeError::Transport(_) => {
-            reused_connection_died_before_first_event(&error)
+            reused_connection_died_before_first_event(error)
         }
         error => error,
     }
 }
 
 pub(super) fn reused_connection_died_before_first_event(
-    error: &CodexWebSocketExchangeError,
+    error: CodexWebSocketExchangeError,
 ) -> CodexWebSocketExchangeError {
     CodexWebSocketExchangeError::ReusedConnectionDiedBeforeFirstEvent {
         message: error.to_string(),
+        source: Some(Box::new(error)),
     }
 }

backend/crates/providers/openai/src/transport/websocket/exchange/stream.rs

@@ -208,14 +208,17 @@ async fn forward_websocket_response_stream(state: WebSocketStreamForwardState) {
                     .await;
                 return;
             }
-            tungstenite::Message::Close(_) => {
+            tungstenite::Message::Close(frame) => {
                 discard_stream_websocket(
                     websocket,
                     pool_return,
                     StreamWebSocketDiscardReason::UpstreamClosed,
                 )
                 .await;
-                let error = CodexWebSocketExchangeError::ClosedBeforeTerminal;
+                let error = CodexWebSocketExchangeError::closed_before_terminal(
+                    frame.as_ref().map(|frame| u16::from(frame.code)),
+                    frame.map(|frame| frame.reason.to_string()),
+                );
                 let error = if reused_connection {
                     reused_stream_receive_error(error)
                 } else {
@@ -285,13 +288,19 @@ async fn forward_websocket_response_stream(state: WebSocketStreamForwardState) {
         }
     }
 
+    let upstream_close = websocket
+        .exit_reason()
+        .and_then(|reason| reason.upstream_close().cloned());
     discard_stream_websocket(
         websocket,
         pool_return,
         StreamWebSocketDiscardReason::UpstreamClosed,
     )
     .await;
-    let error = CodexWebSocketExchangeError::ClosedBeforeTerminal;
+    let error = upstream_close.map_or_else(
+        || CodexWebSocketExchangeError::closed_before_terminal(None, None),
+        CodexWebSocketExchangeError::ClosedBeforeTerminal,
+    );
     let error = if reused_connection {
         reused_stream_receive_error(error)
     } else {

backend/crates/providers/openai/src/transport/websocket/mod.rs

@@ -19,7 +19,7 @@ pub use self::{
         WebSocketOriginBreaker, WebSocketOriginBreakerConfig, WebSocketOriginBreakerDecision,
         WebSocketOriginBreakerPermit,
     },
-    error::{CodexWebSocketExchangeError, CodexWebSocketUpstreamError},
+    error::{CodexWebSocketCloseError, CodexWebSocketExchangeError, CodexWebSocketUpstreamError},
     exchange::{
         CodexWebSocketRateLimitHeaderUpdates, CodexWebSocketSseStream,
         CodexWebSocketStreamingExchange, CodexWebSocketTurnStateUpdate,

backend/crates/providers/openai/src/transport/websocket/pump.rs

@@ -29,6 +29,8 @@ use tokio::{
 use tokio_tungstenite::{MaybeTlsStream, WebSocketStream, tungstenite::Message};
 use uuid::Uuid;
 
+use super::CodexWebSocketCloseError;
+
 /// 底层 tungstenite WebSocket 流。
 pub(crate) type RawWsStream = WebSocketStream<MaybeTlsStream<TcpStream>>;
 
@@ -69,13 +71,25 @@ const PUMP_MESSAGE_BUFFER: usize = 64;
 pub(crate) enum PumpExitReason {
     CommandChannelClosed,
     LocalClose,
-    OutboundTransportError { message: String },
-    UpstreamCloseFrame { frame: Option<String> },
+    OutboundTransportError {
+        message: String,
+    },
+    UpstreamCloseFrame {
+        close: Option<CodexWebSocketCloseError>,
+    },
     UpstreamEof,
-    InboundTransportError { message: String },
-    PongTimeout { timeout: Duration },
-    LivenessTimeout { timeout: Duration },
-    KeepaliveTransportError { message: String },
+    InboundTransportError {
+        message: String,
+    },
+    PongTimeout {
+        timeout: Duration,
+    },
+    LivenessTimeout {
+        timeout: Duration,
+    },
+    KeepaliveTransportError {
+        message: String,
+    },
     MessageReceiverClosed,
 }
 
@@ -100,7 +114,7 @@ impl PumpExitReason {
             Self::OutboundTransportError { message }
             | Self::InboundTransportError { message }
             | Self::KeepaliveTransportError { message } => Some(message.clone()),
-            Self::UpstreamCloseFrame { frame } => frame.clone(),
+            Self::UpstreamCloseFrame { close } => close.as_ref().map(ToString::to_string),
             Self::PongTimeout { timeout } | Self::LivenessTimeout { timeout } => {
                 Some(format!("{timeout:?}"))
             }
@@ -111,6 +125,13 @@ impl PumpExitReason {
         }
     }
 
+    pub(crate) fn upstream_close(&self) -> Option<&CodexWebSocketCloseError> {
+        match self {
+            Self::UpstreamCloseFrame { close: Some(close) } => Some(close),
+            _ => None,
+        }
+    }
+
     fn is_unexpected(&self) -> bool {
         matches!(
             self,
@@ -336,7 +357,12 @@ async fn pump_loop(
                             message => {
                                 let terminal_reason = match &message {
                                     Message::Close(frame) => Some(PumpExitReason::UpstreamCloseFrame {
-                                        frame: frame.as_ref().map(|frame| format!("{frame:?}")),
+                                        close: frame.as_ref().map(|frame| {
+                                            CodexWebSocketCloseError::new(
+                                                Some(u16::from(frame.code)),
+                                                Some(frame.reason.to_string()),
+                                            )
+                                        }),
                                     }),
                                     _ => None,
                                 };

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

@@ -443,7 +443,7 @@ async fn passive_full_percent_with_explicit_allowance_keeps_account_ready() {
 }
 
 #[tokio::test]
-async fn manual_quota_refresh_updates_account_state_from_provider_response() {
+async fn manual_quota_refresh_keeps_confirmed_exhaustion_until_the_recorded_reset() {
     let store = Arc::new(MemoryAccountStore::default());
     let account_id = "acct_manual_quota_state";
     create_account(&store, account_id).await;
@@ -541,24 +541,25 @@ async fn manual_quota_refresh_updates_account_state_from_provider_response() {
             "rate_limit": {
                 "allowed": true,
                 "limit_reached": false,
-                "primary_window": {"used_percent": 20, "reset_at": 1_900_000_000}
+                "primary_window": {"used_percent": 99, "reset_at": 1_900_000_000}
             }
         })))
         .mount(&server)
         .await;
 
-    let recovered = service
+    let unchanged_reset = service
         .refresh_account(account.id())
         .await
-        .expect("refresh recovered quota");
+        .expect("refresh lower usage quota");
 
-    assert!(!recovered.fact().exhausted());
+    assert!(!unchanged_reset.fact().exhausted());
     assert_eq!(
         store
             .account(account_id)
-            .expect("recovered account")
+            .expect("account after lower usage")
             .availability(),
-        AccountAvailability::Ready
+        AccountAvailability::QuotaExhausted,
+        "a lower used_percent before the recorded reset is not quota recovery"
     );
 }
 
@@ -1051,7 +1052,7 @@ async fn periodic_quota_synchronization_rechecks_exhausted_accounts_without_wait
 }
 
 #[tokio::test]
-async fn periodic_quota_synchronization_requires_a_real_snapshot_improvement_to_recover() {
+async fn periodic_quota_synchronization_does_not_recover_from_percent_drop_before_reset() {
     let store = Arc::new(MemoryAccountStore::default());
     let account_id = "acct_periodic_recovery_evidence";
     create_account(&store, account_id).await;
@@ -1066,7 +1067,7 @@ async fn periodic_quota_synchronization_requires_a_real_snapshot_improvement_to_
                     "rate_limit": {
                         "allowed": true,
                         "limit_reached": false,
-                        "primary_window": {"used_percent": 86, "reset_at": reset_at}
+                        "primary_window": {"used_percent": 100, "reset_at": reset_at}
                     }
                 })
                 .as_object()
@@ -1095,7 +1096,7 @@ async fn periodic_quota_synchronization_requires_a_real_snapshot_improvement_to_
             "rate_limit": {
                 "allowed": true,
                 "limit_reached": false,
-                "primary_window": {"used_percent": 86, "reset_at": reset_at}
+                "primary_window": {"used_percent": 100, "reset_at": reset_at}
             }
         })))
         .expect(1)
@@ -1131,7 +1132,7 @@ async fn periodic_quota_synchronization_requires_a_real_snapshot_improvement_to_
             "rate_limit": {
                 "allowed": true,
                 "limit_reached": false,
-                "primary_window": {"used_percent": 20, "reset_at": reset_at}
+                "primary_window": {"used_percent": 99, "reset_at": reset_at}
             }
         })))
         .expect(1)
@@ -1147,21 +1148,73 @@ async fn periodic_quota_synchronization_requires_a_real_snapshot_improvement_to_
         server.uri(),
     );
 
-    let improved = next_cycle
+    let lower_percent = next_cycle
         .synchronize()
         .await
-        .expect("improved quota recheck");
+        .expect("lower percent quota recheck");
 
-    assert_eq!(improved.updated, 1);
+    assert_eq!(lower_percent.updated, 1);
     assert_eq!(
         store
             .account(account_id)
-            .expect("account after improved snapshot")
+            .expect("account after lower percent snapshot")
             .availability(),
-        AccountAvailability::Ready
+        AccountAvailability::QuotaExhausted,
+        "a provider percentage change is not sufficient to release a confirmed exhaustion"
     );
 }
 
+#[tokio::test]
+async fn periodic_quota_synchronization_recovers_after_recorded_reset() {
+    let store = Arc::new(MemoryAccountStore::default());
+    let account_id = "acct_periodic_reset_elapsed";
+    create_account(&store, account_id).await;
+    let account = store.account(account_id).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(60)),
+            observed_at: SystemTime::now(),
+        })
+        .await
+        .expect("mark account exhausted with an elapsed reset");
+    let server = MockServer::start().await;
+    let next_reset = Utc::now().timestamp() + 24 * 60 * 60;
+    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": 1, "reset_at": next_reset}
+            }
+        })))
+        .expect(1)
+        .mount(&server)
+        .await;
+    let service = quota_service_with_base_url(
+        &store,
+        reqwest::Client::builder()
+            .no_proxy()
+            .build()
+            .expect("client"),
+        server.uri(),
+    );
+
+    // 定时复核可提前执行，但只有已记录的 reset 到期后才释放额度耗尽状态。
+    let summary = service.synchronize().await.expect("quota synchronization");
+
+    assert_eq!(summary.updated, 1);
+    let current = store
+        .account(account_id)
+        .expect("account after elapsed reset");
+    assert_eq!(current.availability(), AccountAvailability::Ready);
+    assert_eq!(current.cooldown_until(), None);
+}
+
 #[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

@@ -5,7 +5,7 @@ use std::time::{Duration, SystemTime};
 
 use base64::{Engine as _, engine::general_purpose::STANDARD};
 use chrono::Utc;
-use futures::StreamExt;
+use futures::{SinkExt, StreamExt};
 use gateway_core::engine::credential::{
     AccountAvailability, AccountFeedbackStats, OpaqueProviderData, ProviderAccountId,
     ProviderAccountStore as _, QuotaObservation,
@@ -37,6 +37,8 @@ use tokio::{
     sync::oneshot,
     time::timeout,
 };
+use tokio_tungstenite::tungstenite::Message;
+use tokio_tungstenite::tungstenite::protocol::{CloseFrame, frame::coding::CloseCode};
 use wiremock::matchers::{header, method, path};
 use wiremock::{Mock, MockServer, ResponseTemplate};
 
@@ -44,6 +46,7 @@ use crate::support::{
     MemoryAccountStore, MemorySessionAffinity, MemorySessionExclusions, TestLeaseCoordinator,
     account_policy, agent_identity_service_with_pool, catalog_cache, profile, secret,
 };
+use crate::transport::accept_codex_test_websocket;
 
 const OFFICIAL_FIXTURE: &[u8] =
     include_bytes!("../transport/fixtures/official_models_snapshot.json");
@@ -81,10 +84,36 @@ fn provider_with_base_url(store: &Arc<MemoryAccountStore>, base_url: String) ->
     provider_with_affinity_and_base_url(store, Arc::new(MemorySessionAffinity::default()), base_url)
 }
 
+fn provider_with_leases(
+    store: &Arc<MemoryAccountStore>,
+    leases: Arc<TestLeaseCoordinator>,
+) -> CodexProvider {
+    provider_with_affinity_and_base_url_and_leases(
+        store,
+        Arc::new(MemorySessionAffinity::default()),
+        OFFICIAL_CODEX_BASE_URL.to_owned(),
+        leases,
+    )
+}
+
 fn provider_with_affinity_and_base_url(
     store: &Arc<MemoryAccountStore>,
     session_affinity: Arc<MemorySessionAffinity>,
     base_url: String,
+) -> CodexProvider {
+    provider_with_affinity_and_base_url_and_leases(
+        store,
+        session_affinity,
+        base_url,
+        Arc::new(TestLeaseCoordinator::default()),
+    )
+}
+
+fn provider_with_affinity_and_base_url_and_leases(
+    store: &Arc<MemoryAccountStore>,
+    session_affinity: Arc<MemorySessionAffinity>,
+    base_url: String,
+    leases: Arc<TestLeaseCoordinator>,
 ) -> CodexProvider {
     let profile = wire_profile();
     let http = reqwest::Client::builder()
@@ -112,7 +141,7 @@ fn provider_with_affinity_and_base_url(
     let selector = Arc::new(CodexCredentialSelector::new(
         ProviderKind::new("openai").expect("provider"),
         store.repository(),
-        Arc::new(TestLeaseCoordinator::default()),
+        leases,
         session_affinity,
         Arc::new(MemorySessionExclusions::default()),
         Arc::clone(&catalog),
@@ -517,6 +546,130 @@ async fn generate_without_an_eligible_openai_account_fails_before_network_io() {
     assert_eq!(error.send_state(), UpstreamSendState::NotSent);
 }
 
+#[tokio::test]
+async fn capacity_selection_error_preserves_classification_and_retry_after() {
+    let store = Arc::new(MemoryAccountStore::default());
+    create_account(&store, "acct_capacity_busy").await;
+    let leases = Arc::new(TestLeaseCoordinator::default());
+    *leases.busy.lock().expect("lease busy lock") = true;
+
+    let error = match provider_with_leases(&store, leases)
+        .execute(
+            planned_request("openai", generate_operation()),
+            context("req_capacity_busy", CancellationToken::new()),
+        )
+        .await
+    {
+        Ok(_) => panic!("busy account selection must fail"),
+        Err(error) => error,
+    };
+
+    assert_eq!(
+        (error.kind(), error.send_state(), error.retry_after()),
+        (
+            ProviderErrorKind::AccountCapacityUnavailable,
+            UpstreamSendState::NotSent,
+            Some(Duration::from_millis(25)),
+        )
+    );
+}
+
+#[tokio::test]
+async fn selection_infrastructure_errors_have_a_distinct_classification() {
+    let store = Arc::new(MemoryAccountStore::default());
+    store.fail_provider_listing();
+
+    let error = match provider(&store)
+        .execute(
+            planned_request("openai", generate_operation()),
+            context("req_selection_store_failure", CancellationToken::new()),
+        )
+        .await
+    {
+        Ok(_) => panic!("account store failure must fail selection"),
+        Err(error) => error,
+    };
+
+    assert_eq!(
+        (error.kind(), error.send_state()),
+        (
+            ProviderErrorKind::ProviderInfrastructureUnavailable,
+            UpstreamSendState::NotSent,
+        )
+    );
+}
+
+#[tokio::test]
+async fn websocket_close_details_are_only_exposed_through_the_client_error() {
+    let store = Arc::new(MemoryAccountStore::default());
+    create_account(&store, "acct_websocket_close").await;
+    let listener = TcpListener::bind("127.0.0.1:0")
+        .await
+        .expect("bind WebSocket listener");
+    let base_url = format!(
+        "http://{}",
+        listener.local_addr().expect("listener address")
+    );
+    let server = tokio::spawn(async move {
+        let (stream, _) = listener
+            .accept()
+            .await
+            .expect("accept WebSocket connection");
+        let mut websocket = accept_codex_test_websocket(stream).await;
+        let _request = websocket
+            .next()
+            .await
+            .expect("WebSocket request")
+            .expect("valid WebSocket request");
+        websocket
+            .send(Message::Text(
+                json!({
+                    "type": "response.output_text.delta",
+                    "delta": "partial output"
+                })
+                .to_string()
+                .into(),
+            ))
+            .await
+            .expect("send semantic WebSocket event");
+        websocket
+            .close(Some(CloseFrame {
+                code: CloseCode::Size,
+                reason: "message too big".into(),
+            }))
+            .await
+            .expect("close WebSocket");
+    });
+
+    let mut stream = provider_with_base_url(&store, base_url)
+        .execute(
+            planned_request("openai", generate_operation()),
+            context("req_websocket_close", CancellationToken::new()),
+        )
+        .await
+        .expect("prepare WebSocket provider stream");
+    let error = loop {
+        match stream.next().await {
+            Some(Ok(_)) => {}
+            Some(Err(error)) => break error,
+            None => panic!("WebSocket close must surface a provider error"),
+        }
+    };
+    server.await.expect("WebSocket server");
+
+    assert!(!format!("{error:?}").contains("message too big"));
+    assert!(!error.to_string().contains("message too big"));
+    assert_eq!(error.kind(), ProviderErrorKind::Transport);
+    assert_eq!(error.send_state(), UpstreamSendState::Ambiguous);
+    assert!(error.allows_pre_delivery_retry());
+    let detail = error
+        .client_visible_upstream_error()
+        .expect("WebSocket close detail");
+    assert_eq!(detail.message(), "message too big");
+    assert_eq!(detail.code(), Some("1009"));
+    assert_eq!(detail.error_type(), Some("websocket_close_error"));
+}
+
 #[tokio::test]
 async fn disabled_account_is_excluded_from_normal_scheduling() {
     let store = Arc::new(MemoryAccountStore::default());
@@ -701,6 +854,109 @@ async fn cross_account_scope_sanitizes_only_known_turn_metadata_fields() {
     );
 }
 
+#[tokio::test]
+async fn continuation_replay_should_use_the_client_full_input_without_legacy_transcript() {
+    let store = Arc::new(MemoryAccountStore::default());
+    create_account(&store, "acct_client_history").await;
+    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(CAPTURE_COMPLETED_SSE),
+        )
+        .expect(1)
+        .mount(&server)
+        .await;
+    let client_input = json!([
+        {"role": "user", "content": "complete client history"},
+        {"role": "assistant", "content": "previous answer"},
+        {"role": "user", "content": "current turn"}
+    ]);
+    let generation = GenerateRequest::from_protocol_payload(
+        ProtocolPayload::json_object(
+            "openai",
+            Map::from_iter([
+                ("model".to_owned(), json!("gpt-5.4")),
+                ("input".to_owned(), client_input.clone()),
+            ]),
+        )
+        .expect("OpenAI payload")
+        .with_context(Map::from_iter([("use_websocket".to_owned(), json!(false))])),
+    )
+    .with_provider_session_state(
+        ProviderSessionState::new(
+            "openai",
+            Map::from_iter([
+                ("account_id".to_owned(), json!("acct_client_history")),
+                ("conversation_id".to_owned(), json!("conversation")),
+                ("continuation_scope".to_owned(), json!("replay_required")),
+                (
+                    "transcript".to_owned(),
+                    json!([{"client_input": {"role": "user", "content": "stale proxy copy"}}]),
+                ),
+            ]),
+        )
+        .expect("provider session state"),
+    );
+    let mut stream = provider_with_base_url(&store, server.uri())
+        .execute(
+            planned_request("openai", Operation::Generate(generation)),
+            context("req_client_history", CancellationToken::new())
+                .with_continuation_attempt(ContinuationAttempt::ReplayAny),
+        )
+        .await
+        .expect("prepare replay stream");
+    while let Some(event) = stream.next().await {
+        event.expect("replay response");
+    }
+    let requests = server
+        .received_requests()
+        .await
+        .expect("captured replay request");
+    let body: Value = serde_json::from_slice(&requests[0].body).expect("captured JSON body");
+
+    assert_eq!(body.get("input"), Some(&client_input));
+}
+
+#[tokio::test]
+async fn completed_response_session_state_should_not_copy_the_conversation_transcript() {
+    let store = Arc::new(MemoryAccountStore::default());
+    create_account(&store, "acct_bounded_session_state").await;
+    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(CAPTURE_COMPLETED_SSE),
+        )
+        .expect(1)
+        .mount(&server)
+        .await;
+    let mut stream = provider_with_base_url(&store, server.uri())
+        .execute(
+            planned_request("openai", http_generate_operation()),
+            context("req_bounded_session_state", CancellationToken::new()),
+        )
+        .await
+        .expect("prepare provider stream");
+    let mut session_payload = None;
+    while let Some(event) = stream.next().await {
+        let event = event.expect("provider response");
+        if let Some(update) = event.session_update() {
+            session_payload = Some(update.payload().clone());
+        }
+    }
+
+    assert!(
+        !session_payload
+            .expect("completed response session update")
+            .contains_key("transcript")
+    );
+}
+
 #[tokio::test]
 async fn account_change_drops_only_account_bound_opaque_headers() {
     let protocol_context = Map::from_iter([(

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

@@ -2,7 +2,7 @@
 
 use std::collections::{BTreeMap, BTreeSet};
 use std::num::NonZeroU32;
-use std::sync::atomic::{AtomicUsize, Ordering};
+use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
 use std::sync::{Arc, Mutex};
 use std::time::{Duration, SystemTime};
 
@@ -43,6 +43,7 @@ struct StoredAccount {
 pub(crate) struct MemoryAccountStore {
     accounts: Mutex<BTreeMap<ProviderAccountId, StoredAccount>>,
     quota_reads: AtomicUsize,
+    fail_provider_listing: AtomicBool,
 }
 
 impl MemoryAccountStore {
@@ -115,6 +116,10 @@ impl MemoryAccountStore {
             .get(&id)
             .is_some_and(|stored| stored.quota.is_some())
     }
+
+    pub(crate) fn fail_provider_listing(&self) {
+        self.fail_provider_listing.store(true, Ordering::SeqCst);
+    }
 }
 
 #[async_trait]
@@ -162,6 +167,9 @@ impl ProviderAccountStore for MemoryAccountStore {
         &self,
         provider: &ProviderKind,
     ) -> Result<Vec<ProviderAccount>, StoreError> {
+        if self.fail_provider_listing.load(Ordering::SeqCst) {
+            return Err(store_error(StoreErrorKind::Unavailable));
+        }
         Ok(self
             .accounts
             .lock()

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

@@ -5,6 +5,7 @@ use provider_openai::transport::{
         WebSocketOriginBreaker, WebSocketOriginBreakerConfig, WebSocketOriginBreakerDecision,
     },
 };
+use tokio_tungstenite::tungstenite::protocol::{CloseFrame, frame::coding::CloseCode};
 
 use super::*;
 
@@ -913,14 +914,133 @@ async fn concurrent_same_key_should_singleflight_websocket_opening() {
 }
 
 #[tokio::test]
-async fn payload_send_failure_should_not_open_http_fallback() {
+async fn websocket_failure_before_first_delivery_should_use_same_account_http() {
+    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let addr = listener.local_addr().unwrap();
+    let server = tokio::spawn(async move {
+        let (stream, _) = listener.accept().await.unwrap();
+        let mut websocket = accept_codex_test_websocket_with(stream, |request, response| {
+            assert_eq!(
+                request
+                    .headers()
+                    .get("authorization")
+                    .and_then(|value| value.to_str().ok()),
+                Some("Bearer access-token")
+            );
+            assert_eq!(
+                request
+                    .headers()
+                    .get("chatgpt-account-id")
+                    .and_then(|value| value.to_str().ok()),
+                Some("chatgpt-account")
+            );
+            response.headers_mut().insert(
+                "sec-websocket-extensions",
+                "permessage-deflate".parse().unwrap(),
+            );
+        })
+        .await;
+        let _payload = websocket.next().await.unwrap().unwrap();
+        websocket
+            .send(Message::Text(
+                json!({
+                    "type": "response.created",
+                    "response": {
+                        "id": "resp_abandoned_websocket",
+                        "model": "gpt-test"
+                    }
+                })
+                .to_string()
+                .into(),
+            ))
+            .await
+            .unwrap();
+        websocket
+            .send(Message::Text(
+                json!({
+                    "type": "response.in_progress",
+                    "response": {
+                        "id": "resp_abandoned_websocket",
+                        "model": "gpt-test"
+                    }
+                })
+                .to_string()
+                .into(),
+            ))
+            .await
+            .unwrap();
+        websocket
+            .close(Some(CloseFrame {
+                code: CloseCode::Size,
+                reason: "message too big".into(),
+            }))
+            .await
+            .unwrap();
+
+        let (mut http, _) = listener.accept().await.unwrap();
+        let request = read_http_request(&mut http).await;
+        assert!(request.starts_with("POST /codex/responses HTTP/1.1"));
+        assert_eq!(
+            read_header_value(&request, "authorization"),
+            Some("Bearer access-token")
+        );
+        assert_eq!(
+            read_header_value(&request, "chatgpt-account-id"),
+            Some("chatgpt-account")
+        );
+        write_completed_sse_response(&mut http).await;
+    });
+    let backend = CodexBackendClient::new(
+        reqwest::Client::builder().no_proxy().build().unwrap(),
+        format!("http://{addr}"),
+        test_wire_profile(),
+    )
+    .with_websocket_pool(Arc::new(CodexWebSocketPool::default()));
+
+    let response = backend
+        .create_response(
+            &new_chain_request("conversation-pre-delivery-fallback"),
+            request_context("req_pre_delivery_fallback", Some("chatgpt-account")),
+        )
+        .await
+        .expect("pre-delivery WebSocket failure should use same-account HTTP");
+    server.await.unwrap();
+
+    assert_eq!(response.transport, CodexBackendTransport::HttpSse);
+    assert_eq!(
+        response.transport_metrics.decision,
+        Some(CodexTransportDecision::Http2PreDeliveryFailure)
+    );
+    assert!(response.body.contains("resp_http_fallback"));
+    assert!(!response.body.contains("resp_abandoned_websocket"));
+}
+
+#[tokio::test]
+async fn websocket_failure_after_first_delivery_should_not_open_http_fallback() {
     let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
     let addr = listener.local_addr().unwrap();
     let server = tokio::spawn(async move {
         let (stream, _) = listener.accept().await.unwrap();
         let mut websocket = accept_codex_test_websocket(stream).await;
         let _payload = websocket.next().await.unwrap().unwrap();
-        websocket.close(None).await.unwrap();
+        websocket
+            .send(Message::Text(
+                json!({
+                    "type": "response.output_text.delta",
+                    "delta": "delivered"
+                })
+                .to_string()
+                .into(),
+            ))
+            .await
+            .unwrap();
+        websocket
+            .close(Some(CloseFrame {
+                code: CloseCode::Size,
+                reason: "message too big after output".into(),
+            }))
+            .await
+            .unwrap();
         assert!(
             timeout(Duration::from_millis(100), listener.accept())
                 .await
@@ -936,17 +1056,25 @@ async fn payload_send_failure_should_not_open_http_fallback() {
 
     let error = backend
         .create_response(
-            &new_chain_request("conversation-post-send"),
-            request_context("req_post_send", Some("chatgpt-account")),
+            &new_chain_request("conversation-post-delivery"),
+            request_context("req_post_delivery", Some("chatgpt-account")),
         )
         .await
-        .expect_err("post-send close must not be replayed");
+        .expect_err("post-delivery close must not be replayed");
     server.await.unwrap();
 
-    std::assert_matches!(
-        error,
-        CodexClientError::WebSocket(CodexWebSocketExchangeError::PostSendAmbiguous { .. })
-    );
+    let CodexClientError::WebSocket(CodexWebSocketExchangeError::PostSendAmbiguous {
+        source: Some(source),
+        ..
+    }) = error
+    else {
+        panic!("expected typed post-delivery WebSocket failure");
+    };
+    let CodexWebSocketExchangeError::ClosedBeforeTerminal(close) = *source else {
+        panic!("expected upstream close frame source");
+    };
+    assert_eq!(close.code(), Some(1009));
+    assert_eq!(close.reason(), Some("message too big after output"));
 }
 
 #[tokio::test]

backend/crates/providers/openai/tests/transport/protocol/responses.rs

@@ -218,6 +218,17 @@ fn transport_should_prefer_websocket_when_requested_without_history() {
     assert!(transport_requirement(&request).allows_pre_send_http_fallback());
 }
 
+#[test]
+fn new_chain_should_allow_pre_delivery_http_fallback() {
+    let mut request = codex_request("gpt-test", "be brief", Vec::new());
+    request.use_websocket = true;
+
+    assert!(
+        transport_requirement(&request).allows_pre_delivery_http_fallback(),
+        "new chains may follow the TS same-account fallback behavior"
+    );
+}
+
 #[test]
 fn transport_should_mark_unknown_previous_response_as_external() {
     let mut request = codex_request("gpt-test", "be brief", Vec::new());
@@ -230,6 +241,17 @@ fn transport_should_mark_unknown_previous_response_as_external() {
     assert!(transport_requirement(&request).allows_pre_send_http_fallback());
 }
 
+#[test]
+fn previous_response_should_forbid_pre_delivery_http_fallback() {
+    let mut request = codex_request("gpt-test", "be brief", Vec::new());
+    request.set_previous_response_id(Some("resp_previous".to_owned()));
+
+    assert!(
+        !transport_requirement(&request).allows_pre_delivery_http_fallback(),
+        "a sent continuation must not be replayed on another transport"
+    );
+}
+
 #[test]
 fn transport_should_allow_forced_http_sse() {
     let mut request = codex_request("gpt-test", "be brief", Vec::new());

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

@@ -649,7 +649,15 @@ async fn websocket_execute_response_create_request_should_reject_binary_event()
             .unwrap();
         websocket.close(None).await.unwrap();
     });
-    let prepared = prepared_websocket_request(&format!("http://{addr}"));
+    let mut request = codex_request("gpt-test", "be brief", Vec::new());
+    request.set_previous_response_id(Some("resp_binary_previous".to_owned()));
+    let prepared = CodexWebSocketConnection::responses_create_request(
+        &format!("http://{addr}"),
+        "dGhlIHNhbXBsZSBub25jZQ==",
+        vec![("authorization".to_owned(), "Bearer access-token".to_owned())],
+        &request,
+    )
+    .expect("prepare continuation WebSocket request");
 
     let error = execute_response_create_request(&prepared)
         .await

release/version.yaml

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