From 5c9da8a39dbb4fc1d87f12cd272b896377851fae Mon Sep 17 00:00:00 2001 From: Zyy Date: Fri, 31 Jul 2026 15:32:32 +0800 Subject: [PATCH] fix(openai): preserve upstream failures during fallback Co-authored-by: Codex --- .../crates/gateway-api/src/openai/error.rs | 27 ++- .../crates/gateway-api/tests/openai/error.rs | 75 +++++-- .../gateway-core/src/engine/coordinator.rs | 48 ++--- backend/crates/gateway-core/src/error.rs | 20 ++ .../gateway-core/tests/engine/coordinator.rs | 127 +++++++++++- .../crates/providers/openai/src/provider.rs | 119 ++++++++++-- .../providers/openai/src/transport/client.rs | 2 + .../openai/src/transport/client_sse.rs | 183 ++++++++++++++---- .../src/transport/protocol/responses.rs | 5 + .../openai/src/transport/websocket/error.rs | 87 ++++++++- .../src/transport/websocket/exchange/io.rs | 7 +- .../transport/websocket/exchange/stream.rs | 15 +- .../openai/src/transport/websocket/mod.rs | 2 +- .../openai/src/transport/websocket/pump.rs | 42 +++- .../openai/tests/transport/latency.rs | 146 +++++++++++++- .../tests/transport/protocol/responses.rs | 22 +++ .../openai/tests/transport/websocket.rs | 10 +- 17 files changed, 804 insertions(+), 133 deletions(-) diff --git a/backend/crates/gateway-api/src/openai/error.rs b/backend/crates/gateway-api/src/openai/error.rs index 8cc924f23..e2151359b 100644 --- a/backend/crates/gateway-api/src/openai/error.rs +++ b/backend/crates/gateway-api/src/openai/error.rs @@ -7,8 +7,8 @@ use axum::{ response::{IntoResponse, Response}, }; use gateway_core::{ - engine::{EngineError, UpstreamSendState}, - error::{GatewayError, GatewayErrorKind, ProviderErrorKind}, + engine::EngineError, + error::{GatewayError, GatewayErrorKind}, }; use serde_json::{Value, json}; @@ -72,19 +72,6 @@ pub fn model_not_found_response() -> (StatusCode, Json) { #[must_use] pub fn gateway_error_from_engine(error: &EngineError) -> GatewayError { match error { - EngineError::Provider(error) - if error.kind() == ProviderErrorKind::Unavailable - && error.send_state() == UpstreamSendState::NotSent => - { - let gateway = GatewayError::new( - GatewayErrorKind::NoAvailableProvider, - "no upstream provider is currently available for this request", - ); - match error.client_visible_upstream_error() { - Some(upstream) => gateway.with_client_visible_upstream_error(upstream.clone()), - None => gateway, - } - } EngineError::Provider(error) => GatewayError::from_provider(error), EngineError::Cancelled => { GatewayError::new(GatewayErrorKind::Cancelled, "request was cancelled") @@ -185,6 +172,16 @@ pub const fn gateway_error_contract( "server_error", "no_available_provider", ), + GatewayErrorKind::AccountCapacityUnavailable => ( + StatusCode::SERVICE_UNAVAILABLE, + "server_error", + "account_capacity_unavailable", + ), + GatewayErrorKind::ProviderInfrastructureUnavailable => ( + StatusCode::SERVICE_UNAVAILABLE, + "server_error", + "provider_infrastructure_unavailable", + ), GatewayErrorKind::RateLimited => ( StatusCode::TOO_MANY_REQUESTS, "rate_limit_error", diff --git a/backend/crates/gateway-api/tests/openai/error.rs b/backend/crates/gateway-api/tests/openai/error.rs index 7e59ab659..bce7330d9 100644 --- a/backend/crates/gateway-api/tests/openai/error.rs +++ b/backend/crates/gateway-api/tests/openai/error.rs @@ -58,6 +58,30 @@ fn no_available_provider_error_should_map_to_service_unavailable() { ); } +#[test] +fn account_capacity_error_should_map_to_service_unavailable() { + assert_eq!( + gateway_error_contract(GatewayErrorKind::AccountCapacityUnavailable), + ( + StatusCode::SERVICE_UNAVAILABLE, + "server_error", + "account_capacity_unavailable", + ) + ); +} + +#[test] +fn provider_infrastructure_error_should_map_to_service_unavailable() { + assert_eq!( + gateway_error_contract(GatewayErrorKind::ProviderInfrastructureUnavailable), + ( + StatusCode::SERVICE_UNAVAILABLE, + "server_error", + "provider_infrastructure_unavailable", + ) + ); +} + #[test] fn rate_limited_error_should_map_to_openai_retryable_status() { assert_eq!( @@ -122,29 +146,32 @@ fn engine_provider_error_should_preserve_retry_classification() { #[test] fn local_provider_capacity_exhaustion_should_map_to_service_unavailable() { let error = EngineError::Provider(ProviderError::new( - ProviderErrorKind::Unavailable, + ProviderErrorKind::AccountCapacityUnavailable, UpstreamSendState::NotSent, )); assert_eq!( gateway_error_from_engine(&error), GatewayError::new( - GatewayErrorKind::NoAvailableProvider, - "no upstream provider is currently available for this request" + GatewayErrorKind::AccountCapacityUnavailable, + "all eligible upstream accounts are temporarily busy" ) ); } #[test] -fn local_provider_capacity_exhaustion_should_preserve_safe_provider_detail() { +fn no_eligible_provider_account_should_preserve_safe_provider_detail() { let detail = ClientVisibleUpstreamError::new( "no account is eligible for the requested model", Some("no_eligible_account".to_owned()), Some("account_unavailable_error".to_owned()), ); let error = EngineError::Provider( - ProviderError::new(ProviderErrorKind::Unavailable, UpstreamSendState::NotSent) - .with_client_visible_upstream_error(detail), + ProviderError::new( + ProviderErrorKind::NoEligibleAccount, + UpstreamSendState::NotSent, + ) + .with_client_visible_upstream_error(detail), ); let gateway = gateway_error_from_engine(&error); @@ -164,6 +191,22 @@ fn local_provider_capacity_exhaustion_should_preserve_safe_provider_detail() { ); } +#[test] +fn provider_infrastructure_failure_should_have_a_distinct_client_contract() { + let error = EngineError::Provider(ProviderError::new( + ProviderErrorKind::ProviderInfrastructureUnavailable, + UpstreamSendState::NotSent, + )); + + assert_eq!( + gateway_error_from_engine(&error), + GatewayError::new( + GatewayErrorKind::ProviderInfrastructureUnavailable, + "provider account infrastructure is temporarily unavailable" + ) + ); +} + #[test] fn no_eligible_provider_account_should_map_to_service_unavailable() { let error = EngineError::Provider(ProviderError::new( @@ -181,16 +224,18 @@ fn no_eligible_provider_account_should_map_to_service_unavailable() { } #[test] -fn sent_provider_unavailability_should_remain_bad_gateway() { - let error = EngineError::Provider(ProviderError::new( - ProviderErrorKind::Unavailable, - UpstreamSendState::Sent, - )); +fn provider_unavailability_should_remain_bad_gateway_regardless_of_send_state() { + for send_state in [UpstreamSendState::NotSent, UpstreamSendState::Sent] { + let error = EngineError::Provider(ProviderError::new( + ProviderErrorKind::Unavailable, + send_state, + )); - assert_eq!( - gateway_error_from_engine(&error).kind(), - GatewayErrorKind::UpstreamUnavailable - ); + assert_eq!( + gateway_error_from_engine(&error).kind(), + GatewayErrorKind::UpstreamUnavailable + ); + } } #[test] diff --git a/backend/crates/gateway-core/src/engine/coordinator.rs b/backend/crates/gateway-core/src/engine/coordinator.rs index 5515611a8..08e99b733 100644 --- a/backend/crates/gateway-core/src/engine/coordinator.rs +++ b/backend/crates/gateway-core/src/engine/coordinator.rs @@ -165,8 +165,8 @@ where timings: ModelRequestTimings::default(), client_response_id: None, upstream_response_id: None, - last_account_exhaustion: None, - last_account_exhaustion_events: Vec::new(), + last_retryable_failure: None, + last_retryable_failure_events: Vec::new(), provider_attempt_outcomes: Vec::new(), pending_terminal_failure: None, }; @@ -245,9 +245,10 @@ pub struct ResponseExecutionSession { timings: ModelRequestTimings, client_response_id: Option, upstream_response_id: Option, - last_account_exhaustion: Option, - /// 最近一次账号耗尽在重试时被丢弃的原始失败批次；只用于后续空选路终态。 - last_account_exhaustion_events: Vec, + /// 最近一次为无感恢复而被丢弃的原始上游错误；只在后续空选路时成为终态。 + last_retryable_failure: Option, + /// 与 `last_retryable_failure` 同属一个 attempt 的原始失败批次。 + last_retryable_failure_events: Vec, provider_attempt_outcomes: Vec, /// 原子失败批次已交给协议层、但尚待下游提交后收敛的原 Provider 错误。 pending_terminal_failure: Option, @@ -659,25 +660,30 @@ where ProviderBoundary::Result(result) => match *result { Ok(stream) => stream, Err(error) => { - if error.kind() == ProviderErrorKind::NoEligibleAccount - && let Some(exhaustion) = self.last_account_exhaustion.take() + if matches!( + error.kind(), + ProviderErrorKind::AccountCapacityUnavailable + | ProviderErrorKind::NoEligibleAccount + ) && let Some(last_failure) = self.last_retryable_failure.take() { let send_state = self.current_send_state(); - let mut events = std::mem::take(&mut self.last_account_exhaustion_events); + let mut events = std::mem::take(&mut self.last_retryable_failure_events); if !events.is_empty() { self.observe_atomic_terminal_events(&mut events); return Ok(Some(PullOutcome::TerminalFailure { events, - error: exhaustion, + error: last_failure, send_state, })); } - self.finish_provider_error(&exhaustion).await?; - return Err(provider_engine_error(exhaustion)); + self.finish_provider_error(&last_failure).await?; + return Err(provider_engine_error(last_failure)); } if !(matches!( error.kind(), - ProviderErrorKind::Unavailable | ProviderErrorKind::NoEligibleAccount + ProviderErrorKind::AccountCapacityUnavailable + | ProviderErrorKind::NoEligibleAccount + | ProviderErrorKind::ProviderInfrastructureUnavailable ) && error.send_state() == UpstreamSendState::NotSent) { self.record_provider_failure(candidate.provider().clone(), error.kind()); @@ -909,14 +915,10 @@ where &mut self, mut error: ProviderError, ) -> Result { - // 原始 wire 只能活在本次决策栈内；后续 clone、attempt 记录与终态对象均只 - // 接触已剥离的稳定错误字段。 + // 原始 wire 只活在 request-local 决策状态；clone、attempt 记录与持久化终态 + // 均只接触已剥离的稳定错误字段。 let mut atomic_client_events = error.take_atomic_client_events(); let current = self.current.take().ok_or(EngineError::NoActiveAttempt)?; - let account_exhausted = matches!( - error.kind(), - ProviderErrorKind::RateLimited | ProviderErrorKind::QuotaExhausted - ); self.record_provider_failure(current.metadata.provider().clone(), error.kind()); // attempt_send_state 是本 attempt 自身的发送事实，驱动重试门； // 持久化与终态用请求级水位，二者不可混用（水位会把早先 attempt 的 @@ -976,8 +978,8 @@ where continuation_retry || same_account_retry || ordinary_retry || pre_delivery_retry; if retryable { - // 普通 clone 只保留稳定事实；原始失败响应只能随最后一次客户端可见 - // 的账号耗尽错误继续，不能进入中间失败记录。 + // 普通 clone 只保留稳定事实；原始 wire/HTTP response 由 request-local + // 所有权保留到下一次 attempt 成功，或最终空选路时返回客户端。 let persistence_error = self.request_persisted.then(|| error.clone()); if same_account_retry { if let Some(account) = current.metadata.provider_account_id() { @@ -992,10 +994,8 @@ where { self.excluded_accounts.insert(account.clone()); } - if account_exhausted { - self.last_account_exhaustion_events = atomic_client_events; - self.last_account_exhaustion = Some(error); - } + self.last_retryable_failure_events = atomic_client_events; + self.last_retryable_failure = Some(error); if let Some(error) = persistence_error { best_effort_store_write( "record_intermediate_failure", diff --git a/backend/crates/gateway-core/src/error.rs b/backend/crates/gateway-core/src/error.rs index d0f3ae1fe..9d948ae83 100644 --- a/backend/crates/gateway-core/src/error.rs +++ b/backend/crates/gateway-core/src/error.rs @@ -129,8 +129,12 @@ pub enum ProviderErrorKind { RateLimited, /// Credential 配额耗尽。 QuotaExhausted, + /// 仍有符合条件的账号，但它们暂时没有可调度容量。 + AccountCapacityUnavailable, /// Provider 已确认当前请求无法选出可用账号。 NoEligibleAccount, + /// Provider 的账号存储、租约协调或本地凭据数据不可用。 + ProviderInfrastructureUnavailable, /// 请求超时。 Timeout, /// 网络或 transport 失败。 @@ -156,7 +160,9 @@ impl ProviderErrorKind { Self::PermissionDenied => "permission_denied", Self::RateLimited => "rate_limited", Self::QuotaExhausted => "quota_exhausted", + Self::AccountCapacityUnavailable => "account_capacity_unavailable", Self::NoEligibleAccount => "no_eligible_account", + Self::ProviderInfrastructureUnavailable => "provider_infrastructure_unavailable", Self::Timeout => "timeout", Self::Transport => "transport", Self::Protocol => "protocol", @@ -711,6 +717,10 @@ pub enum GatewayErrorKind { ModelNotFound, /// 当前没有可用 target。 NoAvailableProvider, + /// 符合条件的上游账号暂时没有调度容量。 + AccountCapacityUnavailable, + /// Provider 的本地账号基础设施不可用。 + ProviderInfrastructureUnavailable, /// 上游限流。 RateLimited, /// 上游暂不可用。 @@ -734,6 +744,8 @@ impl GatewayErrorKind { Self::PolicyDenied => "policy_denied", Self::ModelNotFound => "model_not_found", Self::NoAvailableProvider => "no_available_provider", + Self::AccountCapacityUnavailable => "account_capacity_unavailable", + Self::ProviderInfrastructureUnavailable => "provider_infrastructure_unavailable", Self::RateLimited => "rate_limited", Self::UpstreamUnavailable => "upstream_unavailable", Self::Timeout => "timeout", @@ -782,10 +794,18 @@ impl GatewayError { GatewayErrorKind::RateLimited, "upstream capacity is temporarily unavailable", ), + ProviderErrorKind::AccountCapacityUnavailable => Self::new( + GatewayErrorKind::AccountCapacityUnavailable, + "all eligible upstream accounts are temporarily busy", + ), ProviderErrorKind::NoEligibleAccount => Self::new( GatewayErrorKind::NoAvailableProvider, "no upstream provider is currently available for this request", ), + ProviderErrorKind::ProviderInfrastructureUnavailable => Self::new( + GatewayErrorKind::ProviderInfrastructureUnavailable, + "provider account infrastructure is temporarily unavailable", + ), ProviderErrorKind::Timeout => { Self::new(GatewayErrorKind::Timeout, "upstream request timed out") } diff --git a/backend/crates/gateway-core/tests/engine/coordinator.rs b/backend/crates/gateway-core/tests/engine/coordinator.rs index 4e6bdd496..015ad3338 100644 --- a/backend/crates/gateway-core/tests/engine/coordinator.rs +++ b/backend/crates/gateway-core/tests/engine/coordinator.rs @@ -29,8 +29,9 @@ use gateway_core::engine::{ ProviderAttemptOutcome, RecoveryReport, UpstreamSendState, }; use gateway_core::error::{ - ClientVisibleUpstreamResponse, ContinuationFailure, GatewayErrorKind, OpaqueUpstreamValue, - ProviderError, ProviderErrorKind, StoreError, StoreErrorKind, + ClientVisibleUpstreamError, ClientVisibleUpstreamResponse, ContinuationFailure, + GatewayErrorKind, OpaqueUpstreamValue, ProviderError, ProviderErrorKind, StoreError, + StoreErrorKind, }; use gateway_core::event::{ ContentItem, ContentKind, GatewayEvent, ProtocolWireEvent, ProviderEvent, @@ -2554,6 +2555,128 @@ fn pre_delivery_transport_failure_rotates_account_for_non_idempotent_generation( assert_eq!(state.finalizations[0].outcome, ExecutionOutcome::Succeeded); } +#[test] +fn final_capacity_exhaustion_returns_the_last_retryable_upstream_failure() { + let operation = operation(RetrySafety::NonIdempotent); + let route_plan = plan(&operation); + let raw_body = Bytes::from_static(b"{\"error\":\"last upstream body\"}\x00"); + let upstream_error = + ProviderError::new(ProviderErrorKind::Transport, UpstreamSendState::Ambiguous) + .with_pre_delivery_retry() + .with_client_visible_upstream_error(ClientVisibleUpstreamError::new( + "message too big", + Some("1009".to_owned()), + Some("websocket_close_error".to_owned()), + )) + .with_client_visible_upstream_response(ClientVisibleUpstreamResponse::new( + 502, + Some(b"application/json".to_vec()), + raw_body.clone(), + )); + let (coordinator, store, provider) = coordinator(vec![ + Script::Stream { + account_id: "acct_first", + items: vec![Err(upstream_error)], + }, + Script::Error(ProviderError::new( + ProviderErrorKind::AccountCapacityUnavailable, + UpstreamSendState::NotSent, + )), + ]); + + let mut session = block_on(coordinator.start( + model_request(&operation, SystemTime::now() + Duration::from_secs(30)), + operation, + route_plan, + None, + None, + CancellationToken::new(), + )) + .expect("start execution"); + let error = block_on(session.collect_uncommitted()) + .expect_err("empty selection must return the retained upstream failure"); + let EngineError::Provider(error) = error else { + panic!("expected retained provider error"); + }; + + assert_eq!(error.kind(), ProviderErrorKind::Transport); + let detail = error + .client_visible_upstream_error() + .expect("structured upstream detail"); + assert_eq!(detail.message(), "message too big"); + assert_eq!(detail.code(), Some("1009")); + assert_eq!(detail.error_type(), Some("websocket_close_error")); + let response = error + .client_visible_upstream_response() + .expect("original upstream response"); + assert_eq!(response.status(), 502); + assert_eq!(response.body(), &raw_body); + + assert_eq!(provider.contexts.lock().expect("contexts lock").len(), 2); + let state = store.state.lock().expect("store lock"); + assert_eq!(state.intermediate_failures, 1); + assert_eq!(state.finalizations[0].outcome, ExecutionOutcome::Failed); + assert_eq!( + state.finalizations[0].send_state, + UpstreamSendState::Ambiguous + ); +} + +#[test] +fn provider_infrastructure_failure_does_not_restore_the_last_upstream_failure() { + let operation = operation(RetrySafety::NonIdempotent); + let route_plan = plan(&operation); + let upstream_error = + ProviderError::new(ProviderErrorKind::Transport, UpstreamSendState::Ambiguous) + .with_pre_delivery_retry() + .with_client_visible_upstream_error(ClientVisibleUpstreamError::new( + "upstream marker", + Some("upstream_marker".to_owned()), + Some("upstream_error".to_owned()), + )) + .with_client_visible_upstream_response(ClientVisibleUpstreamResponse::new( + 502, + Some(b"application/json".to_vec()), + Bytes::from_static(b"{\"error\":\"upstream marker\"}"), + )); + let (coordinator, store, provider) = coordinator(vec![ + Script::Stream { + account_id: "acct_first", + items: vec![Err(upstream_error)], + }, + Script::Error(ProviderError::new( + ProviderErrorKind::ProviderInfrastructureUnavailable, + UpstreamSendState::NotSent, + )), + ]); + + let mut session = block_on(coordinator.start( + model_request(&operation, SystemTime::now() + Duration::from_secs(30)), + operation, + route_plan, + None, + None, + CancellationToken::new(), + )) + .expect("start execution"); + let error = block_on(session.collect_uncommitted()) + .expect_err("provider infrastructure failure must remain terminal"); + let EngineError::Provider(error) = error else { + panic!("expected provider error"); + }; + + assert_eq!( + error.kind(), + ProviderErrorKind::ProviderInfrastructureUnavailable + ); + assert!(error.client_visible_upstream_error().is_none()); + assert!(error.client_visible_upstream_response().is_none()); + assert_eq!(provider.contexts.lock().expect("contexts lock").len(), 2); + let state = store.state.lock().expect("store lock"); + assert_eq!(state.intermediate_failures, 1); + assert_eq!(state.finalizations[0].outcome, ExecutionOutcome::Failed); +} + #[test] fn pre_delivery_retry_marker_is_ignored_after_downstream_commit() { let operation = operation(RetrySafety::NonIdempotent); diff --git a/backend/crates/providers/openai/src/provider.rs b/backend/crates/providers/openai/src/provider.rs index 499c98e9b..858fdc0aa 100644 --- a/backend/crates/providers/openai/src/provider.rs +++ b/backend/crates/providers/openai/src/provider.rs @@ -1767,7 +1767,10 @@ fn map_request_error(error: CodexRequestEncodeError) -> ProviderError { fn map_selection_error(error: CredentialSelectionError) -> ProviderError { match error { CredentialSelectionError::CapacityUnavailable { retry_after } => { - let error = provider_error(ProviderErrorKind::Unavailable, UpstreamSendState::NotSent); + let error = provider_error( + ProviderErrorKind::AccountCapacityUnavailable, + UpstreamSendState::NotSent, + ); match retry_after { Some(retry) => error.with_retry_after(retry), None => error, @@ -1780,9 +1783,10 @@ fn map_selection_error(error: CredentialSelectionError) -> ProviderError { CredentialSelectionError::InvalidCredential | CredentialSelectionError::Store | CredentialSelectionError::Coordinator - | CredentialSelectionError::CookiePolicy => { - provider_error(ProviderErrorKind::Unavailable, UpstreamSendState::NotSent) - } + | CredentialSelectionError::CookiePolicy => provider_error( + ProviderErrorKind::ProviderInfrastructureUnavailable, + UpstreamSendState::NotSent, + ), } } @@ -2063,7 +2067,7 @@ fn stream_transport_allows_pre_delivery_retry(error: &CodexClientError) -> bool CodexWebSocketExchangeError::Transport(_) | CodexWebSocketExchangeError::PostSendAmbiguous { .. } | CodexWebSocketExchangeError::SendTimeout { .. } - | CodexWebSocketExchangeError::ClosedBeforeTerminal + | CodexWebSocketExchangeError::ClosedBeforeTerminal(_) | CodexWebSocketExchangeError::ReceiveIdleTimeout { .. } | CodexWebSocketExchangeError::ReusedConnectionDiedBeforeFirstEvent { .. } | CodexWebSocketExchangeError::InitialEventTimeout { .. } @@ -2151,10 +2155,19 @@ fn map_client_error( send_state, )) } - CodexClientError::WebSocket(error) => MappedProviderFailure::plain(provider_error( - websocket_error_kind(&error), - websocket_send_state(&error), - )), + CodexClientError::WebSocket(error) => { + let client_visible_error = websocket_client_visible_error(&error); + let mut failure = MappedProviderFailure::plain(provider_error( + websocket_error_kind(&error), + websocket_send_state(&error), + )); + if let Some(client_visible_error) = client_visible_error { + failure.error = failure + .error + .with_client_visible_upstream_error(client_visible_error); + } + failure + } }; if let Some(continuation_failure) = continuation_failure { failure.error = failure @@ -2165,6 +2178,21 @@ fn map_client_error( failure } +fn websocket_client_visible_error( + error: &CodexWebSocketExchangeError, +) -> Option { + let close = error.close_before_terminal()?; + let message = close + .reason() + .filter(|reason| !reason.is_empty()) + .map_or_else(|| close.to_string(), str::to_owned); + Some(ClientVisibleUpstreamError::new( + message, + close.code().map(|code| code.to_string()), + Some("websocket_close_error".to_owned()), + )) +} + fn map_upstream_failure( mut failure: CodexUpstreamFailure, observation: Option, @@ -2333,7 +2361,7 @@ const fn websocket_send_state(error: &CodexWebSocketExchangeError) -> UpstreamSe CodexWebSocketExchangeError::Transport(_) | CodexWebSocketExchangeError::PostSendAmbiguous { .. } | CodexWebSocketExchangeError::SendTimeout { .. } - | CodexWebSocketExchangeError::ClosedBeforeTerminal + | CodexWebSocketExchangeError::ClosedBeforeTerminal(_) | CodexWebSocketExchangeError::ReceiveIdleTimeout { .. } | CodexWebSocketExchangeError::ReusedConnectionDiedBeforeFirstEvent { .. } | CodexWebSocketExchangeError::InitialEventTimeout { .. } => UpstreamSendState::Ambiguous, @@ -2360,7 +2388,7 @@ const fn websocket_error_kind(error: &CodexWebSocketExchangeError) -> ProviderEr CodexWebSocketExchangeError::Transport(_) | CodexWebSocketExchangeError::Connect(_) | CodexWebSocketExchangeError::PostSendAmbiguous { .. } - | CodexWebSocketExchangeError::ClosedBeforeTerminal + | CodexWebSocketExchangeError::ClosedBeforeTerminal(_) | CodexWebSocketExchangeError::ReusedConnectionDiedBeforeFirstEvent { .. } => { ProviderErrorKind::Transport } @@ -2619,3 +2647,72 @@ impl DaemonTask for OpenAiCatalogEtagTask { }) } } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn capacity_selection_error_preserves_classification_and_retry_after() { + let retry_after = Duration::from_secs(7); + let error = map_selection_error(CredentialSelectionError::CapacityUnavailable { + retry_after: Some(retry_after), + }); + + assert_eq!( + (error.kind(), error.send_state(), error.retry_after()), + ( + ProviderErrorKind::AccountCapacityUnavailable, + UpstreamSendState::NotSent, + Some(retry_after), + ) + ); + } + + #[test] + fn selection_infrastructure_errors_have_a_distinct_classification() { + for selection_error in [ + CredentialSelectionError::InvalidCredential, + CredentialSelectionError::Store, + CredentialSelectionError::Coordinator, + CredentialSelectionError::CookiePolicy, + ] { + let error = map_selection_error(selection_error); + + assert_eq!( + (error.kind(), error.send_state()), + ( + ProviderErrorKind::ProviderInfrastructureUnavailable, + UpstreamSendState::NotSent, + ) + ); + } + } + + #[test] + fn websocket_close_details_are_only_exposed_through_the_client_error() { + let close = CodexWebSocketExchangeError::closed_before_terminal( + Some(1009), + Some("message too big".to_owned()), + ); + assert!(!format!("{close:?}").contains("message too big")); + assert!(!close.to_string().contains("message too big")); + + let failure = map_stream_error(CodexClientError::WebSocket( + CodexWebSocketExchangeError::PostSendAmbiguous { + message: close.to_string(), + source: Some(Box::new(close)), + }, + )); + assert_eq!(failure.error.kind(), ProviderErrorKind::Transport); + assert_eq!(failure.error.send_state(), UpstreamSendState::Ambiguous); + assert!(failure.error.allows_pre_delivery_retry()); + let detail = failure + .error + .client_visible_upstream_error() + .expect("WebSocket close detail"); + assert_eq!(detail.message(), "message too big"); + assert_eq!(detail.code(), Some("1009")); + assert_eq!(detail.error_type(), Some("websocket_close_error")); + } +} diff --git a/backend/crates/providers/openai/src/transport/client.rs b/backend/crates/providers/openai/src/transport/client.rs index fddb49467..3c8cfb943 100644 --- a/backend/crates/providers/openai/src/transport/client.rs +++ b/backend/crates/providers/openai/src/transport/client.rs @@ -477,6 +477,7 @@ pub enum CodexTransportDecision { Http2BreakerOpen, Http2PoolUnavailable, Http2PreSendFailure, + Http2PreDeliveryFailure, } impl CodexTransportDecision { @@ -491,6 +492,7 @@ impl CodexTransportDecision { Self::Http2BreakerOpen => "http2_breaker_open", Self::Http2PoolUnavailable => "http2_pool_unavailable", Self::Http2PreSendFailure => "http2_ws_pre_send_failure", + Self::Http2PreDeliveryFailure => "http2_ws_pre_delivery_failure", } } } diff --git a/backend/crates/providers/openai/src/transport/client_sse.rs b/backend/crates/providers/openai/src/transport/client_sse.rs index 885637b66..01f966eba 100644 --- a/backend/crates/providers/openai/src/transport/client_sse.rs +++ b/backend/crates/providers/openai/src/transport/client_sse.rs @@ -32,10 +32,11 @@ use crate::transport::{ }, response_meta, websocket::{ - CodexWebSocketConnection, CodexWebSocketPool, CodexWebSocketPoolKey, - DEFAULT_INITIAL_EVENT_TIMEOUT, WEBSOCKET_FAST_PATH_BUDGET, WebSocketOriginBreaker, - WebSocketPoolDecision, execute_prepared_response_create_request_stream, - post_send_ambiguous, prepare_response_create_request_with_pool, websocket_audit_dir, + CodexWebSocketConnection, CodexWebSocketExchangeError, CodexWebSocketPool, + CodexWebSocketPoolKey, CodexWebSocketStreamingExchange, DEFAULT_INITIAL_EVENT_TIMEOUT, + WEBSOCKET_FAST_PATH_BUDGET, WebSocketOriginBreaker, WebSocketPoolDecision, + execute_prepared_response_create_request_stream, post_send_ambiguous, + prepare_response_create_request_with_pool, websocket_audit_dir, write_websocket_audit_artifact_from_env, }, }; @@ -308,7 +309,7 @@ impl CodexBackendClient { route, metrics, } = prepared; - let result = match route { + match route { PreparedResponseRoute::Http => self .create_response_stream_http_sse(request, context) .await @@ -317,50 +318,124 @@ impl CodexBackendClient { response }), PreparedResponseRoute::WebSocket(route) => { - let PreparedWebSocketRoute { request, prepared } = *route; - execute_prepared_response_create_request_stream(&request, prepared) - .await - .map_err(websocket_exchange_error_to_client_error) - .map(|exchange| { - tracing::info!( - request_id = %context.request_id, - websocket_connection_id = %exchange.websocket_connection_id, - ws_pool = exchange.pool_decision.map_or("unpooled", WebSocketPoolDecision::kind), - "WebSocket response stream established" - ); - CodexBackendStreamingResponse { - body: Box::pin( - exchange - .body - .map_err(post_send_ambiguous) - .map_err(websocket_exchange_error_to_client_error), - ), - transport: CodexBackendTransport::WebSocket, - turn_state: exchange.turn_state, - set_cookie_headers: exchange.set_cookie_headers, - rate_limit_headers: exchange.rate_limit_headers, - rate_limit_header_updates: Some(exchange.rate_limit_header_updates), - turn_state_update: Some(exchange.turn_state_update), - websocket_pool_decision: exchange.pool_decision, - diagnostics: exchange.diagnostics, - response_metadata: exchange.response_metadata, - transport_metrics: metrics, - connection_local_continuation: exchange.connection_local_continuation, - } - }) + let PreparedWebSocketRoute { + request: websocket_request, + prepared, + } = *route; + let delivery_wait_started_at = Instant::now(); + let mut exchange = match execute_prepared_response_create_request_stream( + &websocket_request, + prepared, + ) + .await + { + Ok(exchange) => exchange, + Err(error) + if requirement.allows_pre_delivery_http_fallback() + && error.allows_pre_delivery_http_fallback() => + { + return self + .fallback_to_http_before_websocket_delivery( + request, + context, + requirement, + metrics, + delivery_wait_started_at, + error, + ) + .await; + } + Err(error) => return Err(websocket_exchange_error_to_client_error(error)), + }; + if requirement.allows_pre_delivery_http_fallback() + && let Err(error) = await_websocket_delivery_boundary(&mut exchange).await + { + if error.allows_pre_delivery_http_fallback() { + return self + .fallback_to_http_before_websocket_delivery( + request, + context, + requirement, + metrics, + delivery_wait_started_at, + error, + ) + .await; + } + return Err(websocket_exchange_error_to_client_error( + post_send_ambiguous(error), + )); + } + tracing::info!( + request_id = %context.request_id, + websocket_connection_id = %exchange.websocket_connection_id, + ws_pool = exchange.pool_decision.map_or("unpooled", WebSocketPoolDecision::kind), + "WebSocket response stream established" + ); + Ok(CodexBackendStreamingResponse { + body: Box::pin( + exchange + .body + .map_err(post_send_ambiguous) + .map_err(websocket_exchange_error_to_client_error), + ), + transport: CodexBackendTransport::WebSocket, + turn_state: exchange.turn_state, + set_cookie_headers: exchange.set_cookie_headers, + rate_limit_headers: exchange.rate_limit_headers, + rate_limit_header_updates: Some(exchange.rate_limit_header_updates), + turn_state_update: Some(exchange.turn_state_update), + websocket_pool_decision: exchange.pool_decision, + diagnostics: exchange.diagnostics, + response_metadata: exchange.response_metadata, + transport_metrics: metrics, + connection_local_continuation: exchange.connection_local_continuation, + }) } - }; - result.inspect_err(|error| { + } + .inspect_err(|error| { tracing::warn!( request_id = %context.request_id, transport_requirement = requirement.as_str(), failure_phase = "post_send_or_explicit_response", error = %error, - "Responses stream transport failed after preparation; automatic fallback is disabled" + "Responses stream transport failed after preparation" ); }) } + async fn fallback_to_http_before_websocket_delivery( + &self, + request: &CodexResponsesRequest, + context: CodexRequestContext<'_>, + requirement: TransportRequirement, + mut metrics: CodexTransportMetrics, + delivery_wait_started_at: Instant, + error: CodexWebSocketExchangeError, + ) -> CodexClientResult { + let decision_wait_ms = metrics + .transport_decision_wait_ms + .unwrap_or_default() + .saturating_add(elapsed_duration_millis(delivery_wait_started_at.elapsed())); + metrics.decision = Some(CodexTransportDecision::Http2PreDeliveryFailure); + metrics.transport_decision_wait_ms = Some(decision_wait_ms); + tracing::warn!( + request_id = %context.request_id, + account_id = context.account_id.unwrap_or_default(), + transport_requirement = requirement.as_str(), + transport_decision = CodexTransportDecision::Http2PreDeliveryFailure.as_str(), + transport_decision_wait_ms = decision_wait_ms, + error = %error, + "WebSocket failed before first deliverable event; using same-account HTTP" + ); + self.create_response_stream_http_sse(request, context) + .await + .map(|mut response| { + merge_preparation_metrics(&mut response.transport_metrics, metrics); + response + }) + } + fn websocket_pool_key( &self, request: &CodexResponsesRequest, @@ -563,6 +638,36 @@ impl CodexBackendClient { } } +async fn await_websocket_delivery_boundary( + exchange: &mut CodexWebSocketStreamingExchange, +) -> Result<(), CodexWebSocketExchangeError> { + let mut prelude = Vec::new(); + loop { + match exchange.body.next().await { + Some(Ok(frame)) if is_websocket_lifecycle_prelude(&frame) => prelude.push(frame), + Some(Ok(frame)) => { + prelude.push(frame); + let remaining = + std::mem::replace(&mut exchange.body, Box::pin(futures::stream::empty())); + exchange.body = + Box::pin(futures::stream::iter(prelude.into_iter().map(Ok)).chain(remaining)); + return Ok(()); + } + Some(Err(error)) => return Err(error), + None => { + return Err(CodexWebSocketExchangeError::closed_before_terminal( + None, None, + )); + } + } + } +} + +fn is_websocket_lifecycle_prelude(frame: &[u8]) -> bool { + frame.starts_with(b"event: response.created\n") + || frame.starts_with(b"event: response.in_progress\n") +} + async fn read_model_catalog_body(response: ReqwestResponse) -> CodexClientResult> { if response .content_length() diff --git a/backend/crates/providers/openai/src/transport/protocol/responses.rs b/backend/crates/providers/openai/src/transport/protocol/responses.rs index 0521eac73..3dcb9ee90 100644 --- a/backend/crates/providers/openai/src/transport/protocol/responses.rs +++ b/backend/crates/providers/openai/src/transport/protocol/responses.rs @@ -139,6 +139,11 @@ impl TransportRequirement { ) } + /// WebSocket 已发送 payload、但尚未交付业务事件时，是否允许同账号 HTTP 重放。 + pub fn allows_pre_delivery_http_fallback(self) -> bool { + matches!(self, Self::NewChain) + } + /// 用于审计与遥测的稳定名称。 pub fn as_str(self) -> &'static str { match self { diff --git a/backend/crates/providers/openai/src/transport/websocket/error.rs b/backend/crates/providers/openai/src/transport/websocket/error.rs index 425576ebc..af5d60776 100644 --- a/backend/crates/providers/openai/src/transport/websocket/error.rs +++ b/backend/crates/providers/openai/src/transport/websocket/error.rs @@ -71,8 +71,8 @@ pub enum CodexWebSocketExchangeError { reason: PreviousResponseUnavailableReason, }, /// 上游在 terminal 事件前关闭。 - #[error("websocket closed before terminal event")] - ClosedBeforeTerminal, + #[error("{0}")] + ClosedBeforeTerminal(CodexWebSocketCloseError), /// 上游在指定时间内没有发送任何事件。 #[error("websocket receive idle timeout after {timeout:?}")] ReceiveIdleTimeout { @@ -87,6 +87,9 @@ pub enum CodexWebSocketExchangeError { ReusedConnectionDiedBeforeFirstEvent { /// 底层失效原因。 message: String, + /// 原始 typed transport failure。 + #[source] + source: Option>, }, /// 建连并发送后，上游在配置时间内没有产生任何事件。 #[error("websocket first upstream event not received within {timeout:?}")] @@ -96,6 +99,55 @@ pub enum CodexWebSocketExchangeError { }, } +/// 上游在 terminal 事件前发送的 WebSocket close 信息。 +/// +/// close reason 只供原请求的客户端错误响应使用；`Debug` 与 `Display` 均不会输出它。 +#[derive(Clone, PartialEq, Eq)] +pub struct CodexWebSocketCloseError { + code: Option, + reason: Option, +} + +impl CodexWebSocketCloseError { + pub(crate) fn new(code: Option, reason: Option) -> Self { + Self { code, reason } + } + + /// 返回上游 close code；没有 close frame 时为 `None`。 + #[must_use] + pub const fn code(&self) -> Option { + self.code + } + + /// 返回上游 close reason；仅可用于当前请求的客户端协议响应。 + #[must_use] + pub fn reason(&self) -> Option<&str> { + self.reason.as_deref() + } +} + +impl fmt::Debug for CodexWebSocketCloseError { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter + .debug_struct("CodexWebSocketCloseError") + .field("code", &self.code) + .field("reason", &self.reason.as_ref().map(|_| "")) + .finish() + } +} + +impl fmt::Display for CodexWebSocketCloseError { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + match self.code { + Some(code) => write!( + formatter, + "websocket closed before terminal event (code {code})" + ), + None => formatter.write_str("websocket closed before terminal event"), + } + } +} + /// WebSocket 上游错误帧载荷。 #[derive(Clone, PartialEq, Eq)] pub struct CodexWebSocketUpstreamError { @@ -140,6 +192,25 @@ impl fmt::Display for CodexWebSocketUpstreamError { } impl CodexWebSocketExchangeError { + pub(crate) fn closed_before_terminal(code: Option, reason: Option) -> Self { + Self::ClosedBeforeTerminal(CodexWebSocketCloseError::new(code, reason)) + } + + pub(crate) fn close_before_terminal(&self) -> Option<&CodexWebSocketCloseError> { + match self { + Self::ClosedBeforeTerminal(close) => Some(close), + Self::PostSendAmbiguous { + source: Some(source), + .. + } + | Self::ReusedConnectionDiedBeforeFirstEvent { + source: Some(source), + .. + } => source.close_before_terminal(), + _ => None, + } + } + pub(super) fn upstream( status_code: u16, retry_after_seconds: Option, @@ -173,4 +244,16 @@ impl CodexWebSocketExchangeError { | Self::ContinuationUnavailable { .. } ) } + + /// 首个业务事件交付前，普通 WebSocket 失败可切到同账号 HTTP。 + pub(in crate::transport) fn allows_pre_delivery_http_fallback(&self) -> bool { + match self { + Self::Upstream(_) => false, + Self::PostSendAmbiguous { + source: Some(source), + .. + } => source.allows_pre_delivery_http_fallback(), + _ => true, + } + } } diff --git a/backend/crates/providers/openai/src/transport/websocket/exchange/io.rs b/backend/crates/providers/openai/src/transport/websocket/exchange/io.rs index 43be35693..8f8d1f849 100644 --- a/backend/crates/providers/openai/src/transport/websocket/exchange/io.rs +++ b/backend/crates/providers/openai/src/transport/websocket/exchange/io.rs @@ -40,19 +40,20 @@ pub(super) fn reused_stream_receive_error( error: CodexWebSocketExchangeError, ) -> CodexWebSocketExchangeError { match error { - CodexWebSocketExchangeError::ClosedBeforeTerminal + CodexWebSocketExchangeError::ClosedBeforeTerminal(_) | CodexWebSocketExchangeError::ReceiveIdleTimeout { .. } | CodexWebSocketExchangeError::Transport(_) => { - reused_connection_died_before_first_event(&error) + reused_connection_died_before_first_event(error) } error => error, } } pub(super) fn reused_connection_died_before_first_event( - error: &CodexWebSocketExchangeError, + error: CodexWebSocketExchangeError, ) -> CodexWebSocketExchangeError { CodexWebSocketExchangeError::ReusedConnectionDiedBeforeFirstEvent { message: error.to_string(), + source: Some(Box::new(error)), } } diff --git a/backend/crates/providers/openai/src/transport/websocket/exchange/stream.rs b/backend/crates/providers/openai/src/transport/websocket/exchange/stream.rs index 8f1be9a80..1cb4c6356 100644 --- a/backend/crates/providers/openai/src/transport/websocket/exchange/stream.rs +++ b/backend/crates/providers/openai/src/transport/websocket/exchange/stream.rs @@ -208,14 +208,17 @@ async fn forward_websocket_response_stream(state: WebSocketStreamForwardState) { .await; return; } - tungstenite::Message::Close(_) => { + tungstenite::Message::Close(frame) => { discard_stream_websocket( websocket, pool_return, StreamWebSocketDiscardReason::UpstreamClosed, ) .await; - let error = CodexWebSocketExchangeError::ClosedBeforeTerminal; + let error = CodexWebSocketExchangeError::closed_before_terminal( + frame.as_ref().map(|frame| u16::from(frame.code)), + frame.map(|frame| frame.reason.to_string()), + ); let error = if reused_connection { reused_stream_receive_error(error) } else { @@ -285,13 +288,19 @@ async fn forward_websocket_response_stream(state: WebSocketStreamForwardState) { } } + let upstream_close = websocket + .exit_reason() + .and_then(|reason| reason.upstream_close().cloned()); discard_stream_websocket( websocket, pool_return, StreamWebSocketDiscardReason::UpstreamClosed, ) .await; - let error = CodexWebSocketExchangeError::ClosedBeforeTerminal; + let error = upstream_close.map_or_else( + || CodexWebSocketExchangeError::closed_before_terminal(None, None), + CodexWebSocketExchangeError::ClosedBeforeTerminal, + ); let error = if reused_connection { reused_stream_receive_error(error) } else { diff --git a/backend/crates/providers/openai/src/transport/websocket/mod.rs b/backend/crates/providers/openai/src/transport/websocket/mod.rs index cc07e39db..7fb589411 100644 --- a/backend/crates/providers/openai/src/transport/websocket/mod.rs +++ b/backend/crates/providers/openai/src/transport/websocket/mod.rs @@ -19,7 +19,7 @@ pub use self::{ WebSocketOriginBreaker, WebSocketOriginBreakerConfig, WebSocketOriginBreakerDecision, WebSocketOriginBreakerPermit, }, - error::{CodexWebSocketExchangeError, CodexWebSocketUpstreamError}, + error::{CodexWebSocketCloseError, CodexWebSocketExchangeError, CodexWebSocketUpstreamError}, exchange::{ CodexWebSocketRateLimitHeaderUpdates, CodexWebSocketSseStream, CodexWebSocketStreamingExchange, CodexWebSocketTurnStateUpdate, diff --git a/backend/crates/providers/openai/src/transport/websocket/pump.rs b/backend/crates/providers/openai/src/transport/websocket/pump.rs index 492d19e88..41a2b6259 100644 --- a/backend/crates/providers/openai/src/transport/websocket/pump.rs +++ b/backend/crates/providers/openai/src/transport/websocket/pump.rs @@ -29,6 +29,8 @@ use tokio::{ use tokio_tungstenite::{MaybeTlsStream, WebSocketStream, tungstenite::Message}; use uuid::Uuid; +use super::CodexWebSocketCloseError; + /// 底层 tungstenite WebSocket 流。 pub(crate) type RawWsStream = WebSocketStream>; @@ -69,13 +71,25 @@ const PUMP_MESSAGE_BUFFER: usize = 64; pub(crate) enum PumpExitReason { CommandChannelClosed, LocalClose, - OutboundTransportError { message: String }, - UpstreamCloseFrame { frame: Option }, + OutboundTransportError { + message: String, + }, + UpstreamCloseFrame { + close: Option, + }, UpstreamEof, - InboundTransportError { message: String }, - PongTimeout { timeout: Duration }, - LivenessTimeout { timeout: Duration }, - KeepaliveTransportError { message: String }, + InboundTransportError { + message: String, + }, + PongTimeout { + timeout: Duration, + }, + LivenessTimeout { + timeout: Duration, + }, + KeepaliveTransportError { + message: String, + }, MessageReceiverClosed, } @@ -100,7 +114,7 @@ impl PumpExitReason { Self::OutboundTransportError { message } | Self::InboundTransportError { message } | Self::KeepaliveTransportError { message } => Some(message.clone()), - Self::UpstreamCloseFrame { frame } => frame.clone(), + Self::UpstreamCloseFrame { close } => close.as_ref().map(ToString::to_string), Self::PongTimeout { timeout } | Self::LivenessTimeout { timeout } => { Some(format!("{timeout:?}")) } @@ -111,6 +125,13 @@ impl PumpExitReason { } } + pub(crate) fn upstream_close(&self) -> Option<&CodexWebSocketCloseError> { + match self { + Self::UpstreamCloseFrame { close: Some(close) } => Some(close), + _ => None, + } + } + fn is_unexpected(&self) -> bool { matches!( self, @@ -336,7 +357,12 @@ async fn pump_loop( message => { let terminal_reason = match &message { Message::Close(frame) => Some(PumpExitReason::UpstreamCloseFrame { - frame: frame.as_ref().map(|frame| format!("{frame:?}")), + close: frame.as_ref().map(|frame| { + CodexWebSocketCloseError::new( + Some(u16::from(frame.code)), + Some(frame.reason.to_string()), + ) + }), }), _ => None, }; diff --git a/backend/crates/providers/openai/tests/transport/latency.rs b/backend/crates/providers/openai/tests/transport/latency.rs index 6e4432408..b5342e19b 100644 --- a/backend/crates/providers/openai/tests/transport/latency.rs +++ b/backend/crates/providers/openai/tests/transport/latency.rs @@ -5,6 +5,7 @@ use provider_openai::transport::{ WebSocketOriginBreaker, WebSocketOriginBreakerConfig, WebSocketOriginBreakerDecision, }, }; +use tokio_tungstenite::tungstenite::protocol::{CloseFrame, frame::coding::CloseCode}; use super::*; @@ -913,14 +914,133 @@ async fn concurrent_same_key_should_singleflight_websocket_opening() { } #[tokio::test] -async fn payload_send_failure_should_not_open_http_fallback() { +async fn websocket_failure_before_first_delivery_should_use_same_account_http() { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + let server = tokio::spawn(async move { + let (stream, _) = listener.accept().await.unwrap(); + let mut websocket = accept_codex_test_websocket_with(stream, |request, response| { + assert_eq!( + request + .headers() + .get("authorization") + .and_then(|value| value.to_str().ok()), + Some("Bearer access-token") + ); + assert_eq!( + request + .headers() + .get("chatgpt-account-id") + .and_then(|value| value.to_str().ok()), + Some("chatgpt-account") + ); + response.headers_mut().insert( + "sec-websocket-extensions", + "permessage-deflate".parse().unwrap(), + ); + }) + .await; + let _payload = websocket.next().await.unwrap().unwrap(); + websocket + .send(Message::Text( + json!({ + "type": "response.created", + "response": { + "id": "resp_abandoned_websocket", + "model": "gpt-test" + } + }) + .to_string() + .into(), + )) + .await + .unwrap(); + websocket + .send(Message::Text( + json!({ + "type": "response.in_progress", + "response": { + "id": "resp_abandoned_websocket", + "model": "gpt-test" + } + }) + .to_string() + .into(), + )) + .await + .unwrap(); + websocket + .close(Some(CloseFrame { + code: CloseCode::Size, + reason: "message too big".into(), + })) + .await + .unwrap(); + + let (mut http, _) = listener.accept().await.unwrap(); + let request = read_http_request(&mut http).await; + assert!(request.starts_with("POST /codex/responses HTTP/1.1")); + assert_eq!( + read_header_value(&request, "authorization"), + Some("Bearer access-token") + ); + assert_eq!( + read_header_value(&request, "chatgpt-account-id"), + Some("chatgpt-account") + ); + write_completed_sse_response(&mut http).await; + }); + let backend = CodexBackendClient::new( + reqwest::Client::builder().no_proxy().build().unwrap(), + format!("http://{addr}"), + test_wire_profile(), + ) + .with_websocket_pool(Arc::new(CodexWebSocketPool::default())); + + let response = backend + .create_response( + &new_chain_request("conversation-pre-delivery-fallback"), + request_context("req_pre_delivery_fallback", Some("chatgpt-account")), + ) + .await + .expect("pre-delivery WebSocket failure should use same-account HTTP"); + server.await.unwrap(); + + assert_eq!(response.transport, CodexBackendTransport::HttpSse); + assert_eq!( + response.transport_metrics.decision, + Some(CodexTransportDecision::Http2PreDeliveryFailure) + ); + assert!(response.body.contains("resp_http_fallback")); + assert!(!response.body.contains("resp_abandoned_websocket")); +} + +#[tokio::test] +async fn websocket_failure_after_first_delivery_should_not_open_http_fallback() { let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); let addr = listener.local_addr().unwrap(); let server = tokio::spawn(async move { let (stream, _) = listener.accept().await.unwrap(); let mut websocket = accept_codex_test_websocket(stream).await; let _payload = websocket.next().await.unwrap().unwrap(); - websocket.close(None).await.unwrap(); + websocket + .send(Message::Text( + json!({ + "type": "response.output_text.delta", + "delta": "delivered" + }) + .to_string() + .into(), + )) + .await + .unwrap(); + websocket + .close(Some(CloseFrame { + code: CloseCode::Size, + reason: "message too big after output".into(), + })) + .await + .unwrap(); assert!( timeout(Duration::from_millis(100), listener.accept()) .await @@ -936,17 +1056,25 @@ async fn payload_send_failure_should_not_open_http_fallback() { let error = backend .create_response( - &new_chain_request("conversation-post-send"), - request_context("req_post_send", Some("chatgpt-account")), + &new_chain_request("conversation-post-delivery"), + request_context("req_post_delivery", Some("chatgpt-account")), ) .await - .expect_err("post-send close must not be replayed"); + .expect_err("post-delivery close must not be replayed"); server.await.unwrap(); - std::assert_matches!( - error, - CodexClientError::WebSocket(CodexWebSocketExchangeError::PostSendAmbiguous { .. }) - ); + let CodexClientError::WebSocket(CodexWebSocketExchangeError::PostSendAmbiguous { + source: Some(source), + .. + }) = error + else { + panic!("expected typed post-delivery WebSocket failure"); + }; + let CodexWebSocketExchangeError::ClosedBeforeTerminal(close) = *source else { + panic!("expected upstream close frame source"); + }; + assert_eq!(close.code(), Some(1009)); + assert_eq!(close.reason(), Some("message too big after output")); } #[tokio::test] diff --git a/backend/crates/providers/openai/tests/transport/protocol/responses.rs b/backend/crates/providers/openai/tests/transport/protocol/responses.rs index 0129113a2..2578cccd5 100644 --- a/backend/crates/providers/openai/tests/transport/protocol/responses.rs +++ b/backend/crates/providers/openai/tests/transport/protocol/responses.rs @@ -218,6 +218,17 @@ fn transport_should_prefer_websocket_when_requested_without_history() { assert!(transport_requirement(&request).allows_pre_send_http_fallback()); } +#[test] +fn new_chain_should_allow_pre_delivery_http_fallback() { + let mut request = codex_request("gpt-test", "be brief", Vec::new()); + request.use_websocket = true; + + assert!( + transport_requirement(&request).allows_pre_delivery_http_fallback(), + "new chains may follow the TS same-account fallback behavior" + ); +} + #[test] fn transport_should_mark_unknown_previous_response_as_external() { let mut request = codex_request("gpt-test", "be brief", Vec::new()); @@ -230,6 +241,17 @@ fn transport_should_mark_unknown_previous_response_as_external() { assert!(transport_requirement(&request).allows_pre_send_http_fallback()); } +#[test] +fn previous_response_should_forbid_pre_delivery_http_fallback() { + let mut request = codex_request("gpt-test", "be brief", Vec::new()); + request.set_previous_response_id(Some("resp_previous".to_owned())); + + assert!( + !transport_requirement(&request).allows_pre_delivery_http_fallback(), + "a sent continuation must not be replayed on another transport" + ); +} + #[test] fn transport_should_allow_forced_http_sse() { let mut request = codex_request("gpt-test", "be brief", Vec::new()); diff --git a/backend/crates/providers/openai/tests/transport/websocket.rs b/backend/crates/providers/openai/tests/transport/websocket.rs index a5e8c07de..3f004bb88 100644 --- a/backend/crates/providers/openai/tests/transport/websocket.rs +++ b/backend/crates/providers/openai/tests/transport/websocket.rs @@ -649,7 +649,15 @@ async fn websocket_execute_response_create_request_should_reject_binary_event() .unwrap(); websocket.close(None).await.unwrap(); }); - let prepared = prepared_websocket_request(&format!("http://{addr}")); + let mut request = codex_request("gpt-test", "be brief", Vec::new()); + request.set_previous_response_id(Some("resp_binary_previous".to_owned())); + let prepared = CodexWebSocketConnection::responses_create_request( + &format!("http://{addr}"), + "dGhlIHNhbXBsZSBub25jZQ==", + vec![("authorization".to_owned(), "Bearer access-token".to_owned())], + &request, + ) + .expect("prepare continuation WebSocket request"); let error = execute_response_create_request(&prepared) .await