From ca90f2b586fe075f5ccb5e8d0fee865b0fcd5411 Mon Sep 17 00:00:00 2001 From: niceSprite Date: Sat, 26 Sep 2026 19:03:40 +0800 Subject: [PATCH] fix(openai): recover oversized Codex requests before WebSocket send (#287) Co-authored-by: Codex --- .../openai/responses/websocket/forward.rs | 18 + .../providers/openai/src/provider/mod.rs | 62 +- .../src/transport/protocol/responses.rs | 9 +- .../src/transport/protocol/websocket.rs | 29 + .../openai/tests/provider/contract.rs | 545 ++++++++++++++++++ docs/api.md | 11 + docs/architecture.md | 3 + 7 files changed, 669 insertions(+), 8 deletions(-) diff --git a/backend/crates/gateway-api/tests/openai/responses/websocket/forward.rs b/backend/crates/gateway-api/tests/openai/responses/websocket/forward.rs index e83c17d97..eaf798466 100644 --- a/backend/crates/gateway-api/tests/openai/responses/websocket/forward.rs +++ b/backend/crates/gateway-api/tests/openai/responses/websocket/forward.rs @@ -522,6 +522,24 @@ async fn initial_quota_recovery_delivers_client_projection_instead_of_upstream_s assert!(error["headers"].get("retry-after").is_none()); } +#[tokio::test] +async fn initial_continuation_recovery_delivers_the_official_client_replay_signal() { + let 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 error = initial_error(EngineError::Provider(provider), Vec::new()).await; + assert_eq!(error["type"], "error"); + assert_eq!(error["status"], 400); + assert_eq!(error["error"]["code"], "previous_response_not_found"); + assert_eq!(error["error"]["type"], "invalid_request_error"); +} + #[tokio::test] async fn locally_exhausted_account_pool_sends_one_official_usage_limit_error() { let provider = ProviderError::new( diff --git a/backend/crates/providers/openai/src/provider/mod.rs b/backend/crates/providers/openai/src/provider/mod.rs index 667e6dde7..cd8dfac5f 100644 --- a/backend/crates/providers/openai/src/provider/mod.rs +++ b/backend/crates/providers/openai/src/provider/mod.rs @@ -79,7 +79,9 @@ use crate::transport::protocol::responses::{ CodexResponsesRequest, PREVIOUS_RESPONSE_NOT_FOUND_CODE, PREVIOUS_RESPONSE_NOT_FOUND_MESSAGE, PreviousResponseScope, ResponseEventSignals, TransportRequirement, transport_requirement, }; -use crate::transport::protocol::websocket::WEBSOCKET_CONNECTION_LIMIT_REACHED_CODE; +use crate::transport::protocol::websocket::{ + WEBSOCKET_CONNECTION_LIMIT_REACHED_CODE, websocket_response_create_payload_len, +}; use crate::transport::request::{ CodexRequestEncodeError, RequestAccountScope, align_structured_location_fields, encode_generate_request, scope_request_to_account, @@ -115,6 +117,8 @@ const PROVIDER_NAME: &str = "openai"; const HTTP_SSE_TRANSPORT: &str = "http_sse"; const HTTP_JSON_TRANSPORT: &str = "http_json"; const WEBSOCKET_TRANSPORT: &str = "websocket"; +// 在已观测到的 Codex OAuth 上游 16 MiB 附近消息边界前留出传输 metadata 余量。 +const WEBSOCKET_HTTP_FALLBACK_THRESHOLD_BYTES: usize = 15 * 1024 * 1024; const MAX_COOKIE_HEADER_BYTES: usize = 16 * 1024; /// 提交边界前最多保留 64 KiB 原始上游 chunk；达到阈值后结束无感换号窗口， /// 但不会把上游数据改写成协议失败。 @@ -701,15 +705,29 @@ impl CodexProvider { { return Err(continuation_replay_required_error("scope_unavailable")); } + if upstream_request + .downstream_websocket_connection_id + .is_some() + && upstream_request.previous_response_id().is_some() + && upstream_request.generate() + && previous_session.as_ref().is_some_and(|state| { + state.continuation_scope == OpenAiContinuationScope::ReplayRequired + }) + { + // HTTP store=false 没有原生续链，完整历史仍由客户端持有；不能把 delta + // 当作独立新请求发送，也不能用 previous_response_id 猜测上游存储状态。 + return Err(continuation_replay_required_error("scope_unavailable")); + } scope_request_to_account( &mut upstream_request, lease.installation_id(), account_scope, ); - if matches!( + let codex_oauth = matches!( lease.authentication(), crate::credential::CodexRuntimeAuthentication::OAuth(_) - ) { + ); + if codex_oauth { normalize_selected_codex_downstream_body( upstream_request.body_mut(), generate.protocol_payload().context(), @@ -743,13 +761,49 @@ impl CodexProvider { && session_affinity .as_ref() .is_some_and(|affinity| self.session_transport_recovery.uses_http(affinity.key())); - let transport = if requirement.requires_websocket() { + let mut transport = if requirement.requires_websocket() { CodexProviderTransport::PreferWebSocket } else if context.transport() == AttemptTransport::Fallback || session_http_fallback { CodexProviderTransport::HttpOnly } else { requested_transport }; + if codex_oauth + && transport == CodexProviderTransport::PreferWebSocket + && (matches!( + requirement, + TransportRequirement::NewChain | TransportRequirement::WebSocketNewChain + ) || (requirement == TransportRequirement::ExactWebSocketContinuation + && upstream_request + .downstream_websocket_connection_id + .is_some())) + { + let payload_bytes = + websocket_response_create_payload_len(&upstream_request).map_err(|_| { + provider_error(ProviderErrorKind::Protocol, UpstreamSendState::NotSent) + })?; + if payload_bytes >= WEBSOCKET_HTTP_FALLBACK_THRESHOLD_BYTES { + if requirement == TransportRequirement::ExactWebSocketContinuation { + tracing::info!( + request_id = context.request_id().as_str(), + payload_bytes, + threshold_bytes = WEBSOCKET_HTTP_FALLBACK_THRESHOLD_BYTES, + continuation_recovery_action = "client_replay_required", + "OpenAI oversized WebSocket continuation requires full client input before send" + ); + return Err(continuation_replay_required_error( + "transport_payload_too_large", + )); + } + transport = CodexProviderTransport::HttpOnly; + tracing::info!( + request_id = context.request_id().as_str(), + payload_bytes, + threshold_bytes = WEBSOCKET_HTTP_FALLBACK_THRESHOLD_BYTES, + "OpenAI oversized new-chain request selected HTTP before WebSocket send" + ); + } + } if transport == CodexProviderTransport::PreferWebSocket && middleware_headers.iter().any(|header| { match HeaderValue::from_bytes(header.value()) { diff --git a/backend/crates/providers/openai/src/transport/protocol/responses.rs b/backend/crates/providers/openai/src/transport/protocol/responses.rs index 493a7fb50..b6daed375 100644 --- a/backend/crates/providers/openai/src/transport/protocol/responses.rs +++ b/backend/crates/providers/openai/src/transport/protocol/responses.rs @@ -121,7 +121,8 @@ pub enum TransportRequirement { HttpRequired, /// `generate=false + store=false` 预热必须保留在同一条 WebSocket。 ExplicitWebSocketWarmup, - /// 客户端 WebSocket 的非持久化新链，必须在池化连接上建立后续续接状态。 + /// 客户端 WebSocket 的非持久化新链，默认在池化连接上建立后续续接状态。 + /// Provider 可在发送前对 OAuth 大新链选择 HTTP，后续由客户端完整重放。 WebSocketNewChain, /// 只能使用持有指定 connection-local response 的精确 WebSocket。 ExactWebSocketContinuation, @@ -134,7 +135,7 @@ pub enum TransportRequirement { } impl TransportRequirement { - /// 是否必须使用 WebSocket，且禁止 HTTP fallback。 + /// 默认是否要求 WebSocket；Provider 的 OAuth 大新链预检可在发送前选择 HTTP。 pub fn requires_websocket(self) -> bool { matches!( self, @@ -189,8 +190,8 @@ pub fn transport_requirement(request: &CodexResponsesRequest) -> TransportRequir TransportRequirement::ExternalUnknown } }, - // 客户端会在下一轮提交 response ID；HTTP store=false 的成功响应无法 - // 在池化 WebSocket 上续接，所以首轮也不能按普通快路径降级到 HTTP。 + // HTTP store=false 的成功响应无法在池化 WebSocket 上续接，默认不走 HTTP 快路径； + // Provider 的 OAuth 大新链预检例外通过客户端完整重放恢复后续请求。 None if request.downstream_websocket_connection_id.is_some() && !request.store() => { TransportRequirement::WebSocketNewChain } diff --git a/backend/crates/providers/openai/src/transport/protocol/websocket.rs b/backend/crates/providers/openai/src/transport/protocol/websocket.rs index 3447e59aa..2d1cf2c5a 100644 --- a/backend/crates/providers/openai/src/transport/protocol/websocket.rs +++ b/backend/crates/providers/openai/src/transport/protocol/websocket.rs @@ -225,6 +225,35 @@ pub fn websocket_response_create_payload_text( }) } +pub(crate) fn websocket_response_create_payload_len( + request: &CodexResponsesRequest, +) -> Result { + struct ByteCount(usize); + + impl std::io::Write for ByteCount { + fn write(&mut self, bytes: &[u8]) -> std::io::Result { + self.0 = self + .0 + .checked_add(bytes.len()) + .ok_or_else(|| std::io::Error::other("WebSocket payload size overflow"))?; + Ok(bytes.len()) + } + + fn flush(&mut self) -> std::io::Result<()> { + Ok(()) + } + } + + let mut count = ByteCount(0); + serde_json::to_writer( + &mut count, + &ResponseCreateFrame { + body: request.body(), + }, + )?; + Ok(count.0) +} + fn websocket_payload_keys(request: &CodexResponsesRequest) -> Vec { let mut keys = Vec::with_capacity(request.body().len() + 1); keys.push("type".to_string()); diff --git a/backend/crates/providers/openai/tests/provider/contract.rs b/backend/crates/providers/openai/tests/provider/contract.rs index ac7467375..cd77ac885 100644 --- a/backend/crates/providers/openai/tests/provider/contract.rs +++ b/backend/crates/providers/openai/tests/provider/contract.rs @@ -4804,6 +4804,551 @@ async fn websocket_upgrade_required_immediately_enables_session_http_fallback() server.await.expect("upstream server"); } +#[tokio::test] +async fn oversized_http_new_chain_uses_http_without_disabling_session_websocket() { + let store = Arc::new(MemoryAccountStore::default()); + create_account(&store, "acct_provider_contract").await; + let server = MockServer::start().await; + Mock::given(method("POST")) + .and(path("/codex/responses")) + .respond_with( + ResponseTemplate::new(200) + .insert_header("content-type", "text/event-stream") + .set_body_string(CAPTURE_COMPLETED_SSE), + ) + .expect(1) + .mount(&server) + .await; + let input = json!([ + {"type": "message", "role": "user", "content": [ + {"type": "input_text", "text": "inspect the screenshot"}, + {"type": "input_image", "image_url": format!("data:image/jpeg;base64,{}", "A".repeat(17 * 1024 * 1024))} + ]}, + {"type": "function_call", "call_id": "call_large", "name": "inspect", "arguments": "{}"}, + {"type": "function_call_output", "call_id": "call_large", "output": "done"} + ]); + let tools = json!([{"type": "function", "name": "inspect", "parameters": {"type": "object"}}]); + let body = Map::from_iter([ + ("model".to_owned(), json!("gpt-5.4")), + ("input".to_owned(), input.clone()), + ("tools".to_owned(), tools.clone()), + ("session_id".to_owned(), json!("oversized-http-session")), + ("stream".to_owned(), json!(true)), + ("store".to_owned(), json!(false)), + ]); + let provider = provider_with_base_url(&store, server.uri()); + let mut stream = Arc::clone(&provider) + .execute( + planned_request( + "openai", + Operation::Generate(GenerateRequest::from_protocol_payload( + ProtocolPayload::json_object("openai", body).unwrap(), + )), + ), + context("req_oversized_http", CancellationToken::new()), + ) + .await + .expect("prepare oversized HTTP request"); + assert_eq!(stream.metadata().transport().as_str(), "http_sse"); + let mut completed = false; + while let Some(event) = stream.next().await { + let event = event.expect("oversized HTTP response"); + completed |= event + .wire_event() + .is_some_and(|wire| wire.event_type() == Some("response.completed")); + } + assert!(completed); + drop(stream); + + let requests = server.received_requests().await.unwrap(); + assert_eq!(requests.len(), 1); + assert_eq!(requests[0].method.as_str(), "POST"); + let sent = captured_request_body(&requests[0]); + assert_eq!(sent["input"], input); + assert_eq!(sent["tools"], tools); + assert!(sent.get("previous_response_id").is_none()); + + let next = provider + .execute( + planned_request( + "openai", + Operation::Generate(generate_with_session_context( + "oversized-http-session", + None, + None, + )), + ), + context("req_small_after_oversized_http", CancellationToken::new()), + ) + .await + .expect("prepare small request in the same session"); + assert_eq!(next.metadata().transport().as_str(), "websocket"); +} + +#[tokio::test] +async fn oversized_http_threshold_counts_utf8_and_json_escaping() { + const THRESHOLD: usize = 15 * 1024 * 1024; + let store = Arc::new(MemoryAccountStore::default()); + create_account(&store, "acct_provider_contract").await; + let server = MockServer::start().await; + Mock::given(method("POST")) + .and(path("/codex/responses")) + .respond_with( + ResponseTemplate::new(200) + .insert_header("content-type", "text/event-stream") + .set_body_string(CAPTURE_COMPLETED_SSE), + ) + .expect(1) + .mount(&server) + .await; + let seed_text = "中文\n\"\\"; + let mut body = Map::from_iter([ + ("model".to_owned(), json!("gpt-5.4")), + ( + "input".to_owned(), + json!([{"type": "message", "role": "user", "content": [{"type": "input_text", "text": seed_text}]}]), + ), + ("session_id".to_owned(), json!("oversized-boundary-session")), + ("stream".to_owned(), json!(true)), + ("store".to_owned(), json!(false)), + ]); + let provider = provider_with_base_url(&store, server.uri()); + let mut seed = Arc::clone(&provider) + .execute( + planned_request( + "openai", + Operation::Generate(GenerateRequest::from_protocol_payload( + ProtocolPayload::json_object("openai", body.clone()) + .unwrap() + .with_context(Map::from_iter([("use_websocket".to_owned(), json!(false))])), + )), + ), + context("req_size_baseline", CancellationToken::new()), + ) + .await + .unwrap(); + while let Some(event) = seed.next().await { + event.expect("baseline HTTP response"); + } + drop(seed); + let requests = server.received_requests().await.unwrap(); + let mut frame = captured_request_body(&requests[0]); + frame["type"] = json!("response.create"); + let baseline_len = serde_json::to_vec(&frame).unwrap().len(); + + for (payload_len, expected) in [(THRESHOLD - 1, "websocket"), (THRESHOLD, "http_sse")] { + let text = format!("{seed_text}{}", "x".repeat(payload_len - baseline_len)); + body.get_mut("input").unwrap()[0]["content"][0]["text"] = json!(text); + let stream = Arc::clone(&provider) + .execute( + planned_request( + "openai", + Operation::Generate(GenerateRequest::from_protocol_payload( + ProtocolPayload::json_object("openai", body.clone()).unwrap(), + )), + ), + context( + &format!("req_boundary_{payload_len}"), + CancellationToken::new(), + ), + ) + .await + .expect("prepare boundary request"); + assert_eq!(stream.metadata().transport().as_str(), expected); + } + assert_eq!(server.received_requests().await.unwrap().len(), 1); +} + +#[tokio::test] +async fn oversized_ws_new_chain_uses_http_but_warmup_and_http_continuation_keep_websocket() { + let store = Arc::new(MemoryAccountStore::default()); + create_account(&store, "acct_provider_contract").await; + let server = MockServer::start().await; + let provider = provider_with_base_url(&store, server.uri()); + let body = Map::from_iter([ + ("model".to_owned(), json!("gpt-5.4")), + ("input".to_owned(), json!([])), + ( + "instructions".to_owned(), + json!("x".repeat(16 * 1024 * 1024)), + ), + ("store".to_owned(), json!(false)), + ]); + for case in ["downstream_websocket", "warmup", "native_continuation"] { + let mut body = body.clone(); + let mut request_context = Map::new(); + let mut attempt = context(&format!("req_large_{case}"), CancellationToken::new()); + match case { + "downstream_websocket" => { + request_context.insert( + "downstream_websocket_connection_id".to_owned(), + json!("ws_large"), + ); + } + "warmup" => { + body.insert("generate".to_owned(), json!(false)); + } + "native_continuation" => { + body.insert("previous_response_id".to_owned(), json!("client_previous")); + attempt = pinned_continuation_context( + "req_large_native_continuation", + "acct_provider_contract", + "client_previous", + "upstream_previous", + 1, + ContinuationAttempt::Native, + ); + } + _ => unreachable!(), + } + let mut generate = GenerateRequest::from_protocol_payload( + ProtocolPayload::json_object("openai", body) + .unwrap() + .with_context(request_context), + ); + if case == "native_continuation" { + generate = generate.with_provider_session_state( + ProviderSessionState::new( + "openai", + Map::from_iter([ + ("account_id".to_owned(), json!("acct_provider_contract")), + ("continuation_scope".to_owned(), json!("connection_local")), + ]), + ) + .unwrap(), + ); + } + let stream = Arc::clone(&provider) + .execute( + planned_request("openai", Operation::Generate(generate)), + attempt, + ) + .await + .expect("prepare required WebSocket request"); + let expected = if case == "downstream_websocket" { + "http_sse" + } else { + "websocket" + }; + assert_eq!(stream.metadata().transport().as_str(), expected, "{case}"); + } + assert!(server.received_requests().await.unwrap().is_empty()); +} + +#[tokio::test] +async fn oversized_exact_websocket_continuation_requests_client_replay_before_send() { + let store = Arc::new(MemoryAccountStore::default()); + create_account(&store, "acct_provider_contract").await; + let server = MockServer::start().await; + let provider = provider_with_base_url(&store, server.uri()); + for input_bytes in [128, 16 * 1024 * 1024] { + let payload = ProtocolPayload::json_object("openai", Map::from_iter([ + ("model".to_owned(), json!("gpt-5.4")), + ("input".to_owned(), json!([{"type": "function_call_output", "call_id": "call_large", "output": "x".repeat(input_bytes)}])), + ("previous_response_id".to_owned(), json!("client_previous")), + ("store".to_owned(), json!(false)), + ])).unwrap().with_context(Map::from_iter([ + ("downstream_websocket_connection_id".to_owned(), json!("ws_native_large")), + ])); + let generate = GenerateRequest::from_protocol_payload(payload).with_provider_session_state( + ProviderSessionState::new( + "openai", + Map::from_iter([ + ("account_id".to_owned(), json!("acct_provider_contract")), + ("continuation_scope".to_owned(), json!("connection_local")), + ]), + ) + .unwrap(), + ); + let result = Arc::clone(&provider) + .execute( + planned_request("openai", Operation::Generate(generate)), + pinned_continuation_context( + &format!("req_native_size_{input_bytes}"), + "acct_provider_contract", + "client_previous", + "upstream_previous", + 1, + ContinuationAttempt::Native, + ), + ) + .await; + if input_bytes == 128 { + assert_eq!( + result + .expect("small native continuation") + .metadata() + .transport() + .as_str(), + "websocket" + ); + } else { + let error = result + .err() + .expect("large native continuation requires full client input"); + assert_eq!( + error.kind(), + ProviderErrorKind::ContinuationRecoveryRequired + ); + assert_eq!(error.send_state(), UpstreamSendState::NotSent); + assert_eq!( + error.continuation_recovery_disposition(), + Some(ContinuationRecoveryDisposition::ClientReplayRequired) + ); + assert_eq!( + error.client_visible_upstream_error().unwrap().code(), + Some("previous_response_not_found") + ); + assert!(!error.allows_pre_delivery_retry()); + } + } + assert!(server.received_requests().await.unwrap().is_empty()); +} + +#[tokio::test] +async fn oversized_api_key_new_chains_preserve_explicit_websocket_transport() { + let upstream = MockServer::start().await; + let store = Arc::new(MemoryAccountStore::default()); + store + .seed_api_key( + "acct_provider_contract", + upstream.uri(), + provider_openai::credential::ResponsesTransport::PreferWebsocket, + ) + .await; + let provider = provider_with_base_url(&store, upstream.uri()); + for downstream_websocket in [false, true] { + let mut payload = ProtocolPayload::json_object( + "openai", + Map::from_iter([ + ("model".to_owned(), json!("gpt-5.4")), + ("input".to_owned(), json!([])), + ( + "instructions".to_owned(), + json!("x".repeat(16 * 1024 * 1024)), + ), + ("store".to_owned(), json!(false)), + ]), + ) + .unwrap(); + if downstream_websocket { + payload = payload.with_context(Map::from_iter([( + "downstream_websocket_connection_id".to_owned(), + json!("ws_api_large"), + )])); + } + let stream = Arc::clone(&provider) + .execute( + planned_request( + "openai", + Operation::Generate(GenerateRequest::from_protocol_payload(payload)), + ), + context( + &format!("req_large_api_new_{downstream_websocket}"), + CancellationToken::new(), + ), + ) + .await + .expect("API Key new chain preserves configured transport"); + assert_eq!(stream.metadata().transport().as_str(), "websocket"); + } + assert!(upstream.received_requests().await.unwrap().is_empty()); +} + +#[tokio::test] +async fn oversized_api_key_native_continuation_preserves_websocket_transport() { + let upstream = MockServer::start().await; + let store = Arc::new(MemoryAccountStore::default()); + store + .seed_api_key( + "acct_provider_contract", + upstream.uri(), + provider_openai::credential::ResponsesTransport::PreferWebsocket, + ) + .await; + let payload = ProtocolPayload::json_object("openai", Map::from_iter([ + ("model".to_owned(), json!("gpt-5.4")), + ("input".to_owned(), json!([{"type": "function_call_output", "call_id": "call_api_large", "output": "x".repeat(16 * 1024 * 1024)}])), + ("previous_response_id".to_owned(), json!("client_api_previous")), + ("store".to_owned(), json!(false)), + ])).unwrap().with_context(Map::from_iter([ + ("downstream_websocket_connection_id".to_owned(), json!("ws_api_large")), + ])); + let generate = GenerateRequest::from_protocol_payload(payload).with_provider_session_state( + ProviderSessionState::new( + "openai", + Map::from_iter([ + ("account_id".to_owned(), json!("acct_provider_contract")), + ("continuation_scope".to_owned(), json!("connection_local")), + ]), + ) + .unwrap(), + ); + let stream = provider_with_base_url(&store, upstream.uri()) + .execute( + planned_request("openai", Operation::Generate(generate)), + pinned_continuation_context( + "req_large_api_native", + "acct_provider_contract", + "client_api_previous", + "upstream_api_previous", + 1, + ContinuationAttempt::Native, + ), + ) + .await + .expect("API Key native continuation is not rejected by the Codex size policy"); + assert_eq!(stream.metadata().transport().as_str(), "websocket"); + assert!(upstream.received_requests().await.unwrap().is_empty()); +} + +#[tokio::test] +async fn oversized_websocket_http_response_continues_with_client_tool_history_replay() { + fn ws_operation( + body: Map, + connection_id: &str, + state: Option, + ) -> Operation { + let payload = ProtocolPayload::json_object("openai", body) + .unwrap() + .with_context(Map::from_iter([( + "downstream_websocket_connection_id".to_owned(), + json!(connection_id), + )])); + let mut generate = GenerateRequest::from_protocol_payload(payload); + if let Some(state) = state { + generate = generate.with_provider_session_state(state); + } + Operation::Generate(generate) + } + + let store = Arc::new(MemoryAccountStore::default()); + create_account(&store, "acct_provider_contract").await; + let server = MockServer::start().await; + Mock::given(method("POST")).and(path("/codex/responses")) + .respond_with(|request: &wiremock::Request| { + let body = captured_request_body(request); + let has_tool_output = body["input"].as_array().unwrap().iter().any(|item| item["type"] == "function_call_output"); + let event = if has_tool_output { + CAPTURE_COMPLETED_SSE.to_owned() + } else { + format!("event: response.completed\ndata: {}\n\n", json!({ + "type": "response.completed", + "response": { + "id": "resp_large_ws_http", "model": "gpt-5.4", "status": "completed", "store": false, + "output": [{"type": "function_call", "id": "fc_large", "call_id": "call_large", "name": "inspect", "arguments": "{}", "status": "completed"}], + "usage": {"input_tokens": 1, "output_tokens": 1, "total_tokens": 2} + } + })) + }; + ResponseTemplate::new(200).insert_header("content-type", "text/event-stream").set_body_string(event) + }).expect(2).mount(&server).await; + let mut body = Map::from_iter([ + ("model".to_owned(), json!("gpt-5.4")), + ( + "instructions".to_owned(), + json!("x".repeat(16 * 1024 * 1024)), + ), + ( + "input".to_owned(), + json!([{"type": "message", "role": "user", "content": [{"type": "input_text", "text": "inspect"}]}]), + ), + ( + "tools".to_owned(), + json!([{"type": "function", "name": "inspect", "parameters": {"type": "object"}}]), + ), + ("session_id".to_owned(), json!("large-ws-http-session")), + ("store".to_owned(), json!(false)), + ]); + let provider = provider_with_base_url(&store, server.uri()); + let mut first = Arc::clone(&provider) + .execute( + planned_request("openai", ws_operation(body.clone(), "ws_first", None)), + context("req_large_ws_first", CancellationToken::new()), + ) + .await + .expect("prepare large WS request over HTTP"); + assert_eq!(first.metadata().transport().as_str(), "http_sse"); + let mut state = None; + let mut output = None; + while let Some(event) = first.next().await { + let event = event.expect("HTTP tool response"); + if let Some(update) = event.session_update() { + state = Some(update.clone()); + } + if let Some(wire) = event.wire_event() + && wire.event_type() == Some("response.completed") + { + output = wire.data().pointer("/response/output").cloned(); + } + } + drop(first); + let state = state.expect("HTTP response publishes replay requirement"); + assert_eq!(state.payload()["continuation_scope"], "replay_required"); + let tool_output = + json!({"type": "function_call_output", "call_id": "call_large", "output": "done"}); + let delta = Map::from_iter([ + ("model".to_owned(), json!("gpt-5.4")), + ("input".to_owned(), json!([tool_output.clone()])), + ( + "previous_response_id".to_owned(), + json!("resp_large_ws_http"), + ), + ("store".to_owned(), json!(false)), + ]); + let error = Arc::clone(&provider) + .execute( + planned_request("openai", ws_operation(delta, "ws_first", Some(state))), + pinned_continuation_context( + "req_large_ws_delta", + "acct_provider_contract", + "resp_large_ws_http", + "resp_large_ws_http", + 1, + ContinuationAttempt::Native, + ), + ) + .await + .err() + .expect("HTTP store=false delta requires client replay"); + assert_eq!(error.send_state(), UpstreamSendState::NotSent); + assert_eq!( + error.client_visible_upstream_error().unwrap().code(), + Some("previous_response_not_found") + ); + assert_eq!( + error.continuation_recovery_disposition(), + Some(ContinuationRecoveryDisposition::ClientReplayRequired) + ); + assert_eq!(server.received_requests().await.unwrap().len(), 1); + + let input = body.get_mut("input").unwrap().as_array_mut().unwrap(); + input.extend(output.unwrap().as_array().unwrap().iter().cloned()); + input.push(tool_output); + let mut replay = provider + .execute( + planned_request("openai", ws_operation(body.clone(), "ws_reconnected", None)), + context("req_large_ws_replay", CancellationToken::new()), + ) + .await + .expect("client full history replay over HTTP"); + assert_eq!(replay.metadata().transport().as_str(), "http_sse"); + let mut completed = false; + while let Some(event) = replay.next().await { + completed |= event + .expect("full history HTTP response") + .wire_event() + .is_some_and(|wire| wire.event_type() == Some("response.completed")); + } + assert!(completed); + let requests = server.received_requests().await.unwrap(); + assert_eq!(requests.len(), 2); + let sent = captured_request_body(&requests[1]); + assert_eq!(sent["input"], body["input"]); + assert_eq!(sent["tools"], body["tools"]); + assert!(sent.get("previous_response_id").is_none()); +} + #[tokio::test] async fn fallback_attempt_transport_forces_http_sse_for_a_websocket_request() { let store = Arc::new(MemoryAccountStore::default()); diff --git a/docs/api.md b/docs/api.md index 48bf208b0..63d3232c9 100644 --- a/docs/api.md +++ b/docs/api.md @@ -261,6 +261,17 @@ OAuth 账号默认 `prefer_websocket`，客户端使用 HTTP/SSE 时仍可能选 客户端配置的 `supports_websockets` 只控制第一段连接，不是服务端传输策略开关。 上游在响应终态前发送 Close 1000 仍属于失败，不能按“正常关闭”计为成功。 +Codex OAuth backend 的候选上游为 WS 时，无 `previous_response_id` 的普通新链若规范化 +`response.create` 达到 15 MiB，发送前选择 HTTP/SSE；HTTP 和 WS 入站均适用，下游交付协议不变。 +该阈值为已观察到的上游消息大小边界预留余量，不是网关输入长度上限或 OpenAI 公布的统一限制； +API Key 账号不使用此大小策略。小请求、显式 warmup、未知外部 previous ID 和持久化续接保持原行为。 + +OAuth 的大 connection-local WS 续接，以及 HTTP `store=false` 成功后标记为 `ReplayRequired` +的 WS 增量，均在发送前返回 `status: 400 / previous_response_not_found`。客户端应清除旧 ID， +携带完整历史及工具调用／输出重试。官方 Codex 支持重连 WS 后完整重发，也可能按自身重试预算 +切到 HTTP；其他客户端需要自行实现此合同。代理不缓存 transcript，不把增量输入当作独立新链， +不自动重放发送结果不确定的请求；HTTP 链后续步骤可能增加一次恢复信号、重连及全量上传。 + 已建立模型执行的 Responses、Images 和 Search HTTP 响应按以下规则返回关联 ID： `x-gateway-request-id` 为模型执行 ID；`x-request-id` 保留有效上游值，只有上游 `x-oai-request-id` 时复用其值，没有上游 ID 时使用模型执行 ID。`x-oai-request-id` 不是必需字段， diff --git a/docs/architecture.md b/docs/architecture.md index 341c247e6..f01a5d138 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -473,6 +473,9 @@ Client Key 与账号分组形成授权范围： Continuation 仍受原请求的 Client Key、账号范围、Provider 和发送/交付边界约束： - native continuation 固定创建它的 Provider 与账号； +- OpenAI Provider 的 OAuth 大包传输预检在发送前执行，新链可选择 HTTP；大 connection-local + 续接和 HTTP `store=false` 后不可原生续写的 WS 增量返回 `ClientReplayRequired`，由客户端重发 + 完整历史。体积测量复用 WS 编码结构，Provider session state 仍只携带元数据，不保存 transcript； - OpenAI 在交付前收到可安全重放的明确额度拒绝时，先隔离账号，再投影 `ClientReplayRequired`； 丢弃未交付的原错误帧，由客户端提交完整历史开启新链，不把原增量输入交给其他账号。 真实错误分类、状态码、发送状态和上游诊断保持不变，客户端合同见 [Responses API](api.md#3-openai-数据面与模型目录)；