diff --git a/backend/crates/providers/openai/src/provider/failure.rs b/backend/crates/providers/openai/src/provider/failure.rs index 3b1fa7e1a..9c27244d9 100644 --- a/backend/crates/providers/openai/src/provider/failure.rs +++ b/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), diff --git a/backend/crates/providers/openai/src/transport/diagnostics.rs b/backend/crates/providers/openai/src/transport/diagnostics.rs index b5af42ffd..2ee54a28a 100644 --- a/backend/crates/providers/openai/src/transport/diagnostics.rs +++ b/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> { diff --git a/backend/crates/providers/openai/tests/provider/contract/recovery.rs b/backend/crates/providers/openai/tests/provider/contract/recovery.rs index 7d9a225f1..e5b45e427 100644 --- a/backend/crates/providers/openai/tests/provider/contract/recovery.rs +++ b/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::(); + 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::(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] { diff --git a/backend/crates/providers/openai/tests/transport/diagnostics.rs b/backend/crates/providers/openai/tests/transport/diagnostics.rs index 5f087732d..800cb2ba7 100644 --- a/backend/crates/providers/openai/tests/transport/diagnostics.rs +++ b/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] { diff --git a/docs/architecture.md b/docs/architecture.md index 4c53909b9..6ae2882f7 100644 --- a/docs/architecture.md +++ b/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 只保留单向推进的重放边界，实际输出、终态或未知事件关闭重放窗口。 中间件链中的正常事件同样即时交付，但不保留可替换前导的重试资格。