fix(responses): keep slow websocket uploads alive

Track inbound byte progress before WebSocket decoding so active uploads do not trip the heartbeat timeout. Reuse Axum handshake validation and report actual receive idle time.

Cover successive slow uploads and stalled partial frames, and document the downstream liveness contract.

Validation: 652 API and gateway tests passed; production API check, strict workspace Clippy, and formatting passed.

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

backend/Cargo.lock

@@ -1743,6 +1743,8 @@ dependencies = [
  "http",
  "http-body",
  "http-body-util",
+ "hyper",
+ "hyper-util",
  "regex",
  "secrecy",
  "serde",

backend/crates/gateway-api/Cargo.toml

@@ -25,13 +25,16 @@ gateway-admin = { path = "../gateway-admin" }
 gateway-core = { path = "../gateway-core" }
 gateway-protocol = { path = "../gateway-protocol" }
 headers = "0.4.1"
+hyper = { version = "=1.11.1", default-features = false }
+hyper-util = { version = "=0.1.21", default-features = false, features = ["tokio"] }
 regex = "1.13.1"
 secrecy = "0.10"
 serde = { version = "1.0.228", features = ["derive"] }
 serde_json = { version = "1.0.150", features = ["arbitrary_precision", "preserve_order"] }
 sha2 = "0.11.0"
 thiserror = "2.0.18"
-tokio = { version = "1.52.3", default-features = false, features = ["macros", "rt", "sync", "time"] }
+tokio = { version = "1.52.3", default-features = false, features = ["io-util", "macros", "rt", "sync", "time"] }
+tokio-tungstenite = { version = "0.29.0", default-features = false, features = ["handshake"] }
 tower = { version = "0.5.3", features = ["util"] }
 tower-http = { version = "0.7.0", features = ["compression-gzip", "cors", "fs", "request-id", "timeout", "trace"] }
 tracing = "0.1.44"

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

@@ -4,8 +4,9 @@ use super::{
     frame::*,
     pump::run_pump,
     state::{ConnectionCommand, ConnectionStats},
+    transport::Socket,
 };
-use axum::extract::ws::{CloseFrame, Message, Utf8Bytes, WebSocket, close_code};
+use axum::extract::ws::{CloseFrame, Message, Utf8Bytes, close_code};
 use futures::{Sink, Stream};
 use gateway_core::{
     engine::middleware::{FrozenMiddlewarePlan, MiddlewareHeader},
@@ -21,7 +22,7 @@ use std::{
     time::Duration,
 };
 use tokio::{
-    sync::{mpsc, oneshot},
+    sync::{mpsc, oneshot, watch},
     task::JoinHandle,
     time::{Instant, timeout},
 };
@@ -38,19 +39,21 @@ pub struct ResponsesWebSocketConnection {
     exited: oneshot::Receiver<PumpExitReason>,
     pump_task: Option<JoinHandle<()>>,
     stats: Arc<ConnectionStats>,
+    activity: watch::Sender<Instant>,
     config: ConnectionConfig,
     exit_reason: Option<PumpExitReason>,
 }
 
 impl ResponsesWebSocketConnection {
     pub(in super::super) fn new(
-        socket: WebSocket,
+        socket: Socket,
         connection_id: String,
         cancellation: CancellationToken,
         middleware: Option<FrozenMiddlewarePlan>,
         headers: Arc<[MiddlewareHeader]>,
     ) -> Self {
         spawn_connection(
+            socket.activity.clone(),
             socket,
             Arc::<str>::from(connection_id),
             cancellation,
@@ -278,8 +281,7 @@ impl ResponsesWebSocketConnection {
 
     pub(in super::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 read_idle_ms = self.activity.borrow().elapsed().as_millis();
         let write_idle_ms = connection_age_ms
             .saturating_sub(u128::from(self.stats.last_write_ms.load(Ordering::Relaxed)));
         tracing::info!(
@@ -319,7 +321,9 @@ impl Drop for ResponsesWebSocketConnection {
 
 /// 为 WebSocket transport 启动包含冻结中间件计划的单 owner pump
 /// cancellation 必须属于本连接，连接退出或句柄释放时会取消它
+/// activity 仅由 transport 的实际入站字节或 pump 的完整消息更新，不能由写入刷新
 pub fn spawn_connection<S, E>(
+    activity: watch::Sender<Instant>,
     socket: S,
     connection_id: Arc<str>,
     cancellation: CancellationToken,
@@ -342,6 +346,7 @@ where
         .unwrap_or_default();
     let pump_task = tokio::spawn(run_pump(
         socket,
+        activity.clone(),
         command_rx,
         incoming_tx,
         exit_tx,
@@ -366,6 +371,7 @@ where
         exited: exit_rx,
         pump_task: Some(pump_task),
         stats,
+        activity,
         config,
         exit_reason: None,
     }

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

@@ -5,9 +5,12 @@ mod handle;
 mod middleware;
 mod pump;
 mod state;
+mod transport;
 
 pub use frame::{
     ConnectionConfig, ConnectionEvent, ConnectionWriteError, FramePhase, PumpExitReason,
     WriteContext, WriteOutcome,
 };
 pub use handle::{ResponsesWebSocketConnection, spawn_connection};
+pub(super) use transport::Socket;
+pub(crate) use transport::Upgrade;

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

@@ -26,6 +26,7 @@ use tokio::{
 #[expect(clippy::too_many_arguments)]
 pub(super) async fn run_pump<S, E>(
     socket: S,
+    activity: watch::Sender<Instant>,
     commands: mpsc::Receiver<ConnectionCommand>,
     incoming: mpsc::Sender<PendingConnectionEvent>,
     exited: oneshot::Sender<PumpExitReason>,
@@ -58,7 +59,7 @@ pub(super) async fn run_pump<S, E>(
     let (writer, mut reader) = socket.split();
     let writer = Arc::new(middleware::Writer::new(writer, context.clone()));
     let sender: Arc<dyn contract::Sender> = writer.clone();
-    let (activity, received_activity) = watch::channel(());
+    let received_activity = activity.subscribe();
     // 心跳独立于业务写入等待，但共享唯一 writer，避免长写入或排队阻止失联检测
     let reason = tokio::select! {
         biased;
@@ -81,7 +82,7 @@ async fn read_connection<S, E>(
     socket: &mut S,
     incoming: &mpsc::Sender<PendingConnectionEvent>,
     sender: Arc<dyn contract::Sender>,
-    activity: &watch::Sender<()>,
+    activity: &watch::Sender<Instant>,
     context: &PumpContext,
 ) -> PumpExitReason
 where
@@ -106,10 +107,9 @@ where
                 }
             }
             message = socket.next() => {
-                if matches!(&message, Some(Ok(_))) { context.stats.record_read(context.opened_at); }
-                if matches!(&message, Some(Ok(message)) if !matches!(message, Message::Close(_))) {
+                if matches!(&message, Some(Ok(_))) {
                     // 链路活动以真实入站为准，插件丢弃消息不应抹掉对端已经响应的事实
-                    activity.send_replace(());
+                    activity.send_replace(Instant::now());
                 }
                 let message = match message {
                     Some(Ok(message)) => match middleware::transform(message, contract::Direction::Incoming, sender.clone(), context).await {
@@ -201,7 +201,7 @@ where
 
 async fn heartbeat_connection<S, E>(
     writer: Arc<middleware::Writer<S>>,
-    mut activity: watch::Receiver<()>,
+    mut activity: watch::Receiver<Instant>,
     context: &PumpContext,
 ) -> PumpExitReason
 where

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

@@ -30,18 +30,10 @@ pub(super) struct ConnectionStats {
     pub(super) ping_received_count: AtomicU64,
     pub(super) ping_written_count: AtomicU64,
     pub(super) pong_received_count: AtomicU64,
-    pub(super) last_read_ms: AtomicU64,
     pub(super) last_write_ms: AtomicU64,
 }
 
 impl ConnectionStats {
-    pub(super) 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,
-        );
-    }
-
     pub(super) fn record_write(&self, opened_at: Instant) {
         self.last_write_ms.store(
             u64::try_from(opened_at.elapsed().as_millis()).unwrap_or(u64::MAX),

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

@@ -0,0 +1,204 @@
+//! 保留 Axum 握手校验，在 WS 消息解码前记录下游真实接收进度
+
+use std::{
+    io,
+    pin::Pin,
+    task::{Context, Poll},
+};
+
+use axum::{
+    extract::{
+        FromRequestParts,
+        ws::{self, Message},
+    },
+    http::{HeaderValue, StatusCode, Version, header, request::Parts},
+    response::{IntoResponse, Response},
+};
+use futures::{Sink, Stream, ready};
+use hyper::upgrade::{OnUpgrade, Upgraded};
+use hyper_util::rt::TokioIo;
+use tokio::{
+    io::{AsyncRead, AsyncWrite, ReadBuf},
+    sync::watch,
+    time::Instant,
+};
+use tokio_tungstenite::{
+    WebSocketStream,
+    tungstenite::{
+        self,
+        protocol::{Role, WebSocketConfig},
+    },
+};
+
+pub(crate) struct Upgrade {
+    pending: OnUpgrade,
+    key: Option<HeaderValue>,
+}
+
+impl<S: Send + Sync> FromRequestParts<S> for Upgrade {
+    type Rejection = ws::rejection::WebSocketUpgradeRejection;
+
+    async fn from_request_parts(parts: &mut Parts, state: &S) -> Result<Self, Self::Rejection> {
+        let pending = parts.extensions.get::<OnUpgrade>().cloned();
+        // Axum 的 WebSocket 不暴露底层 I/O；只复用其握手校验，由本层接管唯一升级流
+        let _validated = ws::WebSocketUpgrade::from_request_parts(parts, state).await?;
+        let pending = pending.expect("Axum validated the presence of OnUpgrade");
+        let key = (parts.version <= Version::HTTP_11)
+            .then(|| parts.headers.get(header::SEC_WEBSOCKET_KEY).cloned())
+            .flatten();
+        Ok(Self { pending, key })
+    }
+}
+
+impl Upgrade {
+    pub(in super::super) fn on_upgrade<F, Fut>(self, callback: F) -> Response
+    where
+        F: FnOnce(Socket) -> Fut + Send + 'static,
+        Fut: Future<Output = ()> + Send + 'static,
+    {
+        tokio::spawn(async move {
+            let Ok(upgraded) = self.pending.await else {
+                return;
+            };
+            let (activity, _) = watch::channel(Instant::now());
+            let io = Receiving {
+                inner: TokioIo::new(upgraded),
+                activity: activity.clone(),
+            };
+            // 与原生 Responses 升级保持一致，不引入消息或帧长度上限
+            let config = WebSocketConfig::default()
+                .max_message_size(None)
+                .max_frame_size(None);
+            let inner = WebSocketStream::from_raw_socket(io, Role::Server, Some(config)).await;
+            callback(Socket { inner, activity }).await;
+        });
+        if let Some(key) = self.key {
+            (
+                StatusCode::SWITCHING_PROTOCOLS,
+                [
+                    (header::CONNECTION, "upgrade".to_owned()),
+                    (header::UPGRADE, "websocket".to_owned()),
+                    (
+                        header::SEC_WEBSOCKET_ACCEPT,
+                        tungstenite::handshake::derive_accept_key(key.as_bytes()),
+                    ),
+                ],
+            )
+                .into_response()
+        } else {
+            StatusCode::OK.into_response()
+        }
+    }
+}
+
+struct Receiving {
+    inner: TokioIo<Upgraded>,
+    activity: watch::Sender<Instant>,
+}
+
+impl AsyncRead for Receiving {
+    fn poll_read(
+        mut self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        buffer: &mut ReadBuf<'_>,
+    ) -> Poll<io::Result<()>> {
+        let before = buffer.filled().len();
+        let result = Pin::new(&mut self.inner).poll_read(cx, buffer);
+        if buffer.filled().len() > before {
+            // 完整消息尚未收齐时，客户端可能仍在发送、暂时无法读取 Ping
+            self.activity.send_replace(Instant::now());
+        }
+        result
+    }
+}
+
+impl AsyncWrite for Receiving {
+    fn poll_write(
+        mut self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        buffer: &[u8],
+    ) -> Poll<io::Result<usize>> {
+        Pin::new(&mut self.inner).poll_write(cx, buffer)
+    }
+    fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
+        Pin::new(&mut self.inner).poll_flush(cx)
+    }
+    fn poll_shutdown(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
+        Pin::new(&mut self.inner).poll_shutdown(cx)
+    }
+}
+
+pub(in super::super) struct Socket {
+    inner: WebSocketStream<Receiving>,
+    pub(super) activity: watch::Sender<Instant>,
+}
+
+impl Stream for Socket {
+    type Item = Result<Message, axum::Error>;
+
+    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+        loop {
+            let message = match ready!(Pin::new(&mut self.inner).poll_next(cx)) {
+                Some(Ok(message)) => message,
+                Some(Err(error)) => return Poll::Ready(Some(Err(axum::Error::new(error)))),
+                None => return Poll::Ready(None),
+            };
+            let message = match message {
+                tungstenite::Message::Text(text) => Message::Text(
+                    ws::Utf8Bytes::try_from(bytes::Bytes::from(text)).map_err(axum::Error::new)?,
+                ),
+                tungstenite::Message::Binary(data) => Message::Binary(data),
+                tungstenite::Message::Ping(data) => Message::Ping(data),
+                tungstenite::Message::Pong(data) => Message::Pong(data),
+                tungstenite::Message::Close(frame) => {
+                    Message::Close(frame.map(|frame| ws::CloseFrame {
+                        code: frame.code.into(),
+                        reason: frame.reason.to_string().into(),
+                    }))
+                }
+                tungstenite::Message::Frame(_) => continue,
+            };
+            return Poll::Ready(Some(Ok(message)));
+        }
+    }
+}
+
+impl Sink<Message> for Socket {
+    type Error = axum::Error;
+
+    fn poll_ready(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+        Pin::new(&mut self.inner)
+            .poll_ready(cx)
+            .map_err(axum::Error::new)
+    }
+    fn start_send(mut self: Pin<&mut Self>, message: Message) -> Result<(), Self::Error> {
+        let message = match message {
+            Message::Text(text) => tungstenite::Message::Text(
+                tungstenite::Utf8Bytes::try_from(bytes::Bytes::from(text))
+                    .map_err(axum::Error::new)?,
+            ),
+            Message::Binary(data) => tungstenite::Message::Binary(data),
+            Message::Ping(data) => tungstenite::Message::Ping(data),
+            Message::Pong(data) => tungstenite::Message::Pong(data),
+            Message::Close(frame) => {
+                tungstenite::Message::Close(frame.map(|frame| tungstenite::protocol::CloseFrame {
+                    code: frame.code.into(),
+                    reason: frame.reason.to_string().into(),
+                }))
+            }
+        };
+        Pin::new(&mut self.inner)
+            .start_send(message)
+            .map_err(axum::Error::new)
+    }
+    fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+        Pin::new(&mut self.inner)
+            .poll_flush(cx)
+            .map_err(axum::Error::new)
+    }
+    fn poll_close(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+        Pin::new(&mut self.inner)
+            .poll_close(cx)
+            .map_err(axum::Error::new)
+    }
+}

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

@@ -11,11 +11,7 @@ use std::{
 };
 
 use axum::{
-    extract::{
-        Extension, State,
-        connect_info::ConnectInfo,
-        ws::{WebSocket, WebSocketUpgrade},
-    },
+    extract::{Extension, State, connect_info::ConnectInfo},
     http::{HeaderMap, StatusCode},
     response::{IntoResponse, Response},
 };
@@ -45,7 +41,9 @@ use crate::{
 use super::{
     request::OpenAiRequestHeaders, request_client_context, validation::ResponseValidationFacts,
 };
-use connection::{ConnectionEvent, FramePhase, ResponsesWebSocketConnection, WriteContext};
+use connection::{
+    ConnectionEvent, FramePhase, ResponsesWebSocketConnection, Socket, Upgrade, WriteContext,
+};
 use forward::{
     ConnectionReplaySnapshot, ForwardOutcome, execution_response, forward_control_event,
     forward_response, new_replay_capture, send_control, send_gateway_error, send_middleware_error,
@@ -64,7 +62,7 @@ pub(crate) async fn responses_websocket(
     middleware: Option<Extension<FrozenMiddlewarePlan>>,
     lifetime: Option<Extension<Arc<crate::middleware::CallLifetime>>>,
     headers: HeaderMap,
-    websocket: WebSocketUpgrade,
+    websocket: Upgrade,
 ) -> Response {
     let service = state.openai().clone();
     let client = match authenticate_client(&service, &headers).await {
@@ -104,7 +102,7 @@ impl ResponsesWebSocketAdapter {
     #[expect(clippy::too_many_arguments)]
     fn upgrade_with_client_context(
         &self,
-        websocket: WebSocketUpgrade,
+        websocket: Upgrade,
         client: AuthenticatedClient,
         client_ip: Option<IpAddr>,
         user_agent: Option<String>,
@@ -138,15 +136,10 @@ impl ResponsesWebSocketAdapter {
             middleware,
         };
         let origin = crate::middleware::current();
-        websocket
-            // 覆盖 axum/tungstenite 的私有 64 MiB message 与 16 MiB frame 默认值
-            // Responses JSON 的协议可接受性由上游决定，代理不另设 wire 长度上限
-            .max_message_size(usize::MAX)
-            .max_frame_size(usize::MAX)
-            .on_upgrade(move |socket| async move {
-                let _lifetime = lifetime;
-                crate::middleware::scope(origin, serve_responses_websocket(socket, session)).await;
-            })
+        websocket.on_upgrade(move |socket| async move {
+            let _lifetime = lifetime;
+            crate::middleware::scope(origin, serve_responses_websocket(socket, session)).await;
+        })
     }
 }
 
@@ -163,7 +156,7 @@ struct ResponsesWebSocketSession {
     middleware: Option<FrozenMiddlewarePlan>,
 }
 
-async fn serve_responses_websocket(socket: WebSocket, session: ResponsesWebSocketSession) {
+async fn serve_responses_websocket(socket: Socket, session: ResponsesWebSocketSession) {
     let ResponsesWebSocketSession {
         service,
         client,

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

@@ -66,6 +66,7 @@ async fn client_close_does_not_flush_a_retained_sender_with_an_inflight_write()
         dropped: Arc::new(AtomicBool::new(false)),
     };
     let mut connection = spawn_connection(
+        tokio::sync::watch::channel(tokio::time::Instant::now()).0,
         socket,
         Arc::from("retained-writer"),
         CancellationToken::new(),

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

@@ -1,6 +1,7 @@
 //! Responses WebSocket 连接生命周期与多轮请求边界测试
 
 mod middleware;
+mod transport;
 
 use std::{
     fmt,
@@ -311,6 +312,7 @@ fn test_connection(stall_writes: bool) -> PumpHarness {
         dropped: Arc::clone(&dropped),
     };
     let connection = spawn_connection(
+        tokio::sync::watch::channel(tokio::time::Instant::now()).0,
         socket,
         Arc::from("ws_test"),
         cancellation.clone(),

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

@@ -0,0 +1,181 @@
+//! 实际 HTTP 升级及 WS 编解码下，慢速上传与半帧停传的心跳边界
+
+use std::{
+    io,
+    pin::Pin,
+    sync::{Arc, atomic::Ordering},
+    task::{Context, Poll},
+    time::Duration,
+};
+
+use futures::{SinkExt, StreamExt, ready};
+use hyper_util::rt::TokioIo;
+use serde_json::{Value, json};
+use tokio::{
+    io::{AsyncRead, AsyncWrite, AsyncWriteExt, DuplexStream, ReadBuf},
+    time::{Instant, Sleep},
+};
+use tokio_tungstenite::{
+    WebSocketStream, client_async,
+    tungstenite::{Message, client::IntoClientRequest},
+};
+use tower::ServiceExt;
+
+use super::{CrossingLimitExecution, CrossingLimitTrace, api_router, authenticated_client};
+
+struct SlowUpload {
+    inner: DuplexStream,
+    slow: bool,
+    next_write: Option<Pin<Box<Sleep>>>,
+}
+
+impl AsyncRead for SlowUpload {
+    fn poll_read(
+        mut self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        buffer: &mut ReadBuf<'_>,
+    ) -> Poll<io::Result<()>> {
+        Pin::new(&mut self.inner).poll_read(cx, buffer)
+    }
+}
+
+impl AsyncWrite for SlowUpload {
+    fn poll_write(
+        mut self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        buffer: &[u8],
+    ) -> Poll<io::Result<usize>> {
+        if self.slow {
+            if let Some(delay) = self.next_write.as_mut() {
+                ready!(delay.as_mut().poll(cx));
+            }
+            let size = ready!(
+                Pin::new(&mut self.inner).poll_write(cx, &buffer[..buffer.len().min(4096)])
+            )?;
+            self.next_write = Some(Box::pin(tokio::time::sleep(Duration::from_secs(1))));
+            Poll::Ready(Ok(size))
+        } else {
+            Pin::new(&mut self.inner).poll_write(cx, buffer)
+        }
+    }
+
+    fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
+        Pin::new(&mut self.inner).poll_flush(cx)
+    }
+
+    fn poll_shutdown(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
+        Pin::new(&mut self.inner).poll_shutdown(cx)
+    }
+}
+
+async fn connect() -> (WebSocketStream<SlowUpload>, Arc<CrossingLimitTrace>) {
+    let trace = Arc::new(CrossingLimitTrace::default());
+    let app = api_router(Arc::new(CrossingLimitExecution {
+        client: authenticated_client("sk_ws_crossing_limit"),
+        trace: trace.clone(),
+    }))
+    .await;
+    let (client, server) = tokio::io::duplex(65536);
+    tokio::spawn(async move {
+        hyper::server::conn::http1::Builder::new()
+            .serve_connection(
+                TokioIo::new(server),
+                hyper::service::service_fn(move |request| app.clone().oneshot(request)),
+            )
+            .with_upgrades()
+            .await
+            .expect("serve Responses upgrade");
+    });
+    let mut request = "ws://localhost/v1/responses".into_client_request().unwrap();
+    request.headers_mut().insert(
+        http::header::AUTHORIZATION,
+        "Bearer sk_ws_crossing_limit".parse().unwrap(),
+    );
+    let (socket, response) = client_async(
+        request,
+        SlowUpload {
+            inner: client,
+            slow: false,
+            next_write: None,
+        },
+    )
+    .await
+    .unwrap();
+    assert_eq!(response.status(), http::StatusCode::SWITCHING_PROTOCOLS);
+    (socket, trace)
+}
+
+async fn receive_event(socket: &mut WebSocketStream<SlowUpload>, expected: &str) {
+    tokio::time::timeout(Duration::from_secs(2), async {
+        loop {
+            let message = socket
+                .next()
+                .await
+                .expect("connection stays open")
+                .expect("valid frame");
+            if let Message::Text(text) = message {
+                let event: Value = serde_json::from_str(&text).unwrap();
+                if event["type"] == expected {
+                    break;
+                }
+                assert_eq!(event["type"], "response.metadata");
+            } else {
+                assert!(matches!(message, Message::Ping(_)));
+            }
+        }
+    })
+    .await
+    .expect("receive response after upload");
+}
+
+#[tokio::test(start_paused = true)]
+async fn slow_unfragmented_uploads_survive_heartbeats_on_first_and_later_requests() {
+    let (mut socket, trace) = connect().await;
+    let input = "x".repeat(320 * 1024);
+    for round in 1..=2 {
+        let started = Instant::now();
+        socket.get_mut().slow = true;
+        // 对齐 Codex 的发送语义：整个 send 完成前不读 Ping，也不主动发送 Pong
+        tokio::time::timeout(
+            Duration::from_secs(100),
+            socket.send(Message::Text(
+                json!({"type": "response.create", "model": "model-a", "input": input})
+                    .to_string()
+                    .into(),
+            )),
+        )
+        .await
+        .expect("upload completes")
+        .expect("heartbeat must not interrupt an active upload");
+        assert!(started.elapsed() >= Duration::from_secs(80));
+        socket.get_mut().slow = false;
+        receive_event(&mut socket, "response.created").await;
+        assert_eq!(trace.starts.load(Ordering::Acquire), round);
+        trace.release_terminal.notify_one();
+        receive_event(&mut socket, "response.completed").await;
+    }
+    assert_eq!(
+        *trace.inputs.lock().unwrap(),
+        vec![json!(input), json!(input)]
+    );
+    socket.close(None).await.unwrap();
+}
+
+#[tokio::test(start_paused = true)]
+async fn stalled_partial_frame_without_pongs_still_releases_the_connection() {
+    let (mut socket, trace) = connect().await;
+    // 已掩码的文本帧声明 4096 字节，只交付一个字节，之后不再收发
+    socket
+        .get_mut()
+        .write_all(&[0x81, 0xfe, 0x10, 0x00, 0, 0, 0, 0, b'x'])
+        .await
+        .unwrap();
+    tokio::time::sleep(Duration::from_secs(56)).await;
+    let error = socket
+        .get_mut()
+        .write_all(b"x")
+        .await
+        .expect_err("idle partial frame must be closed");
+    assert_eq!(error.kind(), io::ErrorKind::BrokenPipe);
+    assert_eq!(trace.starts.load(Ordering::Acquire), 0);
+}

deploy/README.md

@@ -306,7 +306,8 @@ Responses 流式请求完成准入后，每 15 秒无输出会发送一次 SSE
 排队、插件显式执行期限与客户端断开的边界见 [请求期限](../docs/api.md#请求期限)
 
 OpenAI 上游 WebSocket 被动回应 Ping，不主动发送心跳；Pong 写入最多等待 5 秒。
-下游 Responses WebSocket 每 25 秒发送 Ping，第一条未响应的 Ping 发出后 30 秒没有 Pong 或其他有效入站帧即断开；
+下游 Responses WebSocket 每 25 秒发送 Ping，第一条未响应的 Ping 发出后 30 秒没有入站活动即断开；
+接收中的消息字节、Pong 和其他有效入站消息均算活动，慢速上传不必等整帧收齐才刷新探活状态。
 后续 Ping 不延长这次截止，心跳入队及写入最多等待 5 秒。两段心跳各自处理，不跨 WS 转发或转换为 SSE 事件。
 心跳不重置上游业务消息的 `openai.ws_pool.stream_idle_timeout_ms`
 

docs/api.md

@@ -468,6 +468,7 @@ Responses WebSocket 接受文本消息，`response.create` 在同一连接串行
 接收队列容量为 32 个事件，超载仍关闭连接；Ping/Pong、客户端关闭和服务关闭不等待队列中的请求执行
 
 网关分别维护两段链路的心跳：上游 WS 被动回复 Pong，下游 WS 独立探活；Ping/Pong 不跨段转发，也不转换为 SSE 事件。
+下游探活包含消息接收中的字节进度，完整消息收齐后才进入中间件与请求处理。
 客户端断开时取消当前执行，并释放仍归该客户端所有的上游连接。`response.completed` 只结束本轮响应，保留客户端 WS 等待下一轮请求
 
 | 入站 → 上游 | 正常回答结束后的连接处理 |