feat(openai): add bounded retries for explicit 503 server errors (#428)

Co-authored-by: Zyy <zyycn@users.noreply.github.com>
Co-authored-by: Codex <noreply@openai.com>

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

@@ -462,7 +462,7 @@ pub(super) fn apply_websocket_recovery_policy(
     if failure.error.retry_is_prohibited() {
         return;
     }
-    // 明确账号拒绝走已有换号路径，容量拒绝走请求内退避；两者都不消耗 WS 传输预算
+    // 明确账号拒绝走已有换号路径，业务瞬态失败走请求内退避，不消耗 WS 传输预算
     if failure.error.replay_is_safe()
         && (failure.account_failure.is_some()
             || matches!(
@@ -1326,6 +1326,7 @@ pub(super) fn map_upstream_failure(
 ) -> MappedProviderFailure {
     let category = failure.category();
     let capacity_unavailable = category == CodexFailureCategory::CapacityUnavailable;
+    let transient_retry = capacity_unavailable || failure.is_retryable_server_error();
     let cyber_policy_failure = failure
         .status
         .is_some_and(|status| status.is_client_error())
@@ -1392,7 +1393,7 @@ pub(super) fn map_upstream_failure(
     if let Some(retry_after) = failure.retry_after_seconds.map(Duration::from_secs) {
         error = error.with_retry_after(retry_after);
     }
-    if capacity_unavailable && error.replay_is_safe() {
+    if transient_retry && error.replay_is_safe() {
         let max_delay = Duration::from_secs(8);
         error = error.with_transient_retry(
             NonZeroU32::new(3).unwrap_or(NonZeroU32::MIN),

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

@@ -180,23 +180,34 @@ impl CodexUpstreamFailure {
         self.category
     }
 
-    /// 只有明确识别的账号或容量拒绝才允许网关重放，普通上游错误交给客户端
+    /// 明确的 503 server_error 允许有界恢复，不把普通 5xx 当作容量或账号故障
+    pub(crate) fn is_retryable_server_error(&self) -> bool {
+        self.category == CodexFailureCategory::Unavailable
+            && self.status == Some(StatusCode::SERVICE_UNAVAILABLE)
+            && self
+                .code
+                .as_deref()
+                .is_some_and(|code| code.trim().eq_ignore_ascii_case("server_error"))
+    }
+
+    /// 只对明确识别的账号、容量拒绝或 503 server_error 提供重放资格
     #[must_use]
-    pub const fn replay_is_safe(&self) -> bool {
+    pub fn replay_is_safe(&self) -> bool {
         !matches!(self.send_phase, CodexUpstreamSendPhase::Ambiguous)
-            && matches!(
-                self.category,
-                CodexFailureCategory::ModelUnsupported
-                    | CodexFailureCategory::CredentialExpired
-                    | CodexFailureCategory::IdentityVerificationRequired
-                    | CodexFailureCategory::Banned
-                    | CodexFailureCategory::UsageLimitExhausted
-                    | CodexFailureCategory::RateLimited
-                    | CodexFailureCategory::QuotaExhausted
-                    | CodexFailureCategory::CloudflareChallenge
-                    | CodexFailureCategory::CloudflarePathBlocked
-                    | CodexFailureCategory::CapacityUnavailable
-            )
+            && (self.is_retryable_server_error()
+                || matches!(
+                    self.category,
+                    CodexFailureCategory::ModelUnsupported
+                        | CodexFailureCategory::CredentialExpired
+                        | CodexFailureCategory::IdentityVerificationRequired
+                        | CodexFailureCategory::Banned
+                        | CodexFailureCategory::UsageLimitExhausted
+                        | CodexFailureCategory::RateLimited
+                        | CodexFailureCategory::QuotaExhausted
+                        | CodexFailureCategory::CloudflareChallenge
+                        | CodexFailureCategory::CloudflarePathBlocked
+                        | CodexFailureCategory::CapacityUnavailable
+                ))
     }
 
     pub(crate) fn upstream_code(&self) -> Option<&str> {

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

@@ -3,6 +3,149 @@
 use super::precommit::stream_failure;
 use super::*;
 
+#[tokio::test]
+async fn server_error_503_stream_recovery_requires_no_semantic_output() {
+    for websocket in [false, true] {
+        for semantic_output in [false, true] {
+            let store = Arc::new(MemoryAccountStore::default());
+            create_account(&store, "acct_provider_contract").await;
+            let failure = json!({"type":"error","status":503,"error":{"code":"server_error","type":"server_error","message":"You can retry your request."}});
+            let mut upstream_events = vec![
+                json!({"type":"codex.response.metadata","metadata":{}}),
+                json!({"type":"response.created","response":{"id":"resp_503","status":"in_progress","output":[]}}),
+                json!({"type":"response.in_progress","response":{"id":"resp_503","status":"in_progress","output":[]}}),
+            ];
+            if semantic_output {
+                upstream_events.push(json!({"type":"response.output_text.delta","response_id":"resp_503","item_id":"msg_503","output_index":0,"content_index":0,"delta":"hello"}));
+            }
+            upstream_events.push(failure.clone());
+            let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+            let base = format!("http://{}", listener.local_addr().unwrap());
+            let server = tokio::spawn(async move {
+                let (mut socket, _) = listener.accept().await.unwrap();
+                if websocket {
+                    let mut socket = accept_codex_test_websocket(socket).await;
+                    socket.next().await.unwrap().unwrap();
+                    for event in upstream_events {
+                        socket
+                            .send(Message::Text(event.to_string().into()))
+                            .await
+                            .unwrap();
+                    }
+                } else {
+                    read_http_request(&mut socket).await;
+                    let body = upstream_events
+                        .iter()
+                        .map(|event| {
+                            gateway_protocol::openai::sse::encode_sse_event(
+                                event["type"].as_str().unwrap(),
+                                &event.to_string(),
+                            )
+                        })
+                        .collect::<String>();
+                    socket.write_all(format!("HTTP/1.1 200 OK\r\nContent-Type: text/event-stream\r\nContent-Length: {}\r\n\r\n{body}", body.len()).as_bytes()).await.unwrap();
+                }
+            });
+            let operation = if websocket {
+                capacity_websocket_operation()
+            } else {
+                http_generate_operation()
+            };
+            let mut stream = provider_with_base_url(&store, base)
+                .execute(
+                    planned_request("openai", operation),
+                    context("req_server_error_503", CancellationToken::new()),
+                )
+                .await
+                .unwrap();
+            let (mut events, mut error) = stream_failure(&mut stream).await;
+            assert_eq!(error.kind(), ProviderErrorKind::Unavailable);
+            assert_eq!(error.upstream_status(), Some(503));
+            assert_eq!(error.send_state(), UpstreamSendState::Sent);
+            assert_eq!(error.replay_is_safe(), !semantic_output);
+            if semantic_output {
+                assert!(error.pre_delivery_retry().is_none());
+                assert!(
+                    events
+                        .iter()
+                        .any(|event| event.has_client_event() && !event.is_replayable_prelude())
+                );
+            } else {
+                assert!(!error.retry_is_prohibited());
+                assert!(
+                    matches!(error.pre_delivery_retry(), Some(PreDeliveryRetry::SameAccountTransientRetry { max_retries, initial_delay, max_delay })
+                    if max_retries.get() == 3 && initial_delay == Duration::from_millis(500) && max_delay == Duration::from_secs(8))
+                );
+                assert!(
+                    events
+                        .iter()
+                        .filter(|event| event.has_client_event())
+                        .all(|event| event.is_replayable_prelude())
+                );
+            }
+            assert_eq!(
+                serde_json::from_str::<Value>(error.raw_upstream_error().unwrap().as_str())
+                    .unwrap(),
+                failure
+            );
+            events.extend(error.take_atomic_client_events());
+            assert_eq!(
+                events.last().unwrap().wire_event().unwrap().data(),
+                &failure
+            );
+            let account = store.account("acct_provider_contract").unwrap();
+            assert_eq!(account.credential_state(), CredentialState::Ready);
+            assert_eq!(account.quota().access(), QuotaAccessState::Unknown);
+            server.await.unwrap();
+        }
+    }
+}
+
+#[tokio::test]
+async fn server_error_503_opening_rejections_request_bounded_retry_and_keep_response() {
+    for websocket in [false, true] {
+        let store = Arc::new(MemoryAccountStore::default());
+        create_account(&store, "acct_provider_contract").await;
+        let server = MockServer::start().await;
+        let body =
+            r#"{ "error":{"code":"server_error","type":"server_error","message":"retry later"} }"#;
+        Mock::given(method(if websocket { "GET" } else { "POST" }))
+            .and(path("/codex/responses"))
+            .respond_with(
+                ResponseTemplate::new(503)
+                    .insert_header("retry-after", "17")
+                    .set_body_string(body),
+            )
+            .expect(1)
+            .mount(&server)
+            .await;
+        let operation = if websocket {
+            capacity_websocket_operation()
+        } else {
+            http_generate_operation()
+        };
+        let mut stream = provider_with_base_url(&store, server.uri())
+            .execute(
+                planned_request("openai", operation),
+                context("req_server_error_opening", CancellationToken::new()),
+            )
+            .await
+            .unwrap();
+        let (_, error) = stream_failure(&mut stream).await;
+        assert!(error.replay_is_safe());
+        assert!(!error.retry_is_prohibited());
+        assert!(matches!(
+            error.pre_delivery_retry(),
+            Some(PreDeliveryRetry::SameAccountTransientRetry { .. })
+        ));
+        assert_eq!(error.retry_after(), Some(Duration::from_secs(17)));
+        let response = error.client_visible_upstream_response().unwrap();
+        assert_eq!(response.status(), 503);
+        assert_eq!(response.body().as_ref(), body.as_bytes());
+        server.verify().await;
+    }
+}
+
 #[tokio::test]
 async fn ordinary_http_and_websocket_opening_errors_remain_client_decisions() {
     for websocket in [false, true] {

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

@@ -1,4 +1,4 @@
-//! 验证上游诊断的容量错误分类、请求标识与响应事实提取
+//! 验证上游失败分类、重放资格、请求标识与响应事实提取
 
 use base64::{Engine as _, engine::general_purpose::STANDARD};
 use provider_openai::transport::CodexUpstreamDiagnostics;
@@ -43,6 +43,61 @@ fn flex_capacity_rejection_should_be_terminal_even_before_payload_send() {
     }
 }
 
+#[test]
+fn server_error_replay_requires_explicit_503_and_unambiguous_send() {
+    for (status, body, retryable) in [
+        (
+            503,
+            r#"{"error":{"code":"server_error","type":"server_error"}}"#,
+            true,
+        ),
+        (
+            500,
+            r#"{"error":{"code":"server_error","type":"server_error"}}"#,
+            false,
+        ),
+        (
+            504,
+            r#"{"error":{"code":"server_error","type":"server_error"}}"#,
+            false,
+        ),
+        (503, r#"{"error":{"type":"server_error"}}"#, false),
+        (
+            503,
+            r#"{"error":{"code":"Vendor.Future.v2","type":"server_error"}}"#,
+            false,
+        ),
+        (
+            503,
+            r#"{"error":{"code":"flex_unavailable","type":"server_error"}}"#,
+            false,
+        ),
+        (503, "server_error", false),
+    ] {
+        for phase in [
+            CodexUpstreamSendPhase::BeforePayload,
+            CodexUpstreamSendPhase::AfterPayload,
+            CodexUpstreamSendPhase::Ambiguous,
+        ] {
+            let failure = CodexUpstreamFailure::from_response(
+                reqwest::StatusCode::from_u16(status).unwrap(),
+                body,
+                None,
+                &CodexUpstreamDiagnostics::default(),
+                None,
+                &[],
+                &[],
+                phase,
+            );
+            assert_eq!(
+                failure.replay_is_safe(),
+                retryable && phase != CodexUpstreamSendPhase::Ambiguous,
+                "status={status}, body={body}, phase={phase:?}"
+            );
+        }
+    }
+}
+
 #[test]
 fn capacity_rejections_are_request_scoped_across_http_statuses() {
     for status in [400, 429, 503] {

docs/architecture.md

@@ -376,8 +376,8 @@ Client Key 鉴权完成后，API adapter 从有界请求头识别客户端，Cor
 
 - 一个客户端请求对应一条 `model_requests`；attempt 是请求内事实，不建立第二张权威表
 - Provider 的一次 `execute` 只选择一个 credential 并返回一个冷流；换号、重试和 fallback 由 Core 决定
-- OpenAI 的自动恢复限于明确识别的账号、容量、续接和传输场景；普通上游拒绝保留响应与重试提示，
-  交由客户端决策。WS opening 尚未发送业务 payload 也不扩大该恢复范围，明确的 `426` 仍允许 HTTP 回退
+- OpenAI 的自动恢复限于明确识别的账号、容量、`503 + error.code=server_error`、续接和传输场景；
+  其他上游拒绝保留响应与重试提示，交由客户端决策。WS opening 尚未发送业务 payload 也不扩大该恢复范围，明确的 `426` 仍允许 HTTP 回退
 - `not_sent`、`sent`、`ambiguous` 是单调的上游发送边界；结果不明确时不能假定上游未收到请求
 - HTTP 明确未发送的瞬时建连失败（含 TLS 握手 EOF）共享最多 4 次 opening、30 秒的请求内预算；
   证书、TLS 协议及配置错误不进入同账号建连恢复，取消和执行 deadline 优先终止等待
@@ -388,8 +388,8 @@ Client Key 鉴权完成后，API adapter 从有界请求头识别客户端，Cor
   仍检查错误分类、续接归属、发送状态、副作用及预算。中间件链不启用此前导例外
 - API 在最终错误编码出口投影客户端恢复信号；该投影不修改 Provider 上游事实，也不改变 Core 的
   重试与提交边界。具体错误合同见 [数据面接口](api.md#3-openai-数据面与模型目录)
-- Provider 可将明确容量拒绝标记为有界同账号退避，Core 在既有安全重放边界内执行，按账号维护请求内
-  预算，耗尽后复用普通换号路径。该退避消耗总路由预算，与 WS 传输恢复、OAuth 刷新及账号额度冷却分开
+- Provider 可将明确容量拒绝或 `503 + error.code=server_error` 标记为有界同账号退避，Core 在既有安全重放边界内执行，
+  按账号维护请求内预算，耗尽后复用普通换号路径。该退避消耗总路由预算，与 WS 传输恢复、OAuth 刷新及账号额度冷却分开
 - OpenAI Provider 与 WS 传输层不暂存前导等待后续事件，正常帧按顺序即时交付；
   Provider 只保留单向推进的重放边界，实际输出、终态或未知事件关闭重放窗口。
   中间件链中的正常事件同样即时交付，但不保留可替换前导的重试资格。