From 61cdaf753e3066b3dab77c7f045c2bb0cd384bdf Mon Sep 17 00:00:00 2001 From: Zyy Date: Fri, 10 Jul 2026 20:30:40 +0800 Subject: [PATCH] fix: harden responses websocket recovery Co-authored-by: Codex --- backend/Cargo.lock | 23 +- backend/Cargo.toml | 4 +- backend/src/api/client/errors.rs | 7 + backend/src/bootstrap/config.rs | 2 +- backend/src/dispatch/errors.rs | 2 + backend/src/dispatch/responses/errors.rs | 11 +- backend/src/dispatch/responses/live_stream.rs | 6 +- backend/src/dispatch/responses/service.rs | 42 ++-- .../dispatch/responses/service/complete.rs | 60 +++-- .../src/dispatch/responses/service/stream.rs | 97 +++++--- .../src/upstream/openai/protocol/responses.rs | 8 +- .../src/upstream/openai/protocol/websocket.rs | 11 - .../openai/transport/response_meta.rs | 4 +- .../upstream/openai/transport/websocket.rs | 22 +- .../openai/transport/websocket/stream.rs | 159 ++++++++---- .../openai/transport/websocket/types.rs | 10 +- .../openai/transport/websocket_pool.rs | 4 +- backend/tests/dispatch/chat_upstream/mod.rs | 134 +++++----- .../dispatch/chat_upstream/responses_http.rs | 24 +- .../chat_upstream/responses_recovery.rs | 228 ++++++++---------- .../chat_upstream/responses_websocket.rs | 130 ++++------ backend/tests/upstream/openai/protocol/mod.rs | 2 +- .../openai/protocol/response_first_event.rs | 40 +-- .../upstream/openai/transport/websocket.rs | 4 +- .../openai/transport/websocket_pool.rs | 6 +- docs/stream-disconnection-remediation.md | 224 +++++++++++++++++ 26 files changed, 786 insertions(+), 478 deletions(-) create mode 100644 docs/stream-disconnection-remediation.md diff --git a/backend/Cargo.lock b/backend/Cargo.lock index 79fe04aef..a0818093b 100644 --- a/backend/Cargo.lock +++ b/backend/Cargo.lock @@ -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" diff --git a/backend/Cargo.toml b/backend/Cargo.toml index 98405de14..77eeec89d 100644 --- a/backend/Cargo.toml +++ b/backend/Cargo.toml @@ -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 } diff --git a/backend/src/api/client/errors.rs b/backend/src/api/client/errors.rs index 311aa0ada..b1c461ea0 100644 --- a/backend/src/api/client/errors.rs +++ b/backend/src/api/client/errors.rs @@ -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, diff --git a/backend/src/bootstrap/config.rs b/backend/src/bootstrap/config.rs index ce56f9296..0a7a973ea 100644 --- a/backend/src/bootstrap/config.rs +++ b/backend/src/bootstrap/config.rs @@ -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` 等纯生命周期帧不计）。 diff --git a/backend/src/dispatch/errors.rs b/backend/src/dispatch/errors.rs index cca448354..221b65d0f 100644 --- a/backend/src/dispatch/errors.rs +++ b/backend/src/dispatch/errors.rs @@ -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", diff --git a/backend/src/dispatch/responses/errors.rs b/backend/src/dispatch/responses/errors.rs index fc96d7299..aefac7b77 100644 --- a/backend/src/dispatch/responses/errors.rs +++ b/backend/src/dispatch/responses/errors.rs @@ -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 => { diff --git a/backend/src/dispatch/responses/live_stream.rs b/backend/src/dispatch/responses/live_stream.rs index 2ccf0176a..eb60ee2e0 100644 --- a/backend/src/dispatch/responses/live_stream.rs +++ b/backend/src/dispatch/responses/live_stream.rs @@ -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, diff --git a/backend/src/dispatch/responses/service.rs b/backend/src/dispatch/responses/service.rs index 4573de2a2..060f3cd2e 100644 --- a/backend/src/dispatch/responses/service.rs +++ b/backend/src/dispatch/responses/service.rs @@ -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, - ) { + 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, 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, + snapshot: ImplicitResumeSnapshot, ) { - if let Some(snapshot) = implicit_resume.take() { - snapshot.restore(request); - } + snapshot.restore(request); strip_request_history(request); } diff --git a/backend/src/dispatch/responses/service/complete.rs b/backend/src/dispatch/responses/service/complete.rs index 547edc597..8a2cb5f41 100644 --- a/backend/src/dispatch/responses/service/complete.rs +++ b/backend/src/dispatch/responses/service/complete.rs @@ -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 = 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, diff --git a/backend/src/dispatch/responses/service/stream.rs b/backend/src/dispatch/responses/service/stream.rs index 7d12924d8..2aa9e3843 100644 --- a/backend/src/dispatch/responses/service/stream.rs +++ b/backend/src/dispatch/responses/service/stream.rs @@ -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 = 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)); } } diff --git a/backend/src/upstream/openai/protocol/responses.rs b/backend/src/upstream/openai/protocol/responses.rs index 9e26ca5ff..692f852ad 100644 --- a/backend/src/upstream/openai/protocol/responses.rs +++ b/backend/src/upstream/openai/protocol/responses.rs @@ -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, ) { - 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)); } } diff --git a/backend/src/upstream/openai/protocol/websocket.rs b/backend/src/upstream/openai/protocol/websocket.rs index d3cf250df..2f2378e2c 100644 --- a/backend/src/upstream/openai/protocol/websocket.rs +++ b/backend/src/upstream/openai/protocol/websocket.rs @@ -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 { let value = serde_json::from_str::(raw).ok()?; diff --git a/backend/src/upstream/openai/transport/response_meta.rs b/backend/src/upstream/openai/transport/response_meta.rs index d63f7458e..f2f511cbd 100644 --- a/backend/src/upstream/openai/transport/response_meta.rs +++ b/backend/src/upstream/openai/transport/response_meta.rs @@ -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, ) { - 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 { diff --git a/backend/src/upstream/openai/transport/websocket.rs b/backend/src/upstream/openai/transport/websocket.rs index a47e49fc0..c33be4532 100644 --- a/backend/src/upstream/openai/transport/websocket.rs +++ b/backend/src/upstream/openai/transport/websocket.rs @@ -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, diff --git a/backend/src/upstream/openai/transport/websocket/stream.rs b/backend/src/upstream/openai/transport/websocket/stream.rs index 1e7e8df07..1fc684c18 100644 --- a/backend/src/upstream/openai/transport/websocket/stream.rs +++ b/backend/src/upstream/openai/transport/websocket/stream.rs @@ -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 = 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, ) -> Result { - 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, ) -> Result { @@ -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, + 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; } diff --git a/backend/src/upstream/openai/transport/websocket/types.rs b/backend/src/upstream/openai/transport/websocket/types.rs index 9b1e0e07a..ecaef922c 100644 --- a/backend/src/upstream/openai/transport/websocket/types.rs +++ b/backend/src/upstream/openai/transport/websocket/types.rs @@ -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, diff --git a/backend/src/upstream/openai/transport/websocket_pool.rs b/backend/src/upstream/openai/transport/websocket_pool.rs index 59544d039..4dfed24d3 100644 --- a/backend/src/upstream/openai/transport/websocket_pool.rs +++ b/backend/src/upstream/openai/transport/websocket_pool.rs @@ -88,7 +88,7 @@ pub struct CodexWebSocketPoolConfig { pub ping_timeout: Duration, /// idle socket 无活动多久后视为失活。 pub liveness_timeout: Option, - /// 首个内容帧到达前的绝对超时；`None` 表示禁用首 token 熔断。 + /// 首个真实输出到达前的绝对超时；`None` 表示禁用首 token 熔断。 pub first_token_timeout: Option, } @@ -147,7 +147,7 @@ impl CodexWebSocketPool { self.config.keepalive() } - /// 首个内容帧到达前的绝对超时；`None` 表示禁用首 token 熔断。 + /// 首个真实输出到达前的绝对超时；`None` 表示禁用首 token 熔断。 pub(crate) fn first_token_timeout(&self) -> Option { self.config.first_token_timeout } diff --git a/backend/tests/dispatch/chat_upstream/mod.rs b/backend/tests/dispatch/chat_upstream/mod.rs index 65f1cfb09..bba69b68e 100644 --- a/backend/tests/dispatch/chat_upstream/mod.rs +++ b/backend/tests/dispatch/chat_upstream/mod.rs @@ -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, tokio::task::JoinHandle) { + 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::(&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) { - 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::(&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::(&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, + response_messages: Vec, +) -> Value { + let followup = tokio::select! { + message = websocket.next() => { + match message { + Some(Ok(message)) if message.is_text() => { + let payload = serde_json::from_str::(&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::(&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, + 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, response_message: String, diff --git a/backend/tests/dispatch/chat_upstream/responses_http.rs b/backend/tests/dispatch/chat_upstream/responses_http.rs index f4ad92fc5..042d5b742 100644 --- a/backend/tests/dispatch/chat_upstream/responses_http.rs +++ b/backend/tests/dispatch/chat_upstream/responses_http.rs @@ -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()); } diff --git a/backend/tests/dispatch/chat_upstream/responses_recovery.rs b/backend/tests/dispatch/chat_upstream/responses_recovery.rs index caf7e45d3..7804a7af1 100644 --- a/backend/tests/dispatch/chat_upstream/responses_recovery.rs +++ b/backend/tests/dispatch/chat_upstream/responses_recovery.rs @@ -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] diff --git a/backend/tests/dispatch/chat_upstream/responses_websocket.rs b/backend/tests/dispatch/chat_upstream/responses_websocket.rs index a08cf43bb..0413b9283 100644 --- a/backend/tests/dispatch/chat_upstream/responses_websocket.rs +++ b/backend/tests/dispatch/chat_upstream/responses_websocket.rs @@ -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")); diff --git a/backend/tests/upstream/openai/protocol/mod.rs b/backend/tests/upstream/openai/protocol/mod.rs index f5a5606b6..03c3a4f11 100644 --- a/backend/tests/upstream/openai/protocol/mod.rs +++ b/backend/tests/upstream/openai/protocol/mod.rs @@ -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, }; diff --git a/backend/tests/upstream/openai/protocol/response_first_event.rs b/backend/tests/upstream/openai/protocol/response_first_event.rs index d32472568..b16e81e6c 100644 --- a/backend/tests/upstream/openai/protocol/response_first_event.rs +++ b/backend/tests/upstream/openai/protocol/response_first_event.rs @@ -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)); } diff --git a/backend/tests/upstream/openai/transport/websocket.rs b/backend/tests/upstream/openai/transport/websocket.rs index e1b6d03fb..6bbf74672 100644 --- a/backend/tests/upstream/openai/transport/websocket.rs +++ b/backend/tests/upstream/openai/transport/websocket.rs @@ -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)); diff --git a/backend/tests/upstream/openai/transport/websocket_pool.rs b/backend/tests/upstream/openai/transport/websocket_pool.rs index d522014f7..ce8774b9f 100644 --- a/backend/tests/upstream/openai/transport/websocket_pool.rs +++ b/backend/tests/upstream/openai/transport/websocket_pool.rs @@ -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)); diff --git a/docs/stream-disconnection-remediation.md b/docs/stream-disconnection-remediation.md new file mode 100644 index 000000000..ea942ba35 --- /dev/null +++ b/docs/stream-disconnection-remediation.md @@ -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`，因为这是上游实际收到的推理强度。 +- 从请求中的 `` 识别主动委派，新增 `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 和流式断连问题的唯一修复方案。