fix(openai): recover oversized Codex requests before WebSocket send (#287)

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

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(

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

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
         }

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<usize, serde_json::Error> {
+    struct ByteCount(usize);
+
+    impl std::io::Write for ByteCount {
+        fn write(&mut self, bytes: &[u8]) -> std::io::Result<usize> {
+            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<String> {
     let mut keys = Vec::with_capacity(request.body().len() + 1);
     keys.push("type".to_string());

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<String, Value>,
+        connection_id: &str,
+        state: Option<ProviderSessionState>,
+    ) -> 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());

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` 不是必需字段，

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-数据面与模型目录)；