refactor(openai): simplify connection and endpoint module files

Restore connection.rs for production and integration tests by folding the
heartbeat tests into the existing connection test module. Keep all heartbeat
coverage while satisfying the workspace layout rules without a subdirectory.

Rename provider_endpoint.rs to endpoint.rs in production and tests. Update
module declarations, imports and frozen file manifests to match.

Validation: gateway application tests (36 passed), all gateway-api tests
(263 passed), workspace rustfmt, and strict gateway-api Clippy with all
targets and features.

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

.github/workflows/release.yml

@@ -626,7 +626,7 @@ jobs:
           cosign sign --yes "${images[@]}"
 
       - name: Publish GitHub Release
-        uses: softprops/action-gh-release@3d0d9888cb7fd7b750713d6e236d1fcb99157228 # v3.0.2
+        uses: softprops/action-gh-release@efb35369e0ad2afab669f228072c1b0d510eae64 # v3.0.3
         with:
           tag_name: ${{ needs.release-info.outputs.tag }}
           generate_release_notes: true

backend/Cargo.lock

@@ -154,9 +154,9 @@ dependencies = [
 
 [[package]]
 name = "aws-lc-rs"
-version = "1.16.2"
+version = "1.18.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "a054912289d18629dc78375ba2c3726a3afe3ff71b4edba9dedfca0e3446d1fc"
+checksum = "b281d307588d634de920874890732659e2e7672f72b5e10e81badc1a8a83621e"
 dependencies = [
  "aws-lc-sys",
  "untrusted 0.7.1",
@@ -165,14 +165,15 @@ dependencies = [
 
 [[package]]
 name = "aws-lc-sys"
-version = "0.39.0"
+version = "0.45.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "1fa7e52a4c5c547c741610a2c6f123f3881e409b714cd27e6798ef020c514f0a"
+checksum = "9bff6c3b54fad79a2e60b8102caf565819711497c1f5f092f49508e2f5c31b27"
 dependencies = [
  "cc",
  "cmake",
  "dunce",
  "fs_extra",
+ "pkg-config",
 ]
 
 [[package]]
@@ -400,9 +401,9 @@ dependencies = [
 
 [[package]]
 name = "aws-smithy-runtime-api"
-version = "1.15.0"
+version = "1.16.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "954c563ce84507722d2679f07a35d21b9c6466b3872d513020d0281fc8112ac9"
+checksum = "9c054752dd9e4dc73d0b75748c99ac2d0feafbf2f25c7b0516f03a3534161223"
 dependencies = [
  "aws-smithy-async",
  "aws-smithy-runtime-api-macros",
@@ -440,9 +441,9 @@ dependencies = [
 
 [[package]]
 name = "aws-smithy-types"
-version = "1.6.2"
+version = "1.6.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "fce83ce9abbb198d25bc7131e468d0f9fe1257125e58c39f3f9fc9f5098c9647"
+checksum = "8f94d16e797ec62cd999fc9d5942b48fa7050c3093ddadff48e4d7528d16fcb9"
 dependencies = [
  "base64-simd",
  "bytes",
@@ -1257,9 +1258,9 @@ checksum = "5baebc0774151f905a1a2cc41989300b1e6fbb29aff0ceffa1064fdd3088d582"
 
 [[package]]
 name = "flate2"
-version = "1.1.9"
+version = "1.1.10"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "843fba2746e448b37e26a819579957415c8cef339bf08564fe8b7ddbd959573c"
+checksum = "6e634e2e0ebac1ee034020da1ca582e17ffe4e0f5e985823721e168928136dcb"
 dependencies = [
  "crc32fast",
  "miniz_oxide",
@@ -1466,7 +1467,7 @@ dependencies = [
  "tokio",
  "tokio-tungstenite 0.29.0",
  "tower",
- "tower-http 0.7.0",
+ "tower-http 0.7.1",
  "tracing",
  "url",
  "uuid",
@@ -1835,9 +1836,9 @@ dependencies = [
 
 [[package]]
 name = "hyper"
-version = "1.8.1"
+version = "1.11.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "2ab2d4f250c3d7b1c9fcdff1cece94ea4e2dfbec68614f7b87cb205f24ca9d11"
+checksum = "27b501faa50e7a26c3d3560ca625132f4078a17771f4810baf70475ae48cbe43"
 dependencies = [
  "atomic-waker",
  "bytes",
@@ -1850,7 +1851,6 @@ dependencies = [
  "httpdate",
  "itoa",
  "pin-project-lite",
- "pin-utils",
  "smallvec",
  "tokio",
  "want",
@@ -2213,9 +2213,9 @@ dependencies = [
 
 [[package]]
 name = "miniz_oxide"
-version = "0.8.9"
+version = "0.9.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "1fa76a2c86f704bdb222d66965fb3d63269ce38518b83cb0575fca855ebb6316"
+checksum = "b63fbc4a50860e98e7b2aa7804ded1db5cbc3aff9193adaff57a6931bf7c4b4c"
 dependencies = [
  "adler2",
  "simd-adler32",
@@ -3026,9 +3026,9 @@ dependencies = [
 
 [[package]]
 name = "rustls-native-certs"
-version = "0.8.3"
+version = "0.8.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "612460d5f7bea540c490b2b6395d8e34a953e52b491accd6c86c8164c5932a63"
+checksum = "dab5152771c58876a2146916e53e35057e1a4dfa2b9df0f0305b07f611fdea4d"
 dependencies = [
  "openssl-probe",
  "rustls-pki-types",
@@ -3038,9 +3038,9 @@ dependencies = [
 
 [[package]]
 name = "rustls-pki-types"
-version = "1.14.0"
+version = "1.15.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "be040f8b0a225e40375822a563fa9524378b9d63112f53e19ffff34df5d33fdd"
+checksum = "2f4925028c7eb5d1fcdaf196971378ed9d2c1c4efc7dc5d011256f76c99c0a96"
 dependencies = [
  "web-time",
  "zeroize",
@@ -3704,7 +3704,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "32497e9a4c7b38532efcdebeef879707aa9f794296a4f0244f6f69e9bc8574bd"
 dependencies = [
  "fastrand",
- "getrandom 0.3.4",
+ "getrandom 0.4.3",
  "once_cell",
  "rustix",
  "windows-sys 0.61.2",
@@ -3805,9 +3805,9 @@ checksum = "1f3ccbac311fea05f86f61904b462b55fb3df8837a366dfc601a0161d0532f20"
 
 [[package]]
 name = "tokio"
-version = "1.52.3"
+version = "1.53.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "8fc7f01b389ac15039e4dc9531aa973a135d7a4135281b12d7c1bc79fd57fffe"
+checksum = "202caea871b69668250d242070849eb495be178ed697a3e98aebce5bc81a0bed"
 dependencies = [
  "bytes",
  "libc",
@@ -3928,9 +3928,9 @@ dependencies = [
 
 [[package]]
 name = "tower-http"
-version = "0.7.0"
+version = "0.7.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "b11f75e912b0c2be01b63d8cf8057b8c3f97cf34abb3d431a3a4c8675498e233"
+checksum = "08a05a66a4fdd61cbbe0a1d755ffe0ca6aba159dd4820936a0ff8a8278245b9c"
 dependencies = [
  "bitflags",
  "bytes",

backend/crates/gateway-api/src/openai/images/http.rs

@@ -15,8 +15,8 @@ use serde_json::Value;
 use crate::ApiState;
 use crate::openai::{
     auth::{authenticate_client, client_access_error_response},
+    endpoint::collect_raw_json_response,
     error::gateway_error_response,
-    provider_endpoint::collect_raw_json_response,
     responses::{OpenAiRequestHeaders, request_client_context},
 };
 

backend/crates/gateway-api/src/openai/mod.rs

@@ -1,10 +1,10 @@
 //! OpenAI 客户端协议 adapter。
 
 pub mod auth;
+mod endpoint;
 pub mod error;
 pub mod images;
 pub mod models;
-mod provider_endpoint;
 pub mod responses;
 pub mod router;
 pub mod search;

backend/crates/gateway-api/src/openai/responses/websocket/connection.rs

@@ -23,6 +23,7 @@ const OUTBOUND_COMMAND_BUFFER: usize = 32;
 const INBOUND_EVENT_BUFFER: usize = 32;
 const DOWNSTREAM_WRITE_TIMEOUT: Duration = Duration::from_secs(5 * 60);
 const CONNECTION_MAX_AGE: Duration = Duration::from_secs(60 * 60);
+const DOWNSTREAM_PING_INTERVAL: Duration = Duration::from_secs(25);
 
 /// WebSocket pump 的写入与生命周期预算。
 #[derive(Clone, Copy)]
@@ -171,9 +172,27 @@ struct ConnectionStats {
     command_backpressure_count: AtomicU64,
     ping_received_count: AtomicU64,
     pong_written_count: AtomicU64,
+    ping_written_count: AtomicU64,
+    pong_received_count: AtomicU64,
+    last_read_ms: AtomicU64,
+    last_write_ms: AtomicU64,
 }
 
 impl ConnectionStats {
+    fn record_read(&self, opened_at: Instant) {
+        self.last_read_ms.store(
+            u64::try_from(opened_at.elapsed().as_millis()).unwrap_or(u64::MAX),
+            Ordering::Relaxed,
+        );
+    }
+
+    fn record_write(&self, opened_at: Instant) {
+        self.last_write_ms.store(
+            u64::try_from(opened_at.elapsed().as_millis()).unwrap_or(u64::MAX),
+            Ordering::Relaxed,
+        );
+    }
+
     fn observe_command_queue(&self, sender: &mpsc::Sender<ConnectionCommand>) {
         let queued = OUTBOUND_COMMAND_BUFFER
             .saturating_sub(sender.capacity())
@@ -400,10 +419,15 @@ impl ResponsesWebSocketConnection {
     }
 
     pub(super) fn log_summary(&self, request_count: u64) {
+        let connection_age_ms = self.age().as_millis();
+        let read_idle_ms = connection_age_ms
+            .saturating_sub(u128::from(self.stats.last_read_ms.load(Ordering::Relaxed)));
+        let write_idle_ms = connection_age_ms
+            .saturating_sub(u128::from(self.stats.last_write_ms.load(Ordering::Relaxed)));
         tracing::info!(
             websocket_connection_id = %self.connection_id,
             request_count,
-            connection_age_ms = self.age().as_millis(),
+            connection_age_ms,
             pump_exit_reason = self
                 .exit_reason
                 .map_or("unknown", PumpExitReason::as_str),
@@ -417,6 +441,10 @@ impl ResponsesWebSocketConnection {
                 .load(Ordering::Relaxed),
             ping_received_count = self.stats.ping_received_count.load(Ordering::Relaxed),
             pong_written_count = self.stats.pong_written_count.load(Ordering::Relaxed),
+            ping_written_count = self.stats.ping_written_count.load(Ordering::Relaxed),
+            pong_received_count = self.stats.pong_received_count.load(Ordering::Relaxed),
+            read_idle_ms,
+            write_idle_ms,
             "Responses WebSocket disconnected"
         );
     }
@@ -493,6 +521,12 @@ async fn run_pump<S, E>(
     let deadline = tokio::time::sleep_until(opened_at + config.max_age);
     tokio::pin!(deadline);
     let mut deadline_elapsed = false;
+    let mut heartbeat = tokio::time::interval_at(
+        opened_at + DOWNSTREAM_PING_INTERVAL,
+        DOWNSTREAM_PING_INTERVAL,
+    );
+    heartbeat.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
+    let mut ping_sequence = 0_u64;
     let reason = loop {
         tokio::select! {
             biased;
@@ -531,6 +565,9 @@ async fn run_pump<S, E>(
                     ConnectionWriteError::Transport { message }
                 });
                 let failed = result.is_err();
+                if !failed {
+                    stats.record_write(opened_at);
+                }
                 let _ = acknowledged.send(result);
                 if failed {
                     break PumpExitReason::WriteError;
@@ -539,45 +576,79 @@ async fn run_pump<S, E>(
                     break PumpExitReason::ServerClose;
                 }
             }
-            message = socket.next() => match message {
-                Some(Ok(Message::Text(payload))) => {
-                    if let Err(reason) = emit_incoming(&incoming, ConnectionEvent::Text(payload.to_string())) {
-                        break reason;
+            // 官方 Codex WsStream 在独立 pump 中回复 Pong，且不将控制帧交给业务流。
+            // 这里只为下游链路保活，不设置 Pong deadline，也不延长业务请求预算。
+            _ = heartbeat.tick() => {
+                ping_sequence = ping_sequence.wrapping_add(1);
+                let ping = Message::Ping(ping_sequence.to_be_bytes().to_vec().into());
+                let result = tokio::select! {
+                    biased;
+                    () = cancellation.cancelled() => break PumpExitReason::LifecycleShutdown,
+                    result = timeout(config.write_timeout, socket.send(ping)) => result,
+                };
+                match result {
+                    Ok(Ok(())) => {
+                        stats.ping_written_count.fetch_add(1, Ordering::Relaxed);
+                        stats.record_write(opened_at);
                     }
-                }
-                Some(Ok(Message::Binary(_))) => {
-                    if let Err(reason) = emit_incoming(&incoming, ConnectionEvent::Binary) {
-                        break reason;
+                    Ok(Err(error)) => {
+                        tracing::info!(
+                            websocket_connection_id = %connection_id,
+                            error = %error,
+                            "Responses WebSocket Ping write failed"
+                        );
+                        break PumpExitReason::WriteError;
                     }
+                    Err(_) => break PumpExitReason::WriteTimeout,
                 }
-                Some(Ok(Message::Ping(payload))) => {
-                    stats.ping_received_count.fetch_add(1, Ordering::Relaxed);
-                    match timeout(config.write_timeout, socket.send(Message::Pong(payload))).await {
-                        Ok(Ok(())) => {
-                            stats.pong_written_count.fetch_add(1, Ordering::Relaxed);
+            }
+            message = socket.next() => {
+                if matches!(&message, Some(Ok(_))) {
+                    stats.record_read(opened_at);
+                }
+                match message {
+                    Some(Ok(Message::Text(payload))) => {
+                        if let Err(reason) = emit_incoming(&incoming, ConnectionEvent::Text(payload.to_string())) {
+                            break reason;
                         }
-                        Ok(Err(error)) => {
-                            tracing::info!(
-                                websocket_connection_id = %connection_id,
-                                error = %error,
-                                "Responses WebSocket Pong write failed"
-                            );
-                            break PumpExitReason::WriteError;
-                        }
-                        Err(_) => break PumpExitReason::WriteTimeout,
                     }
+                    Some(Ok(Message::Binary(_))) => {
+                        if let Err(reason) = emit_incoming(&incoming, ConnectionEvent::Binary) {
+                            break reason;
+                        }
+                    }
+                    Some(Ok(Message::Ping(payload))) => {
+                        stats.ping_received_count.fetch_add(1, Ordering::Relaxed);
+                        match timeout(config.write_timeout, socket.send(Message::Pong(payload))).await {
+                            Ok(Ok(())) => {
+                                stats.pong_written_count.fetch_add(1, Ordering::Relaxed);
+                                stats.record_write(opened_at);
+                            }
+                            Ok(Err(error)) => {
+                                tracing::info!(
+                                    websocket_connection_id = %connection_id,
+                                    error = %error,
+                                    "Responses WebSocket Pong write failed"
+                                );
+                                break PumpExitReason::WriteError;
+                            }
+                            Err(_) => break PumpExitReason::WriteTimeout,
+                        }
+                    }
+                    Some(Ok(Message::Pong(_))) => {
+                        stats.pong_received_count.fetch_add(1, Ordering::Relaxed);
+                    }
+                    Some(Ok(Message::Close(_))) => break PumpExitReason::ClientClose,
+                    Some(Err(error)) => {
+                        tracing::info!(
+                            websocket_connection_id = %connection_id,
+                            error = %error,
+                            "Responses WebSocket receive failed"
+                        );
+                        break PumpExitReason::ReadError;
+                    }
+                    None => break PumpExitReason::PeerEof,
                 }
-                Some(Ok(Message::Pong(_))) => {}
-                Some(Ok(Message::Close(_))) => break PumpExitReason::ClientClose,
-                Some(Err(error)) => {
-                    tracing::info!(
-                        websocket_connection_id = %connection_id,
-                        error = %error,
-                        "Responses WebSocket receive failed"
-                    );
-                    break PumpExitReason::ReadError;
-                }
-                None => break PumpExitReason::PeerEof,
             },
         }
     };

backend/crates/gateway-api/src/openai/responses/websocket/mod.rs

@@ -247,7 +247,7 @@ async fn serve_responses_websocket(socket: WebSocket, session: ResponsesWebSocke
             }
         };
         started.session.trace().record("client.connection", serde_json::json!({
-            "transport": "websocket", "connectionId": connection.id(), "correlationId": correlation_id,
+            "transport": "websocket", "connectionId": connection.id(), "correlationId": correlation_id.as_ref(),
             "requestIndex": request_count,
         }));
         started

backend/crates/gateway-api/src/openai/search/http.rs

@@ -14,8 +14,8 @@ use gateway_core::operation::{Operation, RawJsonPayload, StandaloneSearchRequest
 use crate::ApiState;
 use crate::openai::{
     auth::{authenticate_client, client_access_error_response},
+    endpoint::collect_raw_json_response,
     error::gateway_error_response,
-    provider_endpoint::collect_raw_json_response,
     responses::{OpenAiRequestHeaders, request_client_context},
 };
 

backend/crates/gateway-api/tests/architecture.rs

@@ -61,7 +61,7 @@ fn source_tree_should_match_frozen_machine_manifest() {
         "src/openai/images/mod.rs",
         "src/openai/mod.rs",
         "src/openai/models.rs",
-        "src/openai/provider_endpoint.rs",
+        "src/openai/endpoint.rs",
         "src/openai/responses/error.rs",
         "src/openai/responses/http.rs",
         "src/openai/responses/mod.rs",
@@ -107,7 +107,7 @@ fn test_tree_should_match_frozen_rust_mirror() {
         "tests/openai/images/mod.rs",
         "tests/openai/mod.rs",
         "tests/openai/models.rs",
-        "tests/openai/provider_endpoint.rs",
+        "tests/openai/endpoint.rs",
         "tests/openai/responses/http.rs",
         "tests/openai/responses/mod.rs",
         "tests/openai/responses/websocket/connection.rs",

backend/crates/gateway-api/tests/openai/mod.rs

@@ -1,8 +1,8 @@
 mod auth;
+mod endpoint;
 mod error;
 mod images;
 mod models;
-mod provider_endpoint;
 mod responses;
 mod router;
 mod search;

backend/crates/gateway-api/tests/openai/responses/websocket/connection.rs

@@ -444,21 +444,6 @@ async fn inbound_overload_should_close_without_executing_queued_frames() {
     ));
 }
 
-#[tokio::test(start_paused = true)]
-async fn pump_does_not_emit_an_active_ping() {
-    let PumpHarness {
-        connection,
-        incoming: _incoming,
-        mut written,
-        ..
-    } = test_connection(false);
-    tokio::time::advance(Duration::from_secs(5 * 60)).await;
-    tokio::task::yield_now().await;
-
-    assert!(matches!(written.try_recv(), Err(TryRecvError::Empty)));
-    drop(connection);
-}
-
 #[tokio::test]
 async fn outbound_commands_are_written_in_acknowledged_order() {
     let PumpHarness {
@@ -582,13 +567,17 @@ async fn idle_connection_reaches_the_official_limit_without_starting_an_executio
 
     tokio::time::advance(Duration::from_secs(60 * 60)).await;
     tokio::task::yield_now().await;
-    let message = tokio::time::timeout(Duration::from_secs(1), socket.next())
-        .await
-        .expect("connection limit response timeout")
-        .expect("connection remains available for the limit error")
-        .expect("valid WebSocket frame");
-    let ClientMessage::Text(text) = message else {
-        panic!("connection limit must be a text error event");
+    let text = loop {
+        let message = tokio::time::timeout(Duration::from_secs(1), socket.next())
+            .await
+            .expect("connection limit response timeout")
+            .expect("connection remains available for the limit error")
+            .expect("valid WebSocket frame");
+        match message {
+            ClientMessage::Text(text) => break text,
+            ClientMessage::Ping(_) => {}
+            other => panic!("unexpected frame before connection limit: {other:?}"),
+        }
     };
     let value = serde_json::from_str::<Value>(&text).expect("connection limit JSON");
 
@@ -606,7 +595,8 @@ async fn idle_connection_reaches_the_official_limit_without_starting_an_executio
     );
     assert_eq!(trace.starts.load(Ordering::Acquire), 0);
 
-    socket.close(None).await.expect("close WebSocket");
+    // 服务端已因生命周期到期关闭；不再向关闭中的连接写入。
+    drop(socket);
     server.abort();
 }
 
@@ -949,3 +939,134 @@ async fn client_close_should_cancel_active_execution_without_starting_queued_req
     assert!(trace.cancelled.load(Ordering::Acquire));
     server.abort();
 }
+
+#[tokio::test(start_paused = true)]
+async fn idle_pump_sends_heartbeat_without_bursting_after_a_delay() {
+    let PumpHarness {
+        connection,
+        incoming: _incoming,
+        mut written,
+        ..
+    } = test_connection(false);
+    tokio::task::yield_now().await;
+    assert!(matches!(written.try_recv(), Err(TryRecvError::Empty)));
+
+    tokio::time::advance(Duration::from_secs(25)).await;
+    let first = written.recv().await.expect("first heartbeat");
+    assert!(matches!(first, Message::Ping(_)));
+
+    tokio::time::advance(Duration::from_secs(120)).await;
+    let second = written.recv().await.expect("heartbeat after delay");
+    assert!(matches!(&second, Message::Ping(_)));
+    assert_ne!(first, second);
+    tokio::task::yield_now().await;
+    assert!(matches!(written.try_recv(), Err(TryRecvError::Empty)));
+    drop(connection);
+}
+
+#[tokio::test(start_paused = true)]
+async fn missing_pongs_do_not_cancel_the_connection_or_block_business_frames() {
+    let PumpHarness {
+        mut connection,
+        incoming,
+        mut written,
+        ..
+    } = test_connection(false);
+    for _ in 0..13 {
+        tokio::time::advance(Duration::from_secs(25)).await;
+        assert!(matches!(written.recv().await, Some(Message::Ping(_))));
+    }
+
+    incoming
+        .send(Ok(Message::Text("still connected".into())))
+        .expect("send business frame after 325 seconds without Pong");
+    assert!(matches!(
+        connection.next_event().await,
+        Some(ConnectionEvent::Text(text)) if text == "still connected"
+    ));
+    incoming.send(Ok(Message::Close(None))).expect("close peer");
+    assert_eq!(
+        connection.wait_for_exit().await,
+        PumpExitReason::ClientClose
+    );
+    tokio::time::advance(Duration::from_secs(25)).await;
+    assert!(written.recv().await.is_none());
+}
+
+#[tokio::test(start_paused = true)]
+async fn stalled_heartbeat_can_be_cancelled_without_waiting_for_write_timeout() {
+    let PumpHarness {
+        mut connection,
+        incoming: _incoming,
+        written: _written,
+        dropped,
+        cancellation,
+    } = test_connection(true);
+    tokio::time::advance(Duration::from_secs(25)).await;
+    tokio::task::yield_now().await;
+    cancellation.cancel();
+    let reason = tokio::time::timeout(Duration::from_secs(1), connection.wait_for_exit())
+        .await
+        .expect("cancel a blocked heartbeat promptly");
+    assert_eq!(reason, PumpExitReason::LifecycleShutdown);
+    assert!(dropped.load(Ordering::Acquire));
+}
+
+#[tokio::test(start_paused = true)]
+async fn stalled_heartbeat_obeys_the_existing_write_timeout() {
+    let PumpHarness {
+        mut connection,
+        incoming: _incoming,
+        written: _written,
+        dropped,
+        ..
+    } = test_connection(true);
+    let started = tokio::time::Instant::now();
+    assert_eq!(
+        connection.wait_for_exit().await,
+        PumpExitReason::WriteTimeout
+    );
+    assert_eq!(started.elapsed(), Duration::from_secs(25 + 300));
+    assert!(dropped.load(Ordering::Acquire));
+}
+
+#[tokio::test]
+async fn active_response_receives_heartbeats_during_two_hundred_seconds_of_silence() {
+    let (trace, mut socket, server) = start_active_response().await;
+    for _ in 0..8 {
+        tokio::time::pause();
+        tokio::time::advance(Duration::from_secs(25)).await;
+        tokio::task::yield_now().await;
+        tokio::time::resume();
+        let message = tokio::time::timeout(Duration::from_secs(2), socket.next())
+            .await
+            .expect("heartbeat while the upstream response is pending")
+            .expect("connection remains open")
+            .expect("valid heartbeat frame");
+        let ClientMessage::Ping(payload) = message else {
+            panic!("expected heartbeat while business stream is silent: {message:?}");
+        };
+        // 与官方 WsStream 一样，在业务响应未完成时回复同 payload 的 Pong。
+        socket
+            .send(ClientMessage::Pong(payload))
+            .await
+            .expect("reply to heartbeat");
+    }
+    assert_eq!(trace.starts.load(Ordering::Acquire), 1);
+    assert!(!trace.cancelled.load(Ordering::Acquire));
+
+    trace.release_terminal.notify_one();
+    let message = tokio::time::timeout(Duration::from_secs(2), socket.next())
+        .await
+        .expect("terminal response after heartbeats")
+        .expect("connection remains open")
+        .expect("valid terminal frame");
+    let ClientMessage::Text(text) = message else {
+        panic!("expected completed response: {message:?}");
+    };
+    let value: Value = serde_json::from_str(&text).expect("terminal JSON");
+    assert_eq!(value["type"], "response.completed");
+    assert_eq!(trace.starts.load(Ordering::Acquire), 1);
+    socket.close(None).await.expect("close WebSocket");
+    server.abort();
+}

backend/crates/gateway-api/tests/openai/search/mod.rs

@@ -22,7 +22,7 @@ use gateway_core::upstream::UpstreamSendState;
 use serde_json::{Value, json};
 use tower::ServiceExt;
 
-use super::provider_endpoint::BufferedJsonSession;
+use super::endpoint::BufferedJsonSession;
 use super::{api_router, authenticated_client};
 
 const SEARCH_RESPONSE: &[u8] = br#"{ "encrypted_output":"ciphertext", "output":"search result", "results":[{"type":"text_result","ref_id":"turn0search0","future":9007199254740993}] }"#;

backend/crates/gateway-store/src/postgres/observability/queries/ops.rs

@@ -13,12 +13,12 @@ const REQUEST_ERROR_SELECT: &str = "select 'model_request'::text as source,
        mr.provider_account_authentication_kind_snapshot
          as provider_account_authentication_kind,
        mr.upstream_model_id, mr.upstream_transport,
-       coalesce(mr.error_kind, 'failed') as failure_kind,
+       mr.error_kind as failure_kind,
        mr.upstream_send_state,
        mr.client_status_code, mr.upstream_status_code,
        mr.provider_error_code, mr.client_response_id, mr.upstream_request_id,
        mr.latency_ms,
-       coalesce(mr.error_message, mr.error_kind, 'request failed') as message,
+       coalesce(mr.error_message, mr.error_kind) as message,
        mr.raw_upstream_error,
        host(mr.client_ip) as client_ip, mr.user_agent,
        mr.reasoning_effort, mr.reasoning_preset, mr.request_kind,
@@ -32,7 +32,7 @@ const REQUEST_ERROR_SELECT: &str = "select 'model_request'::text as source,
        mr.completed_at as occurred_at,
        'model_request:' || mr.id as stable_sort_id
 from model_requests mr
-where mr.outcome = 'failed'";
+where true";
 
 const OPS_EVENT_SELECT: &str = "select 'ops_event'::text as source,
        oe.id as event_id, oe.model_request_id as request_id, oe.attempt_index,
@@ -106,7 +106,7 @@ pub(crate) async fn count_ops_errors(
     filter: &OpsErrorFilter,
 ) -> StoreResult<u64> {
     let mut statement = QueryBuilder::<Postgres>::new(
-        "select coalesce(sum(source_count), 0)::bigint from (select count(*)::bigint as source_count from model_requests mr where mr.outcome = 'failed'",
+        "select coalesce(sum(source_count), 0)::bigint from (select count(*)::bigint as source_count from model_requests mr where true",
     );
     push_request_error_predicates(&mut statement, range, filter);
     statement.push(
@@ -127,6 +127,9 @@ fn push_request_error_predicates(
     range: ObservabilityRange,
     filter: &OpsErrorFilter,
 ) {
+    // 错误事实独立于请求结束状态；主动取消不属于需要排查的错误。
+    // 列表和总数共用此条件，避免流式响应中的错误因 outcome 被漏掉。
+    statement.push(" and mr.error_kind is not null and mr.error_kind <> 'cancelled'");
     push_range(statement, "mr.completed_at", range);
     for (column, value) in [
         ("mr.client_api_key_ref", &filter.client_api_key_ref),

backend/crates/gateway-store/tests/postgres/observability/mod.rs

@@ -258,6 +258,119 @@ async fn ops_search_should_treat_sql_wildcards_as_literals() {
     database.close().await;
 }
 
+#[tokio::test]
+async fn ops_should_include_incomplete_upstream_errors() {
+    let Some(database) = TestDatabase::create("ops_incomplete_error").await else {
+        return;
+    };
+    let now = Utc::now();
+    seed_observability_facts(&database.pool, now).await.unwrap();
+    let raw_error = r#"{"type":"error","status":400,"error":{"type":"invalid_request_error","message":"Invalid `previous_response_id`."}}"#;
+    sqlx::query(
+        "update model_requests
+            set outcome = 'incomplete', error_kind = 'invalid_request',
+                client_transport = 'websocket', upstream_transport = 'websocket',
+                downstream_committed_at = completed_at,
+                client_status_code = null, upstream_status_code = 400,
+                provider_error_code = null, raw_upstream_error = $1
+          where id = 'req_observe_failed'",
+    )
+    .bind(raw_error)
+    .execute(&database.pool)
+    .await
+    .unwrap();
+    let page = admin_observability_store(&database.pool)
+        .list_ops_errors(admin_observability::OpsErrorQuery {
+            range: admin_observability::TimeRange::new(
+                now - TimeDelta::hours(1),
+                now + TimeDelta::hours(1),
+            )
+            .unwrap(),
+            filter: admin_observability::OpsErrorFilter {
+                request_id: Some("req_observe_failed".to_owned()),
+                status_code: Some(400),
+                ..admin_observability::OpsErrorFilter::default()
+            },
+            current_page: 1,
+            page_size: PageSize::new(10).unwrap(),
+        })
+        .await
+        .expect("incomplete request must be available in admin error troubleshooting");
+    database.close().await;
+
+    assert_eq!(page.total, 1);
+    assert_eq!(page.items.len(), 1);
+    let error = &page.items[0];
+    assert_eq!(error.source, "model_request");
+    assert_eq!(error.failure_kind, "invalid_request");
+    assert_eq!(error.upstream_status_code, Some(400));
+    assert_eq!(error.client_status_code, None);
+    assert_eq!(error.raw_upstream_error.as_deref(), Some(raw_error));
+}
+
+#[tokio::test]
+async fn ops_should_select_errors_independently_of_request_outcome() {
+    let Some(database) = TestDatabase::create("ops_error_outcomes").await else {
+        return;
+    };
+    let now = Utc::now();
+    seed_observability_facts(&database.pool, now).await.unwrap();
+    let range =
+        ObservabilityRange::new(now - TimeDelta::hours(1), now + TimeDelta::hours(1)).unwrap();
+    let repository = observability_repository(&database.pool);
+    let mut results = Vec::new();
+    for (outcome, error_kind, expected_count) in [
+        ("incomplete", Some("upstream_unavailable"), 1),
+        ("succeeded", Some("upstream_unavailable"), 1),
+        ("failed", Some("upstream_unavailable"), 1),
+        ("failed", None, 0),
+        ("incomplete", None, 0),
+        ("cancelled", Some("cancelled"), 0),
+        ("incomplete", Some("cancelled"), 0),
+        ("failed", Some("cancelled"), 0),
+        ("succeeded", None, 0),
+    ] {
+        sqlx::query(
+            "update model_requests set outcome = $1, error_kind = $2
+              where id = 'req_observe_success'",
+        )
+        .bind(outcome)
+        .bind(error_kind)
+        .execute(&database.pool)
+        .await
+        .unwrap();
+        let page = repository
+            .list_ops_errors(OpsErrorQuery {
+                range,
+                filter: OpsErrorFilter {
+                    request_id: Some("req_observe_success".to_owned()),
+                    ..OpsErrorFilter::default()
+                },
+                current_page: 1,
+                page_size: ObservabilityPageSize::new(10).unwrap(),
+            })
+            .await
+            .unwrap();
+        results.push((outcome, error_kind, expected_count, page));
+    }
+    database.close().await;
+
+    for (outcome, error_kind, expected_count, page) in results {
+        assert_eq!(
+            page.total, expected_count,
+            "total for outcome={outcome}, error_kind={error_kind:?}"
+        );
+        assert_eq!(
+            page.items.len() as u64,
+            expected_count,
+            "items for outcome={outcome}, error_kind={error_kind:?}"
+        );
+        if let Some(error) = page.items.first() {
+            assert_eq!(Some(error.failure_kind.as_str()), error_kind);
+        }
+    }
+}
+
 #[tokio::test]
 async fn recovered_continuation_failure_should_be_visible_in_ops_but_hidden_from_business_metrics()
 {

backend/crates/providers/openai/Cargo.toml

@@ -23,8 +23,8 @@ getrandom = "0.4.3"
 hex = "0.4.3"
 # TLS 依赖按官方 Core 锁文件固定，包含用于约束传递依赖的直接声明。
 # 升级前须重复采样 ClientHello，核对算法、扩展集合及排序约束、key_share 与 ALPN。
-aws-lc-rs = "=1.16.2"
-hyper = { version = "=1.8.1", default-features = false }
+aws-lc-rs = "=1.18.1"
+hyper = { version = "=1.11.1", default-features = false }
 hyper-rustls = "=0.27.7"
 # 与官方 Codex Desktop 一致：reqwest 不启用压缩特性，上游请求不带 accept-encoding。
 reqwest = { version = "=0.12.28", default-features = false, features = ["json", "stream", "cookies", "rustls-tls-native-roots", "http2"] }
@@ -33,15 +33,15 @@ semver = "1.0.27"
 # 与官方 rustls 路径一致：aws-lc 支持 P-521，优先提供 X25519MLKEM768。
 # reqwest 的 roots feature 仍会编译 ring；实际 provider 由 TLS 初始化显式选择。
 rustls = { version = "=0.23.36", default-features = false, features = ["aws_lc_rs", "prefer-post-quantum", "std", "tls12"] }
-rustls-native-certs = "=0.8.3"
-rustls-pki-types = "=1.14.0"
+rustls-native-certs = "=0.8.4"
+rustls-pki-types = "=1.15.1"
 secrecy = "0.10.3"
 serde = { version = "1.0.228", features = ["derive"] }
 serde_json = { version = "1.0.150", features = ["arbitrary_precision", "preserve_order"] }
 sha2 = "0.11.0"
 subtle = "2.6.1"
 thiserror = "2.0.18"
-tokio = { version = "=1.52.3", default-features = false, features = ["fs", "macros", "rt", "sync", "time"] }
+tokio = { version = "=1.53.1", default-features = false, features = ["fs", "macros", "rt", "sync", "time"] }
 tokio-rustls = { version = "=0.26.4", default-features = false }
 tokio-util = { version = "0.7.18", features = ["rt"] }
 # 官方 Codex WebSocket 需要上游尚未发布的 proxy 与 permessage-deflate 实现，固定到已审计提交。

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

@@ -851,7 +851,9 @@ pub(super) fn websocket_success_decision(
 ) -> CodexTransportDecision {
     match requirement {
         TransportRequirement::ExactWebSocketContinuation => CodexTransportDecision::ExactWebSocket,
-        TransportRequirement::ExplicitWebSocketWarmup => CodexTransportDecision::RequiredWebSocket,
+        TransportRequirement::ExplicitWebSocketWarmup | TransportRequirement::WebSocketNewChain => {
+            CodexTransportDecision::RequiredWebSocket
+        }
         _ if prepared.reused() => CodexTransportDecision::ReusedWebSocket,
         _ => CodexTransportDecision::ConnectedWebSocket,
     }

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

@@ -272,6 +272,7 @@ impl CodexBackendClient {
                 Some(WEBSOCKET_FAST_PATH_BUDGET)
             }
             TransportRequirement::ExplicitWebSocketWarmup
+            | TransportRequirement::WebSocketNewChain
             | TransportRequirement::ExactWebSocketContinuation
             | TransportRequirement::ExternalUnknown => None,
             TransportRequirement::HttpRequired => None,

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

@@ -121,6 +121,8 @@ pub enum TransportRequirement {
     HttpRequired,
     /// `generate=false + store=false` 预热必须保留在同一条 WebSocket。
     ExplicitWebSocketWarmup,
+    /// 客户端 WebSocket 的非持久化新链，必须在池化连接上建立后续续接状态。
+    WebSocketNewChain,
     /// 只能使用持有指定 connection-local response 的精确 WebSocket。
     ExactWebSocketContinuation,
     /// previous response 已持久化，允许 WebSocket 或 HTTP/2。
@@ -136,7 +138,9 @@ impl TransportRequirement {
     pub fn requires_websocket(self) -> bool {
         matches!(
             self,
-            Self::ExplicitWebSocketWarmup | Self::ExactWebSocketContinuation
+            Self::ExplicitWebSocketWarmup
+                | Self::WebSocketNewChain
+                | Self::ExactWebSocketContinuation
         )
     }
 
@@ -150,7 +154,7 @@ impl TransportRequirement {
 
     /// 无续接依赖的新请求可以在明确的连接级拒绝后重新建连。
     pub fn allows_connection_restart(self) -> bool {
-        matches!(self, Self::NewChain)
+        matches!(self, Self::NewChain | Self::WebSocketNewChain)
     }
 
     /// 用于审计与遥测的稳定名称。
@@ -158,6 +162,7 @@ impl TransportRequirement {
         match self {
             Self::HttpRequired => "http_required",
             Self::ExplicitWebSocketWarmup => "explicit_websocket_warmup",
+            Self::WebSocketNewChain => "websocket_new_chain",
             Self::ExactWebSocketContinuation => "exact_websocket_continuation",
             Self::PersistedContinuation => "persisted_continuation",
             Self::ExternalUnknown => "external_unknown",
@@ -184,6 +189,11 @@ pub fn transport_requirement(request: &CodexResponsesRequest) -> TransportRequir
                 TransportRequirement::ExternalUnknown
             }
         },
+        // 客户端会在下一轮提交 response ID；HTTP store=false 的成功响应无法
+        // 在池化 WebSocket 上续接，所以首轮也不能按普通快路径降级到 HTTP。
+        None if request.downstream_websocket_connection_id.is_some() && !request.store() => {
+            TransportRequirement::WebSocketNewChain
+        }
         None => TransportRequirement::NewChain,
     }
 }

backend/crates/providers/openai/src/transport/websocket/exchange/stream.rs

@@ -151,6 +151,9 @@ async fn forward_websocket_response_stream(state: WebSocketStreamForwardState) {
         .map(|pool_return| std::mem::take(&mut pool_return.continuation))
         .unwrap_or_default();
     let mut last_event_type = None;
+    // 官方 run_websocket_response_stream 按响应消费 metadata 事件；它们不属于握手头。
+    // 复用连接时恢复握手快照，避免上一轮模型信息及普通响应头随每次请求累积。
+    let opening_response_metadata = metadata.response_metadata.clone();
     loop {
         let message = tokio::select! {
             biased;
@@ -323,6 +326,7 @@ async fn forward_websocket_response_stream(state: WebSocketStreamForwardState) {
             );
             match terminal {
                 WebSocketTerminalKind::Completed => {
+                    metadata.response_metadata = opening_response_metadata;
                     finish_stream_websocket(websocket, metadata, continuation, pool_return.take())
                         .await;
                 }

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

@@ -2521,6 +2521,127 @@ async fn fallback_attempt_transport_forces_http_sse_for_a_websocket_request() {
     }
 }
 
+#[tokio::test]
+async fn downstream_websocket_new_chain_should_override_session_and_attempt_http_fallback() {
+    const SESSION_ID: &str = "downstream-required-session";
+
+    let store = Arc::new(MemoryAccountStore::default());
+    create_account(&store, "acct_http_sse_exhausted").await;
+    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let base_url = format!("http://{}", listener.local_addr().unwrap());
+    let server = tokio::spawn(async move {
+        let (mut opening, _) = listener.accept().await.unwrap();
+        let request = capture_http_request(&mut opening).await;
+        assert!(String::from_utf8_lossy(&request).starts_with("GET /codex/responses"));
+        opening
+            .write_all(
+                b"HTTP/1.1 426 Upgrade Required\r\ncontent-length: 0\r\nconnection: close\r\n\r\n",
+            )
+            .await
+            .unwrap();
+
+        let (stream, _) = listener.accept().await.unwrap();
+        let mut websocket = accept_codex_test_websocket(stream).await;
+        for id in ["resp_required_session", "resp_required_attempt"] {
+            let request = websocket.next().await.unwrap().unwrap();
+            let payload: Value = serde_json::from_str(request.to_text().unwrap()).unwrap();
+            assert_eq!(payload["store"], false);
+            websocket
+                .send(Message::Text(
+                    json!({
+                        "type": "response.completed",
+                        "response": {
+                            "id": id,
+                            "model": "gpt-5.4",
+                            "status": "completed",
+                            "store": false,
+                            "output": [],
+                            "usage": {"input_tokens": 1, "output_tokens": 1, "total_tokens": 2}
+                        }
+                    })
+                    .to_string()
+                    .into(),
+                ))
+                .await
+                .unwrap();
+        }
+    });
+    let provider = provider_with_base_url(&store, base_url);
+    let mut seed = provider
+        .execute(
+            planned_request(
+                "openai",
+                Operation::Generate(generate_with_session_context(SESSION_ID, None, None)),
+            ),
+            context("req_disable_optional_websocket", CancellationToken::new()),
+        )
+        .await
+        .unwrap();
+    let error = loop {
+        match seed.next().await {
+            Some(Ok(_)) => {}
+            Some(Err(error)) => break error,
+            None => panic!("426 opening must enable session HTTP fallback"),
+        }
+    };
+    assert_eq!(
+        error.pre_delivery_retry(),
+        Some(PreDeliveryRetry::SameAccountTransportFallback)
+    );
+    drop(seed);
+
+    for attempt in [
+        context("req_required_session", CancellationToken::new()),
+        fallback_transport_context("req_required_attempt"),
+    ] {
+        let payload = ProtocolPayload::json_object(
+            "openai",
+            Map::from_iter([
+                ("model".to_owned(), json!("gpt-5.4")),
+                ("input".to_owned(), json!("hello")),
+                ("store".to_owned(), json!(false)),
+                ("session_id".to_owned(), json!(SESSION_ID)),
+            ]),
+        )
+        .unwrap()
+        .with_context(Map::from_iter([(
+            "downstream_websocket_connection_id".to_owned(),
+            json!("ws_required_session"),
+        )]));
+        let request = GenerateRequest::from_protocol_payload(payload).with_provider_session_state(
+            ProviderSessionState::new(
+                "openai",
+                Map::from_iter([
+                    ("account_id".to_owned(), json!("acct_http_sse_exhausted")),
+                    ("conversation_id".to_owned(), json!(SESSION_ID)),
+                    ("continuation_scope".to_owned(), json!("persisted")),
+                ]),
+            )
+            .unwrap(),
+        );
+        let mut stream = provider
+            .execute(
+                planned_request("openai", Operation::Generate(request)),
+                attempt,
+            )
+            .await
+            .unwrap();
+        assert_eq!(stream.metadata().transport().as_str(), "websocket");
+        let mut session = None;
+        while let Some(event) = stream.next().await {
+            let event = event.expect("required WebSocket response");
+            if let Some(update) = event.session_update() {
+                session = Some(update.clone());
+            }
+        }
+        assert_eq!(
+            session.unwrap().payload().get("continuation_scope"),
+            Some(&json!("connection_local"))
+        );
+    }
+    server.await.unwrap();
+}
+
 #[tokio::test]
 async fn websocket_turn_state_metadata_is_exposed_through_response_observation() {
     let store = Arc::new(MemoryAccountStore::default());

backend/crates/providers/openai/tests/transport/latency.rs

@@ -115,6 +115,79 @@ async fn cold_websocket_should_fall_back_without_recording_a_successful_connect(
     pool.shutdown().await;
 }
 
+#[tokio::test(start_paused = true)]
+async fn downstream_websocket_new_chain_should_preserve_continuation_after_slow_opening() {
+    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let addr = listener.local_addr().unwrap();
+    let server = tokio::spawn(async move {
+        let (stream, _) = listener.accept().await.unwrap();
+        tokio::select! {
+            () = tokio::time::sleep(Duration::from_secs(1)) => {}
+            accepted = listener.accept() => {
+                let (mut http, _) = accepted.unwrap();
+                let request = read_http_request(&mut http).await;
+                assert!(request.starts_with("POST /codex/responses HTTP/1.1"));
+                write_completed_sse_response(&mut http).await;
+                return;
+            }
+        }
+        let mut websocket = accept_codex_test_websocket(stream).await;
+        let _first = websocket.next().await.unwrap().unwrap();
+        websocket
+            .send(Message::Text(
+                completed_websocket_response("resp_slow_seed", 2, 1).into(),
+            ))
+            .await
+            .unwrap();
+
+        let second = websocket.next().await.unwrap().unwrap();
+        let second: Value = serde_json::from_str(second.to_text().unwrap()).unwrap();
+        assert_eq!(second["previous_response_id"], "resp_slow_seed");
+        websocket
+            .send(Message::Text(
+                completed_websocket_response("resp_slow_continuation", 2, 1).into(),
+            ))
+            .await
+            .unwrap();
+    });
+    let pool = Arc::new(CodexWebSocketPool::default());
+    let backend = CodexBackendClient::new(
+        reqwest::Client::builder().no_proxy().build().unwrap(),
+        format!("http://{addr}"),
+        test_wire_profile(),
+    )
+    .with_websocket_pool(Arc::clone(&pool));
+    let mut request = new_chain_request("conversation-downstream-slow");
+    request.downstream_websocket_connection_id = Some("ws_downstream_slow".to_owned());
+
+    let first = backend
+        .create_response(
+            &request,
+            request_context("req_slow_seed", Some("chatgpt-account")),
+        )
+        .await
+        .unwrap();
+    assert_eq!(first.transport, CodexBackendTransport::WebSocket);
+    assert!(first.connection_local_continuation);
+
+    request.set_previous_response_id(Some("resp_slow_seed".to_owned()));
+    request.previous_response_scope = Some(PreviousResponseScope::ConnectionLocal);
+    let second = backend
+        .create_response(
+            &request,
+            request_context("req_slow_continuation", Some("chatgpt-account")),
+        )
+        .await
+        .expect("the next turn must use the socket that generated the previous response");
+    assert_eq!(
+        second.transport_metrics.decision,
+        Some(CodexTransportDecision::ExactWebSocket)
+    );
+    assert!(second.body.contains("resp_slow_continuation"));
+    server.await.unwrap();
+    pool.shutdown().await;
+}
+
 #[tokio::test(start_paused = true)]
 async fn external_continuation_should_wait_for_a_cold_websocket() {
     let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();

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

@@ -273,6 +273,34 @@ fn transport_should_prefer_websocket_when_requested_without_history() {
     assert!(transport_requirement(&request).allows_pre_send_http_fallback());
 }
 
+#[test]
+fn downstream_websocket_new_chain_should_require_pooled_websocket_without_storage() {
+    for store in [None, Some(false)] {
+        let mut body = serde_json::Map::new();
+        if let Some(store) = store {
+            body.insert("store".to_owned(), json!(store));
+        }
+        let mut request = CodexResponsesRequest::from_body(body);
+        request.downstream_websocket_connection_id = Some("ws_downstream".to_owned());
+
+        let requirement = transport_requirement(&request);
+        assert!(requirement.requires_websocket());
+        assert!(!requirement.allows_pre_send_http_fallback());
+        assert!(requirement.allows_connection_restart());
+    }
+}
+
+#[test]
+fn downstream_websocket_new_chain_should_allow_http_when_response_is_stored() {
+    let mut request = CodexResponsesRequest::from_body(serde_json::Map::from_iter([(
+        "store".to_owned(),
+        json!(true),
+    )]));
+    request.downstream_websocket_connection_id = Some("ws_downstream".to_owned());
+
+    assert!(transport_requirement(&request).allows_pre_send_http_fallback());
+}
+
 #[test]
 fn unspecified_transport_without_history_should_still_be_a_new_chain() {
     let request = codex_request("gpt-test", "be brief", Vec::new());

backend/crates/providers/openai/tests/transport/websocket/metadata.rs

@@ -0,0 +1,92 @@
+use super::*;
+
+#[tokio::test]
+async fn reused_websocket_should_keep_response_metadata_scoped_to_each_exchange() {
+    const ROUNDS: usize = 40;
+    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let addr = listener.local_addr().unwrap();
+    let server = tokio::spawn(async move {
+        let (stream, _) = listener.accept().await.unwrap();
+        let mut websocket = accept_codex_test_websocket_with(stream, |_, response| {
+            for value in ["first", "second"] {
+                response
+                    .headers_mut()
+                    .append("x-opening-multi", value.parse().unwrap());
+            }
+        })
+        .await;
+        for round in 0..ROUNDS {
+            websocket.next().await.unwrap().unwrap();
+            websocket
+                .send(Message::Text(
+                    json!({
+                        "type": "codex.response.metadata",
+                        "headers": {
+                            "x-models-etag": format!("etag-{round}"),
+                            "openai-model": format!("model-{round}"),
+                            "x-reasoning-included": "true",
+                            "x-codex-safety-buffering-enabled": "true",
+                            "x-codex-safety-buffering-faster-model": "faster-model",
+                            "x-future-response-metadata": format!("future-{round}"),
+                            "x-codex-turn-state": format!("turn-{round}")
+                        }
+                    })
+                    .to_string()
+                    .into(),
+                ))
+                .await
+                .unwrap();
+            websocket
+                .send(Message::Text(
+                    completed_websocket_response(&format!("resp_metadata_{round}"), 1, 1).into(),
+                ))
+                .await
+                .unwrap();
+        }
+    });
+    let pool = Arc::new(CodexWebSocketPool::new(Duration::from_mins(1)));
+    let backend = CodexBackendClient::new(
+        reqwest::Client::builder().no_proxy().build().unwrap(),
+        format!("http://{addr}"),
+        test_wire_profile(),
+    )
+    .with_websocket_pool(pool);
+    let request = pooled_websocket_request("conversation-response-metadata");
+    let expected = CodexResponseMetadata {
+        client_headers: vec![
+            (
+                "x-opening-multi".to_owned(),
+                bytes::Bytes::from_static(b"first"),
+            ),
+            (
+                "x-opening-multi".to_owned(),
+                bytes::Bytes::from_static(b"second"),
+            ),
+        ],
+        ..CodexResponseMetadata::default()
+    };
+
+    for round in 0..ROUNDS {
+        let response = timeout(
+            Duration::from_secs(10),
+            backend.create_response(
+                &request,
+                request_context("req_response_metadata", Some("chatgpt-account")),
+            ),
+        )
+        .await
+        .expect("response within timeout")
+        .expect("response should complete on the same socket");
+        assert_eq!(response.response_metadata, expected, "exchange {round}");
+        assert!(response.body.contains(&format!("etag-{round}")));
+        assert_eq!(response.turn_state, Some(format!("turn-{round}")));
+        if round > 0 {
+            assert!(
+                response
+                    .websocket_pool_decision
+                    .is_some_and(WebSocketPoolDecision::is_reuse)
+            );
+        }
+    }
+    server.await.unwrap();
+}

backend/crates/providers/openai/tests/transport/websocket/mod.rs

@@ -4,6 +4,8 @@ use provider_openai::transport::{
     websocket::write_websocket_audit_artifact_for_dir,
 };
 
+mod metadata;
+
 fn rate_limit_event(used_percent: u64) -> String {
     json!({
         "type": "codex.rate_limits",

frontend/package.json

@@ -19,7 +19,7 @@
     "@boxicons/vue": "1.0.1",
     "@fontsource-variable/inter": "^5.3.0",
     "@fontsource-variable/jetbrains-mono": "^5.3.0",
-    "@lucide/vue": "^1.37.0",
+    "@lucide/vue": "^1.41.0",
     "@vueuse/core": "^14.4.0",
     "axios": "^1.20.0",
     "dayjs": "^1.11.23",
@@ -31,15 +31,15 @@
     "pinia": "^4.0.3",
     "pinia-plugin-persistedstate": "^4.7.1",
     "vue": "^3.5.42",
-    "vue-router": "^5.3.0"
+    "vue-router": "^5.3.1"
   },
   "devDependencies": {
-    "@antfu/eslint-config": "^9.3.0",
+    "@antfu/eslint-config": "^9.5.1",
     "@tailwindcss/vite": "^4.3.3",
-    "@types/node": "^26.4.0",
+    "@types/node": "^26.4.1",
     "@vitejs/plugin-vue": "^6.0.8",
     "@vue/tsconfig": "^0.9.1",
-    "eslint": "^10.9.1",
+    "eslint": "^10.10.0",
     "eslint-plugin-format": "^2.0.1",
     "eslint-plugin-vuejs-accessibility": "^2.6.0",
     "tailwindcss": "^4.3.3",

release/version.yaml

@@ -1 +1 @@
-version: "3.3.26"
+version: "3.3.27"