fix(openai): preserve Retry-After when converting WS errors to SSE (#531)

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

backend/crates/gateway-api/tests/openai/responses/http.rs

@@ -2164,109 +2164,132 @@ async fn streaming_upstream_wire_failure_should_not_be_rewritten_as_a_gateway_er
 
 #[tokio::test]
 async fn streaming_upstream_error_event_should_be_translated_to_response_failed_for_sse() {
-    let trace = Arc::new(Trace::default());
-    let error_data = json!({
-        "type": "error",
-        "error": {
-            "type": "service_unavailable_error",
-            "code": "server_is_overloaded",
-            "message": "Our servers are currently overloaded. Please try again later.",
-            "param": null,
-            "future_error_field": {"keep": true}
-        },
-        "sequence_number": 2,
-        "future_event_field": {"keep": true}
-    });
-    let projected_error_frame = Bytes::from(encode_sse_event("error", &error_data.to_string()));
-    let error_wire = ProviderEvent::wire(
-        ProtocolWireEvent::json_with_raw_sse_metadata(
-            "openai",
-            Some("error".to_owned()),
-            error_data,
-            projected_error_frame,
-            None,
-            None,
-        )
-        .expect("valid upstream error wire"),
-    );
-    let started_wire = ProviderEvent::canonical_with_wire(
-        vec![started()],
-        ProtocolWireEvent::json(
-            "openai",
-            Some("response.created".to_owned()),
-            json!({
-                "type": "response.created",
-                "response": {
-                    "id": "resp_test",
-                    "model": "public-model",
-                    "status": "in_progress",
-                    "future_response_field": {"keep": true}
-                }
-            }),
-        )
-        .expect("valid upstream started wire"),
-    );
-    let session = FakeSession::streaming(
-        Arc::clone(&trace),
-        vec![
-            NextStep::Event(CoordinatedEvent::single(
-                started_wire,
-                CommitRequirement::CommitBeforeDelivery,
-            )),
-            NextStep::Event(CoordinatedEvent::single(
-                error_wire,
-                CommitRequirement::AlreadyCommitted,
-            )),
-            NextStep::Error(EngineError::Provider(ProviderError::new(
-                ProviderErrorKind::UpstreamCapacityUnavailable,
-                UpstreamSendState::Sent,
-            ))),
-        ],
-    );
-
-    let response = stream_execution_response(Box::new(session), None).await;
-    let body = to_bytes(response.into_body(), usize::MAX)
-        .await
-        .expect("read SSE body");
-    let body = String::from_utf8(body.to_vec()).expect("SSE is UTF-8");
-    let events = parse_sse_events(&body).expect("translated SSE should parse");
-    let failed = events
-        .iter()
-        .find(|event| event.event.as_deref() == Some("response.failed"))
-        .expect("translated response.failed event");
-    let failed: Value = serde_json::from_str(&failed.data).expect("response.failed JSON");
-
-    assert_eq!(
+    for (error_type, code, client_code, kind, client_status) in [
         (
-            failed,
-            body.matches("event: response.failed").count(),
-            body.contains("event: error\n"),
-            body.ends_with("data: [DONE]\n\n"),
+            "service_unavailable_error",
+            "server_is_overloaded",
+            "server_error",
+            ProviderErrorKind::UpstreamCapacityUnavailable,
+            503,
         ),
         (
-            json!({
-                "type": "response.failed",
-                "response": {
-                    "id": "resp_test",
-                    "model": "public-model",
-                    "status": "failed",
-                    "error": {
-                        "type": "service_unavailable_error",
-                        "code": "server_error",
-                        "message": "Our servers are currently overloaded. Please try again later.",
-                        "param": null,
-                        "future_error_field": {"keep": true}
+            "rate_limit_error",
+            "turn_rate_limit_exceeded",
+            "turn_rate_limit_exceeded",
+            ProviderErrorKind::RateLimited,
+            429,
+        ),
+    ] {
+        let trace = Arc::new(Trace::default());
+        let error_data = json!({
+            "type": "error",
+            "status": 429,
+            "headers": {"Retry-After": "24"},
+            "error": {
+                "type": error_type,
+                "code": code,
+                "message": "Our servers are currently overloaded. Please try again later.",
+                "param": null,
+                "future_error_field": {"keep": true}
+            },
+            "sequence_number": 2,
+            "future_event_field": {"keep": true}
+        });
+        let projected_error_frame = Bytes::from(encode_sse_event("error", &error_data.to_string()));
+        let error_wire = ProviderEvent::wire(
+            ProtocolWireEvent::json_with_raw_sse_metadata(
+                "openai",
+                Some("error".to_owned()),
+                error_data,
+                projected_error_frame,
+                None,
+                None,
+            )
+            .expect("valid upstream error wire"),
+        );
+        let started_wire = ProviderEvent::canonical_with_wire(
+            vec![started()],
+            ProtocolWireEvent::json(
+                "openai",
+                Some("response.created".to_owned()),
+                json!({
+                    "type": "response.created",
+                    "response": {
+                        "id": "resp_test",
+                        "model": "public-model",
+                        "status": "in_progress",
+                        "future_response_field": {"keep": true}
+                    }
+                }),
+            )
+            .expect("valid upstream started wire"),
+        );
+        let session = FakeSession::streaming(
+            Arc::clone(&trace),
+            vec![
+                NextStep::Event(CoordinatedEvent::single(
+                    started_wire,
+                    CommitRequirement::CommitBeforeDelivery,
+                )),
+                NextStep::Event(CoordinatedEvent::single(
+                    error_wire,
+                    CommitRequirement::AlreadyCommitted,
+                )),
+                NextStep::Error(EngineError::Provider(ProviderError::new(
+                    kind,
+                    UpstreamSendState::Sent,
+                ))),
+            ],
+        );
+
+        let response = stream_execution_response(Box::new(session), None).await;
+        assert_eq!(response.status(), StatusCode::OK);
+        let body = to_bytes(response.into_body(), usize::MAX)
+            .await
+            .expect("read SSE body");
+        let body = String::from_utf8(body.to_vec()).expect("SSE is UTF-8");
+        let events = parse_sse_events(&body).expect("translated SSE should parse");
+        let failed = events
+            .iter()
+            .find(|event| event.event.as_deref() == Some("response.failed"))
+            .expect("translated response.failed event");
+        let failed: Value = serde_json::from_str(&failed.data).expect("response.failed JSON");
+
+        assert_eq!(
+            (
+                failed,
+                body.matches("event: response.failed").count(),
+                body.contains("event: error\n"),
+                body.ends_with("data: [DONE]\n\n"),
+            ),
+            (
+                json!({
+                    "type": "response.failed",
+                    "status": client_status,
+                    "headers": {"Retry-After": "24"},
+                    "response": {
+                        "id": "resp_test",
+                        "model": "public-model",
+                        "status": "failed",
+                        "error": {
+                            "type": error_type,
+                            "code": client_code,
+                            "message": "Our servers are currently overloaded. Please try again later.",
+                            "param": null,
+                            "future_error_field": {"keep": true},
+                            "headers": {"retry-after": "24"}
+                        },
+                        "future_response_field": {"keep": true}
                     },
-                    "future_response_field": {"keep": true}
-                },
-                "sequence_number": 2,
-                "future_event_field": {"keep": true}
-            }),
-            1,
-            false,
-            true,
-        )
-    );
+                    "sequence_number": 2,
+                    "future_event_field": {"keep": true}
+                }),
+                1,
+                false,
+                true,
+            )
+        );
+    }
 }
 
 #[tokio::test]

backend/crates/gateway-api/tests/openai/responses/mod.rs

@@ -948,8 +948,8 @@ fn transparent_encoder_should_translate_error_wire_to_response_failed_for_websoc
         json!({
             "type": "error",
             "status": 429,
-            "headers": {"x-request-id": "req_raw"},
-            "error": {"type": "custom_type", "code": "custom_code", "message": "raw marker"},
+            "headers": {"x-request-id": "req_raw", "Retry-After": "24"},
+            "error": {"type": "rate_limit_error", "code": "turn_rate_limit_exceeded", "message": "raw marker"},
             "future": {"keep": true}
         }),
     );

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

