From a704fbc3f34d25c7aca7580a4340c4b4fedfff52 Mon Sep 17 00:00:00 2001 From: Zyy Date: Sun, 6 Sep 2026 13:55:14 +0800 Subject: [PATCH] fix(websocket): serialize queued response requests Queue incoming response.create frames until the active response finishes instead of closing the connection, aligning execution with the Codex client serial request flow. Separate pump exit notifications from the bounded business queue so disconnects and shutdown remain prompt. Preserve expiry handling with a full queue, retain protocol validation, and document the serial execution contract. Validated with 253 API tests, 12 architecture tests, strict workspace Clippy and Rust formatting. Co-authored-by: Codex --- .../openai/responses/websocket/connection.rs | 40 +- .../src/openai/responses/websocket/forward.rs | 120 +++--- .../openai/responses/websocket/connection.rs | 359 ++++++++++++++++-- docs/api.md | 5 + 4 files changed, 418 insertions(+), 106 deletions(-) diff --git a/backend/crates/gateway-api/src/openai/responses/websocket/connection.rs b/backend/crates/gateway-api/src/openai/responses/websocket/connection.rs index 3bc77f3fb..c3f0d0450 100644 --- a/backend/crates/gateway-api/src/openai/responses/websocket/connection.rs +++ b/backend/crates/gateway-api/src/openai/responses/websocket/connection.rs @@ -195,6 +195,7 @@ pub struct ResponsesWebSocketConnection { expired: Arc, commands: Option>, incoming: mpsc::Receiver, + exited: oneshot::Receiver, pump_task: Option>, stats: Arc, config: ConnectionConfig, @@ -231,7 +232,16 @@ impl ResponsesWebSocketConnection { /// 等待下一个需要业务层处理的客户端事件。 pub async fn next_event(&mut self) -> Option { - let event = self.incoming.recv().await; + if let Some(reason) = self.exit_reason { + return Some(ConnectionEvent::Exited(reason)); + } + let event = tokio::select! { + biased; + reason = &mut self.exited => { + Some(ConnectionEvent::Exited(reason.unwrap_or(PumpExitReason::PumpStopped))) + } + event = self.incoming.recv() => event, + }; if let Some(ConnectionEvent::Exited(reason)) = event.as_ref() { self.exit_reason.get_or_insert(*reason); } else if event.is_none() { @@ -240,6 +250,18 @@ impl ResponsesWebSocketConnection { event } + /// 等待连接退出，不消费留给后续串行请求的业务帧。 + pub async fn wait_for_exit(&mut self) -> PumpExitReason { + if let Some(reason) = self.exit_reason { + return reason; + } + let reason = (&mut self.exited) + .await + .unwrap_or(PumpExitReason::PumpStopped); + self.exit_reason = Some(reason); + reason + } + /// 串行写入文本帧，并等待 pump 确认 transport 写入结果。 /// /// # Errors @@ -425,10 +447,12 @@ where let stats = Arc::new(ConnectionStats::default()); let (command_tx, command_rx) = mpsc::channel(OUTBOUND_COMMAND_BUFFER); let (incoming_tx, incoming_rx) = mpsc::channel(INBOUND_EVENT_BUFFER); + let (exit_tx, exit_rx) = oneshot::channel(); let pump_task = tokio::spawn(run_pump( socket, command_rx, incoming_tx, + exit_tx, Arc::clone(&connection_id), cancellation, opened_at, @@ -442,6 +466,7 @@ where expired, commands: Some(command_tx), incoming: incoming_rx, + exited: exit_rx, pump_task: Some(pump_task), stats, config, @@ -454,6 +479,7 @@ async fn run_pump( mut socket: S, mut commands: mpsc::Receiver, incoming: mpsc::Sender, + exited: oneshot::Sender, connection_id: Arc, cancellation: CancellationToken, opened_at: Instant, @@ -474,8 +500,13 @@ async fn run_pump( () = &mut deadline, if !deadline_elapsed => { deadline_elapsed = true; expired.store(true, Ordering::Release); - if let Err(reason) = emit_incoming(&incoming, ConnectionEvent::Expired) { - break reason; + // 满队列本身已能唤醒空闲协调层；到期标志仍阻止启动下一轮。 + // 不能因无法追加 Expired 通知而中断正在收尾的响应。 + match incoming.try_send(ConnectionEvent::Expired) { + Ok(()) | Err(mpsc::error::TrySendError::Full(_)) => {} + Err(mpsc::error::TrySendError::Closed(_)) => { + break PumpExitReason::CoordinatorDropped; + } } } command = commands.recv() => { @@ -550,7 +581,8 @@ async fn run_pump( }, } }; - let _ = incoming.try_send(ConnectionEvent::Exited(reason)); + // 退出不能排在待执行请求之后，也不能因业务队列已满而丢失。 + let _ = exited.send(reason); tracing::debug!( websocket_connection_id = %connection_id, pump_exit_reason = reason.as_str(), diff --git a/backend/crates/gateway-api/src/openai/responses/websocket/forward.rs b/backend/crates/gateway-api/src/openai/responses/websocket/forward.rs index b61532f02..207e0f6c7 100644 --- a/backend/crates/gateway-api/src/openai/responses/websocket/forward.rs +++ b/backend/crates/gateway-api/src/openai/responses/websocket/forward.rs @@ -13,12 +13,10 @@ use crate::openai::error::{gateway_error_contract, gateway_error_from_engine}; use super::{ super::{DecodedResponsesRequest, OpenAiResponsesEncoder, PendingExecution, ProtocolErrorBody}, - connection::{ConnectionEvent, FramePhase, ResponsesWebSocketConnection, WriteContext}, + connection::{FramePhase, ResponsesWebSocketConnection, WriteContext}, protocol::{error_event, response_metadata_event}, }; -const ACTIVE_RESPONSE_VIOLATION: &str = "Only one response.create may be active per connection"; - #[derive(Clone, Copy, PartialEq, Eq)] pub(super) enum ForwardOutcome { Continue, @@ -66,29 +64,20 @@ pub(super) async fn forward_execution( ); return send_gateway_error(connection, &error, &request_id).await; } - let first = loop { - match next_active_input(connection, &mut execution).await { - ActiveInput::Event(Ok(Some(event))) => break event, - ActiveInput::Event(Ok(None)) => { - let error = GatewayError::new( - GatewayErrorKind::Internal, - "gateway response ended before its first event", - ); - return send_gateway_error(connection, &error, &request_id).await; - } - ActiveInput::Event(Err(error)) => { - return send_initial_engine_error(connection, &mut execution, &error, &request_id) - .await; - } - ActiveInput::Control => continue, - ActiveInput::Disconnect => return ForwardOutcome::Disconnect, - ActiveInput::ProtocolViolation => { - connection - .close_policy(ACTIVE_RESPONSE_VIOLATION, Some(&request_id)) - .await; - return ForwardOutcome::Disconnect; - } + let first = match next_active_input(connection, &mut execution).await { + ActiveInput::Event(Ok(Some(event))) => event, + ActiveInput::Event(Ok(None)) => { + let error = GatewayError::new( + GatewayErrorKind::Internal, + "gateway response ended before its first event", + ); + return send_gateway_error(connection, &error, &request_id).await; } + ActiveInput::Event(Err(error)) => { + return send_initial_engine_error(connection, &mut execution, &error, &request_id) + .await; + } + ActiveInput::Disconnect => return ForwardOutcome::Disconnect, }; let requirement = first.commit_requirement(); let mut first = first.into_provider_events(); @@ -259,14 +248,7 @@ pub(super) async fn forward_execution( let error = gateway_error_from_engine(&error); return send_gateway_error(connection, &error, &request_id).await; } - ActiveInput::Control => {} ActiveInput::Disconnect => return ForwardOutcome::Disconnect, - ActiveInput::ProtocolViolation => { - connection - .close_policy(ACTIVE_RESPONSE_VIOLATION, Some(&request_id)) - .await; - return ForwardOutcome::Disconnect; - } } } } @@ -310,50 +292,39 @@ async fn confirm_completed_execution( execution: &mut PendingExecution, request_id: &Arc, ) -> Result<(), ForwardOutcome> { - loop { - match next_active_input(connection, execution).await { - ActiveInput::Event(Ok(None)) - if execution - .session_mut() - .is_some_and(|session| session.is_finalized()) => - { - return Ok(()); - } - ActiveInput::Event(Ok(None)) => { - let error = GatewayError::new( - GatewayErrorKind::Internal, - "gateway response was not finalized after its terminal event", - ); - return Err(send_gateway_error(connection, &error, request_id).await); - } - ActiveInput::Event(Ok(Some(_))) => { - let error = GatewayError::new( - GatewayErrorKind::Internal, - "gateway response continued after its terminal event", - ); - return Err(send_gateway_error(connection, &error, request_id).await); - } - ActiveInput::Event(Err(error)) => { - let error = gateway_error_from_engine(&error); - return Err(send_gateway_error(connection, &error, request_id).await); - } - ActiveInput::Control => {} - ActiveInput::Disconnect => return Err(ForwardOutcome::Disconnect), - ActiveInput::ProtocolViolation => { - connection - .close_policy(ACTIVE_RESPONSE_VIOLATION, Some(request_id)) - .await; - return Err(ForwardOutcome::Disconnect); - } + match next_active_input(connection, execution).await { + ActiveInput::Event(Ok(None)) + if execution + .session_mut() + .is_some_and(|session| session.is_finalized()) => + { + Ok(()) } + ActiveInput::Event(Ok(None)) => { + let error = GatewayError::new( + GatewayErrorKind::Internal, + "gateway response was not finalized after its terminal event", + ); + Err(send_gateway_error(connection, &error, request_id).await) + } + ActiveInput::Event(Ok(Some(_))) => { + let error = GatewayError::new( + GatewayErrorKind::Internal, + "gateway response continued after its terminal event", + ); + Err(send_gateway_error(connection, &error, request_id).await) + } + ActiveInput::Event(Err(error)) => { + let error = gateway_error_from_engine(&error); + Err(send_gateway_error(connection, &error, request_id).await) + } + ActiveInput::Disconnect => Err(ForwardOutcome::Disconnect), } } enum ActiveInput { Event(Result, EngineError>), - Control, Disconnect, - ProtocolViolation, } async fn next_active_input( @@ -363,15 +334,12 @@ async fn next_active_input( let Some(session) = execution.session_mut() else { return ActiveInput::Disconnect; }; + // 与 Codex stream_request 的连接锁一致：本轮结束前不消费下一条请求。 + // pump 继续接收有界业务帧和处理 Ping/Pong；退出通过独立通知取消本轮。 tokio::select! { + biased; + _ = connection.wait_for_exit() => ActiveInput::Disconnect, event = session.next_event() => ActiveInput::Event(event), - message = connection.next_event() => match message { - Some(ConnectionEvent::Expired) => ActiveInput::Control, - Some(ConnectionEvent::Text(_) | ConnectionEvent::Binary) => { - ActiveInput::ProtocolViolation - } - Some(ConnectionEvent::Exited(_)) | None => ActiveInput::Disconnect, - }, } } diff --git a/backend/crates/gateway-api/tests/openai/responses/websocket/connection.rs b/backend/crates/gateway-api/tests/openai/responses/websocket/connection.rs index 7631dfdee..d3a7efc6c 100644 --- a/backend/crates/gateway-api/tests/openai/responses/websocket/connection.rs +++ b/backend/crates/gateway-api/tests/openai/responses/websocket/connection.rs @@ -2,8 +2,8 @@ use std::{ fmt, pin::Pin, sync::{ - Arc, - atomic::{AtomicBool, Ordering}, + Arc, Mutex, + atomic::{AtomicBool, AtomicUsize, Ordering}, }, task::{Context, Poll}, time::{Duration, SystemTime}, @@ -25,6 +25,7 @@ use gateway_core::{ error::GatewayError, event::{GatewayEvent, ProtocolWireEvent, ProviderEvent, ProviderResponseHeader, ResponseMeta}, lifecycle::CancellationToken, + operation::Operation, routing::PublicModelId, }; use serde_json::{Value, json}; @@ -32,7 +33,7 @@ use tokio::sync::mpsc::{ UnboundedReceiver, UnboundedSender, error::TryRecvError, unbounded_channel, }; use tokio_tungstenite::{ - connect_async, + MaybeTlsStream, WebSocketStream, connect_async, tungstenite::{ Message as ClientMessage, client::IntoClientRequest, protocol::frame::coding::CloseCode, }, @@ -44,27 +45,33 @@ use crate::openai::{api_router, authenticated_client}; #[derive(Default)] struct CrossingLimitTrace { release_terminal: tokio::sync::Notify, - started_emitted: AtomicBool, - terminal_emitted: AtomicBool, + starts: AtomicUsize, + inputs: Mutex>, + cancelled: AtomicBool, + cancellation_observed: tokio::sync::Notify, finalized: AtomicBool, } struct CrossingLimitSession { trace: Arc, + started_emitted: bool, + terminal_emitted: bool, + finalized: bool, } impl ExecutionSession for CrossingLimitSession { fn next_event(&mut self) -> BoxFuture<'_, Result, EngineError>> { Box::pin(async move { - if !self.trace.started_emitted.swap(true, Ordering::AcqRel) { + if !self.started_emitted { + self.started_emitted = true; return Ok(Some(crossing_limit_started_batch())); } - if !self.trace.terminal_emitted.load(Ordering::Acquire) { + if !self.terminal_emitted { self.trace.release_terminal.notified().await; - if !self.trace.terminal_emitted.swap(true, Ordering::AcqRel) { - return Ok(Some(crossing_limit_terminal_batch())); - } + self.terminal_emitted = true; + return Ok(Some(crossing_limit_terminal_batch())); } + self.finalized = true; self.trace.finalized.store(true, Ordering::Release); Ok(None) }) @@ -87,10 +94,13 @@ impl ExecutionSession for CrossingLimitSession { } fn is_finalized(&self) -> bool { - self.trace.finalized.load(Ordering::Acquire) + self.finalized } - fn cancel(&self) {} + fn cancel(&self) { + self.trace.cancelled.store(true, Ordering::Release); + self.trace.cancellation_observed.notify_one(); + } fn detach_finalize(self: Box) -> BoxFuture<'static, ()> { Box::pin(async move { @@ -127,6 +137,14 @@ impl ExecutionService for CrossingLimitExecution { request: StartExecution, ) -> BoxFuture<'_, Result> { Box::pin(async move { + self.trace.starts.fetch_add(1, Ordering::AcqRel); + if let Operation::Generate(generate) = &request.operation { + self.trace + .inputs + .lock() + .expect("input trace lock") + .push(generate.protocol_payload().body()["input"].clone()); + } Ok(StartedExecution { request_id: gateway_core::engine::ModelRequestId::new("req_ws_crossing_limit") .expect("request id"), @@ -134,6 +152,9 @@ impl ExecutionService for CrossingLimitExecution { stream: request.metadata.stream, session: Box::new(CrossingLimitSession { trace: Arc::clone(&self.trace), + started_emitted: false, + terminal_emitted: false, + finalized: false, }), }) }) @@ -192,7 +213,7 @@ fn crossing_limit_terminal_batch() -> CoordinatedEvent { } use gateway_api::openai::responses::websocket::connection::{ - ConnectionConfig, ConnectionEvent, ConnectionWriteError, FramePhase, + ConnectionConfig, ConnectionEvent, ConnectionWriteError, FramePhase, PumpExitReason, ResponsesWebSocketConnection, WriteContext, spawn_connection, }; @@ -255,12 +276,14 @@ struct PumpHarness { incoming: UnboundedSender>, written: UnboundedReceiver, dropped: Arc, + cancellation: CancellationToken, } fn test_connection(stall_writes: bool) -> PumpHarness { let (incoming_tx, incoming_rx) = unbounded_channel(); let (written_tx, written_rx) = unbounded_channel(); let dropped = Arc::new(AtomicBool::new(false)); + let cancellation = CancellationToken::new(); let socket = TestSocket { incoming: incoming_rx, written: written_tx, @@ -270,7 +293,7 @@ fn test_connection(stall_writes: bool) -> PumpHarness { let connection = spawn_connection( socket, Arc::from("ws_test"), - CancellationToken::new(), + cancellation.clone(), ConnectionConfig::PRODUCTION, ); PumpHarness { @@ -278,6 +301,7 @@ fn test_connection(stall_writes: bool) -> PumpHarness { incoming: incoming_tx, written: written_rx, dropped, + cancellation, } } @@ -316,6 +340,110 @@ async fn pump_consumes_pong_without_forwarding_a_business_event() { assert!(result.is_err(), "Pong unexpectedly reached the coordinator"); } +#[tokio::test(start_paused = true)] +async fn waiting_for_exit_should_preserve_queued_business_frames() { + let PumpHarness { + mut connection, + incoming, + .. + } = test_connection(false); + for payload in ["first", "second"] { + incoming + .send(Ok(Message::Text(payload.into()))) + .expect("queue text"); + } + + assert!( + tokio::time::timeout(Duration::from_secs(1), connection.wait_for_exit()) + .await + .is_err() + ); + for expected in ["first", "second"] { + assert!( + matches!(connection.next_event().await, Some(ConnectionEvent::Text(payload)) if payload == expected) + ); + } +} + +#[tokio::test] +async fn client_close_should_take_priority_over_queued_business_frames() { + let PumpHarness { + mut connection, + incoming, + .. + } = test_connection(false); + incoming + .send(Ok(Message::Text("queued".into()))) + .expect("queue text"); + incoming + .send(Ok(Message::Close(None))) + .expect("close socket"); + + assert_eq!( + tokio::time::timeout(Duration::from_secs(1), connection.wait_for_exit()) + .await + .expect("observe close"), + PumpExitReason::ClientClose + ); + assert!(matches!( + connection.next_event().await, + Some(ConnectionEvent::Exited(PumpExitReason::ClientClose)) + )); +} + +#[tokio::test] +async fn full_business_queue_should_not_block_lifecycle_shutdown() { + let PumpHarness { + mut connection, + incoming, + mut written, + cancellation, + .. + } = test_connection(false); + for _ in 0..32 { + incoming + .send(Ok(Message::Text("queued".into()))) + .expect("queue text"); + } + incoming + .send(Ok(Message::Ping(vec![1].into()))) + .expect("send Ping after queue fills"); + assert!(matches!(written.recv().await, Some(Message::Pong(_)))); + cancellation.cancel(); + + assert_eq!( + tokio::time::timeout(Duration::from_secs(1), connection.wait_for_exit()) + .await + .expect("observe shutdown"), + PumpExitReason::LifecycleShutdown + ); +} + +#[tokio::test] +async fn inbound_overload_should_close_without_executing_queued_frames() { + let PumpHarness { + mut connection, + incoming, + .. + } = test_connection(false); + for _ in 0..33 { + incoming + .send(Ok(Message::Text("queued".into()))) + .expect("queue text"); + } + + assert_eq!( + tokio::time::timeout(Duration::from_secs(1), connection.wait_for_exit()) + .await + .expect("observe overload"), + PumpExitReason::InboundOverload + ); + assert!(matches!( + connection.next_event().await, + Some(ConnectionEvent::Exited(PumpExitReason::InboundOverload)) + )); +} + #[tokio::test(start_paused = true)] async fn pump_does_not_emit_an_active_ping() { let PumpHarness { @@ -371,6 +499,7 @@ async fn stalled_write_times_out_at_three_hundred_seconds_and_aborts_the_pump() incoming: _incoming, written: _written, dropped, + .. } = test_connection(true); let request_id = Arc::::from("req_test"); @@ -413,6 +542,7 @@ async fn dropping_the_connection_aborts_and_drops_the_socket_owner() { incoming: _incoming, written: _written, dropped, + .. } = test_connection(false); drop(connection); @@ -540,6 +670,27 @@ async fn active_response_crossing_the_limit_writes_terminal_before_the_limit_err } } + for _ in 0..32 { + socket + .send(ClientMessage::Text( + json!({"type": "response.create", "model": "model-a", "input": "queued after limit"}) + .to_string() + .into(), + )) + .await + .expect("fill request queue before expiry"); + } + socket + .send(ClientMessage::Ping(vec![1].into())) + .await + .expect("send Ping after queued request"); + let pong = tokio::time::timeout(Duration::from_secs(1), socket.next()) + .await + .expect("Pong timeout") + .expect("connection remains open") + .expect("valid frame"); + assert!(matches!(pong, ClientMessage::Pong(_))); + tokio::time::pause(); tokio::time::advance(Duration::from_secs(60 * 60)).await; tokio::task::yield_now().await; @@ -568,15 +719,21 @@ async fn active_response_crossing_the_limit_writes_terminal_before_the_limit_err assert_eq!(event_types, ["response.completed", "error"]); assert!(trace.finalized.load(Ordering::Acquire)); + assert_eq!(trace.starts.load(Ordering::Acquire), 1); server.abort(); } -#[tokio::test] -async fn second_text_frame_during_an_active_response_keeps_the_single_response_contract() { +type TestClientSocket = WebSocketStream>; + +async fn start_active_response() -> ( + Arc, + TestClientSocket, + tokio::task::JoinHandle<()>, +) { let trace = Arc::new(CrossingLimitTrace::default()); let execution = Arc::new(CrossingLimitExecution { client: authenticated_client("sk_ws_crossing_limit"), - trace, + trace: Arc::clone(&trace), }); let app = api_router(execution).await; let listener = tokio::net::TcpListener::bind("127.0.0.1:0") @@ -608,7 +765,7 @@ async fn second_text_frame_during_an_active_response_keeps_the_single_response_c .into(), ); socket - .send(response_create.clone()) + .send(response_create) .await .expect("send first response.create"); loop { @@ -626,19 +783,169 @@ async fn second_text_frame_during_an_active_response_keeps_the_single_response_c } } + (trace, socket, server) +} + +#[tokio::test] +async fn requests_received_during_an_active_response_should_execute_in_order() { + let (trace, mut socket, server) = start_active_response().await; + for input in ["second", "third"] { + socket + .send(ClientMessage::Text( + json!({"type": "response.create", "model": "model-a", "input": input}) + .to_string() + .into(), + )) + .await + .expect("send queued response.create"); + } socket - .send(response_create) + .send(ClientMessage::Ping(vec![1, 2, 3].into())) .await - .expect("send overlapping response.create"); + .expect("send Ping after queued requests"); + let message = tokio::time::timeout(Duration::from_secs(2), socket.next()) + .await + .expect("Pong timeout while response is active") + .expect("WebSocket remains open") + .expect("valid WebSocket frame"); + assert!(matches!(message, ClientMessage::Pong(_))); + assert_eq!(trace.starts.load(Ordering::Acquire), 1); + + for completed_count in 1..=3 { + trace.release_terminal.notify_one(); + let expected_types = if completed_count < 3 { + vec![ + "response.completed", + "response.metadata", + "response.created", + ] + } else { + vec!["response.completed"] + }; + let mut event_types = Vec::new(); + for _ in &expected_types { + let message = tokio::time::timeout(Duration::from_secs(2), socket.next()) + .await + .expect("serial response timeout") + .expect("WebSocket remains open") + .expect("valid WebSocket frame"); + let ClientMessage::Text(text) = message else { + panic!("unexpected frame during serial execution: {message:?}"); + }; + let event: Value = serde_json::from_str(&text).expect("response event JSON"); + event_types.push(event["type"].as_str().expect("event type").to_owned()); + } + assert_eq!(event_types, expected_types); + assert_eq!( + trace.starts.load(Ordering::Acquire), + (completed_count + 1).min(3) + ); + } + assert_eq!( + *trace.inputs.lock().expect("input trace lock"), + vec![json!("hello"), json!("second"), json!("third")] + ); + assert!(!trace.cancelled.load(Ordering::Acquire)); + socket.close(None).await.expect("close WebSocket"); + server.abort(); +} + +#[tokio::test] +async fn queued_unknown_message_should_be_rejected_before_the_next_valid_request() { + let (trace, mut socket, server) = start_active_response().await; + for payload in [ + json!({"type": "response.future"}), + json!({"type": "response.create", "model": "model-a", "input": "next"}), + ] { + socket + .send(ClientMessage::Text(payload.to_string().into())) + .await + .expect("queue frame"); + } + trace.release_terminal.notify_one(); + + let mut event_types = Vec::new(); + for _ in 0..4 { + let message = tokio::time::timeout(Duration::from_secs(2), socket.next()) + .await + .expect("response timeout") + .expect("connection remains open") + .expect("valid frame"); + let ClientMessage::Text(text) = message else { + panic!("unexpected frame: {message:?}"); + }; + let event: Value = serde_json::from_str(&text).expect("event JSON"); + event_types.push(event["type"].as_str().expect("event type").to_owned()); + } + assert_eq!( + event_types, + [ + "response.completed", + "error", + "response.metadata", + "response.created" + ] + ); + assert_eq!(trace.starts.load(Ordering::Acquire), 2); + assert_eq!( + *trace.inputs.lock().expect("input trace lock"), + vec![json!("hello"), json!("next")] + ); + socket.close(None).await.expect("close WebSocket"); + server.abort(); +} + +#[tokio::test] +async fn queued_binary_frame_should_be_rejected_after_the_active_response_finishes() { + let (trace, mut socket, server) = start_active_response().await; + socket + .send(ClientMessage::Binary(b"{}".to_vec().into())) + .await + .expect("queue binary frame"); + trace.release_terminal.notify_one(); + + let message = tokio::time::timeout(Duration::from_secs(2), socket.next()) + .await + .expect("terminal timeout") + .expect("connection remains open") + .expect("valid frame"); + let ClientMessage::Text(text) = message else { + panic!("active response did not finish before binary rejection: {message:?}"); + }; + let event: Value = serde_json::from_str(&text).expect("event JSON"); + assert_eq!(event["type"], "response.completed"); let message = tokio::time::timeout(Duration::from_secs(2), socket.next()) .await .expect("policy close timeout") - .expect("WebSocket returns a policy close") - .expect("valid WebSocket frame"); - let ClientMessage::Close(Some(frame)) = message else { - panic!("overlapping response.create must close the connection"); - }; + .expect("connection closes") + .expect("valid frame"); + assert!( + matches!(message, ClientMessage::Close(Some(frame)) if frame.code == CloseCode::Policy) + ); + assert_eq!(trace.starts.load(Ordering::Acquire), 1); + server.abort(); +} - assert_eq!(frame.code, CloseCode::Policy); +#[tokio::test] +async fn client_close_should_cancel_active_execution_without_starting_queued_requests() { + let (trace, mut socket, server) = start_active_response().await; + socket + .send(ClientMessage::Text( + json!({"type": "response.create", "model": "model-a", "input": "queued"}) + .to_string() + .into(), + )) + .await + .expect("queue request"); + socket.close(None).await.expect("close WebSocket"); + + tokio::time::timeout( + Duration::from_secs(2), + trace.cancellation_observed.notified(), + ) + .await + .expect("cancel active execution promptly"); + assert_eq!(trace.starts.load(Ordering::Acquire), 1); + assert!(trace.cancelled.load(Ordering::Acquire)); server.abort(); } diff --git a/docs/api.md b/docs/api.md index ac6241328..5a66aa77a 100644 --- a/docs/api.md +++ b/docs/api.md @@ -124,6 +124,11 @@ Responses、Images 和 standalone Search HTTP body、WebSocket message 和 frame Codex 的 review 等子代理请求仍使用 `/v1/responses`，并通过 `x-openai-subagent` 请求头携带子代理类型； 网关不提供独立的子代理请求路径。 +Responses WebSocket 仅接受文本 `response.create`，同一连接串行执行。当前响应期间收到的后续业务帧 +留在有界接收队列中，待当前响应完成终结和写出后再逐条校验、准入与执行，不因请求提前到达而断开。 +这对齐 Codex 客户端 `stream_request` 持锁至本轮结束的串行行为，不表示支持额外控制消息类型。 +接收队列容量为 32 个事件，超载仍关闭连接；Ping/Pong、客户端关闭和服务关闭不等待队列中的请求执行。 + `GET /v1/models` 默认返回 OpenAI 兼容列表 `{"object": "list", "data": [...]}`；请求携带非空 `client_version` query 参数（Codex 客户端）时改为返回 Codex 专用目录合同 `{"models": [...]}`。