fix(openai): keep turn rate limits scoped to the request

Preserve upstream turn_rate_limit_exceeded errors and retry advice without
cooling the account or replaying the request inside the proxy.

Cover HTTP and WebSocket rejections plus streamed failures, and document
client-owned waiting and retries.

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

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

@@ -1464,7 +1464,9 @@ pub(super) const fn provider_error_kind(category: CodexFailureCategory) -> Provi
             ProviderErrorKind::PermissionDenied
         }
         CodexFailureCategory::UsageLimitExhausted => ProviderErrorKind::QuotaExhausted,
-        CodexFailureCategory::RateLimited => ProviderErrorKind::RateLimited,
+        CodexFailureCategory::RateLimited | CodexFailureCategory::TurnRateLimited => {
+            ProviderErrorKind::RateLimited
+        }
         CodexFailureCategory::QuotaExhausted => ProviderErrorKind::QuotaExhausted,
         CodexFailureCategory::CloudflareChallenge
         | CodexFailureCategory::CloudflarePathBlocked
@@ -1519,6 +1521,7 @@ pub(super) fn account_failure(
             Some(CodexAccountFailure::CloudflarePathBlocked)
         }
         CodexFailureCategory::ModelUnsupported
+        | CodexFailureCategory::TurnRateLimited
         | CodexFailureCategory::InvalidRequest
         | CodexFailureCategory::PermissionDenied
         | CodexFailureCategory::Timeout

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

@@ -36,7 +36,7 @@ pub enum CodexUpstreamSendPhase {
     Ambiguous,
 }
 
