fix(responses): reconnect idle WebSockets for replay regardless of client identity

Refs #596

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

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

@@ -179,12 +179,10 @@ async fn serve_responses_websocket(socket: Socket, session: ResponsesWebSocketSe
         crate::middleware::websocket::CancelOnDrop(Some(cancellation.clone()));
     let request_headers =
         request_headers.with_downstream_websocket_connection_id(connection_id.clone());
-    // Codex Core 观察正常关闭后会重建完整输入；原始帧由中间件改写时无法证明同一终态已交付
-    let reconnect_for_replay = crate::openai::auth::identify_codex_client(&raw_headers)
-        .is_some_and(|client| client.kind() == gateway_core::policy::CodexClientKind::Cli)
-        && !middleware
-            .as_ref()
-            .is_some_and(FrozenMiddlewarePlan::has_websocket);
+    // 帧中间件可能改写终态，此时不能据原始响应关闭下游连接
+    let reconnect_for_replay = !middleware
+        .as_ref()
+        .is_some_and(FrozenMiddlewarePlan::has_websocket);
     let mut connection = ResponsesWebSocketConnection::new(
         socket,
         connection_id,

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

@@ -388,6 +388,7 @@ async fn rewrite_upgrade_and_control_active_response_keep_http_context_until_con
     assert!(matches!(reply, ClientMessage::Binary(bytes) if bytes.as_ref() == [0,255,1]));
     assert_eq!(trace.starts.load(Ordering::SeqCst), 1);
     assert!(!trace.cancelled.load(Ordering::SeqCst));
+    trace.reconnect_for_replay.store(true, Ordering::Release);
     trace.release_terminal.notify_one();
     let reply = tokio::time::timeout(Duration::from_secs(2), socket.next())
         .await
@@ -400,6 +401,16 @@ async fn rewrite_upgrade_and_control_active_response_keep_http_context_until_con
     let event: Value = serde_json::from_str(&text).unwrap();
     assert_eq!(event["type"], "response.completed");
     assert_eq!(event["plugin"], true);
+    socket
+        .send(ClientMessage::Ping(vec![9].into()))
+        .await
+        .unwrap();
+    let reply = tokio::time::timeout(Duration::from_secs(2), socket.next())
+        .await
+        .expect("Pong after middleware rewrote the terminal")
+        .unwrap()
+        .unwrap();
+    assert!(matches!(reply, ClientMessage::Pong(_)));
     assert!(plan.incoming.load(Ordering::SeqCst) >= 2);
     assert!(plan.outgoing.load(Ordering::SeqCst) >= 3);
     socket.close(None).await.unwrap();

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

@@ -921,17 +921,7 @@ async fn start_active_response_with(
 
 #[tokio::test]
 async fn replay_reconnect_waits_for_queued_requests_and_writes_the_last_terminal() {
-    assert_replay_queue("codex-tui/0.166.0", true).await;
-}
-
-#[tokio::test]
-async fn third_party_client_keeps_its_socket_after_replay_required_responses() {
-    assert_replay_queue("integration-test", false).await;
-}
-
-async fn assert_replay_queue(user_agent: &str, expects_close: bool) {
-    let (trace, mut socket, server) =
-        start_active_response_with(None, "/v1/responses", user_agent).await;
+    let (trace, mut socket, server) = start_active_response().await;
     trace.reconnect_for_replay.store(true, Ordering::Release);
     for input in ["second", "third"] {
         socket
@@ -990,25 +980,13 @@ async fn assert_replay_queue(user_agent: &str, expects_close: bool) {
         vec![json!("hello"), json!("second"), json!("third")]
     );
     assert!(!trace.cancelled.load(Ordering::Acquire));
-    if !expects_close {
-        socket
-            .send(ClientMessage::Ping(vec![9].into()))
-            .await
-            .unwrap();
-    }
     let next = tokio::time::timeout(Duration::from_secs(2), socket.next())
         .await
-        .expect("close or pong")
+        .expect("replay close timeout")
         .unwrap()
         .unwrap();
-    if expects_close {
-        assert!(
-            matches!(next, ClientMessage::Close(Some(frame)) if frame.code == CloseCode::Normal)
-        );
-    } else {
-        assert!(matches!(next, ClientMessage::Pong(_)));
-        socket.close(None).await.unwrap();
-    }
+    assert!(matches!(next, ClientMessage::Close(Some(frame))
+        if frame.code == CloseCode::Normal && frame.reason == "Reconnect with full input."));
     server.abort();
 }
 

docs/api.md

@@ -475,7 +475,7 @@ Responses WebSocket 接受文本消息，`response.create` 在同一连接串行
 
 网关分别维护两段链路的心跳：上游 WS 定时发送 Ping 保活并回复 Pong，下游 WS 独立探活；Ping/Pong 不跨段转发，也不转换为 SSE 事件。
 下游探活包含消息接收中的字节进度，完整消息收齐后才进入中间件与请求处理。
-客户端断开时取消当前执行，并释放仍归该客户端所有的上游连接。`response.completed` 结束本轮响应。通常保留客户端 WS 等待下一轮请求，大包回退后的 Codex 重连规则见下文
+客户端断开时取消当前执行，并释放仍归该客户端所有的上游连接。`response.completed` 结束本轮响应。通常保留客户端 WS 等待下一轮请求，大包回退后的重连规则见下文
 
 | 入站 → 上游 | 正常回答结束后的连接处理 |
 | --- | --- |
@@ -531,9 +531,9 @@ OAuth 的大 connection-local WS 续接，以及 HTTP `store=false` 成功后标
 切到 HTTP；其他客户端需要自行实现此合同。代理不缓存 transcript，不把增量输入当作独立新链，
 连接内续接失去原连接时仍需客户端恢复
 
-因大小走 HTTP/SSE 且需要完整重放时，网关向可识别的 Codex CLI/Core 写出成功终态后，
-下游空闲则正常关闭 WS。客户端观察到关闭后，下一步在新连接上发送完整输入。
-已接收的后续请求按序处理；启用 WS 帧中间件时保留连接。若增量先到，仍按上述 400 合同恢复
+因大小走 HTTP/SSE 且需要完整重放时，网关写出成功终态后，会以关闭码 1000 正常关闭空闲的下游 WS。
+此规则适用于所有客户端。下一步请求应在新连接上发送完整输入，省略 `previous_response_id`。
+已接收的后续请求按序处理，若增量先到，仍按上述 400 合同恢复。启用 WS 帧中间件时保留连接
 
 已建立模型执行的 Responses、Images 和 Search HTTP 响应按以下规则返回关联 ID：
 `x-gateway-request-id` 为模型执行 ID；`x-request-id` 保留有效上游值，只有上游