fix(openai): normalize HTTP previous response errors

Co-authored-by: Codex <noreply@openai.com>

backend/crates/providers/openai/src/provider.rs

@@ -69,10 +69,8 @@ use crate::transport::profile::{
     APPCAST_POLL_INTERVAL, CodexDesktopReleaseService, CodexWireProfileState,
 };
 use crate::transport::protocol::responses::{
-    CodexResponsesRequest, PreviousResponseScope, ResponseEventSignals,
-};
-use crate::transport::protocol::websocket::{
-    PREVIOUS_RESPONSE_NOT_FOUND_CODE, PREVIOUS_RESPONSE_NOT_FOUND_MESSAGE,
+    CodexResponsesRequest, PREVIOUS_RESPONSE_NOT_FOUND_CODE, PREVIOUS_RESPONSE_NOT_FOUND_MESSAGE,
+    PreviousResponseScope, ResponseEventSignals,
 };
 use crate::transport::request::{
     CodexRequestEncodeError, RequestAccountScope, encode_generate_request, scope_request_to_account,

backend/crates/providers/openai/src/transport/diagnostics.rs

@@ -1,7 +1,10 @@
 //! 在 transport 边界采集的上游响应诊断信息。
 
 use super::client::CodexClientVisibleUpstreamResponse;
-use super::protocol::responses::ResponsesSseFailure;
+use super::protocol::responses::{
+    PREVIOUS_RESPONSE_NOT_FOUND_CODE, ResponsesSseFailure,
+    is_bare_invalid_previous_response_id_error,
+};
 use base64::{Engine as _, engine::general_purpose::STANDARD};
 use chrono::{DateTime, Utc};
 use reqwest::{StatusCode, header::HeaderMap};
@@ -140,7 +143,7 @@ impl CodexUpstreamFailure {
         rate_limit_headers: &[(String, String)],
         send_phase: CodexUpstreamSendPhase,
     ) -> Self {
-        let fields = ParsedUpstreamError::from_body(body);
+        let fields = ParsedUpstreamError::from_http_response(status, body);
         let category = classify_upstream_failure(
             UpstreamFailureSource::HttpResponse,
             Some(status),
@@ -260,7 +263,7 @@ struct ParsedUpstreamError {
 }
 
 impl ParsedUpstreamError {
-    fn from_body(body: &str) -> Self {
+    fn from_http_response(status: StatusCode, body: &str) -> Self {
         let Ok(value) = serde_json::from_str::<Value>(body) else {
             return Self {
                 code: None,
@@ -275,11 +278,7 @@ impl ParsedUpstreamError {
             .or_else(|| value.get("error"))
             .or_else(|| value.get("detail"))
             .unwrap_or(&value);
-        let code = error
-            .get("code")
-            .or_else(|| value.get("code"))
-            .and_then(Value::as_str)
-            .and_then(non_empty_owned);
+        let code_value = error.get("code").or_else(|| value.get("code"));
         let error_type = error
             .get("type")
             .or_else(|| value.get("type"))
@@ -290,6 +289,16 @@ impl ParsedUpstreamError {
             .or_else(|| value.get("message"))
             .and_then(Value::as_str)
             .and_then(non_empty_owned);
+        let code = if status == StatusCode::BAD_REQUEST
+            && is_bare_invalid_previous_response_id_error(
+                code_value,
+                error_type.as_deref(),
+                client_message.as_deref(),
+            ) {
+            Some(PREVIOUS_RESPONSE_NOT_FOUND_CODE.to_owned())
+        } else {
+            code_value.and_then(Value::as_str).and_then(non_empty_owned)
+        };
         let message = client_message
             .clone()
             .or_else(|| error.as_str().and_then(non_empty_owned))

backend/crates/providers/openai/src/transport/protocol/responses.rs

@@ -8,6 +8,24 @@ use reqwest::header::HeaderMap;
 use serde::Serialize;
 use serde_json::{Map, Value};
 
+/// 官方 Codex 客户端据此触发完整历史重放的稳定错误码。
+pub(crate) const PREVIOUS_RESPONSE_NOT_FOUND_CODE: &str = "previous_response_not_found";
+/// 本地生成 history unavailable 错误时使用的官方提示文本。
+pub(crate) const PREVIOUS_RESPONSE_NOT_FOUND_MESSAGE: &str =
+    "Previous response was not found. Retrying the full request.";
+const INVALID_PREVIOUS_RESPONSE_ID_MESSAGE: &str = "Invalid `previous_response_id`.";
+
+/// 精确匹配上游省略错误码的旧版 previous response 拒绝形状。
+pub(crate) fn is_bare_invalid_previous_response_id_error(
+    code: Option<&Value>,
+    error_type: Option<&str>,
+    message: Option<&str>,
+) -> bool {
+    code.is_none_or(Value::is_null)
+        && error_type == Some("invalid_request_error")
+        && message == Some(INVALID_PREVIOUS_RESPONSE_ID_MESSAGE)
+}
+
 /// Codex Responses 上游请求体。
 ///
 /// 发往上游的 Responses 请求。`body` 持有客户端原始 JSON object，逐字段（含顺序、

backend/crates/providers/openai/src/transport/protocol/websocket.rs

@@ -2,13 +2,12 @@ use gateway_protocol::openai::sse::encode_sse_event;
 use serde::{Deserialize, Serialize};
 use serde_json::{Map, Value};
 
-use crate::transport::protocol::responses::{CodexResponsesRequest, transport_requirement};
+use crate::transport::protocol::responses::{
+    CodexResponsesRequest, PREVIOUS_RESPONSE_NOT_FOUND_CODE,
+    is_bare_invalid_previous_response_id_error, transport_requirement,
+};
 
 const REDACTED_PAYLOAD_VALUE: &str = "<redacted>";
-pub(crate) const PREVIOUS_RESPONSE_NOT_FOUND_CODE: &str = "previous_response_not_found";
-pub(crate) const PREVIOUS_RESPONSE_NOT_FOUND_MESSAGE: &str =
-    "Previous response was not found. Retrying the full request.";
-const INVALID_PREVIOUS_RESPONSE_ID_MESSAGE: &str = "Invalid `previous_response_id`.";
 
 /// WebSocket 握手审计快照。
 #[derive(Debug, Clone, Default, Serialize, Deserialize, PartialEq, Eq)]
@@ -87,11 +86,11 @@ fn previous_response_not_found_event(value: &Value) -> Option<String> {
         return None;
     }
     let error = value.get("error")?.as_object()?;
-    if error.get("code").is_some_and(|code| !code.is_null())
-        || error.get("type").and_then(Value::as_str) != Some("invalid_request_error")
-        || error.get("message").and_then(Value::as_str)
-            != Some(INVALID_PREVIOUS_RESPONSE_ID_MESSAGE)
-    {
+    if !is_bare_invalid_previous_response_id_error(
+        error.get("code"),
+        error.get("type").and_then(Value::as_str),
+        error.get("message").and_then(Value::as_str),
+    ) {
         return None;
     }
 

backend/crates/providers/openai/tests/provider/contract.rs

@@ -412,6 +412,7 @@ fn contract_account_scope() -> Arc<FrozenAccountScope> {
         "acct_truncated_stream",
         "acct_usage_limit_request_path",
         "acct_websocket_close",
+        "acct_websocket_busy_replay",
         "acct_websocket_turn_state",
     ]
     .into_iter()
@@ -510,6 +511,40 @@ fn replay_any_context(request_id: &str, owner_account_id: &str) -> AttemptContex
     .with_continuation_attempt(ContinuationAttempt::ReplayAny)
 }
 
+fn pinned_continuation_context(
+    request_id: &str,
+    account_id: &str,
+    client_previous_response_id: &str,
+    upstream_previous_response_id: &str,
+    attempt_index: u32,
+    continuation_attempt: ContinuationAttempt,
+) -> AttemptContext {
+    let account = ProviderAccountId::new(account_id).expect("account id");
+    let provider = ProviderKind::new("openai").expect("provider");
+    let owner = ProviderAccountStateOwner::new(provider.clone(), account.clone());
+    let continuation = NativeContinuationPin::new(
+        PreviousResponseId::new(client_previous_response_id),
+        PreviousResponseId::new(upstream_previous_response_id),
+        ClientApiKeyId::new("key_openai_contract").expect("client key id"),
+        provider,
+        account,
+    );
+    AttemptContext::new(
+        RequestAttemptContext::new(
+            ModelRequestId::new(request_id).expect("request id"),
+            ClientApiKeyId::new("key_openai_contract").expect("client key id"),
+        ),
+        NonZeroU32::new(attempt_index).expect("attempt index"),
+        SystemTime::now() + Duration::from_secs(30),
+        account_policy(),
+        AccountAttemptContext::new(BTreeSet::new(), None, Some(owner))
+            .with_account_scope(contract_account_scope()),
+        Some(ContinuationBinding::Pinned(continuation)),
+        CancellationToken::new(),
+    )
+    .with_continuation_attempt(continuation_attempt)
+}
+
 fn external_continuation_context(request_id: &str) -> AttemptContext {
     AttemptContext::new(
         RequestAttemptContext::new(
@@ -2919,6 +2954,288 @@ async fn same_account_previous_response_not_found_should_remain_client_visible()
     );
 }
 
+#[tokio::test]
+async fn exact_websocket_busy_then_http_invalid_previous_response_should_request_official_replay() {
+    const ACCOUNT_ID: &str = "acct_websocket_busy_replay";
+    const CONVERSATION_ID: &str = "conversation-websocket-busy-replay";
+    const CLIENT_PREVIOUS_RESPONSE_ID: &str = "client-resp-busy-seed";
+    const UPSTREAM_PREVIOUS_RESPONSE_ID: &str = "resp_busy_seed";
+
+    fn operation(
+        previous_response_id: Option<&str>,
+        session_state: ProviderSessionState,
+    ) -> Operation {
+        let mut body = Map::from_iter([
+            ("model".to_owned(), json!("gpt-5.4")),
+            ("input".to_owned(), json!("hello")),
+        ]);
+        if let Some(previous_response_id) = previous_response_id {
+            body.insert(
+                "previous_response_id".to_owned(),
+                json!(previous_response_id),
+            );
+        }
+        Operation::Generate(
+            GenerateRequest::from_protocol_payload(
+                ProtocolPayload::json_object("openai", body).expect("OpenAI payload"),
+            )
+            .with_provider_session_state(session_state),
+        )
+    }
+
+    let store = Arc::new(MemoryAccountStore::default());
+    create_account(&store, ACCOUNT_ID).await;
+    let listener = TcpListener::bind("127.0.0.1:0")
+        .await
+        .expect("bind upstream listener");
+    let base_url = format!(
+        "http://{}",
+        listener.local_addr().expect("listener address")
+    );
+    let (release_busy_sender, release_busy_receiver) = oneshot::channel();
+    let server = tokio::spawn(async move {
+        let (stream, _) = listener.accept().await.expect("accept WebSocket");
+        let mut websocket = accept_codex_test_websocket(stream).await;
+
+        let seed = websocket
+            .next()
+            .await
+            .expect("seed request")
+            .expect("valid seed request");
+        let Message::Text(seed) = seed else {
+            panic!("seed request must be text");
+        };
+        let seed: Value = serde_json::from_str(&seed).expect("seed request JSON");
+        assert_eq!(seed.get("previous_response_id"), None);
+        websocket
+            .send(Message::Text(
+                json!({
+                    "type": "response.completed",
+                    "response": {
+                        "id": UPSTREAM_PREVIOUS_RESPONSE_ID,
+                        "model": "gpt-5.4",
+                        "status": "completed",
+                        "output": [],
+                        "usage": {
+                            "input_tokens": 1,
+                            "output_tokens": 1,
+                            "total_tokens": 2
+                        }
+                    }
+                })
+                .to_string()
+                .into(),
+            ))
+            .await
+            .expect("complete seed request");
+
+        let busy = websocket
+            .next()
+            .await
+            .expect("busy continuation request")
+            .expect("valid busy continuation request");
+        let Message::Text(busy) = busy else {
+            panic!("busy continuation request must be text");
+        };
+        let busy: Value = serde_json::from_str(&busy).expect("busy continuation JSON");
+        assert_eq!(
+            busy.get("previous_response_id").and_then(Value::as_str),
+            Some(UPSTREAM_PREVIOUS_RESPONSE_ID)
+        );
+        websocket
+            .send(Message::Text(
+                json!({
+                    "type": "response.output_text.delta",
+                    "delta": "the exact WebSocket is busy"
+                })
+                .to_string()
+                .into(),
+            ))
+            .await
+            .expect("send busy stream output");
+
+        let (mut http, _) = listener.accept().await.expect("accept HTTP fallback");
+        read_http_request(&mut http).await;
+        let body = json!({
+            "error": {
+                "code": null,
+                "type": "invalid_request_error",
+                "message": "Invalid `previous_response_id`."
+            }
+        })
+        .to_string();
+        http.write_all(
+            format!(
+                "HTTP/1.1 400 Bad Request\r\ncontent-type: application/json\r\ncontent-length: {}\r\nconnection: close\r\n\r\n{body}",
+                body.len()
+            )
+            .as_bytes(),
+        )
+        .await
+        .expect("write HTTP 400");
+
+        release_busy_receiver.await.expect("release busy stream");
+        websocket
+            .send(Message::Text(
+                json!({
+                    "type": "response.completed",
+                    "response": {
+                        "id": "resp_busy_finished",
+                        "model": "gpt-5.4",
+                        "status": "completed",
+                        "output": [],
+                        "usage": {
+                            "input_tokens": 1,
+                            "output_tokens": 1,
+                            "total_tokens": 2
+                        }
+                    }
+                })
+                .to_string()
+                .into(),
+            ))
+            .await
+            .expect("complete busy stream");
+    });
+    let provider = provider_with_base_url(&store, base_url);
+    let initial_session_state = ProviderSessionState::new(
+        "openai",
+        Map::from_iter([
+            ("account_id".to_owned(), json!(ACCOUNT_ID)),
+            ("conversation_id".to_owned(), json!(CONVERSATION_ID)),
+            ("continuation_scope".to_owned(), json!("connection_local")),
+        ]),
+    )
+    .expect("initial provider session state");
+
+    let mut seed = provider
+        .execute(
+            planned_request("openai", operation(None, initial_session_state)),
+            context("req_ws_busy_seed", CancellationToken::new()),
+        )
+        .await
+        .expect("prepare seed stream");
+    let mut session_state = None;
+    while let Some(event) = seed.next().await {
+        let event = event.expect("seed event");
+        if let Some(update) = event.session_update() {
+            session_state = Some(update.clone());
+        }
+    }
+    let session_state = session_state.expect("seed session update");
+    assert_eq!(
+        session_state
+            .payload()
+            .get("continuation_scope")
+            .and_then(Value::as_str),
+        Some("connection_local")
+    );
+
+    let continuation = operation(Some(CLIENT_PREVIOUS_RESPONSE_ID), session_state);
+    let mut busy = provider
+        .execute(
+            planned_request("openai", continuation.clone()),
+            pinned_continuation_context(
+                "req_ws_busy_owner",
+                ACCOUNT_ID,
+                CLIENT_PREVIOUS_RESPONSE_ID,
+                UPSTREAM_PREVIOUS_RESPONSE_ID,
+                1,
+                ContinuationAttempt::Native,
+            ),
+        )
+        .await
+        .expect("prepare busy continuation stream");
+    loop {
+        let event = busy
+            .next()
+            .await
+            .expect("busy stream event")
+            .expect("valid busy stream event");
+        if event.has_client_event() {
+            break;
+        }
+    }
+
+    let mut exact = provider
+        .execute(
+            planned_request("openai", continuation.clone()),
+            pinned_continuation_context(
+                "req_ws_busy_exact",
+                ACCOUNT_ID,
+                CLIENT_PREVIOUS_RESPONSE_ID,
+                UPSTREAM_PREVIOUS_RESPONSE_ID,
+                1,
+                ContinuationAttempt::Native,
+            ),
+        )
+        .await
+        .expect("prepare competing exact continuation");
+    let exact_error = loop {
+        match exact.next().await {
+            Some(Ok(_)) => {}
+            Some(Err(error)) => break error,
+            None => panic!("competing exact continuation must fail while its socket is busy"),
+        }
+    };
+    assert_eq!(
+        exact_error.continuation_failure(),
+        Some(ContinuationFailure::Busy)
+    );
+
+    let mut replay = provider
+        .execute(
+            planned_request("openai", continuation),
+            pinned_continuation_context(
+                "req_ws_busy_replay",
+                ACCOUNT_ID,
+                CLIENT_PREVIOUS_RESPONSE_ID,
+                UPSTREAM_PREVIOUS_RESPONSE_ID,
+                2,
+                ContinuationAttempt::ReplayAny,
+            ),
+        )
+        .await
+        .expect("prepare replay-any continuation");
+    let replay_error = loop {
+        match replay.next().await {
+            Some(Ok(_)) => {}
+            Some(Err(error)) => break error,
+            None => panic!("HTTP fallback 400 must produce a provider error"),
+        }
+    };
+    let detail = replay_error
+        .client_visible_upstream_error()
+        .expect("client-visible replay error");
+
+    assert_eq!(
+        (
+            replay_error.kind(),
+            replay_error.continuation_failure(),
+            detail.code(),
+            detail.error_type(),
+            detail.message(),
+        ),
+        (
+            ProviderErrorKind::InvalidRequest,
+            Some(ContinuationFailure::HistoryUnavailable),
+            Some("previous_response_not_found"),
+            Some("invalid_request_error"),
+            "Invalid `previous_response_id`.",
+        )
+    );
+
+    release_busy_sender.send(()).expect("release busy stream");
+    while busy
+        .next()
+        .await
+        .transpose()
+        .expect("busy stream event")
+        .is_some()
+    {}
+    server.await.expect("upstream server");
+}
+
 #[tokio::test]
 async fn continuation_prefetch_over_64_kib_should_commit_wire_without_protocol_failure() {
     let store = Arc::new(MemoryAccountStore::default());