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": [...]}`。