From f2e91bbdaaedbcd28c566ac3334b1f7a2fea120a Mon Sep 17 00:00:00 2001 From: Zyy Date: Wed, 2 Sep 2026 18:21:30 +0800 Subject: [PATCH] fix(openai): preserve websocket continuation routing Limit session-sticky HTTP fallback to new chains, keep continuation requests on their safe transport, and classify rebuild-required failures separately from account health. Co-authored-by: Codex --- .../crates/gateway-api/tests/openai/error.rs | 35 ++++++ backend/crates/gateway-core/src/error.rs | 7 ++ .../gateway-core/tests/engine/coordinator.rs | 13 ++- .../gateway-core/tests/engine/execution.rs | 3 + backend/crates/gateway-core/tests/error.rs | 24 ++++ .../providers/openai/src/provider/failure.rs | 9 +- .../providers/openai/src/provider/mod.rs | 2 +- .../src/transport/protocol/responses.rs | 8 +- .../openai/tests/provider/contract.rs | 103 ++++++++++++++---- .../openai/tests/provider/failure.rs | 4 + .../tests/transport/protocol/responses.rs | 29 +++++ 11 files changed, 206 insertions(+), 31 deletions(-) diff --git a/backend/crates/gateway-api/tests/openai/error.rs b/backend/crates/gateway-api/tests/openai/error.rs index 17cd3c051..1ba85001d 100644 --- a/backend/crates/gateway-api/tests/openai/error.rs +++ b/backend/crates/gateway-api/tests/openai/error.rs @@ -398,3 +398,38 @@ async fn gateway_error_response_should_expose_only_structured_client_visible_ups }) ); } + +#[tokio::test] +async fn continuation_recovery_should_preserve_the_official_retry_signal() { + let error = GatewayError::from_provider( + &ProviderError::new( + ProviderErrorKind::ContinuationRecoveryRequired, + UpstreamSendState::NotSent, + ) + .with_client_visible_upstream_error(ClientVisibleUpstreamError::new( + "Previous response was not found. Retrying the full request.", + Some("previous_response_not_found".to_owned()), + Some("invalid_request_error".to_owned()), + )), + ); + + let response = gateway_error_response(&error); + let status = response.status(); + let body = to_bytes(response.into_body(), usize::MAX) + .await + .expect("read continuation recovery response"); + let body: serde_json::Value = + serde_json::from_slice(&body).expect("continuation recovery JSON"); + + assert_eq!(status, StatusCode::BAD_REQUEST); + assert_eq!( + body, + serde_json::json!({ + "error": { + "message": "Previous response was not found. Retrying the full request.", + "type": "invalid_request_error", + "code": "previous_response_not_found" + } + }) + ); +} diff --git a/backend/crates/gateway-core/src/error.rs b/backend/crates/gateway-core/src/error.rs index 7e4d7da06..42e9d04cb 100644 --- a/backend/crates/gateway-core/src/error.rs +++ b/backend/crates/gateway-core/src/error.rs @@ -135,6 +135,8 @@ pub enum MeteringError { pub enum ProviderErrorKind { /// 请求语义错误。 InvalidRequest, + /// previous-response 续接状态已不可用，需要客户端重建请求链。 + ContinuationRecoveryRequired, /// Provider 不支持所需能力。 Unsupported, /// Credential 认证失败。 @@ -171,6 +173,7 @@ impl ProviderErrorKind { pub const fn as_str(self) -> &'static str { match self { Self::InvalidRequest => "invalid_request", + Self::ContinuationRecoveryRequired => "continuation_recovery_required", Self::Unsupported => "unsupported", Self::Unauthorized => "unauthorized", Self::PermissionDenied => "permission_denied", @@ -968,6 +971,10 @@ impl GatewayError { ProviderErrorKind::InvalidRequest => { Self::new(GatewayErrorKind::InvalidRequest, "invalid upstream request") } + ProviderErrorKind::ContinuationRecoveryRequired => Self::new( + GatewayErrorKind::InvalidRequest, + "conversation continuation must be rebuilt", + ), ProviderErrorKind::Unsupported => Self::new( GatewayErrorKind::Unsupported, "requested capability is unsupported", diff --git a/backend/crates/gateway-core/tests/engine/coordinator.rs b/backend/crates/gateway-core/tests/engine/coordinator.rs index 37c3d60d2..fc951bdda 100644 --- a/backend/crates/gateway-core/tests/engine/coordinator.rs +++ b/backend/crates/gateway-core/tests/engine/coordinator.rs @@ -501,10 +501,13 @@ fn atomic_history_unavailable(response_id: &str) -> ProviderError { ) .expect("history unavailable wire"), ); - ProviderError::new(ProviderErrorKind::InvalidRequest, UpstreamSendState::Sent) - .with_status(400) - .with_continuation_failure(ContinuationFailure::HistoryUnavailable) - .with_atomic_client_events(vec![failed]) + ProviderError::new( + ProviderErrorKind::ContinuationRecoveryRequired, + UpstreamSendState::Sent, + ) + .with_status(400) + .with_continuation_failure(ContinuationFailure::HistoryUnavailable) + .with_atomic_client_events(vec![failed]) } fn image_stream(image_output_tokens: Option) -> Vec> { @@ -1591,7 +1594,7 @@ fn native_continuation_client_replay_required_is_terminal() { Script::Stream { account_id: "acct_one", items: vec![Err(ProviderError::new( - ProviderErrorKind::InvalidRequest, + ProviderErrorKind::ContinuationRecoveryRequired, UpstreamSendState::NotSent, ) .with_continuation_failure(ContinuationFailure::HistoryUnavailable) diff --git a/backend/crates/gateway-core/tests/engine/execution.rs b/backend/crates/gateway-core/tests/engine/execution.rs index 23f4b81bf..86c7e55cb 100644 --- a/backend/crates/gateway-core/tests/engine/execution.rs +++ b/backend/crates/gateway-core/tests/engine/execution.rs @@ -62,6 +62,9 @@ fn only_provider_attributable_failures_should_affect_circuit() { assert!(!provider_failure_affects_circuit( ProviderErrorKind::InvalidRequest )); + assert!(!provider_failure_affects_circuit( + ProviderErrorKind::ContinuationRecoveryRequired + )); } #[test] diff --git a/backend/crates/gateway-core/tests/error.rs b/backend/crates/gateway-core/tests/error.rs index 3fdc66e2a..7c3dab529 100644 --- a/backend/crates/gateway-core/tests/error.rs +++ b/backend/crates/gateway-core/tests/error.rs @@ -168,6 +168,30 @@ fn gateway_error_should_keep_client_visible_upstream_fields_out_of_safe_diagnost assert!(!format!("{gateway:?}").contains(message)); } +#[test] +fn continuation_recovery_should_keep_its_internal_classification_and_client_contract() { + let error = ProviderError::new( + ProviderErrorKind::ContinuationRecoveryRequired, + UpstreamSendState::NotSent, + ) + .with_client_visible_upstream_error(ClientVisibleUpstreamError::new( + "Previous response was not found. Retrying the full request.", + Some("previous_response_not_found".to_owned()), + Some("invalid_request_error".to_owned()), + )); + let gateway = GatewayError::from_provider(&error); + + assert_eq!(error.kind().as_str(), "continuation_recovery_required"); + assert_eq!( + gateway.safe_message(), + "conversation continuation must be rebuilt" + ); + assert_eq!( + gateway.client_error_code(), + Some("previous_response_not_found") + ); +} + #[test] fn client_visible_upstream_error_should_preserve_opaque_structured_fields() { let message = format!("\0{}\n", "m".repeat(9_000)); diff --git a/backend/crates/providers/openai/src/provider/failure.rs b/backend/crates/providers/openai/src/provider/failure.rs index 2c1706962..ab89dcc81 100644 --- a/backend/crates/providers/openai/src/provider/failure.rs +++ b/backend/crates/providers/openai/src/provider/failure.rs @@ -617,7 +617,7 @@ pub(super) fn websocket_retry_backoff(retry_index: NonZeroU32) -> Duration { pub(super) fn continuation_replay_required_error() -> ProviderError { ProviderError::new( - ProviderErrorKind::InvalidRequest, + ProviderErrorKind::ContinuationRecoveryRequired, UpstreamSendState::NotSent, ) .with_continuation_failure(ContinuationFailure::HistoryUnavailable) @@ -934,7 +934,12 @@ pub(super) fn map_upstream_failure( .filter(|code| is_history_failure_code(code)) .map(|_| ContinuationFailure::HistoryUnavailable); let send_state = upstream_send_state(failure.send_phase); - let mut error = provider_error(provider_error_kind(category), send_state); + let error_kind = if continuation_failure.is_some() { + ProviderErrorKind::ContinuationRecoveryRequired + } else { + provider_error_kind(category) + }; + let mut error = provider_error(error_kind, send_state); error = error.with_raw_upstream_error(RawUpstreamError::new(failure.raw_body.clone())); if let Some(message) = failure.client_message.as_ref() { error = error.with_client_visible_upstream_error(ClientVisibleUpstreamError::new( diff --git a/backend/crates/providers/openai/src/provider/mod.rs b/backend/crates/providers/openai/src/provider/mod.rs index 033699050..3f38acd71 100644 --- a/backend/crates/providers/openai/src/provider/mod.rs +++ b/backend/crates/providers/openai/src/provider/mod.rs @@ -423,7 +423,7 @@ impl Provider for CodexProvider { let requested_transport = selected_transport(&upstream_request); let sticky_http_fallback = context.transport() == AttemptTransport::Default && requested_transport == CodexProviderTransport::PreferWebSocket - && !requirement.requires_websocket() + && requirement.allows_sticky_http_fallback() && session_affinity.as_ref().is_some_and(|affinity| { self.session_transport_fallbacks .is_http_only(affinity.key()) diff --git a/backend/crates/providers/openai/src/transport/protocol/responses.rs b/backend/crates/providers/openai/src/transport/protocol/responses.rs index 8418218ec..2447b25e5 100644 --- a/backend/crates/providers/openai/src/transport/protocol/responses.rs +++ b/backend/crates/providers/openai/src/transport/protocol/responses.rs @@ -162,6 +162,11 @@ impl TransportRequirement { matches!(self, Self::NewChain) } + /// 会话级 sticky 状态是否可以把首选 WebSocket 改为 HTTP。 + pub fn allows_sticky_http_fallback(self) -> bool { + matches!(self, Self::NewChain) + } + /// 用于审计与遥测的稳定名称。 pub fn as_str(self) -> &'static str { match self { @@ -193,8 +198,7 @@ pub fn transport_requirement(request: &CodexResponsesRequest) -> TransportRequir TransportRequirement::ExternalUnknown } }, - None if request.use_websocket => TransportRequirement::NewChain, - None => TransportRequirement::HttpRequired, + None => TransportRequirement::NewChain, } } diff --git a/backend/crates/providers/openai/tests/provider/contract.rs b/backend/crates/providers/openai/tests/provider/contract.rs index 39c08de94..b75cc803e 100644 --- a/backend/crates/providers/openai/tests/provider/contract.rs +++ b/backend/crates/providers/openai/tests/provider/contract.rs @@ -1295,7 +1295,7 @@ async fn websocket_close_after_delivery_preserves_details_without_enabling_stick } #[tokio::test] -async fn ambiguous_websocket_close_is_terminal_and_sticky_only_for_the_current_session() { +async fn ambiguous_websocket_close_only_sticks_new_chains_in_the_current_session() { let store = Arc::new(MemoryAccountStore::default()); create_account(&store, "acct_websocket_close").await; let listener = TcpListener::bind("127.0.0.1:0") @@ -1306,6 +1306,26 @@ async fn ambiguous_websocket_close_is_terminal_and_sticky_only_for_the_current_s listener.local_addr().expect("listener address") ); let server = tokio::spawn(async move { + let completed_response = |response_id: &str| { + Message::Text( + json!({ + "type": "response.completed", + "response": { + "id": response_id, + "model": "gpt-5.4", + "status": "completed", + "output": [], + "usage": { + "input_tokens": 1, + "output_tokens": 1, + "total_tokens": 2 + } + } + }) + .to_string() + .into(), + ) + }; let (stream, _) = listener .accept() .await @@ -1371,6 +1391,31 @@ async fn ambiguous_websocket_close_is_terminal_and_sticky_only_for_the_current_s .await .expect("write sticky HTTP response"); + let (stream, _) = listener + .accept() + .await + .expect("accept same-session continuation WebSocket connection"); + let mut websocket = accept_codex_test_websocket(stream).await; + let request = websocket + .next() + .await + .expect("same-session continuation request") + .expect("valid same-session continuation request"); + let payload: Value = serde_json::from_str( + request + .to_text() + .expect("same-session continuation request text"), + ) + .expect("same-session continuation request JSON"); + assert_eq!( + payload.get("previous_response_id"), + Some(&json!("external-previous-response")) + ); + websocket + .send(completed_response("resp_external_continuation")) + .await + .expect("complete same-session continuation request"); + let (stream, _) = listener .accept() .await @@ -1382,24 +1427,7 @@ async fn ambiguous_websocket_close_is_terminal_and_sticky_only_for_the_current_s .expect("other-session WebSocket request") .expect("valid other-session WebSocket request"); websocket - .send(Message::Text( - json!({ - "type": "response.completed", - "response": { - "id": "resp_other_session", - "model": "gpt-5.4", - "status": "completed", - "output": [], - "usage": { - "input_tokens": 1, - "output_tokens": 1, - "total_tokens": 2 - } - } - }) - .to_string() - .into(), - )) + .send(completed_response("resp_other_session")) .await .expect("complete other-session WebSocket request"); }); @@ -1491,6 +1519,27 @@ async fn ambiguous_websocket_close_is_terminal_and_sticky_only_for_the_current_s } drop(sticky_stream); + let continuation_operation = Operation::Generate(generate_with_session_context( + "sticky-websocket-session", + Some("thread-continuation"), + None, + )); + let mut continuation_stream = provider + .execute( + planned_request("openai", continuation_operation), + external_continuation_context("req_sticky_external_continuation"), + ) + .await + .expect("sticky state must not force a continuation onto HTTP"); + assert_eq!( + continuation_stream.metadata().transport().as_str(), + "websocket" + ); + while let Some(event) = continuation_stream.next().await { + event.expect("same-session continuation WebSocket response"); + } + drop(continuation_stream); + let other_operation = Operation::Generate(generate_with_session_context( "other-websocket-session", Some("thread-first"), @@ -2044,7 +2093,7 @@ async fn cross_account_continuation_should_require_client_replay_without_an_upst detail.error_type(), ), ( - ProviderErrorKind::InvalidRequest, + ProviderErrorKind::ContinuationRecoveryRequired, UpstreamSendState::NotSent, Some(ContinuationFailure::HistoryUnavailable), Some("previous_response_not_found"), @@ -2094,6 +2143,10 @@ async fn missing_affinity_continuation_should_require_client_replay_without_an_u .await .expect("captured upstream requests"); + assert_eq!( + error.kind(), + ProviderErrorKind::ContinuationRecoveryRequired + ); assert_eq!( error .client_visible_upstream_error() @@ -3396,6 +3449,10 @@ async fn same_account_previous_response_not_found_should_remain_client_visible() } }; + assert_eq!( + failure.kind(), + ProviderErrorKind::ContinuationRecoveryRequired + ); assert_eq!( failure.continuation_failure(), Some(ContinuationFailure::HistoryUnavailable) @@ -3625,6 +3682,10 @@ async fn exact_websocket_busy_then_replay_scope_relaxation_is_rejected_before_se None => panic!("competing exact continuation must fail while its socket is busy"), } }; + assert_eq!( + exact_error.kind(), + ProviderErrorKind::ContinuationRecoveryRequired + ); assert_eq!( exact_error.continuation_failure(), Some(ContinuationFailure::Busy) @@ -3673,7 +3734,7 @@ async fn exact_websocket_busy_then_replay_scope_relaxation_is_rejected_before_se detail.message(), ), ( - ProviderErrorKind::InvalidRequest, + ProviderErrorKind::ContinuationRecoveryRequired, UpstreamSendState::NotSent, Some(ContinuationFailure::HistoryUnavailable), Some(ContinuationRecoveryDisposition::ClientReplayRequired), diff --git a/backend/crates/providers/openai/tests/provider/failure.rs b/backend/crates/providers/openai/tests/provider/failure.rs index 5a5122562..ae7842e3a 100644 --- a/backend/crates/providers/openai/tests/provider/failure.rs +++ b/backend/crates/providers/openai/tests/provider/failure.rs @@ -104,6 +104,10 @@ fn account_score_failure_filter_should_prefer_a_structured_code_over_type() { fn account_score_failure_filter_should_reject_client_and_unknown_failures() { for error in [ sent_error(ProviderErrorKind::InvalidRequest, Some("invalid_request")), + sent_error( + ProviderErrorKind::ContinuationRecoveryRequired, + Some("previous_response_not_found"), + ), sent_error(ProviderErrorKind::Unsupported, Some("unsupported")), sent_error(ProviderErrorKind::Unauthorized, Some("token_expired")), sent_error(ProviderErrorKind::QuotaExhausted, Some("quota_exhausted")), diff --git a/backend/crates/providers/openai/tests/transport/protocol/responses.rs b/backend/crates/providers/openai/tests/transport/protocol/responses.rs index 346611d93..b58a03fba 100644 --- a/backend/crates/providers/openai/tests/transport/protocol/responses.rs +++ b/backend/crates/providers/openai/tests/transport/protocol/responses.rs @@ -273,6 +273,16 @@ fn transport_should_prefer_websocket_when_requested_without_history() { assert!(transport_requirement(&request).allows_pre_send_http_fallback()); } +#[test] +fn unspecified_transport_without_history_should_still_be_a_new_chain() { + let request = codex_request("gpt-test", "be brief", Vec::new()); + + assert_eq!( + transport_requirement(&request), + TransportRequirement::NewChain + ); +} + #[test] fn new_chain_should_allow_pre_delivery_http_fallback() { let mut request = codex_request("gpt-test", "be brief", Vec::new()); @@ -284,6 +294,25 @@ fn new_chain_should_allow_pre_delivery_http_fallback() { ); } +#[test] +fn sticky_http_fallback_should_only_apply_to_new_chains() { + assert!(TransportRequirement::NewChain.allows_sticky_http_fallback()); + + for requirement in [ + TransportRequirement::HttpRequired, + TransportRequirement::ExplicitWebSocketWarmup, + TransportRequirement::ExactWebSocketContinuation, + TransportRequirement::PersistedContinuation, + TransportRequirement::ExternalUnknown, + ] { + assert!( + !requirement.allows_sticky_http_fallback(), + "sticky HTTP must not override {}", + requirement.as_str() + ); + } +} + #[test] fn transport_should_mark_unknown_previous_response_as_external() { let mut request = codex_request("gpt-test", "be brief", Vec::new());