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

fix: harden responses websocket recovery

Browse Source 
Co-authored-by: Codex <noreply@openai.com>
This commit is contained in:
Zyy
2026-07-10 20:30:40 +08:00
co-authored by Codex
parent f06530278d
commit 61cdaf753e
26 changed files with 786 additions and 478 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/Cargo.lock
Generated
+9 -14
View File
@@ -844,8 +844,8 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "843fba2746e448b37e26a819579957415c8cef339bf08564fe8b7ddbd959573c"
dependencies = [
"crc32fast",
"libz-sys",
"miniz_oxide",
"zlib-rs",
]
[[package]]
@@ -1475,17 +1475,6 @@ dependencies = [
"vcpkg",
]
[[package]]
name = "libz-sys"
version = "1.1.29"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "85bc9657773828b90eeb625adff10eeac83cc21bbfd8e23a03eaa8a33c9e28d9"
dependencies = [
"cc",
"pkg-config",
"vcpkg",
]
[[package]]
name = "linux-raw-sys"
version = "0.12.1"
@@ -2863,7 +2852,7 @@ dependencies = [
[[package]]
name = "tokio-tungstenite"
version = "0.28.0"
source = "git+https://github.com/openai-oss-forks/tokio-tungstenite?rev=132f5b39c862e3a970f731d709608b3e6276d5f6#132f5b39c862e3a970f731d709608b3e6276d5f6"
source = "git+https://github.com/openai-oss-forks/tokio-tungstenite?rev=0e5b2d73aa18dd9f0a50ee9ff199d5aef7594186#0e5b2d73aa18dd9f0a50ee9ff199d5aef7594186"
dependencies = [
"futures-util",
"log",
@@ -3074,7 +3063,7 @@ checksum = "e421abadd41a4225275504ea4d6566923418b7f05506fbc9c0fe86ba7396114b"
[[package]]
name = "tungstenite"
version = "0.27.0"
source = "git+https://github.com/openai-oss-forks/tungstenite-rs?rev=9200079d3b54a1ff51072e24d81fd354f085156f#9200079d3b54a1ff51072e24d81fd354f085156f"
source = "git+https://github.com/openai-oss-forks/tungstenite-rs?rev=4fffad30fe373adbdcffab9545e9e9bf4f2fc19f#4fffad30fe373adbdcffab9545e9e9bf4f2fc19f"
dependencies = [
"bytes",
"data-encoding",
@@ -3790,6 +3779,12 @@ dependencies = [
"syn",
]
[[package]]
name = "zlib-rs"
version = "0.6.6"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "b142a20ec14a91d5bc708c1dc21b080c550113d8aa77afa29635673a65dd02c5"
[[package]]
name = "zmij"
version = "1.0.21"
backend/Cargo.toml
+2 -2
View File
@@ -50,8 +50,8 @@ tempfile = "3.27.0"
wiremock = "0.6.5"
[patch.crates-io]
tokio-tungstenite = { git = "https://github.com/openai-oss-forks/tokio-tungstenite", rev = "132f5b39c862e3a970f731d709608b3e6276d5f6" }
tungstenite = { git = "https://github.com/openai-oss-forks/tungstenite-rs", rev = "9200079d3b54a1ff51072e24d81fd354f085156f" }
tokio-tungstenite = { git = "https://github.com/openai-oss-forks/tokio-tungstenite", rev = "0e5b2d73aa18dd9f0a50ee9ff199d5aef7594186" }
tungstenite = { git = "https://github.com/openai-oss-forks/tungstenite-rs", rev = "4fffad30fe373adbdcffab9545e9e9bf4f2fc19f" }
[lints.rust]
future_incompatible = { level = "warn", priority = -1 }
backend/src/api/client/errors.rs
+7
View File
@@ -18,6 +18,7 @@ const NO_AVAILABLE_RESPONSES_ACCOUNTS_MESSAGE: &str =
const UPSTREAM_CODEX_REQUEST_FAILED_MESSAGE: &str = "Upstream Codex request failed";
const INVALID_UPSTREAM_CODEX_RESPONSE_MESSAGE: &str = "Invalid upstream Codex response";
const UPSTREAM_CODEX_RESPONSE_FAILED_MESSAGE: &str = "Upstream Codex response failed";
const HISTORY_UNAVAILABLE_MESSAGE: &str = "Previous response context is unavailable. Start a new conversation or resend the complete input without previous_response_id.";
#[derive(Clone, Copy)]
pub enum ResponseDispatchMessageStyle {
@@ -73,6 +74,10 @@ const MODEL_NOT_FOUND_ERROR: OpenAiErrorKind = OpenAiErrorKind {
error_type: "invalid_request_error",
code: "model_not_found",
};
const HISTORY_UNAVAILABLE_ERROR: OpenAiErrorKind = OpenAiErrorKind {
error_type: "invalid_request_error",
code: "previous_response_unavailable",
};
const INVALID_UPSTREAM_RESPONSE_ERROR: OpenAiErrorKind = OpenAiErrorKind {
error_type: "server_error",
code: "invalid_upstream_response",
@@ -320,6 +325,7 @@ fn dispatch_failure_message(
INVALID_UPSTREAM_CODEX_RESPONSE_MESSAGE.to_owned()
}
DispatchFailureClass::ResponseFailed => UPSTREAM_CODEX_RESPONSE_FAILED_MESSAGE.to_owned(),
DispatchFailureClass::HistoryUnavailable => HISTORY_UNAVAILABLE_MESSAGE.to_owned(),
_ => UPSTREAM_CODEX_REQUEST_FAILED_MESSAGE.to_owned(),
}
}
@@ -336,6 +342,7 @@ fn dispatch_failure_openai_error_kind(
| DispatchFailureClass::Disabled
| DispatchFailureClass::Banned => AUTHENTICATION_ERROR,
DispatchFailureClass::ModelUnsupported => MODEL_NOT_FOUND_ERROR,
DispatchFailureClass::HistoryUnavailable => HISTORY_UNAVAILABLE_ERROR,
DispatchFailureClass::InvalidSse
| DispatchFailureClass::MissingCompleted
| DispatchFailureClass::EmptyUpstreamResponse => INVALID_UPSTREAM_RESPONSE_ERROR,
backend/src/bootstrap/config.rs
+1 -1
View File
@@ -168,7 +168,7 @@ pub struct WebSocketPoolConfig {
pub max_age_ms: u64,
/// 单账号最大连接数。
pub max_per_account: usize,
/// 首个内容帧（首 token）到达前的绝对超时；`0` 表示禁用。
/// 首个真实输出（首 token）到达前的绝对超时；`0` 表示禁用。
///
/// 从发出 `response.create` 起算，覆盖建连/发送/上游排队/思考的全程，直到收到
/// 首个真实内容帧（`response.created`/`response.in_progress` 等纯生命周期帧不计）。
backend/src/dispatch/errors.rs
+2
View File
@@ -31,6 +31,7 @@ pub(crate) enum DispatchFailureClass {
CloudflareChallenge,
CloudflarePathBlocked,
ModelUnsupported,
HistoryUnavailable,
Upstream,
InvalidSse,
MissingCompleted,
@@ -50,6 +51,7 @@ impl DispatchFailureClass {
Self::CloudflareChallenge => "cloudflare_challenge",
Self::CloudflarePathBlocked => "cloudflare_path_blocked",
Self::ModelUnsupported => "model_unsupported",
Self::HistoryUnavailable => "history_unavailable",
Self::Upstream => "upstream",
Self::InvalidSse => "invalid_sse",
Self::MissingCompleted => "missing_completed",
backend/src/dispatch/responses/errors.rs
+10 -1
View File
@@ -63,6 +63,8 @@ pub enum ResponseDispatchError {
count: usize,
upstream_error: String,
},
#[error("previous response context is unavailable: {upstream_error}")]
HistoryUnavailable { upstream_error: String },
#[error("upstream request failed: {0}")]
Upstream(#[from] CodexClientError),
#[error("invalid upstream SSE response: {0}")]
@@ -127,7 +129,7 @@ impl ResponseDispatchError {
| Self::MissingCompleted
| Self::EmptyUpstreamResponse => 502,
Self::Failed(failure) => stream_failure_http_status(failure),
Self::ModelUnsupported { .. } => 400,
Self::ModelUnsupported { .. } | Self::HistoryUnavailable { .. } => 400,
Self::Upstream(error) => upstream_error_http_status(error),
}
}
@@ -186,6 +188,13 @@ impl ResponseDispatchError {
*count,
upstream_error,
),
Self::HistoryUnavailable { upstream_error } => DispatchErrorMetadata {
failure_class: DispatchFailureClass::HistoryUnavailable,
exhausted_count: None,
upstream_error: Some(upstream_error.clone()),
upstream_status: Some(400),
diagnostics: None,
},
Self::Upstream(error) => DispatchErrorMetadata::upstream(error),
Self::InvalidSse(_) => DispatchErrorMetadata::simple(DispatchFailureClass::InvalidSse),
Self::MissingCompleted => {
backend/src/dispatch/responses/live_stream.rs
+3 -3
View File
@@ -21,7 +21,7 @@ use crate::{
protocol::{
responses::{
reconvert_responses_sse_event_tuple_values, response_sse_event_is_terminal,
update_first_response_event_ms,
update_first_response_output_ms,
},
sse::{
encode_sse_event, parse_sse_events, sse_body_has_done, sse_frame_end,
@@ -78,7 +78,7 @@ pub(super) fn spawn_live_response_stream(
context.account_pool.release(&context.account_id).await;
return;
}
update_first_response_event_ms(context.started_at, &body_bytes, &mut first_token_ms);
update_first_response_output_ms(context.started_at, &body_bytes, &mut first_token_ms);
loop {
let next = tokio::select! {
@@ -104,7 +104,7 @@ pub(super) fn spawn_live_response_stream(
context.account_pool.release(&context.account_id).await;
return;
}
update_first_response_event_ms(
update_first_response_output_ms(
context.started_at,
&body_bytes,
&mut first_token_ms,
backend/src/dispatch/responses/service.rs
+27 -15
View File
@@ -331,15 +331,24 @@ impl ResponseDispatchService {
.await
}
async fn recover_request_history(
async fn try_recover_implicit_resume(
&self,
request: &mut CodexResponsesRequest,
implicit_resume: &mut Option<ImplicitResumeSnapshot>,
) {
account_id: &str,
evict_reasoning_replay: bool,
) -> bool {
let Some(snapshot) = implicit_resume.take() else {
return false;
};
if evict_reasoning_replay {
self.evict_reasoning_replay(request, account_id).await;
}
if let Some(previous_response_id) = request.previous_response_id() {
self.session_affinity.forget(previous_response_id).await;
}
restore_request_without_history(request, implicit_resume);
restore_full_request_without_history(request, snapshot);
true
}
fn strip_history_if_account_changed(
@@ -354,7 +363,10 @@ impl ResponseDispatchService {
return;
}
restore_request_without_history(request, implicit_resume);
let Some(snapshot) = implicit_resume.take() else {
return;
};
restore_full_request_without_history(request, snapshot);
}
async fn apply_cascading_ban_defense(
@@ -363,7 +375,6 @@ impl ResponseDispatchService {
implicit_resume: &mut Option<ImplicitResumeSnapshot>,
preferred_account_id: Option<&str>,
acquired_account_id: &str,
explicit_previous_response_id: Option<&str>,
) -> bool {
let Some(preferred_account_id) =
preferred_account_id.filter(|account_id| *account_id != acquired_account_id)
@@ -392,14 +403,17 @@ impl ResponseDispatchService {
return false;
}
let response_id_to_forget = explicit_previous_response_id
.or(request.previous_response_id())
.map(str::to_string);
restore_request_without_history(request, implicit_resume);
let response_id_to_forget = request.previous_response_id().map(str::to_string);
let restored_full_request = if let Some(snapshot) = implicit_resume.take() {
restore_full_request_without_history(request, snapshot);
true
} else {
false
};
if let Some(response_id) = response_id_to_forget {
self.session_affinity.forget(&response_id).await;
}
true
restored_full_request
}
async fn evict_reasoning_replay(&self, request: &CodexResponsesRequest, account_id: &str) {
@@ -436,12 +450,10 @@ fn strip_request_history(request: &mut CodexResponsesRequest) {
request.turn_state = None;
}
fn restore_request_without_history(
fn restore_full_request_without_history(
request: &mut CodexResponsesRequest,
implicit_resume: &mut Option<ImplicitResumeSnapshot>,
snapshot: ImplicitResumeSnapshot,
) {
if let Some(snapshot) = implicit_resume.take() {
snapshot.restore(request);
}
snapshot.restore(request);
strip_request_history(request);
}
backend/src/dispatch/responses/service/complete.rs
+36 -24
View File
@@ -16,7 +16,6 @@ impl ResponseDispatchService {
let tuple_schema = request.tuple_schema.clone();
let image_generation_requested = request.expects_image_generation();
let now = Utc::now();
let explicit_previous_response_id = request.previous_response_id().map(ToString::to_string);
let mut implicit_resume = self.prepare_response_session(&mut request).await;
let account_affinity = self.account_affinity_for_request(&request, now).await;
let mut acquire_request = AccountAcquireRequest::new(request.model(), now);
@@ -25,7 +24,6 @@ impl ResponseDispatchService {
}
let mut excluded_account_ids = Vec::new();
let mut exhausted_accounts = AccountExhaustionTracker::default();
let mut history_recovery_used = false;
let mut next_required_account_id: Option<String> = None;
let mut empty_response_retries = 0u8;
let mut quota_verify_attempts = 0usize;
@@ -135,7 +133,6 @@ impl ResponseDispatchService {
&mut implicit_resume,
account_affinity.preferred_account_id(),
&acquired.account.id,
explicit_previous_response_id.as_deref(),
)
.await;
Self::strip_history_if_account_changed(
@@ -216,14 +213,16 @@ impl ResponseDispatchService {
}
}
if let CollectedResponse::Failed(failure) = &collected_response {
if is_history_recovery_sse_failure(failure) && !history_recovery_used {
if sse_failure_invalid_reasoning_replay(failure) {
self.evict_reasoning_replay(&request, &release_account_id)
.await;
}
self.recover_request_history(&mut request, &mut implicit_resume)
.await;
history_recovery_used = true;
if is_history_recovery_sse_failure(failure)
&& self
.try_recover_implicit_resume(
&mut request,
&mut implicit_resume,
&release_account_id,
sse_failure_invalid_reasoning_replay(failure),
)
.await
{
next_required_account_id = Some(release_account_id);
continue;
}
@@ -308,18 +307,6 @@ impl ResponseDispatchService {
.await;
excluded_account_ids.push(release_account_id);
}
Err(error)
if is_history_recovery_upstream_error(&error) && !history_recovery_used =>
{
if client_error_invalid_reasoning_replay(&error) {
self.evict_reasoning_replay(&request, &release_account_id)
.await;
}
self.recover_request_history(&mut request, &mut implicit_resume)
.await;
history_recovery_used = true;
next_required_account_id = Some(release_account_id);
}
Err(error) if is_auth_upstream_error(&error) => {
let upstream_error = upstream_error_body(&error);
let account_status = auth_failure_account_status(&error);
@@ -396,6 +383,20 @@ impl ResponseDispatchService {
excluded_account_ids.push(release_account_id);
}
Err(error) => {
let history_unavailable = is_history_recovery_upstream_error(&error);
if history_unavailable
&& self
.try_recover_implicit_resume(
&mut request,
&mut implicit_resume,
&release_account_id,
client_error_invalid_reasoning_replay(&error),
)
.await
{
next_required_account_id = Some(release_account_id);
continue;
}
record_response_upstream_error_event(ResponseUpstreamErrorEventRecord {
recorder: &self.recorder,
request_id,
@@ -412,6 +413,11 @@ impl ResponseDispatchService {
attempt: Some(&attempt),
})
.await;
if history_unavailable {
return Err(ResponseDispatchError::HistoryUnavailable {
upstream_error: upstream_error_body(&error),
});
}
return Err(ResponseDispatchError::Upstream(error));
}
}
@@ -477,7 +483,13 @@ impl ResponseDispatchService {
Ok(body)
}
CollectedResponse::Failed(failure) => {
let error = ResponseDispatchError::Failed(failure);
let error = if is_history_recovery_sse_failure(&failure) {
ResponseDispatchError::HistoryUnavailable {
upstream_error: sse_failure_error_body(&failure),
}
} else {
ResponseDispatchError::Failed(failure)
};
self.record_response_dispatch_error(
request_id,
route,
backend/src/dispatch/responses/service/stream.rs
+57 -40
View File
@@ -16,7 +16,6 @@ impl ResponseDispatchService {
request.set_stream(true);
let tuple_schema = request.tuple_schema.clone();
let now = Utc::now();
let explicit_previous_response_id = request.previous_response_id().map(ToString::to_string);
let mut implicit_resume = self.prepare_response_session(&mut request).await;
let account_affinity = self.account_affinity_for_request(&request, now).await;
let mut acquire_request = AccountAcquireRequest::new(request.model(), now);
@@ -25,7 +24,6 @@ impl ResponseDispatchService {
}
let mut excluded_account_ids = Vec::new();
let mut exhausted_accounts = AccountExhaustionTracker::default();
let mut history_recovery_used = false;
let mut next_required_account_id: Option<String> = None;
let mut quota_verify_attempts = 0usize;
let mut trace = ResponseDispatchTrace::default();
@@ -140,7 +138,6 @@ impl ResponseDispatchService {
&mut implicit_resume,
account_affinity.preferred_account_id(),
&acquired.account.id,
explicit_previous_response_id.as_deref(),
)
.await;
Self::strip_history_if_account_changed(
@@ -190,21 +187,6 @@ impl ResponseDispatchService {
.await;
let (prefetched, body) = match prefetch_first_sse_chunk(response.body).await {
Ok(prefetched) => prefetched,
Err(ResponseDispatchError::Upstream(error))
if is_history_recovery_upstream_error(&error)
&& !history_recovery_used =>
{
self.account_pool.release(&release_account_id).await;
if client_error_invalid_reasoning_replay(&error) {
self.evict_reasoning_replay(&request, &release_account_id)
.await;
}
self.recover_request_history(&mut request, &mut implicit_resume)
.await;
history_recovery_used = true;
next_required_account_id = Some(release_account_id);
continue;
}
Err(ResponseDispatchError::Upstream(error))
if is_rate_limit_upstream_error(&error) =>
{
@@ -320,6 +302,21 @@ impl ResponseDispatchService {
Err(error) => {
self.account_pool.release(&release_account_id).await;
if let ResponseDispatchError::Upstream(upstream_error) = &error {
let history_unavailable =
is_history_recovery_upstream_error(upstream_error);
if history_unavailable
&& self
.try_recover_implicit_resume(
&mut request,
&mut implicit_resume,
&release_account_id,
client_error_invalid_reasoning_replay(upstream_error),
)
.await
{
next_required_account_id = Some(release_account_id);
continue;
}
record_response_upstream_error_event(
ResponseUpstreamErrorEventRecord {
recorder: &self.recorder,
@@ -338,6 +335,11 @@ impl ResponseDispatchService {
},
)
.await;
if history_unavailable {
return Err(ResponseDispatchError::HistoryUnavailable {
upstream_error: upstream_error_body(upstream_error),
});
}
return Err(error);
}
return_stream_dispatch_error!(
@@ -359,15 +361,17 @@ impl ResponseDispatchService {
}
};
if let Some(failure) = first_failure {
if is_history_recovery_sse_failure(&failure) && !history_recovery_used {
if is_history_recovery_sse_failure(&failure)
&& self
.try_recover_implicit_resume(
&mut request,
&mut implicit_resume,
&release_account_id,
sse_failure_invalid_reasoning_replay(&failure),
)
.await
{
self.account_pool.release(&release_account_id).await;
if sse_failure_invalid_reasoning_replay(&failure) {
self.evict_reasoning_replay(&request, &release_account_id)
.await;
}
self.recover_request_history(&mut request, &mut implicit_resume)
.await;
history_recovery_used = true;
next_required_account_id = Some(release_account_id);
continue;
}
@@ -440,7 +444,14 @@ impl ResponseDispatchService {
},
)
.await;
return Err(ResponseDispatchError::Failed(failure.clone()));
let error = if is_history_recovery_sse_failure(&failure) {
ResponseDispatchError::HistoryUnavailable {
upstream_error: sse_failure_error_body(&failure),
}
} else {
ResponseDispatchError::Failed(failure.clone())
};
return Err(error);
}
let context = LiveResponseStreamContext {
@@ -495,19 +506,6 @@ impl ResponseDispatchService {
.await;
excluded_account_ids.push(release_account_id);
}
Err(error)
if is_history_recovery_upstream_error(&error) && !history_recovery_used =>
{
self.account_pool.release(&release_account_id).await;
if client_error_invalid_reasoning_replay(&error) {
self.evict_reasoning_replay(&request, &release_account_id)
.await;
}
self.recover_request_history(&mut request, &mut implicit_resume)
.await;
history_recovery_used = true;
next_required_account_id = Some(release_account_id);
}
Err(error) if is_auth_upstream_error(&error) => {
self.account_pool.release(&release_account_id).await;
let upstream_error = upstream_error_body(&error);
@@ -578,6 +576,20 @@ impl ResponseDispatchService {
}
Err(error) => {
self.account_pool.release(&release_account_id).await;
let history_unavailable = is_history_recovery_upstream_error(&error);
if history_unavailable
&& self
.try_recover_implicit_resume(
&mut request,
&mut implicit_resume,
&release_account_id,
client_error_invalid_reasoning_replay(&error),
)
.await
{
next_required_account_id = Some(release_account_id);
continue;
}
record_response_upstream_error_event(ResponseUpstreamErrorEventRecord {
recorder: &self.recorder,
request_id,
@@ -594,6 +606,11 @@ impl ResponseDispatchService {
attempt: Some(&attempt),
})
.await;
if history_unavailable {
return Err(ResponseDispatchError::HistoryUnavailable {
upstream_error: upstream_error_body(&error),
});
}
return Err(ResponseDispatchError::Upstream(error));
}
}
backend/src/upstream/openai/protocol/responses.rs
+4 -4
View File
@@ -104,8 +104,8 @@ pub fn http_sse_fallback_allowed(request: &CodexResponsesRequest) -> bool {
)
}
/// 判断已收到的 Responses SSE 内容是否包含首个完整的有效数据事件。
pub fn response_body_has_first_event(body_bytes: &[u8]) -> bool {
/// 判断已收到的 Responses SSE 内容是否包含首个完整的真实输出事件。
pub fn response_body_has_first_output(body_bytes: &[u8]) -> bool {
let body = String::from_utf8_lossy(body_bytes);
let Some(complete_body) = complete_sse_body_prefix(&body) else {
return false;
@@ -115,12 +115,12 @@ pub fn response_body_has_first_event(body_bytes: &[u8]) -> bool {
}
/// 已收到首个有效 Responses SSE 输出事件时记录首 token 耗时。
pub fn update_first_response_event_ms(
pub fn update_first_response_output_ms(
started_at: Instant,
body_bytes: &[u8],
first_token_ms: &mut Option<i64>,
) {
if first_token_ms.is_none() && response_body_has_first_event(body_bytes) {
if first_token_ms.is_none() && response_body_has_first_output(body_bytes) {
*first_token_ms = Some(elapsed_millis_i64(started_at).max(1));
}
}
backend/src/upstream/openai/protocol/websocket.rs
-11
View File
@@ -224,17 +224,6 @@ pub fn is_terminal_websocket_event(event: &str) -> bool {
matches!(event, "response.completed" | "response.failed" | "error")
}
/// 判断是否为「首个内容帧之前的纯生命周期帧」。
///
/// `response.created` / `response.in_progress` 仅表示上游已受理请求，尚未开始产出
/// 真实内容（含 reasoning）。首 token 熔断计时不应被它们解除，否则连接一收到
/// `response.created`（通常 1s 内）就被误判为「已开始产出」，使真正的首内容帧
/// 可以无限拖延而不触发保护。其余任何帧（reasoning/delta/output_item 等）到达即视为
/// 上游已开始真实产出，解除熔断——对慢思考模型友好。
pub fn is_pre_content_lifecycle_event(event: &str) -> bool {
matches!(event, "response.created" | "response.in_progress")
}
/// 将 WebSocket 错误帧映射为上游 HTTP 状态分类。
pub fn classify_websocket_error_frame(raw: &str) -> Option<ClassifiedWebSocketError> {
let value = serde_json::from_str::<Value>(raw).ok()?;
backend/src/upstream/openai/transport/response_meta.rs
+2 -2
View File
@@ -4,7 +4,7 @@ use std::time::Instant;
use reqwest::header::{HeaderMap, SET_COOKIE};
use crate::upstream::openai::protocol::responses::update_first_response_event_ms;
use crate::upstream::openai::protocol::responses::update_first_response_output_ms;
use super::diagnostics::CodexUpstreamDiagnostics;
@@ -48,7 +48,7 @@ pub(super) fn update_first_token_ms(
body_bytes: &[u8],
first_token_ms: &mut Option<i64>,
) {
update_first_response_event_ms(started_at, body_bytes, first_token_ms);
update_first_response_output_ms(started_at, body_bytes, first_token_ms);
}
fn is_rate_limit_header(name: &str) -> bool {
backend/src/upstream/openai/transport/websocket.rs
+12 -10
View File
@@ -27,10 +27,12 @@ use tungstenite::{
use crate::infra::time::china_filename_timestamp_millis;
use crate::upstream::openai::protocol::events::{self, TokenUsage};
use crate::upstream::openai::protocol::responses::CodexResponsesRequest;
use crate::upstream::openai::protocol::responses::{
response_body_has_first_output, CodexResponsesRequest,
};
use crate::upstream::openai::protocol::sse::SseError;
use crate::upstream::openai::protocol::websocket::{
classify_websocket_error_frame, is_pre_content_lifecycle_event, is_terminal_websocket_event,
classify_websocket_error_frame, is_terminal_websocket_event,
retry_after_seconds_from_wrapped_error_headers, websocket_event_to_sse_frame,
websocket_event_type, websocket_incomplete_response_reason, websocket_metadata_turn_state,
websocket_response_completed_parse_error, websocket_response_create_payload_text,
@@ -66,7 +68,7 @@ mod types;
pub use stream::execute_response_create_request;
use stream::{
audit_header_value, collect_websocket_response, is_first_token_timeout,
prefetch_first_content_stream_frames, reusable_websocket_metadata,
prefetch_stream_frames_until_output_or_terminal, reusable_websocket_metadata,
reused_stream_prefetch_error, stream_websocket_response, websocket_audit_file_name,
websocket_connection_metadata, WebSocketStreamPoolReturn,
};
@@ -317,7 +319,7 @@ async fn execute_fresh_response_create_request_stream(
.await?;
let mut metadata = websocket_connection_metadata(&response);
let prefetched_frames = match prefetch_first_content_stream_frames(
let prefetched_frames = match prefetch_stream_frames_until_output_or_terminal(
&mut websocket,
&mut metadata,
first_token_started_at,
@@ -399,7 +401,7 @@ pub(crate) async fn execute_response_create_request_with_pool(
execute_pooled_response_create_request(request, pool, key, *connection, started_at)
.await;
match result {
Err(CodexWebSocketExchangeError::ReusedConnectionDiedBeforeFirstFrame {
Err(CodexWebSocketExchangeError::ReusedConnectionDiedBeforeFirstOutput {
..
})
| Err(CodexWebSocketExchangeError::FirstTokenTimeout { .. }) => {
@@ -489,7 +491,7 @@ pub(crate) async fn execute_response_create_request_stream_with_pool(
)
.await;
match result {
Err(CodexWebSocketExchangeError::ReusedConnectionDiedBeforeFirstFrame {
Err(CodexWebSocketExchangeError::ReusedConnectionDiedBeforeFirstOutput {
..
})
| Err(CodexWebSocketExchangeError::FirstTokenTimeout { .. }) => {
@@ -601,7 +603,7 @@ async fn execute_fresh_pooled_response_create_request_stream(
let now = Instant::now();
let mut metadata = websocket_connection_metadata(&response);
let prefetched_frames = match prefetch_first_content_stream_frames(
let prefetched_frames = match prefetch_stream_frames_until_output_or_terminal(
&mut websocket,
&mut metadata,
first_token_started_at,
@@ -642,7 +644,7 @@ async fn execute_pooled_response_create_request(
{
pool.discard(&key).await;
return Err(
CodexWebSocketExchangeError::ReusedConnectionDiedBeforeFirstFrame {
CodexWebSocketExchangeError::ReusedConnectionDiedBeforeFirstOutput {
message: error.to_string(),
},
);
@@ -694,13 +696,13 @@ async fn execute_pooled_response_create_request_stream(
{
pool.discard(&key).await;
return Err(
CodexWebSocketExchangeError::ReusedConnectionDiedBeforeFirstFrame {
CodexWebSocketExchangeError::ReusedConnectionDiedBeforeFirstOutput {
message: error.to_string(),
},
);
}
let prefetched_frames = match prefetch_first_content_stream_frames(
let prefetched_frames = match prefetch_stream_frames_until_output_or_terminal(
&mut websocket,
&mut metadata,
first_token_started_at,
backend/src/upstream/openai/transport/websocket/stream.rs
+116 -43
View File
@@ -27,25 +27,25 @@ pub(super) async fn collect_websocket_response(
CodexWebSocketExchangeError,
> {
let mut body = String::new();
let mut saw_content_frame = false;
let mut saw_first_output = false;
let mut first_token_ms = None;
loop {
let receive_timeout = receive_idle_timeout_with_first_token_deadline(
saw_content_frame,
saw_first_output,
first_token_started_at,
first_token_timeout,
)?;
let message = match next_websocket_message(&mut websocket, receive_timeout).await {
Ok(message) => message,
Err(error) if !saw_content_frame => {
Err(error) if !saw_first_output => {
if let Some(timeout) =
elapsed_first_token_timeout(first_token_started_at, first_token_timeout)
{
return Err(CodexWebSocketExchangeError::FirstTokenTimeout { timeout });
}
if reused_connection {
return Err(reused_connection_died_before_first_frame(&error));
return Err(reused_connection_died_before_first_output(&error));
}
return Err(error);
}
@@ -57,9 +57,9 @@ pub(super) async fn collect_websocket_response(
let text = match message {
Message::Text(text) => text,
Message::Binary(_) => return Err(CodexWebSocketExchangeError::UnexpectedBinaryEvent),
Message::Close(_) if reused_connection && !saw_content_frame => {
Message::Close(_) if reused_connection && !saw_first_output => {
return Err(
CodexWebSocketExchangeError::ReusedConnectionDiedBeforeFirstFrame {
CodexWebSocketExchangeError::ReusedConnectionDiedBeforeFirstOutput {
message: "websocket closed".to_string(),
},
);
@@ -95,9 +95,10 @@ pub(super) async fn collect_websocket_response(
}
let event = websocket_event_type(&raw);
let forwarded = if let Some(frame) = websocket_event_to_sse_frame(&raw) {
let has_first_output = response_body_has_first_output(frame.as_bytes());
body.push_str(&frame);
if event_is_first_content(event.as_deref()) {
saw_content_frame = true;
if has_first_output {
saw_first_output = true;
response_meta::update_first_token_ms(
started_at,
body.as_bytes(),
@@ -124,9 +125,9 @@ pub(super) async fn collect_websocket_response(
}
}
if reused_connection && !saw_content_frame {
if reused_connection && !saw_first_output {
return Err(
CodexWebSocketExchangeError::ReusedConnectionDiedBeforeFirstFrame {
CodexWebSocketExchangeError::ReusedConnectionDiedBeforeFirstOutput {
message: "websocket closed before terminal event".to_string(),
},
);
@@ -135,7 +136,7 @@ pub(super) async fn collect_websocket_response(
Err(CodexWebSocketExchangeError::ClosedBeforeTerminal)
}
pub(super) async fn prefetch_first_content_stream_frames(
pub(super) async fn prefetch_stream_frames_until_output_or_terminal(
websocket: &mut PumpedWebSocket,
metadata: &mut CodexWebSocketConnectionMetadata,
first_token_started_at: Instant,
@@ -144,7 +145,7 @@ pub(super) async fn prefetch_first_content_stream_frames(
let mut prefetched_frames = Vec::new();
loop {
let receive_timeout =
receive_timeout_before_first_content(first_token_started_at, first_token_timeout)?;
receive_timeout_before_first_output(first_token_started_at, first_token_timeout)?;
let message = match next_websocket_message(websocket, receive_timeout).await {
Ok(Some(message)) => message,
Ok(None) => return Err(CodexWebSocketExchangeError::ClosedBeforeTerminal),
@@ -192,12 +193,13 @@ pub(super) async fn prefetch_first_content_stream_frames(
}
let event = websocket_event_type(&raw);
if websocket_event_to_sse_frame(&raw).is_none() {
let Some(frame) = websocket_event_to_sse_frame(&raw) else {
continue;
}
let is_first_content = event_is_first_content(event.as_deref());
};
let has_first_output = response_body_has_first_output(frame.as_bytes());
let terminal = event.as_deref().is_some_and(is_terminal_websocket_event);
prefetched_frames.push(raw);
if is_first_content {
if has_first_output || terminal {
return Ok(prefetched_frames);
}
}
@@ -210,16 +212,16 @@ pub(super) fn reused_stream_prefetch_error(
CodexWebSocketExchangeError::ClosedBeforeTerminal
| CodexWebSocketExchangeError::ReceiveIdleTimeout { .. }
| CodexWebSocketExchangeError::Transport(_) => {
reused_connection_died_before_first_frame(&error)
reused_connection_died_before_first_output(&error)
}
error => error,
}
}
fn reused_connection_died_before_first_frame(
fn reused_connection_died_before_first_output(
error: &CodexWebSocketExchangeError,
) -> CodexWebSocketExchangeError {
CodexWebSocketExchangeError::ReusedConnectionDiedBeforeFirstFrame {
CodexWebSocketExchangeError::ReusedConnectionDiedBeforeFirstOutput {
message: error.to_string(),
}
}
@@ -255,6 +257,33 @@ pub(super) struct WebSocketStreamPoolReturn {
pub(super) created_at: Instant,
}
#[derive(Debug, Clone, Copy)]
enum StreamWebSocketDiscardReason {
ClientCancelled,
DownstreamSendFailed,
IncompleteResponse,
InvalidCompletedResponse,
UnexpectedBinaryEvent,
UpstreamClosed,
UpstreamError,
UpstreamReceiveFailed,
}
impl StreamWebSocketDiscardReason {
fn as_str(self) -> &'static str {
match self {
Self::ClientCancelled => "client_cancelled",
Self::DownstreamSendFailed => "downstream_send_failed",
Self::IncompleteResponse => "incomplete_response",
Self::InvalidCompletedResponse => "invalid_completed_response",
Self::UnexpectedBinaryEvent => "unexpected_binary_event",
Self::UpstreamClosed => "upstream_closed",
Self::UpstreamError => "upstream_error",
Self::UpstreamReceiveFailed => "upstream_receive_failed",
}
}
}
pub(super) fn stream_websocket_response(
websocket: PumpedWebSocket,
metadata: CodexWebSocketConnectionMetadata,
@@ -307,7 +336,7 @@ async fn forward_websocket_response_stream(
) {
let mut pool_return = pool_return;
let mut prefetched_frames: VecDeque<String> = prefetched_frames.into();
let mut saw_content_frame = false;
let mut saw_first_output = false;
loop {
let raw = if let Some(text) = prefetched_frames.pop_front() {
text
@@ -317,18 +346,27 @@ async fn forward_websocket_response_stream(
// 客户端断开下游 SSE 流：立即丢弃连接并释放池 slot，
// 不再傻等上游 idle 超时（否则同会话后续请求会一直 bypass/busy）。
() = tx.closed() => {
discard_stream_websocket(websocket, pool_return).await;
discard_stream_websocket(
websocket,
pool_return,
StreamWebSocketDiscardReason::ClientCancelled,
).await;
return;
}
message = next_websocket_message(
&mut websocket,
receive_idle_timeout(saw_content_frame),
receive_idle_timeout(saw_first_output),
) => message,
};
let message = match message {
Ok(message) => message,
Err(error) => {
discard_stream_websocket(websocket, pool_return).await;
discard_stream_websocket(
websocket,
pool_return,
StreamWebSocketDiscardReason::UpstreamReceiveFailed,
)
.await;
let _ = tx.send(Err(error)).await;
return;
}
@@ -339,14 +377,24 @@ async fn forward_websocket_response_stream(
match message {
Message::Text(text) => text.to_string(),
Message::Binary(_) => {
discard_stream_websocket(websocket, pool_return).await;
discard_stream_websocket(
websocket,
pool_return,
StreamWebSocketDiscardReason::UnexpectedBinaryEvent,
)
.await;
let _ = tx
.send(Err(CodexWebSocketExchangeError::UnexpectedBinaryEvent))
.await;
return;
}
Message::Close(_) => {
discard_stream_websocket(websocket, pool_return).await;
discard_stream_websocket(
websocket,
pool_return,
StreamWebSocketDiscardReason::UpstreamClosed,
)
.await;
let _ = tx
.send(Err(CodexWebSocketExchangeError::ClosedBeforeTerminal))
.await;
@@ -356,7 +404,12 @@ async fn forward_websocket_response_stream(
}
};
if let Some(classified) = classify_websocket_error_frame(&raw) {
discard_stream_websocket(websocket, pool_return).await;
discard_stream_websocket(
websocket,
pool_return,
StreamWebSocketDiscardReason::UpstreamError,
)
.await;
let retry_after_seconds = retry_after_seconds_from_wrapped_error_headers(&raw)
.or_else(|| events::retry_after_seconds_from_body(&raw));
let _ = tx
@@ -371,7 +424,12 @@ async fn forward_websocket_response_stream(
return;
}
if let Some(reason) = websocket_incomplete_response_reason(&raw) {
discard_stream_websocket(websocket, pool_return).await;
discard_stream_websocket(
websocket,
pool_return,
StreamWebSocketDiscardReason::IncompleteResponse,
)
.await;
let _ = tx
.send(Err(CodexWebSocketExchangeError::IncompleteResponse {
reason,
@@ -380,7 +438,12 @@ async fn forward_websocket_response_stream(
return;
}
if let Some(message) = websocket_response_completed_parse_error(&raw) {
discard_stream_websocket(websocket, pool_return).await;
discard_stream_websocket(
websocket,
pool_return,
StreamWebSocketDiscardReason::InvalidCompletedResponse,
)
.await;
let _ = tx
.send(Err(CodexWebSocketExchangeError::InvalidCompletedResponse {
message,
@@ -403,12 +466,18 @@ async fn forward_websocket_response_stream(
continue;
};
let terminal = event.as_deref().is_some_and(is_terminal_websocket_event);
let has_first_output = response_body_has_first_output(frame.as_bytes());
if tx.send(Ok(Bytes::from(frame))).await.is_err() {
discard_stream_websocket(websocket, pool_return).await;
discard_stream_websocket(
websocket,
pool_return,
StreamWebSocketDiscardReason::DownstreamSendFailed,
)
.await;
return;
}
if event_is_first_content(event.as_deref()) {
saw_content_frame = true;
if has_first_output {
saw_first_output = true;
}
if terminal {
finish_stream_websocket(websocket, metadata, pool_return.take()).await;
@@ -416,7 +485,12 @@ async fn forward_websocket_response_stream(
}
}
discard_stream_websocket(websocket, pool_return).await;
discard_stream_websocket(
websocket,
pool_return,
StreamWebSocketDiscardReason::UpstreamClosed,
)
.await;
let _ = tx
.send(Err(CodexWebSocketExchangeError::ClosedBeforeTerminal))
.await;
@@ -431,17 +505,17 @@ fn receive_idle_timeout(saw_response_frame: bool) -> Duration {
}
fn receive_idle_timeout_with_first_token_deadline(
saw_content_frame: bool,
saw_first_output: bool,
first_token_started_at: Instant,
first_token_timeout: Option<Duration>,
) -> Result<Duration, CodexWebSocketExchangeError> {
if saw_content_frame {
if saw_first_output {
return Ok(WEBSOCKET_ACTIVE_STREAM_IDLE_TIMEOUT);
}
receive_timeout_before_first_content(first_token_started_at, first_token_timeout)
receive_timeout_before_first_output(first_token_started_at, first_token_timeout)
}
fn receive_timeout_before_first_content(
fn receive_timeout_before_first_output(
first_token_started_at: Instant,
first_token_timeout: Option<Duration>,
) -> Result<Duration, CodexWebSocketExchangeError> {
@@ -469,13 +543,6 @@ fn elapsed_first_token_timeout(
(first_token_started_at.elapsed() >= first_token_timeout).then_some(first_token_timeout)
}
fn event_is_first_content(event: Option<&str>) -> bool {
match event {
Some(event) => !is_pre_content_lifecycle_event(event),
None => true,
}
}
pub(super) fn is_first_token_timeout(error: &CodexWebSocketExchangeError) -> bool {
matches!(error, CodexWebSocketExchangeError::FirstTokenTimeout { .. })
}
@@ -517,7 +584,13 @@ async fn finish_stream_websocket(
async fn discard_stream_websocket(
mut websocket: PumpedWebSocket,
pool_return: Option<WebSocketStreamPoolReturn>,
reason: StreamWebSocketDiscardReason,
) {
tracing::info!(
reason = reason.as_str(),
pooled = pool_return.is_some(),
"discarding Responses WebSocket stream"
);
if let Some(pool_return) = pool_return {
pool_return.pool.discard(&pool_return.key).await;
}
backend/src/upstream/openai/transport/websocket/types.rs
+5 -5
View File
@@ -131,14 +131,14 @@ pub enum CodexWebSocketExchangeError {
/// 上游返回非文本事件帧。
#[error("unexpected binary websocket event")]
UnexpectedBinaryEvent,
/// 复用的池连接在首个响应帧前失效。
#[error("reused websocket connection died before first response frame: {message}")]
ReusedConnectionDiedBeforeFirstFrame {
/// 复用的池连接在首个真实输出前失效。
#[error("reused websocket connection died before first output: {message}")]
ReusedConnectionDiedBeforeFirstOutput {
/// 底层失效原因。
message: String,
},
/// 首个内容帧在配置的绝对超时内未到达（连接落到病态上游后端）。
#[error("websocket first content frame not received within {timeout:?}")]
/// 首个真实输出在配置的绝对超时内未到达（连接落到病态上游后端）。
#[error("websocket first output not received within {timeout:?}")]
FirstTokenTimeout {
/// 首 token 超时时长。
timeout: Duration,
backend/src/upstream/openai/transport/websocket_pool.rs
+2 -2
View File
@@ -88,7 +88,7 @@ pub struct CodexWebSocketPoolConfig {
pub ping_timeout: Duration,
/// idle socket 无活动多久后视为失活。
pub liveness_timeout: Option<Duration>,
/// 首个内容帧到达前的绝对超时；`None` 表示禁用首 token 熔断。
/// 首个真实输出到达前的绝对超时；`None` 表示禁用首 token 熔断。
pub first_token_timeout: Option<Duration>,
}
@@ -147,7 +147,7 @@ impl CodexWebSocketPool {
self.config.keepalive()
}
/// 首个内容帧到达前的绝对超时；`None` 表示禁用首 token 熔断。
/// 首个真实输出到达前的绝对超时；`None` 表示禁用首 token 熔断。
pub(crate) fn first_token_timeout(&self) -> Option<Duration> {
self.config.first_token_timeout
}
backend/tests/dispatch/chat_upstream/mod.rs
+73 -61
View File
@@ -222,8 +222,6 @@ const WEBSOCKET_INVALID_ENCRYPTED_CONTENT: &str =
include_str!("../../fixtures/responses/websocket/invalid_encrypted_content.json");
const WEBSOCKET_PREVIOUS_RESPONSE_NOT_FOUND: &str =
include_str!("../../fixtures/responses/websocket/previous_response_not_found.json");
const WEBSOCKET_UNANSWERED_FUNCTION_CALL: &str =
include_str!("../../fixtures/responses/websocket/unanswered_function_call.json");
const REASONING_REPLAY_REQUEST_GOLDEN: &str =
include_str!("../../fixtures/responses/golden/reasoning_replay_request.json");
@@ -777,11 +775,35 @@ struct CapturedWebSocketRequest {
payload: Value,
}
struct HistoryRecoveryCapture {
first_ws_headers: Vec<(String, String)>,
first_ws_payload: Value,
second_ws_headers: Vec<(String, String)>,
second_ws_payload: Value,
struct WebSocketSequenceCapture {
payload: Value,
retry_attempted: bool,
}
async fn spawn_single_websocket_sequence_upstream(
messages: Vec<String>,
) -> (String, tokio::task::JoinHandle<WebSocketSequenceCapture>) {
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
let base_url = format!("http://{}", listener.local_addr().unwrap());
let server = tokio::spawn(async move {
let (stream, _) = listener.accept().await.unwrap();
let mut websocket = accept_async(stream).await.unwrap();
let payload = websocket.next().await.unwrap().unwrap();
let payload = serde_json::from_str::<Value>(&payload.into_text().unwrap())
.expect("websocket payload should be json");
send_websocket_messages(&mut websocket, &messages).await;
drop(websocket);
let retry_attempted = timeout(StdDuration::from_millis(500), listener.accept())
.await
.is_ok();
WebSocketSequenceCapture {
payload,
retry_attempted,
}
});
(base_url, server)
}
async fn spawn_single_websocket_completed_upstream(
@@ -820,60 +842,6 @@ async fn spawn_single_websocket_completed_upstream(
(base_url, server)
}
async fn spawn_websocket_failure_then_websocket_success_upstream(
failure: String,
) -> (String, tokio::task::JoinHandle<HistoryRecoveryCapture>) {
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
let base_url = format!("http://{}", listener.local_addr().unwrap());
let server = tokio::spawn(async move {
let (first_stream, _) = listener.accept().await.unwrap();
let first_headers = Arc::new(Mutex::new(Vec::new()));
let first_headers_for_callback = Arc::clone(&first_headers);
let mut first_websocket = accept_hdr_async(first_stream, move |request, _response| {
*first_headers_for_callback.lock().unwrap() = request_headers(request);
})
.await
.unwrap();
let first_message = first_websocket.next().await.unwrap().unwrap();
let first_ws_payload = serde_json::from_str::<Value>(&first_message.into_text().unwrap())
.expect("websocket payload should be json");
first_websocket
.send(Message::Text(failure.into()))
.await
.unwrap();
drop(first_websocket);
let (second_stream, _) = listener.accept().await.unwrap();
let second_headers = Arc::new(Mutex::new(Vec::new()));
let second_headers_for_callback = Arc::clone(&second_headers);
let mut second_websocket = accept_hdr_async(second_stream, move |request, _response| {
*second_headers_for_callback.lock().unwrap() = request_headers(request);
})
.await
.unwrap();
let second_message = second_websocket.next().await.unwrap().unwrap();
let second_ws_payload = serde_json::from_str::<Value>(&second_message.into_text().unwrap())
.expect("second websocket payload should be json");
second_websocket
.send(Message::Text(
response_completed_websocket_message("resp_after_history_recovery").into(),
))
.await
.unwrap();
second_websocket.close(None).await.unwrap();
let first_ws_headers = first_headers.lock().unwrap().clone();
let second_ws_headers = second_headers.lock().unwrap().clone();
HistoryRecoveryCapture {
first_ws_headers,
first_ws_payload,
second_ws_headers,
second_ws_payload,
}
});
(base_url, server)
}
async fn spawn_chunked_websocket_upstream() -> (
String,
oneshot::Receiver<()>,
@@ -1132,6 +1100,50 @@ async fn accept_followup_websocket_response(
}
}
async fn accept_followup_websocket_response_sequence(
listener: &TcpListener,
websocket: &mut tokio_tungstenite::WebSocketStream<TcpStream>,
response_messages: Vec<String>,
) -> Value {
let followup = tokio::select! {
message = websocket.next() => {
match message {
Some(Ok(message)) if message.is_text() => {
let payload = serde_json::from_str::<Value>(&message.into_text().unwrap())
.expect("websocket payload should be json");
send_websocket_messages(websocket, &response_messages).await;
return payload;
}
_ => None,
}
}
accepted = listener.accept() => Some(accepted.unwrap().0),
};
let stream = followup.expect("follow-up websocket connection should be available");
let mut followup = accept_async(stream).await.unwrap();
let message = followup.next().await.unwrap().unwrap();
let payload = serde_json::from_str::<Value>(&message.into_text().unwrap())
.expect("websocket payload should be json");
send_websocket_messages(&mut followup, &response_messages).await;
payload
}
async fn send_websocket_messages(
websocket: &mut tokio_tungstenite::WebSocketStream<TcpStream>,
messages: &[String],
) {
for (index, message) in messages.iter().enumerate() {
websocket
.send(Message::Text(message.clone().into()))
.await
.unwrap();
if index + 1 < messages.len() {
tokio::time::sleep(StdDuration::from_millis(25)).await;
}
}
}
async fn send_websocket_response_and_capture_payload(
websocket: &mut tokio_tungstenite::WebSocketStream<TcpStream>,
response_message: String,
backend/tests/dispatch/chat_upstream/responses_http.rs
+18 -6
View File
@@ -1201,7 +1201,7 @@ async fn responses_should_prefer_session_affinity_account_for_previous_response(
}
#[tokio::test]
async fn responses_should_strip_banned_affinity_history_when_switching_to_active_account() {
async fn responses_should_preserve_explicit_history_when_banned_affinity_switches_account() {
let (base_url, upstream) =
spawn_single_websocket_completed_upstream("resp_after_banned_affinity").await;
let (app, state, api_key, _pool, _dir) =
@@ -1244,13 +1244,19 @@ async fn responses_should_strip_banned_affinity_history_when_switching_to_active
captured_header(&captured.headers, "chatgpt-account-id"),
Some("chatgpt-primary")
);
assert!(captured_header(&captured.headers, "x-codex-turn-state").is_none());
assert!(captured.payload.get("previous_response_id").is_none());
assert_eq!(
captured_header(&captured.headers, "x-codex-turn-state"),
Some("turn_affinity_risk")
);
assert_eq!(
captured.payload["previous_response_id"],
"resp_affinity_risk"
);
assert!(affinity.is_none());
}
#[tokio::test]
async fn responses_should_strip_quota_history_when_switching_to_active_account() {
async fn responses_should_preserve_explicit_history_when_quota_affinity_switches_account() {
let (base_url, upstream) =
spawn_single_websocket_completed_upstream("resp_after_quota_affinity").await;
let (app, state, api_key, _pool, _dir) =
@@ -1289,7 +1295,13 @@ async fn responses_should_strip_quota_history_when_switching_to_active_account()
captured_header(&captured.headers, "authorization"),
Some("Bearer access-primary")
);
assert!(captured_header(&captured.headers, "x-codex-turn-state").is_none());
assert!(captured.payload.get("previous_response_id").is_none());
assert_eq!(
captured_header(&captured.headers, "x-codex-turn-state"),
Some("turn_affinity_risk")
);
assert_eq!(
captured.payload["previous_response_id"],
"resp_affinity_risk"
);
assert!(affinity.is_some());
}
backend/tests/dispatch/chat_upstream/responses_recovery.rs
+98 -130
View File
@@ -1040,13 +1040,13 @@ async fn responses_stream_should_return_model_unsupported_error_when_fallback_is
}
#[tokio::test]
async fn responses_should_strip_history_after_previous_response_not_found() {
async fn responses_explicit_previous_response_should_fail_without_transparent_retry() {
let (base_url, upstream) =
spawn_websocket_failure_then_websocket_success_upstream(response_failed_websocket_message(
spawn_single_websocket_sequence_upstream(vec![response_failed_websocket_message(
"resp_previous_missing_failed",
"previous_response_not_found",
"Previous response with id resp_stale was not found",
))
)])
.await;
let (app, api_key, _pool, _dir) =
test_app_with_account_pool_and_affinity(base_url, "resp_stale").await;
@@ -1075,31 +1075,24 @@ async fn responses_should_strip_history_after_previous_response_not_found() {
let body = response_json(response).await;
let captured = upstream.await.unwrap();
assert_eq!(status, StatusCode::OK);
assert_eq!(body["id"], "resp_after_history_recovery");
assert_eq!(
captured_header(&captured.first_ws_headers, "x-codex-turn-state"),
Some("turn-stale")
);
assert!(captured_header(&captured.second_ws_headers, "x-codex-turn-state").is_none());
assert_eq!(
captured.first_ws_payload["previous_response_id"],
"resp_stale"
);
assert!(captured
.second_ws_payload
.get("previous_response_id")
.is_none());
assert_eq!(status, StatusCode::BAD_REQUEST);
assert_eq!(body["error"]["code"], "previous_response_unavailable");
assert!(body["error"]["message"]
.as_str()
.unwrap()
.contains("resend the complete input"));
assert_eq!(captured.payload["previous_response_id"], "resp_stale");
assert!(!captured.retry_attempted);
}
#[tokio::test]
async fn responses_should_strip_history_after_unanswered_function_call() {
async fn responses_explicit_unanswered_function_call_should_not_retry_without_full_context() {
let (base_url, upstream) =
spawn_websocket_failure_then_websocket_success_upstream(response_failed_websocket_message(
spawn_single_websocket_sequence_upstream(vec![response_failed_websocket_message(
"resp_unanswered_call_failed",
"invalid_request",
"No tool output found for function call call_1",
))
)])
.await;
let (app, api_key, _pool, _dir) =
test_app_with_account_pool_and_affinity(base_url, "resp_stale").await;
@@ -1128,23 +1121,20 @@ async fn responses_should_strip_history_after_unanswered_function_call() {
let body = response_json(response).await;
let captured = upstream.await.unwrap();
assert_eq!(status, StatusCode::OK);
assert_eq!(body["id"], "resp_after_history_recovery");
assert!(captured_header(&captured.second_ws_headers, "x-codex-turn-state").is_none());
assert!(captured
.second_ws_payload
.get("previous_response_id")
.is_none());
assert_eq!(status, StatusCode::BAD_REQUEST);
assert_eq!(body["error"]["code"], "previous_response_unavailable");
assert_eq!(captured.payload["previous_response_id"], "resp_stale");
assert!(!captured.retry_attempted);
}
#[tokio::test]
async fn responses_should_strip_history_after_websocket_invalid_encrypted_reasoning_replay() {
async fn responses_explicit_invalid_reasoning_replay_should_not_retry_without_full_context() {
let (base_url, upstream) =
spawn_websocket_failure_then_websocket_success_upstream(response_failed_websocket_message(
spawn_single_websocket_sequence_upstream(vec![response_failed_websocket_message(
"resp_invalid_reasoning_failed",
"invalid_encrypted_content",
"Invalid encrypted content in reasoning replay",
))
)])
.await;
let (app, api_key, _pool, _dir) =
test_app_with_account_pool_and_affinity(base_url, "resp_stale").await;
@@ -1173,73 +1163,44 @@ async fn responses_should_strip_history_after_websocket_invalid_encrypted_reason
let body = response_json(response).await;
let captured = upstream.await.unwrap();
assert_eq!(status, StatusCode::OK);
assert_eq!(body["id"], "resp_after_history_recovery");
assert!(captured_header(&captured.second_ws_headers, "x-codex-turn-state").is_none());
assert!(captured
.second_ws_payload
.get("previous_response_id")
.is_none());
assert_eq!(status, StatusCode::BAD_REQUEST);
assert_eq!(body["error"]["code"], "previous_response_unavailable");
assert_eq!(captured.payload["previous_response_id"], "resp_stale");
assert!(!captured.retry_attempted);
}
#[tokio::test]
async fn responses_should_strip_history_after_sse_invalid_encrypted_reasoning_replay() {
let (base_url, upstream) =
spawn_websocket_failure_then_websocket_success_upstream(response_failed_websocket_message(
"resp_invalid_reasoning_failed",
"invalid_encrypted_content",
"Invalid encrypted content in reasoning replay",
))
.await;
let (app, api_key, _pool, _dir) =
test_app_with_account_pool_and_affinity(base_url, "resp_stale").await;
let response = app
.oneshot(
Request::builder()
.method("POST")
.uri("/v1/responses")
.header("authorization", format!("Bearer {api_key}"))
.header("content-type", "application/json")
.body(Body::from(
json!({
"model": "gpt-5.5",
"previous_response_id": "resp_stale",
"turnState": "turn-stale",
"input": [{"role": "user", "content": "Continue"}],
"stream": false
})
.to_string(),
))
.unwrap(),
)
.await
.unwrap();
let status = response.status();
let body = response_json(response).await;
let captured = upstream.await.unwrap();
assert_eq!(status, StatusCode::OK);
assert_eq!(body["id"], "resp_after_history_recovery");
assert!(captured_header(&captured.second_ws_headers, "x-codex-turn-state").is_none());
assert!(captured
.second_ws_payload
.get("previous_response_id")
.is_none());
}
#[tokio::test]
async fn responses_stream_should_strip_history_after_previous_response_not_found() {
let (base_url, upstream) =
spawn_websocket_failure_then_websocket_success_upstream(response_failed_websocket_message(
async fn responses_stream_structural_events_should_not_commit_before_history_failure() {
let (base_url, upstream) = spawn_single_websocket_sequence_upstream(vec![
json!({
"type": "response.created",
"response": {
"id": "resp_previous_missing_failed",
"object": "response",
"status": "in_progress"
}
})
.to_string(),
json!({
"type": "response.content_part.added",
"response_id": "resp_previous_missing_failed",
"output_index": 0,
"content_index": 0,
"part": {"type": "output_text", "text": "", "annotations": []}
})
.to_string(),
response_failed_websocket_message(
"resp_previous_missing_failed",
"previous_response_not_found",
"Previous response with id resp_stale was not found",
))
.await;
),
])
.await;
let (app, api_key, _pool, _dir) =
test_app_with_account_pool_and_affinity(base_url, "resp_stale").await;
let response = app
.oneshot(
let response = timeout(
StdDuration::from_secs(2),
app.oneshot(
Request::builder()
.method("POST")
.uri("/v1/responses")
@@ -1256,40 +1217,57 @@ async fn responses_stream_should_strip_history_after_previous_response_not_found
.to_string(),
))
.unwrap(),
)
.await
.unwrap();
),
)
.await
.expect("dispatch should classify the failure before starting the live stream")
.unwrap();
let status = response.status();
let body = response_text(response).await;
let captured = upstream.await.unwrap();
assert_eq!(status, StatusCode::OK);
assert!(body.contains("resp_after_history_recovery"));
assert_eq!(
captured_header(&captured.first_ws_headers, "x-codex-turn-state"),
Some("turn-stale")
);
assert_eq!(
captured.first_ws_payload["previous_response_id"],
"resp_stale"
);
assert!(captured_header(&captured.second_ws_headers, "x-codex-turn-state").is_none());
assert!(captured
.second_ws_payload
.get("previous_response_id")
.is_none());
assert!(body.contains("previous_response_unavailable"));
assert!(!body.contains("response.created"));
assert_eq!(captured.payload["previous_response_id"], "resp_stale");
assert!(!captured.retry_attempted);
}
#[tokio::test]
async fn responses_stream_should_strip_history_after_websocket_invalid_encrypted_reasoning_replay()
{
let (base_url, upstream) =
spawn_websocket_failure_then_websocket_success_upstream(response_failed_websocket_message(
"resp_invalid_reasoning_failed",
"invalid_encrypted_content",
"Invalid encrypted content in reasoning replay",
))
.await;
async fn responses_stream_should_not_retry_history_failure_after_real_output() {
let (base_url, upstream) = spawn_single_websocket_sequence_upstream(vec![
json!({
"type": "response.created",
"response": {
"id": "resp_output_then_failed",
"object": "response",
"status": "in_progress"
}
})
.to_string(),
json!({
"type": "response.content_part.added",
"response_id": "resp_output_then_failed",
"output_index": 0,
"content_index": 0,
"part": {"type": "output_text", "text": "", "annotations": []}
})
.to_string(),
json!({
"type": "response.output_text.delta",
"response_id": "resp_output_then_failed",
"output_index": 0,
"content_index": 0,
"delta": "visible output"
})
.to_string(),
response_failed_websocket_message(
"resp_output_then_failed",
"previous_response_not_found",
"Previous response with id resp_stale was not found",
),
])
.await;
let (app, api_key, _pool, _dir) =
test_app_with_account_pool_and_affinity(base_url, "resp_stale").await;
let response = app
@@ -1318,20 +1296,10 @@ async fn responses_stream_should_strip_history_after_websocket_invalid_encrypted
let captured = upstream.await.unwrap();
assert_eq!(status, StatusCode::OK);
assert!(body.contains("resp_after_history_recovery"));
assert_eq!(
captured_header(&captured.first_ws_headers, "x-codex-turn-state"),
Some("turn-stale")
);
assert_eq!(
captured.first_ws_payload["previous_response_id"],
"resp_stale"
);
assert!(captured_header(&captured.second_ws_headers, "x-codex-turn-state").is_none());
assert!(captured
.second_ws_payload
.get("previous_response_id")
.is_none());
assert!(body.contains("visible output"));
assert!(body.contains("response.failed"));
assert_eq!(captured.payload["previous_response_id"], "resp_stale");
assert!(!captured.retry_attempted);
}
#[tokio::test]
backend/tests/dispatch/chat_upstream/responses_websocket.rs
+52 -78
View File
@@ -348,7 +348,7 @@ async fn responses_websocket_should_reuse_connection_for_recorded_conversation()
}
#[tokio::test]
async fn responses_websocket_should_retry_fresh_connection_when_reused_connection_dies_before_first_frame(
async fn responses_websocket_should_retry_fresh_connection_when_reused_connection_dies_before_first_output(
) {
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
let base_url = format!("http://{}", listener.local_addr().unwrap());
@@ -1405,10 +1405,29 @@ async fn responses_websocket_should_restore_full_history_when_implicit_resume_pr
WEBSOCKET_COMPLETED_WITH_REASONING_REPLAY.trim().to_string(),
)
.await;
let implicit_payload = accept_followup_websocket_response(
let implicit_payload = accept_followup_websocket_response_sequence(
&listener,
&mut websocket,
WEBSOCKET_PREVIOUS_RESPONSE_NOT_FOUND.trim().to_string(),
vec![
json!({
"type": "response.created",
"response": {
"id": "resp_implicit_resume_failed",
"object": "response",
"status": "in_progress"
}
})
.to_string(),
json!({
"type": "response.content_part.added",
"response_id": "resp_implicit_resume_failed",
"output_index": 0,
"content_index": 0,
"part": {"type": "output_text", "text": "", "annotations": []}
})
.to_string(),
WEBSOCKET_PREVIOUS_RESPONSE_NOT_FOUND.trim().to_string(),
],
)
.await;
let _ = websocket.close(None).await;
@@ -1442,7 +1461,7 @@ async fn responses_websocket_should_restore_full_history_when_implicit_resume_pr
&json!({
"model": "gpt-5.5",
"instructions": "answer briefly",
"stream": false,
"stream": true,
"input": [
{"role": "user", "content": "remember this"},
{"role": "assistant", "content": "cached answer"},
@@ -1452,9 +1471,10 @@ async fn responses_websocket_should_restore_full_history_when_implicit_resume_pr
))
.await
.unwrap();
let body = response_json(second_response).await;
assert_eq!(second_response.status(), StatusCode::OK);
let body = response_text(second_response).await;
let (_first_payload, implicit_payload, restored_payload) = upstream.await.unwrap();
assert_eq!(body["id"], "resp_implicit_resume_restored");
assert!(body.contains("resp_implicit_resume_restored"));
assert_eq!(
implicit_payload["previous_response_id"],
"resp_implicit_resume_first"
@@ -1610,15 +1630,14 @@ async fn responses_websocket_should_prefer_conversation_account_without_previous
}
#[tokio::test]
async fn responses_websocket_non_stream_unknown_previous_response_should_strip_history_before_upstream(
) {
async fn responses_websocket_non_stream_should_forward_unknown_previous_response() {
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
let base_url = format!("http://{}", listener.local_addr().unwrap());
let upstream = tokio::spawn(async move {
accept_websocket_response_with_authorization_and_message(
&listener,
"Bearer access-secret",
websocket_completed_response("resp_after_history_strip", 3, 1),
websocket_completed_response("resp_unknown_previous_forwarded", 3, 1),
)
.await
});
@@ -1639,19 +1658,19 @@ async fn responses_websocket_non_stream_unknown_previous_response_should_strip_h
let body = response_json(response).await;
let payload = upstream.await.unwrap();
assert_eq!(body["id"], "resp_after_history_strip");
assert!(payload.get("previous_response_id").is_none());
assert_eq!(body["id"], "resp_unknown_previous_forwarded");
assert_eq!(payload["previous_response_id"], "resp_missing");
}
#[tokio::test]
async fn responses_websocket_should_strip_known_history_when_preferred_account_is_full() {
async fn responses_websocket_should_preserve_known_history_on_fallback_account() {
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
let base_url = format!("http://{}", listener.local_addr().unwrap());
let upstream = tokio::spawn(async move {
accept_websocket_response_with_authorization_and_message(
&listener,
"Bearer access-secondary",
websocket_completed_response("resp_after_fallback_history_strip", 3, 1),
websocket_completed_response("resp_history_forwarded_to_fallback", 3, 1),
)
.await
});
@@ -1677,19 +1696,19 @@ async fn responses_websocket_should_strip_known_history_when_preferred_account_i
let body = response_json(response).await;
let payload = upstream.await.unwrap();
assert_eq!(body["id"], "resp_after_fallback_history_strip");
assert!(payload.get("previous_response_id").is_none());
assert_eq!(body["id"], "resp_history_forwarded_to_fallback");
assert_eq!(payload["previous_response_id"], "resp_owned_by_primary");
}
#[tokio::test]
async fn responses_websocket_stream_should_strip_known_history_when_preferred_account_is_full() {
async fn responses_websocket_stream_should_preserve_known_history_on_fallback_account() {
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
let base_url = format!("http://{}", listener.local_addr().unwrap());
let upstream = tokio::spawn(async move {
accept_websocket_response_with_authorization_and_message(
&listener,
"Bearer access-secondary",
websocket_completed_response("resp_stream_after_fallback_history_strip", 3, 1),
websocket_completed_response("resp_stream_history_forwarded_to_fallback", 3, 1),
)
.await
});
@@ -1715,20 +1734,19 @@ async fn responses_websocket_stream_should_strip_known_history_when_preferred_ac
let body = response_text(response).await;
let payload = upstream.await.unwrap();
assert!(body.contains("resp_stream_after_fallback_history_strip"));
assert!(payload.get("previous_response_id").is_none());
assert!(body.contains("resp_stream_history_forwarded_to_fallback"));
assert_eq!(payload["previous_response_id"], "resp_owned_by_primary");
}
#[tokio::test]
async fn responses_websocket_stream_unknown_previous_response_should_strip_history_before_upstream()
{
async fn responses_websocket_stream_should_forward_unknown_previous_response() {
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
let base_url = format!("http://{}", listener.local_addr().unwrap());
let upstream = tokio::spawn(async move {
accept_websocket_response_with_authorization_and_message(
&listener,
"Bearer access-secret",
websocket_completed_response("resp_stream_after_history_strip", 3, 1),
websocket_completed_response("resp_stream_unknown_previous_forwarded", 3, 1),
)
.await
});
@@ -1749,51 +1767,8 @@ async fn responses_websocket_stream_unknown_previous_response_should_strip_histo
let body = response_text(response).await;
let payload = upstream.await.unwrap();
assert!(body.contains("\"id\":\"resp_stream_after_history_strip\""));
assert!(payload.get("previous_response_id").is_none());
}
#[tokio::test]
async fn responses_websocket_non_stream_unanswered_function_call_should_strip_history_and_retry_same_account(
) {
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
let base_url = format!("http://{}", listener.local_addr().unwrap());
let upstream = tokio::spawn(async move {
let first_payload = accept_websocket_response_with_authorization_and_message(
&listener,
"Bearer access-secret",
WEBSOCKET_UNANSWERED_FUNCTION_CALL.trim().to_string(),
)
.await;
let second_payload = accept_websocket_response_with_authorization_and_message(
&listener,
"Bearer access-secret",
websocket_completed_response("resp_after_function_call_strip", 3, 1),
)
.await;
(first_payload, second_payload)
});
let (app, api_key, _pool, _dir) =
test_app_with_account_pool_and_affinity(base_url, "resp_with_call").await;
let response = app
.oneshot(responses_json_request(
&api_key,
&json!({
"model": "gpt-5.5",
"input": [],
"stream": false,
"previous_response_id": "resp_with_call"
}),
))
.await
.unwrap();
let body = response_json(response).await;
let (first_payload, second_payload) = upstream.await.unwrap();
assert_eq!(body["id"], "resp_after_function_call_strip");
assert_eq!(first_payload["previous_response_id"], "resp_with_call");
assert!(second_payload.get("previous_response_id").is_none());
assert!(body.contains("\"id\":\"resp_stream_unknown_previous_forwarded\""));
assert_eq!(payload["previous_response_id"], "resp_missing");
}
#[tokio::test]
@@ -1840,9 +1815,7 @@ async fn responses_websocket_previous_response_id_should_retry_fallback_account_
.unwrap();
assert!(body.contains("\"id\":\"resp_history_fallback\""));
// 亲和缓存无 `resp_prev` 记录（Unknown owner），换账号前历史即被剥离，
// 由 `input` 让备用账号重建上下文，不把上一账号作用域的 id 透传给它。
assert_eq!(fallback_payload.get("previous_response_id"), None);
assert_eq!(fallback_payload["previous_response_id"], "resp_prev");
assert_eq!(secondary_usage.0, 1);
}
@@ -1892,9 +1865,7 @@ async fn responses_websocket_non_stream_previous_response_id_should_retry_fallba
.unwrap();
assert_eq!(body["id"], "resp_history_fallback_non_stream");
// 亲和缓存无 `resp_prev` 记录（Unknown owner），换账号前历史即被剥离，
// 由 `input` 让备用账号重建上下文，不把上一账号作用域的 id 透传给它。
assert_eq!(fallback_payload.get("previous_response_id"), None);
assert_eq!(fallback_payload["previous_response_id"], "resp_prev");
assert_eq!(secondary_usage.0, 1);
}
@@ -2340,9 +2311,10 @@ async fn responses_stream_with_previous_response_id_should_forward_websocket_chu
captured_header(&captured.headers, "authorization"),
Some("Bearer access-secret")
);
// 亲和缓存无该 previous_response_id 的记录（Unknown），故上游 payload 里应被剥离，
// 靠已展开的 input 让选中账号自行重建续链——与 sub2api/codex2api 的默认行为一致。
assert_eq!(captured.payload["previous_response_id"], Value::Null);
assert_eq!(
captured.payload["previous_response_id"],
"resp_runtime_pool_previous"
);
}
#[tokio::test]
@@ -2370,8 +2342,10 @@ async fn responses_stream_with_previous_response_id_should_record_websocket_audi
let metadata: Value = serde_json::from_str(&event.metadata_json).unwrap();
assert!(body.contains("resp_live_websocket_stream"));
// 亲和缓存无记录（Unknown）→ 上游 payload 里 previous_response_id 被剥离。
assert_eq!(captured.payload["previous_response_id"], Value::Null);
assert_eq!(
captured.payload["previous_response_id"],
"resp_runtime_pool_previous"
);
assert_eq!(event.level, "info");
assert_eq!(metadata["stream"], true);
assert_eq!(event.transport.as_deref(), Some("websocket"));
backend/tests/upstream/openai/protocol/mod.rs
+1 -1
View File
@@ -5,7 +5,7 @@ use codex_proxy_rs::upstream::openai::protocol::events::{
rate_limit_quota, retry_after_seconds_from_body, RateLimitWindow, TokenUsage,
};
use codex_proxy_rs::upstream::openai::protocol::responses::{
http_sse_fallback_allowed, response_body_has_first_event, response_from_codex_sse,
http_sse_fallback_allowed, response_body_has_first_output, response_from_codex_sse,
response_sse_event_is_terminal, transport_for_request, CodexResponsesRequest, CodexTransport,
CollectedResponse,
};
backend/tests/upstream/openai/protocol/response_first_event.rs
+20 -20
View File
@@ -1,74 +1,74 @@
use super::*;
#[test]
fn first_event_detection_should_match_text_delta() {
fn first_output_detection_should_match_text_delta() {
let body = include_bytes!("../../../fixtures/responses/http_sse/text_delta_hello.sse");
assert!(response_body_has_first_event(body));
assert!(response_body_has_first_output(body));
}
#[test]
fn first_event_detection_should_match_reasoning_delta() {
fn first_output_detection_should_match_reasoning_delta() {
let body = include_bytes!("../../../fixtures/responses/http_sse/reasoning_delta.sse");
assert!(response_body_has_first_event(body));
assert!(response_body_has_first_output(body));
}
#[test]
fn first_event_detection_should_match_function_call_arguments_delta() {
fn first_output_detection_should_match_function_call_arguments_delta() {
let body =
include_bytes!("../../../fixtures/responses/http_sse/function_call_arguments_delta.sse");
assert!(response_body_has_first_event(body));
assert!(response_body_has_first_output(body));
}
#[test]
fn first_event_detection_should_match_type_when_event_name_is_missing() {
fn first_output_detection_should_match_type_when_event_name_is_missing() {
let body = include_bytes!("../../../fixtures/responses/http_sse/data_type_text_delta.sse");
assert!(response_body_has_first_event(body));
assert!(response_body_has_first_output(body));
}
#[test]
fn first_event_detection_should_ignore_response_created() {
fn first_output_detection_should_ignore_response_created() {
let body = include_bytes!("../../../fixtures/responses/http_sse/response_created.sse");
assert!(!response_body_has_first_event(body));
assert!(!response_body_has_first_output(body));
}
#[test]
fn first_event_detection_should_ignore_metadata_and_rate_limit_events() {
fn first_output_detection_should_ignore_metadata_and_rate_limit_events() {
let body = include_bytes!(
"../../../fixtures/responses/http_sse/metadata_rate_limits_no_first_output.sse"
);
assert!(!response_body_has_first_event(body));
assert!(!response_body_has_first_output(body));
}
#[test]
fn first_event_detection_should_ignore_empty_delta() {
fn first_output_detection_should_ignore_empty_delta() {
let body = include_bytes!("../../../fixtures/responses/http_sse/empty_text_delta.sse");
assert!(!response_body_has_first_event(body));
assert!(!response_body_has_first_output(body));
}
#[test]
fn first_event_detection_should_ignore_done_frame() {
fn first_output_detection_should_ignore_done_frame() {
let body = include_bytes!("../../../fixtures/responses/http_sse/done_only.sse");
assert!(!response_body_has_first_event(body));
assert!(!response_body_has_first_output(body));
}
#[test]
fn first_event_detection_should_ignore_incomplete_frames() {
fn first_output_detection_should_ignore_incomplete_frames() {
let body = include_bytes!("../../../fixtures/responses/http_sse/incomplete_text_delta.sse");
assert!(!response_body_has_first_event(body));
assert!(!response_body_has_first_output(body));
}
#[test]
fn first_event_detection_should_match_crlf_frames() {
fn first_output_detection_should_match_crlf_frames() {
let body = include_bytes!("../../../fixtures/responses/http_sse/text_delta_crlf.sse");
assert!(response_body_has_first_event(body));
assert!(response_body_has_first_output(body));
}
backend/tests/upstream/openai/transport/websocket.rs
+2 -2
View File
@@ -1101,7 +1101,7 @@ async fn codex_backend_client_stream_should_reject_binary_websocket_event() {
)
.await
else {
panic!("binary event before first content should reject stream open");
panic!("binary event before first output should reject stream open");
};
server.await.unwrap();
@@ -1112,7 +1112,7 @@ async fn codex_backend_client_stream_should_reject_binary_websocket_event() {
}
#[tokio::test]
async fn codex_backend_client_stream_should_retry_without_pool_when_first_content_times_out() {
async fn codex_backend_client_stream_should_retry_without_pool_when_first_output_times_out() {
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
let addr = listener.local_addr().unwrap();
let accepted_connections = Arc::new(AtomicUsize::new(0));
backend/tests/upstream/openai/transport/websocket_pool.rs
+3 -3
View File
@@ -156,7 +156,7 @@ async fn codex_backend_client_should_open_fresh_socket_when_idle_pooled_websocke
}
#[tokio::test]
async fn codex_backend_client_stream_should_retry_fresh_socket_when_first_content_times_out() {
async fn codex_backend_client_stream_should_retry_fresh_socket_when_first_output_times_out() {
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
let addr = listener.local_addr().unwrap();
let accepted_connections = Arc::new(AtomicUsize::new(0));
@@ -241,7 +241,7 @@ async fn codex_backend_client_stream_should_retry_fresh_socket_when_first_conten
}
#[tokio::test]
async fn codex_backend_client_stream_should_retry_reused_socket_when_first_content_times_out() {
async fn codex_backend_client_stream_should_retry_reused_socket_when_first_output_times_out() {
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
let addr = listener.local_addr().unwrap();
let accepted_connections = Arc::new(AtomicUsize::new(0));
@@ -342,7 +342,7 @@ async fn codex_backend_client_stream_should_retry_reused_socket_when_first_conte
}
#[tokio::test]
async fn codex_backend_client_stream_should_retry_bypass_socket_when_first_content_times_out() {
async fn codex_backend_client_stream_should_retry_bypass_socket_when_first_output_times_out() {
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
let addr = listener.local_addr().unwrap();
let accepted_connections = Arc::new(AtomicUsize::new(0));
docs/stream-disconnection-remediation.md
+224
View File
@@ -0,0 +1,224 @@
# Responses 流式断连与历史续接失败排查
日期：2026-07-10
状态：已完成只读排查，未修改业务代码或线上配置
## 一、排查过程
### 1. 确认实际运行版本与基础设施
- Docker 镜像 tag 仍是旧版本，但在线升级状态确认当前进程运行 `v1.0.11`；本次故障发生在升级完成之后。
- Nginx 已关闭请求和响应缓冲，读写超时为 7200 秒。
- 故障窗口内应用 healthy，无 OOM、panic、重启或 Nginx upstream error。
因此，本次故障不是镜像版本判断错误、常规反代超时或服务进程崩溃。
### 2. 反查 `resp_0af2da5fcb22adab016a508f8f35a481958ce857bc359c73b9`
| 北京时间 | 请求 | 服务端事实 |
| --- | --- | --- |
| 14:22:17 | `req_6b1fa095...` | 原响应正常完成，`completed=true`、`store=false`；账号 `acct_0b5ed92a...`；WebSocket 为 `reuse` |
| 14:23:51 | `req_a085bb96...` | 同账号、同 conversation hash、同 pool key，再次 `reuse`；HTTP 200 已提交，但之后没有 `live response stream finalized` |
| 14:24:20 | `req_4f265b8e...` | 客户端仍携带原 response ID 重试；账号和 pool key 不变，但 WebSocket 已变为 `new` |
| 14:24:20.755 | 同上 | 服务先向客户端提交 HTTP 200 |
| 14:24:20.830 | 同上 | 75 ms 后上游返回 `previous_response_not_found`；没有真实首 token |
| 14:24:20–14:25:56 | 17 个请求 | 同一个失效 response ID 被连续重试，全部失败 |
14:23:51 的请求没有流终结记录，且服务进程没有重启。结合代码中的下游取消路径，可以确定该次下游流被丢弃：响应 body 被 drop 后触发 cancel，随后上游 WebSocket 被 discard。下一次请求因此只能创建新连接。
### 3. 核对智能调度与会话亲和性
- 原响应和 17 次失败全部选择同一个上游账号 `acct_0b5ed92a...`。
- 调度日志虽然显示 `rotation_strategy=smart`，但选择来源为 `preferred`。
- 原请求和失败请求的 conversation hash、WebSocket pool key 均一致。
- 如果真的切换了账号，`strip_history_if_account_changed` 会在发送上游前移除旧 history。
结论：智能调度和账号亲和性工作正常，不是本次 `previous_response_not_found` 的原因。
### 4. 核对 Responses WebSocket 续接语义
原响应使用 `store=false`，不会作为持久化 Response 保存。OpenAI 官方说明：WebSocket 模式只在连接本地缓存最近的 previous response；连接缓存无法解析该 ID 时，应将 `previous_response_id` 设为 `null`，并重新提交完整输入上下文。
参考：[OpenAI Conversation state / WebSocket mode](https://developers.openai.com/api/docs/guides/conversation-state#previous_response_id-in-websocket-mode)
因此，“仍是同一账号”不足以保证旧 ID 可用；对于未持久化响应，还依赖原 WebSocket 的连接级缓存。
### 5. 核对 1.0.11 的恢复路径
代码已经实现 `previous_response_not_found` 恢复：
1. `recover_request_history` 忘记失效 ID，并移除 `previous_response_id` 与 turn state；
2. 将下一次尝试固定到原账号；
3. 在同一个入站请求内重新调用上游。
但该恢复只在 HTTP 响应尚未提交时有效。本次请求在真正的 400 到达前 75 ms 已提交 HTTP 200。
原因是 `event_is_first_content` 只排除 `response.created` 和 `response.in_progress`，其余可转发事件都会结束预取。日志同时显示没有真实首 token，说明某个非输出结构事件提前越过了预取边界。400 随后进入 live stream，只能被包装为 `response.failed`，不能再回到调度循环重试。
现有测试 `responses_stream_should_strip_history_after_previous_response_not_found` 让失败帧作为 WebSocket 第一帧返回，因此能够通过，但没有覆盖“非输出结构帧在前、失败帧稍后到达”的真实时序。
## 二、问题原因
本次故障是以下链路叠加造成的：
```text
客户端/TUN 下游断流
-> 服务取消 live stream 并销毁原 WebSocket
-> store=false 的连接级 previous response 缓存随连接消失
-> 客户端在新 WebSocket 上继续发送旧 previous_response_id
-> 上游正确返回 previous_response_not_found
-> 代理预取过早提交 HTTP 200，错过已有的 history recovery
-> 客户端不断重试同一个失效 ID
```
根因分层如下：
1. **触发条件**：下游连接中断导致原 WebSocket 被 discard。该现象与此前客户端侧 `error decoding response body`、TUN/fake-IP 路径一致。
2. **上游行为**：对 `store=false` 且连接缓存已丢失的 response ID 返回 `previous_response_not_found`，符合 WebSocket 续接语义。
3. **服务端缺陷**：WebSocket “首内容”判定与真实首 token 判定不一致，导致可恢复错误在 HTTP 200 之后才暴露。
4. **重试放大**：Codex 继续在原线程发送相同 ID；第一次失败又会销毁新 WebSocket，随后每次请求都从 `new` 连接开始。
5. **非根因**：智能调度、账号轮换、Nginx 超时、应用重启以及运行版本均已排除。
外层错误中的 `stream_disconnected` 是代理在 live stream 阶段合成的包装；真正的上游原因是内部的 HTTP 400 `previous_response_not_found`。
## 三、修复方案
### 1. 立即止损
- 受影响客户端应开启新线程，或者移除 `previous_response_id` 后重新提交完整上下文。
- 不要在原线程连续执行“继续”；它只会重复发送同一个失效 ID。
- 对客户端到 `codex-proxy.aivify.cc` 的路径做 DIRECT/TUN 绕过 A/B，先解决导致原 WebSocket 被销毁的下游断流。
### 2. 修复 WebSocket 预取边界
WebSocket 预取和 `firstTokenMs` 必须共用同一套语义判定：
- 真实文本、reasoning、工具参数 delta 或 output item：视为首输出，可以提交下游响应；
- `response.completed`、`response.failed`、`error`：视为终态，在提交前完成分类；
- created、in-progress、part-added 等没有实际输出的结构事件：只缓存，不结束预取。
这样 `previous_response_not_found` 会在 HTTP 200 提交前进入现有 `recover_request_history` 分支。
### 3. 保证恢复语义安全
- 如果代理保留了完整输入快照，清除旧 ID 后用完整上下文重试。
- 如果是客户端显式传入的 ID，而代理没有完整历史，不能假装无损恢复；应返回明确错误，要求客户端新建线程或重发完整上下文。
- 不应静默把所有请求从 `store=false` 改成 `store=true`，否则会改变数据留存语义。若要启用持久化，必须作为显式配置。
### 4. 补回归测试
- `response.created -> 非输出结构事件 -> previous_response_not_found`：断言 HTTP 200 前完成恢复，第二次上游 payload 不含旧 ID。
- 下游取消后 pooled WebSocket 被 discard，再用旧 ID 续接：断言不会形成重试风暴。
- 真实 delta 之后再出现错误：断言禁止透明重试，避免重复输出、工具调用和计费。
- 显式 previous ID 且没有完整上下文：断言不会以残缺输入伪装成功恢复。
### 5. 调度与部署建议
- 保持智能调度和会话亲和性；切换成轮询不能解决连接级缓存问题，反而更容易跨账号。
- 不要通过禁用 WebSocket pool 规避；`store=false + previous_response_id` 在每次新建连接时更容易失败。
- 修复后先在 next 实例验证上述时序，再切换 Nginx upstream，并保留旧实例回滚。
- Nginx access log 补充 request ID、请求时长和协议；应用补充 `client_cancelled`、`downstream_send_failed` 与 pool discard 原因，便于后续直接串联同一事件。
## 四、`gpt-5.6-sol` 的 `ultra` 记录问题
### 1. 排查过程
- 查询线上最近 80 条 `gpt-5.6-sol` 使用记录，`reasoningEffort` 全部为 `max`。
- 核对代理代码：请求体中的 `reasoning` 原样透传；使用记录直接读取 `reasoning.effort`；管理端直接展示该记录值，没有 `ultra -> max` 转换。
- 使用 Codex 0.144.1 和本地 mock 端点分别抓取 `ultra`、`max` 两种配置的实际请求：
```text
客户端选择 ultra -> reasoning.effort = max
客户端选择 max -> reasoning.effort = max
```
- 两份请求的实际差异在协作模式：`ultra` 注入“主动多 Agent 委派”，`max` 注入“仅在明确要求时委派”。
- 因此该现象与智能调度、账号选择和代理归一化无关。
### 2. 问题原因
Codex 0.144.1 将 `ultra` 实现为组合预设：
```text
ultra = max 推理强度 + 主动多 Agent 委派
```
发往 Responses API 的请求中没有字面值 `ultra`，只有 `reasoning.effort = max` 和单独的多 Agent 模式指令。代理目前只记录前者，因此管理端显示 `max`。这是记录维度不足，不是记录值被错误改写，也不是上游降级。
### 3. 修复方案
- 保留 `reasoningEffort = max`，因为这是上游实际收到的推理强度。
- 从请求中的 `<multi_agent_mode>` 识别主动委派，新增 `multiAgentMode = proactive` 或 `reasoningPreset = ultra` 元数据。
- 管理端组合展示为 `ultra（max + 自动委派）`，不要直接把原始 `reasoningEffort` 覆盖为 `ultra`。
- 长期方案是由客户端在 `client_metadata` 中显式携带原始预设；在此之前，服务端只能根据多 Agent 模式指令推断。
## 五、Codex 0.144.1 与 WebSocket 依赖更新核对
### 1. 排查过程
截至 2026-07-10，OpenAI Codex 最新稳定版本为 `rust-v0.144.1`。本次核对了该版本、`main` 分支、两个 OpenAI fork 的提交历史以及本项目 `v1.0.11` 的实际依赖锁定。
本项目当前锁定：
```toml
tokio-tungstenite = { git = "https://github.com/openai-oss-forks/tokio-tungstenite", rev = "132f5b39c862e3a970f731d709608b3e6276d5f6" }
tungstenite = { git = "https://github.com/openai-oss-forks/tungstenite-rs", rev = "9200079d3b54a1ff51072e24d81fd354f085156f" }
```
Codex `0.144.1` 和当前 `main` 均已更新为：
```toml
tokio-tungstenite = { git = "https://github.com/openai-oss-forks/tokio-tungstenite", rev = "0e5b2d73aa18dd9f0a50ee9ff199d5aef7594186" }
tungstenite = { git = "https://github.com/openai-oss-forks/tungstenite-rs", rev = "4fffad30fe373adbdcffab9545e9e9bf4f2fc19f" }
```
参考：
- [Codex 0.144.1 Cargo.toml](https://github.com/openai/codex/blob/rust-v0.144.1/codex-rs/Cargo.toml#L559-L568)
- [`tokio-tungstenite` 两个版本的完整差异](https://github.com/openai-oss-forks/tokio-tungstenite/compare/132f5b39c862e3a970f731d709608b3e6276d5f6...0e5b2d73aa18dd9f0a50ee9ff199d5aef7594186)
- [`tungstenite` 两个版本的完整差异](https://github.com/openai-oss-forks/tungstenite-rs/compare/9200079d3b54a1ff51072e24d81fd354f085156f...4fffad30fe373adbdcffab9545e9e9bf4f2fc19f)
版本变化如下：
| 项目 | 实际变化 | 对当前服务的影响 |
| --- | --- | --- |
| `tokio-tungstenite` | 新增 Happy Eyeballs：交错尝试 IPv4/IPv6，首个地址连接停滞 250 ms 后竞争备用地址；直连和代理连接均使用该逻辑 | 可改善首次建连和断线重连，尤其是 IPv6 可解析但不可达的场景 |
| `tungstenite` | 更新 WASM 下 `permessage-deflate` 的压缩后端、最低 Rust 版本及代理代码整理 | 对本服务 Linux 环境的帧解析、Ping/Pong 和连接存活语义没有实质变化 |
| Codex `0.143.0` | 加入 GPT-5.6 模型；将 `max` 作为正式推理强度；增量 WebSocket 请求比较时忽略 response metadata | 影响模型能力声明和客户端是否选择增量请求，不代表上游改变了 `previous_response_id` 的失效规则 |
| Codex `0.144.0` | Responses WebSocket 继续走低延迟通道，同时支持系统代理和自定义 CA | 主要改变 Codex 客户端的建连路径 |
| Codex `0.144.1` | 安装器和 Code Mode 可靠性修复 | 没有新的 Responses 或 WebSocket 协议变化 |
参考：[Codex 0.143.0](https://github.com/openai/codex/releases/tag/rust-v0.143.0)、[Codex 0.144.0](https://github.com/openai/codex/releases/tag/rust-v0.144.0)、[Codex 0.144.1](https://github.com/openai/codex/releases/tag/rust-v0.144.1)。
同时核对了官方客户端的续接实现：
1. `previous_response_id` 由当前 WebSocket session 的上一条响应生成，只用于增量请求；
2. 检测到连接关闭并新建 WebSocket 前，官方客户端会清除 `last_request` 和 `last_response`；
3. 新连接发送完整请求，不会把旧连接缓存中的 response ID 继续带过去；
4. 官方实现没有针对 `previous_response_not_found` 的专门 400 重试分支，安全性来自新连接前主动清理增量状态。
参考：[增量请求构造](https://github.com/openai/codex/blob/rust-v0.144.1/codex-rs/core/src/client.rs#L1223-L1254)、[新连接状态清理](https://github.com/openai/codex/blob/rust-v0.144.1/codex-rs/core/src/client.rs#L1323-L1354)。
本项目已经对齐的部分包括 `responses_websockets=2026-02-06` beta header、自定义 CA 和 `permessage-deflate`。尚未完全对齐的是：底层 transport 创建新 WebSocket 后仍会原样发送已经准备好的 payload，其中可能包含旧 `previous_response_id`；`v1.0.11` 依靠上层的一次 history recovery 事后补救。
### 2. 问题原因
这两个旧依赖不是本次 `previous_response_not_found`、`stream closed before response.completed` 或 15 分钟后连接失效的直接原因：
- 新版 `tokio-tungstenite` 只改善 DNS 解析后的 TCP 建连竞争，不能延长已建立 WebSocket 的上游 response 缓存寿命；
- 新版 `tungstenite` 没有增加主动 Ping、Pong 超时或断流后的业务请求恢复；
- 官方 Codex `0.144.1` 也没有周期性主动 Ping；它在读取期间处理 Ping/Pong，并依靠 idle timeout 和断线后的新连接状态重置；
- `previous_response_not_found` 的关键差异仍是连接状态管理：官方新连接不继承旧增量状态，本项目目前会先尝试发送旧 ID，再依赖上层识别 400 后恢复；
- 智能调度只有在切换账号或打破会话亲和性时才会放大问题，本次日志已确认没有切换账号，因此不是根因。
因此，依赖升级能够降低部分“建连或重连失败”，但不能单独解决当前红色断流错误。
### 3. 修复方案
1. **优先对齐连接状态语义**：将 `previous_response_id` 与上游账号、具体 WebSocket session 共同绑定。连接已失效时，不能直接在新连接上发送旧 ID。
2. **安全重建请求**：如果代理持有完整历史，新连接应清除旧 ID 并重放完整 input；如果只有客户端增量输入，必须返回明确的续接失效错误，不能把残缺输入伪装成无损恢复。
3. **更新两个依赖 pin**：将两个 rev 更新到 Codex `0.144.1` 使用的版本，并在同一次变更中更新。新版 `tokio-tungstenite` 自身已经依赖新的 `tungstenite` pin，分开更新可能引入重复依赖和类型不一致。
4. **保留独立的保活修复**：主动 Ping、Pong deadline、连接 idle 生命周期和业务层重试应继续作为本项目自己的连接池策略，不能依赖本次库升级自动解决。
5. **升级后验证**：覆盖 IPv4/IPv6 建连、custom CA、`permessage-deflate`、空闲 15 分钟后复用、旧连接断开后完整历史重放，以及 `previous_response_not_found` 在首输出前后的两种时序。
依赖升级优先级低于新连接状态修复：可以升级，但不能把它作为现有 400 和流式断连问题的唯一修复方案。