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 <noreply@openai.com>

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"
+            }
+        })
+    );
+}

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",

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<u64>) -> Vec<Result<GatewayEvent, ProviderError>> {
@@ -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)

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]

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));

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(

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())

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,
     }
 }
 

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),

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")),

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());