diff --git a/backend/crates/providers/openai/src/provider.rs b/backend/crates/providers/openai/src/provider.rs index ece8ce8b9..5068d3377 100644 --- a/backend/crates/providers/openai/src/provider.rs +++ b/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, diff --git a/backend/crates/providers/openai/src/transport/diagnostics.rs b/backend/crates/providers/openai/src/transport/diagnostics.rs index 8d2d6ce16..b317b299d 100644 --- a/backend/crates/providers/openai/src/transport/diagnostics.rs +++ b/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::(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)) diff --git a/backend/crates/providers/openai/src/transport/protocol/responses.rs b/backend/crates/providers/openai/src/transport/protocol/responses.rs index bb8149c85..30ea5ec0b 100644 --- a/backend/crates/providers/openai/src/transport/protocol/responses.rs +++ b/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，逐字段（含顺序、 diff --git a/backend/crates/providers/openai/src/transport/protocol/websocket.rs b/backend/crates/providers/openai/src/transport/protocol/websocket.rs index fe7b4b121..caed6cedc 100644 --- a/backend/crates/providers/openai/src/transport/protocol/websocket.rs +++ b/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 = ""; -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 { 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; } diff --git a/backend/crates/providers/openai/tests/provider/contract.rs b/backend/crates/providers/openai/tests/provider/contract.rs index 89d14e825..64eb61bf7 100644 --- a/backend/crates/providers/openai/tests/provider/contract.rs +++ b/backend/crates/providers/openai/tests/provider/contract.rs @@ -412,6 +412,7 @@ fn contract_account_scope() -> Arc { "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());