diff --git a/backend/crates/gateway-api/src/openai/responses/websocket/mod.rs b/backend/crates/gateway-api/src/openai/responses/websocket/mod.rs index 9a067c806..96c81aafb 100644 --- a/backend/crates/gateway-api/src/openai/responses/websocket/mod.rs +++ b/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, diff --git a/backend/crates/gateway-api/tests/openai/responses/websocket/connection/middleware.rs b/backend/crates/gateway-api/tests/openai/responses/websocket/connection/middleware.rs index 10c3595d4..d511da5e3 100644 --- a/backend/crates/gateway-api/tests/openai/responses/websocket/connection/middleware.rs +++ b/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(); diff --git a/backend/crates/gateway-api/tests/openai/responses/websocket/connection/mod.rs b/backend/crates/gateway-api/tests/openai/responses/websocket/connection/mod.rs index 1fb6c8b96..4e3146e6e 100644 --- a/backend/crates/gateway-api/tests/openai/responses/websocket/connection/mod.rs +++ b/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(); } diff --git a/docs/api.md b/docs/api.md index 5df1f6aee..cb0317efc 100644 --- a/docs/api.md +++ b/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` 保留有效上游值，只有上游