@@ -199,8 +199,9 @@ pub fn retry_after_seconds_from_value(value: &Value) -> Option<u64> {
         .pointer("/response/error")
         .or_else(|| value.get("error"))
         .unwrap_or(value);
-    retry_after_seconds_header(error)
-        .or_else(|| retry_after_seconds_header(value))
+    retry_after_header(error)
+        .or_else(|| retry_after_header(value))
+        .map(|(_, seconds)| seconds)
         .or_else(|| retry_after_seconds_field(error))
         .or_else(|| {
             error
@@ -225,22 +226,27 @@ fn retry_after_seconds_field(value: &Value) -> Option<u64> {
         .filter(|seconds| *seconds > 0)
 }
 
-fn retry_after_seconds_header(value: &Value) -> Option<u64> {
+/// 同时保留原始头值与解析秒数，供事件转换和重试观测共用官方头部规则
+pub(super) fn retry_after_header(value: &Value) -> Option<(&Value, u64)> {
     let headers = value.get("headers")?.as_object()?;
     // 官方先按 HTTP 头值校验，再覆盖大小写不同的同名头；最后才解释重试时间
     // 合法但无法解析为时间的末项也会覆盖前值，非法头值和数组不参与覆盖
-    let header = headers
+    let (value, header) = headers
         .iter()
         .filter(|(name, _)| name.eq_ignore_ascii_case("retry-after"))
-        .filter_map(|(_, value)| match value {
-            Value::String(value) => http::HeaderValue::from_str(value).ok(),
-            Value::Number(_) | Value::Bool(_) => {
-                http::HeaderValue::from_str(&value.to_string()).ok()
-            }
-            Value::Null | Value::Array(_) | Value::Object(_) => None,
+        .filter_map(|(_, value)| {
+            let header = match value {
+                Value::String(value) => http::HeaderValue::from_str(value).ok(),
+                Value::Number(_) | Value::Bool(_) => {
+                    http::HeaderValue::from_str(&value.to_string()).ok()
+                }
+                Value::Null | Value::Array(_) | Value::Object(_) => None,
+            }?;
+            Some((value, header))
         })
         .next_back()?;
-    super::headers::parse_retry_after_seconds(header.to_str().ok()?)
+    let seconds = super::headers::parse_retry_after_seconds(header.to_str().ok()?)?;
+    Some((value, seconds))
 }
 
 /// 标准化的单个限流窗口

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

@@ -295,7 +295,7 @@ pub fn response_failed_sse_data_with_id(
 /// 将携带嵌套错误的上游事件投影为 OpenAI Responses `response.failed` 数据
 ///
 /// 该投影供客户端 SSE 兼容转换使用
-/// 原始错误对象和事件级元数据保持不透明；
+/// 保留原始错误与事件级元数据，并补齐官方 SSE 读取的错误重试头
 /// 已有 response 快照只更新失败终态字段
 pub fn response_failed_sse_data_from_error_event(
     response: Option<&Value>,
@@ -303,7 +303,23 @@ pub fn response_failed_sse_data_from_error_event(
     error_event: &Value,
 ) -> Option<Value> {
     let mut event = error_event.as_object()?.clone();
-    let error = event.remove("error").filter(Value::is_object)?;
+    let mut error = event.remove("error").filter(Value::is_object)?;
+    // 官方 WS 先读错误头、再读事件头，SSE 只读 response.error.headers
+    // 仅补充有效的等待建议，保留原始日期或秒数，不复制其他事件级响应头
+    if super::events::retry_after_header(&error).is_none()
+        && let Some((retry_after, _)) = super::events::retry_after_header(error_event)
+    {
+        let headers = error
+            .as_object_mut()?
+            .entry("headers")
+            .or_insert_with(|| json!({}));
+        if !headers.is_object() {
+            *headers = json!({});
+        }
+        let headers = headers.as_object_mut()?;
+        headers.retain(|name, _| !name.eq_ignore_ascii_case("retry-after"));
+        headers.insert("retry-after".to_owned(), retry_after.clone());
+    }
     let mut response = response
         .and_then(Value::as_object)
         .cloned()

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

@@ -442,6 +442,76 @@ fn response_failed_projection_should_preserve_response_error_and_event_metadata(
     );
 }
 
+#[test]
+fn response_failed_projection_should_preserve_retry_advice_for_sse_clients() {
+    let cases = [
+        (
+            json!({"Retry-After": "24"}),
+            None,
+            Some(json!({"retry-after": "24"})),
+        ),
+        (
+            json!({"ReTrY-AfTeR": 15}),
+            None,
+            Some(json!({"retry-after": 15})),
+        ),
+        (
+            json!({"Retry-After": "Wed, 21 Oct 2015 07:28:00 GMT"}),
+            None,
+            Some(json!({"retry-after": "Wed, 21 Oct 2015 07:28:00 GMT"})),
+        ),
+        (
+            json!({"Retry-After": "24"}),
+            Some(json!({"ReTrY-AfTeR": "7", "x-error-id": "keep"})),
+            Some(json!({"ReTrY-AfTeR": "7", "x-error-id": "keep"})),
+        ),
+        (
+            json!({"Retry-After": "24"}),
+            Some(json!({"Retry-After": "0"})),
+            Some(json!({"Retry-After": "0"})),
+        ),
+        (
+            json!({"Retry-After": "24", "x-event-id": "event-only"}),
+            Some(json!({"Retry-After": "later", "x-error-id": "keep"})),
+            Some(json!({"retry-after": "24", "x-error-id": "keep"})),
+        ),
+        (
+            json!({"Retry-After": "24"}),
+            Some(json!(null)),
+            Some(json!({"retry-after": "24"})),
+        ),
+        (json!({"Retry-After": ["24"]}), None, None),
+        (json!({"Retry-After": "later"}), None, None),
+        (json!({}), None, None),
+    ];
+    for (event_headers, error_headers, expected_headers) in cases {
+        let mut error_event = json!({
+            "type": "error",
+            "status": 429,
+            "headers": event_headers,
+            "error": {
+                "type": "rate_limit_error",
+                "code": "turn_rate_limit_exceeded",
+                "message": "Codex is at capacity, please try again later"
+            }
+        });
+        if let Some(headers) = error_headers {
+            error_event["error"]["headers"] = headers;
+        }
+        let projected =
+            response_failed_sse_data_from_error_event(None, Some("resp_retry"), &error_event)
+                .expect("wrapped error event should project");
+
+        // 官方 SSE 解析器仅接收 response，无法回读事件顶层的等待提示
+        assert_eq!(
+            projected.pointer("/response/error/headers"),
+            expected_headers.as_ref(),
+            "upstream event: {error_event}"
+        );
+        assert_eq!(projected["headers"], error_event["headers"]);
+    }
+}
+
 #[test]
 fn done_frame_detection_should_require_exact_single_data_field() {
     assert_eq!(

docs/api.md

@@ -509,6 +509,9 @@ OpenAI HTTP 明确未发送的瞬时建连失败（含 TLS 握手 EOF）在单
 `turn_rate_limit_exceeded` 保留原始错误和 `Retry-After`，由客户端按自身重试预算等待。
 代理不因此写入账号冷却、换号或透明重试，其他会话仍可使用该账号
 
+上游 WS `error` 转为 SSE `response.failed` 时，事件级的有效 `Retry-After` 同时放入
+`response.error.headers`，供客户端读取；错误对象已有有效重试头时优先保留
+
 Responses 的 HTTP/SSE 与 WebSocket 按顺序即时交付正常事件，不暂存前导等待后续输出。
 未启用中间件链时，无输出的 `response.created`、`response.in_progress`
 及 `codex.response.metadata`、`codex.rate_limits` 元数据交付后仍可按已有安全条件重试，