diff --git a/backend/crates/gateway-api/tests/openai/responses/http.rs b/backend/crates/gateway-api/tests/openai/responses/http.rs index 48405478c..9f8c85cc4 100644 --- a/backend/crates/gateway-api/tests/openai/responses/http.rs +++ b/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] diff --git a/backend/crates/gateway-api/tests/openai/responses/mod.rs b/backend/crates/gateway-api/tests/openai/responses/mod.rs index f9c86bc48..2de007cc8 100644 --- a/backend/crates/gateway-api/tests/openai/responses/mod.rs +++ b/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} }), ); diff --git a/backend/crates/gateway-protocol/src/openai/events.rs b/backend/crates/gateway-protocol/src/openai/events.rs index 845dc3363..985ab1f7e 100644 --- a/backend/crates/gateway-protocol/src/openai/events.rs +++ b/backend/crates/gateway-protocol/src/openai/events.rs @@ -199,8 +199,9 @@ pub fn retry_after_seconds_from_value(value: &Value) -> Option { .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 { .filter(|seconds| *seconds > 0) } -fn retry_after_seconds_header(value: &Value) -> Option { +/// 同时保留原始头值与解析秒数,供事件转换和重试观测共用官方头部规则 +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)) } /// 标准化的单个限流窗口 diff --git a/backend/crates/gateway-protocol/src/openai/sse.rs b/backend/crates/gateway-protocol/src/openai/sse.rs index e5e151f59..aa92566fd 100644 --- a/backend/crates/gateway-protocol/src/openai/sse.rs +++ b/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 { 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() diff --git a/backend/crates/gateway-protocol/tests/openai/sse.rs b/backend/crates/gateway-protocol/tests/openai/sse.rs index 649db3028..fbbf43b01 100644 --- a/backend/crates/gateway-protocol/tests/openai/sse.rs +++ b/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!( diff --git a/docs/api.md b/docs/api.md index f9be45736..1771a9cbc 100644 --- a/docs/api.md +++ b/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` 元数据交付后仍可按已有安全条件重试,