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 d32181b38..0647f4355 100644 --- a/backend/crates/gateway-api/src/openai/responses/websocket/forward.rs +++ b/backend/crates/gateway-api/src/openai/responses/websocket/forward.rs @@ -770,6 +770,21 @@ pub(super) async fn send_control( payload: &str, request_id: &Arc, ) -> ForwardOutcome { + // HTTP 桥接没有原连接可中断;普通 error 会使 Codex 丢弃仍可正常完成的响应流 + // 仅在发送前确认无绑定时拒绝中断,不把发送结果不明的失败伪装成明确拒绝 + if control.binding().is_none() + && let Some(message) = super::interrupt::ResponseInterrupt::unavailable_event(payload) + { + return if connection + .send_text(message, WriteContext::request(request_id, FramePhase::Data)) + .await + .is_ok() + { + ForwardOutcome::Continue + } else { + ForwardOutcome::Disconnect + }; + } let result = tokio::select! { _ = connection.wait_for_exit() => return ForwardOutcome::Disconnect, result = control.send(payload) => result, diff --git a/backend/crates/gateway-api/src/openai/responses/websocket/interrupt.rs b/backend/crates/gateway-api/src/openai/responses/websocket/interrupt.rs index c4ebcea14..0d6a449c0 100644 --- a/backend/crates/gateway-api/src/openai/responses/websocket/interrupt.rs +++ b/backend/crates/gateway-api/src/openai/responses/websocket/interrupt.rs @@ -1,4 +1,4 @@ -//! 当前逻辑响应的中断意图,跨重试连接保留并随新响应 ID 转发 +//! 当前逻辑响应的中断意图、无控制通道时的拒绝与跨重试转发 use std::{borrow::Cow, collections::BTreeSet}; @@ -53,14 +53,12 @@ impl ResponseInterrupt { /// 未知控制消息、模式和其他逻辑响应的 ID 仍走原始透明通道 pub(super) fn retain(&mut self, payload: &str) -> bool { - let Ok(event) = serde_json::from_str::(payload) else { + let Some(event) = interrupt_request(payload) else { return false; }; - if event["type"] != "response.interrupt" - || event["mode"] != "discard_partial_items" - || !event["response_id"] - .as_str() - .is_some_and(|id| self.delivered_ids.contains(id)) + if !event["response_id"] + .as_str() + .is_some_and(|id| self.delivered_ids.contains(id)) { return false; } @@ -69,6 +67,18 @@ impl ResponseInterrupt { true } + /// 中断失败不代表模型响应失败,客户端仍可收完真实终态后续接 + pub(super) fn unavailable_event(payload: &str) -> Option { + let event = interrupt_request(payload)?; + Some( + serde_json::json!({ + "type": "response.interrupt.failed", + "response_id": event["response_id"], + }) + .to_string(), + ) + } + pub(super) async fn forward_pending(&mut self, control: &ResponseControl) { if self.sent { return; @@ -89,3 +99,13 @@ impl ResponseInterrupt { } } } + +fn interrupt_request(payload: &str) -> Option { + let event: Value = serde_json::from_str(payload).ok()?; + (event["type"] == "response.interrupt" + && event["mode"] == "discard_partial_items" + && event["response_id"] + .as_str() + .is_some_and(|id| !id.is_empty())) + .then_some(event) +} diff --git a/backend/crates/gateway-api/tests/openai/responses/websocket/interrupt.rs b/backend/crates/gateway-api/tests/openai/responses/websocket/interrupt.rs index 4f07d0a62..d5d75f8d5 100644 --- a/backend/crates/gateway-api/tests/openai/responses/websocket/interrupt.rs +++ b/backend/crates/gateway-api/tests/openai/responses/websocket/interrupt.rs @@ -1,4 +1,4 @@ -//! 验证未知控制消息、迟到中断、结算期间收发和创建请求的串行准入 +//! 验证控制消息透传、HTTP 中断拒绝、结算期间收发和创建请求的串行准入 use super::*; use gateway_core::engine::client_retry::{ClientRetryKey, ClientRetryStore}; @@ -244,7 +244,9 @@ impl ResponseControlTransport for ControlTransport { #[derive(Default)] struct InterruptProvider { calls: AtomicUsize, + store: Arc, fail_first: bool, + http_only: bool, retry_started: tokio::sync::Notify, retry_release: tokio::sync::Notify, transport: Mutex>>, @@ -284,7 +286,9 @@ impl Provider for InterruptProvider { let id = format!("resp_interrupt_{number}"); let transport = Arc::new(ControlTransport::new(id)); let owner: Arc = transport.clone(); - context.response_control().unwrap().bind(&owner); + if !self.http_only { + context.response_control().unwrap().bind(&owner); + } *self.transport.lock().unwrap() = Some(transport.clone()); let candidate = request.candidate(); let metadata = ProviderCallMetadata::new( @@ -296,7 +300,12 @@ impl Provider for InterruptProvider { "acct_api_test" }) .unwrap(), - UpstreamTransport::new("websocket").unwrap(), + UpstreamTransport::new(if self.http_only { + "http_sse" + } else { + "websocket" + }) + .unwrap(), ); let body = futures::stream::unfold( (0, transport, self), @@ -457,7 +466,7 @@ async fn connect( } else { snapshot("sk_ws_interrupt", "openai") }), - Arc::new(SettlementPorts::default()), + provider.store.clone(), ProviderRegistry::new([provider as Arc]).unwrap(), admissions, Arc::new(UnusedContinuation), @@ -489,6 +498,140 @@ fn interrupt(id: &str, mode: &str) -> Value { json!({"type":"response.interrupt","response_id":id,"mode":mode,"future_extension":true}) } +#[tokio::test] +async fn http_response_rejects_interrupt_without_aborting_or_admitting_queued_requests() { + let provider = Arc::new(InterruptProvider { + http_only: true, + ..Default::default() + }); + let admissions = Arc::new(InterruptAdmissions::default()); + let (mut socket, _server) = connect(provider.clone(), admissions.clone()).await; + let create = json!({"type":"response.create","model":"model-a","input":"hello"}); + send(&mut socket, create.clone()).await; + assert_eq!(next(&mut socket).await["type"], "response.created"); + send(&mut socket, create).await; + send( + &mut socket, + interrupt("resp_interrupt_0", "discard_partial_items"), + ) + .await; + assert_eq!( + next(&mut socket).await, + json!({ + "type":"response.interrupt.failed", "response_id":"resp_interrupt_0" + }) + ); + assert_eq!(provider.calls.load(Ordering::SeqCst), 1); + assert_eq!(admissions.active.load(Ordering::SeqCst), 1); + assert!( + provider + .transport + .lock() + .unwrap() + .as_ref() + .unwrap() + .requests + .lock() + .unwrap() + .is_empty() + ); + + provider.complete.notify_one(); + let completed = next(&mut socket).await; + assert_eq!(completed["type"], "response.completed"); + assert_eq!(completed["response"]["id"], "resp_interrupt_0"); + assert_eq!( + next(&mut socket).await["response"]["id"], + "resp_interrupt_1" + ); + assert_eq!(provider.calls.load(Ordering::SeqCst), 2); + assert_eq!( + provider.store.finalizations.lock().unwrap()[0].outcome, + gateway_core::engine::ExecutionOutcome::Succeeded + ); + provider.complete.notify_one(); + assert_eq!(next(&mut socket).await["type"], "response.completed"); + socket.close(None).await.unwrap(); +} + +#[tokio::test] +async fn http_response_rejects_late_interrupt_during_settlement_and_when_idle() { + let provider = Arc::new(InterruptProvider { + http_only: true, + ..Default::default() + }); + let (release, wait_for_release) = tokio::sync::oneshot::channel(); + let admissions = Arc::new(InterruptAdmissions { + release_gate: Mutex::new(Some(wait_for_release)), + ..Default::default() + }); + let (mut socket, _server) = connect(provider.clone(), admissions.clone()).await; + let create = json!({"type":"response.create","model":"model-a","input":"hello"}); + send(&mut socket, create.clone()).await; + assert_eq!(next(&mut socket).await["type"], "response.created"); + provider.complete.notify_one(); + assert_eq!(next(&mut socket).await["type"], "response.completed"); + tokio::time::timeout( + Duration::from_secs(5), + admissions.release_started.notified(), + ) + .await + .unwrap(); + let control = interrupt("resp_interrupt_0", "discard_partial_items"); + let rejection = json!({"type":"response.interrupt.failed","response_id":"resp_interrupt_0"}); + send(&mut socket, control.clone()).await; + assert_eq!(next(&mut socket).await, rejection); + assert_eq!(admissions.active.load(Ordering::SeqCst), 1); + release.send(()).unwrap(); + tokio::time::timeout(Duration::from_secs(5), async { + while admissions.active.load(Ordering::SeqCst) != 0 { + tokio::task::yield_now().await; + } + }) + .await + .unwrap(); + send(&mut socket, control).await; + assert_eq!(next(&mut socket).await, rejection); + assert_eq!(provider.calls.load(Ordering::SeqCst), 1); + send(&mut socket, create).await; + assert_eq!( + next(&mut socket).await["response"]["id"], + "resp_interrupt_1" + ); + provider.complete.notify_one(); + assert_eq!(next(&mut socket).await["type"], "response.completed"); + socket.close(None).await.unwrap(); +} + +#[tokio::test] +async fn missing_control_channel_keeps_unknown_and_invalid_controls_as_errors() { + let provider = Arc::new(InterruptProvider::default()); + let (mut socket, _server) = + connect(provider.clone(), Arc::new(InterruptAdmissions::default())).await; + for control in [ + interrupt("resp_unknown", "future_mode"), + interrupt("", "discard_partial_items"), + json!({"type":"response.interrupt", "mode":"discard_partial_items"}), + json!({"type":"future.control"}), + ] { + send(&mut socket, control).await; + assert_eq!(next(&mut socket).await["type"], "error"); + } + send( + &mut socket, + interrupt("resp_unknown", "discard_partial_items"), + ) + .await; + assert_eq!( + next(&mut socket).await, + json!({ + "type":"response.interrupt.failed", "response_id":"resp_unknown" + }) + ); + assert_eq!(provider.calls.load(Ordering::SeqCst), 0); + socket.close(None).await.unwrap(); +} + #[tokio::test] async fn late_interrupt_after_terminal_is_sent_upstream_and_keeps_the_connection_reusable() { let provider = Arc::new(InterruptProvider::default()); diff --git a/docs/api.md b/docs/api.md index af5c5c870..a4b3c0936 100644 --- a/docs/api.md +++ b/docs/api.md @@ -482,8 +482,12 @@ Responses WebSocket 接受文本消息,`response.create` 在同一连接串行 跨账号缺少原连接上下文时,仍通过恢复错误要求客户端完整重放 控制帧不重新选号、不创建推理 attempt。响应结束后原连接仍可接收控制帧并返回上游事件; -下一轮执行使用新绑定的控制入口。除上述重试期间保留中断的情况外,尚未建立原连接、原连接已失效、 -实际走 HTTP 或 Provider 未提供控制通道时返回 `400`,不能通过控制帧新建上游连接 +下一轮执行使用新绑定的控制入口。没有绑定上游控制通道时(包括实际走 HTTP),有效的 +`response.interrupt` / `discard_partial_items` 请求返回带原 `response_id` 的 +`response.interrupt.failed`,当前模型响应继续交付真实事件与终态,客户端可收完后再续接。 +该事件只表示网关无法转发中断,不表示上游已停止生成或丢弃部分输出。 +未知控制消息、未知模式或无效响应 ID 字段在无控制通道时仍返回 `400`。 +除上述重试期间保留中断的情况外,已绑定的原连接失效或发送失败仍返回 `400`,不能通过控制帧新建上游连接 中断后继续转发上游事件与终态;只有 `response.incomplete` 的 `incomplete_details.reason` 为 `interrupted` 时,才按官方中断语义保留原连接续接能力。部分输出是否被丢弃以上游终态为准。