starred/codex-proxy-rs
1
0
Fork 0
mirror of https://github.com/zyycn/codex-proxy-rs.git synced 2026-11-06 08:06:19 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(responses): reject interrupts without aborting HTTP fallback (#558)

Browse Source 
Return response.interrupt.failed for supported interrupts when no upstream control channel is bound, so clients can drain the genuine response and continue. Preserve existing errors for unsupported controls and ambiguous sends.

Cover active, settling, and idle HTTP responses and queued request admission. Document the rejection contract while retaining the oversized request protection.

Co-authored-by: Zyy <zyycn@users.noreply.github.com>
Co-authored-by: Codex <noreply@openai.com>
This commit is contained in:
Zyy
2026-10-30 12:17:09 +08:00
committed by GitHub
co-authored by Zyy Codex
parent b3dd9b5337
commit 7e7acad6a3
4 changed files with 195 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/crates/gateway-api/src/openai/responses/websocket/forward.rs
+15
View File
@@ -770,6 +770,21 @@ pub(super) async fn send_control(
payload: &str,
request_id: &Arc<str>,
) -> 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,
backend/crates/gateway-api/src/openai/responses/websocket/interrupt.rs
+27 -7
View File
@@ -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::<Value>(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<String> {
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<Value> {
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)
}
backend/crates/gateway-api/tests/openai/responses/websocket/interrupt.rs
+147 -4
View File
@@ -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<SettlementPorts>,
fail_first: bool,
http_only: bool,
retry_started: tokio::sync::Notify,
retry_release: tokio::sync::Notify,
transport: Mutex<Option<Arc<ControlTransport>>>,
@@ -284,7 +286,9 @@ impl Provider for InterruptProvider {
let id = format!("resp_interrupt_{number}");
let transport = Arc::new(ControlTransport::new(id));
let owner: Arc<dyn ResponseControlTransport> = 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<dyn Provider>]).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());
docs/api.md
+6 -2
View File
@@ -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` 时，才按官方中断语义保留原连接续接能力。部分输出是否被丢弃以上游终态为准。