-/// Transport 边界对上游失败的账号级分类
+/// Transport 边界对上游失败的分类
 #[derive(Clone, Copy, Debug, PartialEq, Eq)]
 pub enum CodexFailureCategory {
     ModelUnsupported,
@@ -46,6 +46,8 @@ pub enum CodexFailureCategory {
     /// 当前用量窗口已经耗尽，但存在自动恢复的重置时间
     UsageLimitExhausted,
     RateLimited,
+    /// 当前 turn 的限流交给客户端等待，不据此推导账号级冷却
+    TurnRateLimited,
     QuotaExhausted,
     CloudflareChallenge,
     CloudflarePathBlocked,
@@ -174,7 +176,7 @@ impl CodexUpstreamFailure {
         }
     }
 
-    /// 返回账号级失败分类
+    /// 返回上游失败分类
     #[must_use]
     pub const fn category(&self) -> CodexFailureCategory {
         self.category
@@ -389,6 +391,9 @@ fn classify_upstream_failure(
     if code == "flex_unavailable" {
         return CodexFailureCategory::FlexUnavailable;
     }
+    if code == "turn_rate_limit_exceeded" {
+        return CodexFailureCategory::TurnRateLimited;
+    }
 
     // 容量拒绝可能带 400/429/503；仅用结构化错误字段识别，不能扫描任意正文
     if is_capacity_error(

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

@@ -8498,15 +8498,27 @@ async fn capacity_http_and_websocket_opening_rejections_use_bounded_business_ret
 }
 
 #[tokio::test]
-async fn flex_http_and_websocket_opening_rejections_should_not_retry_or_cool_down_accounts() {
+async fn request_rejections_should_not_retry_or_cool_down_accounts() {
     use gateway_core::provider_ports::ProviderCooldownPort as _;
-    for use_websocket in [false, true] {
+    for (use_websocket, (code, expected_kind)) in [false, true].into_iter().flat_map(|websocket| {
+        [
+            ("flex_unavailable", ProviderErrorKind::Unavailable),
+            ("turn_rate_limit_exceeded", ProviderErrorKind::RateLimited),
+        ]
+        .map(|case| (websocket, case))
+    }) {
         let store = Arc::new(MemoryAccountStore::default());
         let account_id = "acct_provider_contract";
         create_account(&store, account_id).await;
         let cooldowns = Arc::new(MemoryCooldownPort::new());
         let server = MockServer::start().await;
-        let body = json!({"error":{"type":"resource_unavailable","code":"flex_unavailable","message":"Flex capacity unavailable."}});
+        let error_type = if code == "turn_rate_limit_exceeded" {
+            "rate_limit_error"
+        } else {
+            "resource_unavailable"
+        };
+        let body =
+            json!({"error":{"type":error_type,"code":code,"message":"Retry this request later."}});
         Mock::given(method(if use_websocket { "GET" } else { "POST" }))
             .and(path("/codex/responses"))
             .respond_with(
@@ -8527,7 +8539,7 @@ async fn flex_http_and_websocket_opening_rejections_should_not_retry_or_cool_dow
         let mut stream = provider
             .execute(
                 planned_request("openai", operation),
-                context("req_flex", CancellationToken::new()),
+                context("req_terminal_rejection", CancellationToken::new()),
             )
             .await
             .unwrap();
@@ -8535,14 +8547,12 @@ async fn flex_http_and_websocket_opening_rejections_should_not_retry_or_cool_dow
             match stream.next().await {
                 Some(Ok(_)) => {}
                 Some(Err(error)) => break error,
-                None => panic!("expected Flex failure"),
+                None => panic!("expected request rejection"),
             }
         };
-        assert_eq!(error.kind(), ProviderErrorKind::Unavailable);
-        assert_eq!(
-            error.upstream_code().map(|code| code.as_str()),
-            Some("flex_unavailable")
-        );
+        assert_eq!(error.kind(), expected_kind);
+        assert_eq!(error.upstream_code().map(|code| code.as_str()), Some(code));
+        assert_eq!(error.retry_after(), Some(Duration::from_secs(300)));
         assert!(!error.replay_is_safe());
         assert!(error.pre_delivery_retry().is_none());
         assert!(error.retry_is_prohibited());
@@ -8552,6 +8562,11 @@ async fn flex_http_and_websocket_opening_rejections_should_not_retry_or_cool_dow
         let response = error.client_visible_upstream_response().unwrap();
         assert_eq!(response.status(), 429);
         assert_eq!(response.body().as_ref(), body.to_string().as_bytes());
+        assert!(
+            response.headers().iter().any(|header| {
+                header.name() == "retry-after" && header.value().as_ref() == b"300"
+            })
+        );
         let account = store.account(account_id).unwrap();
         assert_eq!(account.credential_state(), CredentialState::Ready);
         assert_eq!(account.quota().access(), QuotaAccessState::Unknown);
@@ -8815,6 +8830,12 @@ async fn capacity_feedback_in_stream_only_counts_explicit_overload() {
                     ProviderErrorKind::Unavailable,
                     false,
                 ),
+                (
+                    "turn_rate_limit_exceeded",
+                    "Retry this turn later.",
+                    ProviderErrorKind::RateLimited,
+                    false,
+                ),
                 (
                     "invalid_prompt",
                     "Invalid prompt: we've limited access to this content for safety reasons.",
@@ -8856,7 +8877,10 @@ async fn capacity_feedback_in_stream_only_counts_explicit_overload() {
             if semantic_output {
                 events.push(json!({"type": "response.output_text.delta", "output_index": 0, "content_index": 0, "delta": "hello"}));
             }
-            let original = json!({"type": "response.failed", "response": {"id": "resp_capacity", "error": {"code": code, "message": message}}});
+            let mut original = json!({"type": "response.failed", "response": {"id": "resp_capacity", "error": {"code": code, "message": message}}});
+            if code == "turn_rate_limit_exceeded" {
+                original["response"]["error"]["headers"] = json!({"retry-after": "7"});
+            }
             events.push(original.clone());
             let (base_url, _http_server, websocket_server) = if use_websocket {
                 let listener = TcpListener::bind("127.0.0.1:0").await.expect("listener");
@@ -8928,6 +8952,10 @@ async fn capacity_feedback_in_stream_only_counts_explicit_overload() {
             );
             assert_eq!(error.replay_is_safe(), capacity && !semantic_output);
             assert_eq!(error.upstream_status(), None);
+            if code == "turn_rate_limit_exceeded" {
+                assert!(error.retry_is_prohibited());
+                assert_eq!(error.retry_after(), Some(Duration::from_secs(7)));
+            }
             assert_eq!(
                 error.pre_delivery_retry().is_some(),
                 capacity && !semantic_output

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

@@ -43,6 +43,32 @@ fn flex_capacity_rejection_should_be_terminal_even_before_payload_send() {
     }
 }
 
+#[test]
+fn turn_rate_limit_does_not_grant_proxy_replay_or_account_cooldown() {
+    for phase in [
+        CodexUpstreamSendPhase::BeforePayload,
+        CodexUpstreamSendPhase::AfterPayload,
+        CodexUpstreamSendPhase::Ambiguous,
+    ] {
+        let failure = CodexUpstreamFailure::from_response(
+            reqwest::StatusCode::TOO_MANY_REQUESTS,
+            r#"{"error":{"code":"turn_rate_limit_exceeded","type":"rate_limit_error","message":"Retry this turn later."}}"#,
+            Some(53),
+            &CodexUpstreamDiagnostics::default(),
+            None,
+            &[],
+            &[],
+            phase,
+        );
+        assert_eq!(failure.category(), CodexFailureCategory::TurnRateLimited);
+        assert!(!failure.replay_is_safe());
+    }
+    assert_eq!(
+        classify(429, r#"{"error":{"message":"turn_rate_limit_exceeded"}}"#),
+        CodexFailureCategory::RateLimited
+    );
+}
+
 #[test]
 fn server_error_replay_requires_explicit_503_and_unambiguous_send() {
     for (status, body, retryable) in [

docs/api.md

@@ -443,6 +443,9 @@ xAI 的 `reasoning.effort` 接受 `none / minimal / low / medium / high / xhigh
 
 #### 容量拒绝与重试
 
+`turn_rate_limit_exceeded` 保留原始错误和 `Retry-After`，由客户端按自身重试预算等待。
+代理不因此写入账号冷却、换号或透明重试，其他会话仍可使用该账号
+
 Responses 的 HTTP/SSE 与 WebSocket 按顺序即时交付正常事件，不暂存前导等待后续输出。
 未启用中间件链时，无输出的 `response.created`、`response.in_progress`
 及 `codex.response.metadata`、`codex.rate_limits` 元数据交付后仍可按已有安全条件重试，