diff --git a/docs/OPENAI_CHAIN_PARITY_REAUDIT.md b/docs/OPENAI_CHAIN_PARITY_REAUDIT.md index 00754f363..dc9bfe38b 100644 --- a/docs/OPENAI_CHAIN_PARITY_REAUDIT.md +++ b/docs/OPENAI_CHAIN_PARITY_REAUDIT.md @@ -27,7 +27,7 @@ Rust 仓库：`/home/zyy/Codes/codex-proxy-rs` 最高优先级未对齐项： - 安全模拟仍缺 TLS 指纹证明；WS permessage-deflate offer、协商响应和服务端压缩 frame 解码已补齐测试与实现。 -- response 转换和错误恢复仍缺原版级集中状态机；tuple schema request conversion/reconvert、streaming premature close 合成、reasoning replay、implicit resume failure restore、`response.failed`/`error` 的 429/402/403 分类、账号 fallback、显式 `previous_response_id` 的 429 fallback、`previous_response_not_found`/unanswered function call strip-and-retry、5xx same-account retry、CF path-block 404、model unsupported fallback、401 token invalid fallback、Responses/compact 最终错误外壳已补齐。 +- response 转换和错误恢复仍缺完整 golden 覆盖与副作用 transition 集中化；tuple schema request conversion/reconvert、streaming premature close 合成、reasoning replay、implicit resume failure restore、`response.failed`/`error` 的 429/402/403 分类、账号 fallback、显式 `previous_response_id` 的 429 fallback、`previous_response_not_found`/unanswered function call strip-and-retry、5xx same-account retry、CF path-block 404、model unsupported fallback、401 token invalid fallback、Responses/compact 最终错误外壳、集中 `UpstreamRecoveryAction` 分类入口已补齐。 明确不纳入缺口：IP 代理、VPN、本地代理探测、`HttpsProxyAgent`、账号代理池。 明确不移植：非 OpenAI upstream family 的兼容路由、translator、cache usage hint 估算及其 `clientConversationId` 来源不属于本项目目标；后续只按 OpenAI `/v1/responses`、`/v1/responses/compact` 与 `/v1/chat/completions` 链路核对。 @@ -371,9 +371,11 @@ Rust 证据： - 本轮修正后，`src/codex/serving/responses.rs` 会复用第一次 SSE collect 结果生成最终响应，不再对同一个 upstream body 二次解析。 - 本轮修正后，`src/codex/serving/http/errors.rs` 增加 Responses 专用错误外壳：普通上游错误返回 `type:error` + `server_error/codex_api_error`，429 返回 `rate_limit_error/rate_limit_exceeded`，无可用账号返回 `no_available_accounts`；`src/codex/serving/responses.rs` 的 `/v1/responses` 与 `/v1/responses/compact` 非流式错误路径已切到该外壳，Chat 仍保留 OpenAI Chat 错误格式。 - 本轮修正后，`src/codex/serving/dispatch/fallback.rs` 增加 request-level recovery 分类，识别 `previous_response_not_found` 和 `No tool output found for function call`；`src/codex/serving/responses.rs`、`src/codex/serving/dispatch/mod.rs` 在账号 fallback 前先清除 stale `previous_response_id`/`turn_state` 并同账号重试一次。 +- 本轮修正后，`src/codex/serving/dispatch/fallback.rs` 增加集中 `UpstreamRecoveryAction` 与 `UpstreamRecoveryState` 分类入口，优先级为 request recovery，再 account retry，最后返回错误；`src/codex/serving/responses.rs`、HTTP SSE streaming 和 WebSocket streaming 的请求建立/collect 错误路径已改为复用该入口。 - `tests/codex_serving/upstream_errors.rs:67-194` 覆盖 non-streaming `response.failed` 按 `rate_limit_exceeded` 返回 429、streaming 透传 `response.failed` 并记录失败审计。 - 本轮修正后，`tests/codex_serving/upstream_errors.rs` 覆盖 non-streaming `error` 事件返回 `codex_api_error`，`response.failed` 429 返回 `rate_limit_exceeded`；`tests/codex_serving/responses_http_sse.rs` 覆盖 `/v1/responses` 和 `/v1/responses/compact` 无可用账号返回 `no_available_accounts`，以及 compact fallback exhausted 429 返回 `rate_limit_exceeded`。 - 本轮新增 `tests/fixtures/responses/http_sse/response_failed_quota.sse` 与 `tests/codex_serving/upstream_fallback.rs::v1_responses_should_classify_non_stream_sse_failure_and_retry_next_account`，覆盖 non-streaming mid-SSE `quota_exceeded` 会标记当前账号 `quota_exhausted` 并 fallback 到下一账号。 +- 本轮新增 `tests/fixtures/responses/http_sse/previous_response_not_found.sse`、`tests/fixtures/responses/http_sse/unanswered_function_call.sse` 与 `tests/codex_serving/upstream_fallback.rs::v1_responses_should_strip_history_after_http_sse_previous_response_not_found`、`v1_responses_should_strip_history_after_http_sse_unanswered_function_call`，覆盖 HTTP SSE non-streaming collect 阶段收到 stale-history 类 `error` 事件后会走同账号 strip-and-retry。 - 本轮新增 `tests/fixtures/responses/websocket/previous_response_not_found.json`、`tests/fixtures/responses/websocket/unanswered_function_call.json`，并用 `tests/codex_serving/responses_websocket.rs` 覆盖 non-streaming 与 streaming `previous_response_not_found`、non-streaming unanswered function call 的同账号 strip-and-retry。 - `tests/codex_gateway/usage_events.rs` 覆盖 multiline data、非标准 JSON 续行、`[DONE]` 忽略和非 SSE JSON body 转 `non_sse_response`。 - `tests/codex_gateway/websocket.rs:376-760` 覆盖 WS 内部 rate-limit event 不转发、首个 rotatable error、mid-stream close error 和 fallback。 @@ -389,13 +391,15 @@ Rust 证据： - streaming 路径对客户端保持 SSE 格式，HTTP SSE 与 WebSocket 两种上游传输都能以 SSE 返回给下游。 - `response.failed`/`error` 在 streaming 路径会被透传并写入审计；non-streaming 路径也能识别为上游失败，并已对 429/402/403 进入账号状态机和 fallback。 - request-level `previous_response_not_found` 与 unanswered function call 已按原版优先于账号 fallback 处理：忘记 stale affinity，清除 `previous_response_id`/`turn_state`，同账号重试一次。 +- retry/recovery 分类优先级已集中为 `UpstreamRecoveryAction`：request recovery 优先于 account retry，model unsupported 只允许一次，再进入最终错误响应。 +- HTTP SSE non-streaming collect 阶段的 `previous_response_not_found` 与 unanswered function call 已有端到端覆盖，确认 mid-SSE error event 会复用同账号 strip-and-retry。 - streaming premature close 已按原版对齐：HTTP SSE 正常 EOF、HTTP SSE body stream error、WebSocket body stream error 在缺少 terminal event 时都会合成 `event: response.failed`，错误码 `stream_disconnected`，并进入现有 stream 审计失败记录。 - 单个未解析 SSE event buffer 已按原版限制为 64 MiB，避免异常上游响应在解析阶段无界增长。 - OpenAI Responses 与 OpenAI Chat 的 tuple schema request conversion/reconvert 已对齐原版，不涉及非 OpenAI translator。 - Responses pooled path 的非流式最终错误外壳已按原版 `PASSTHROUGH_FORMAT` 对齐：普通上游错误为 `codex_api_error`，429 为 `rate_limit_exceeded`，无账号为 `no_available_accounts`；compact 路径复用同一 Responses 外壳。 未完全对齐： -- mid-SSE/request failure 尚未完全覆盖原版恢复集合。Rust 已补齐 429/402/403 分类与 fallback、401 token invalid fallback，以及 `previous_response_not_found`/unanswered function call strip-and-retry。 +- response/SSE/WS 转换链路仍缺完整 golden fixture 矩阵。主恢复集合已覆盖 429/402/403、401 token invalid、model unsupported、`previous_response_not_found`/unanswered function call、premature close 和最终错误外壳，但还需要更系统的 golden tests 防止格式细节回退。 缺口/后续动作： - 增加 golden tests：标准 SSE、漂亮打印 JSON 续行、非 SSE body、premature close、`response.failed` auth recovery、reasoning replay、WS frame to SSE chunk。 @@ -418,8 +422,9 @@ Rust 证据： - `src/codex/serving/dispatch/fallback.rs:331-446` 对 429 设置 quota cooldown 并记录 request attempt；402 设置 `QuotaExhausted`；Cloudflare challenge 删除 cookies 并设置 persisted cooldown；空 body 404 清理当前账号 cookies、记录 1 小时滑窗内 path-block 次数，连续 3 次设置 `Disabled`；401 token invalid 设置 `Expired`，401 deactivated 设置 `Banned`；403 ban 设置 `Banned`；状态变化会 evict WS pool。 - 本轮修正后，`src/codex/serving/dispatch/account_refresh.rs` 已删除；serving request path 不再对上游 401 做同账号 OAuth refresh，refresh 仍保留在账号健康检查、后台调度和 admin/manual refresh 链路。 - 本轮修正后，`src/codex/serving/dispatch/mod.rs` 在 HTTP SSE/WS streaming 请求建立前，遇到可分类错误会尝试 acquire fallback account；显式 `previous_response_id` 仍保持 WebSocket required，不会降级 HTTP SSE，但 429/402/403/401/model unsupported 等账号级错误会按原版通用 error handler 尝试备用账号。 -- 本轮修正后，`src/codex/serving/responses.rs` non-streaming 请求前错误与 mid-SSE `error`/`response.failed` 都会先尝试 request-level recovery，再进入 `classify_upstream_account_retry()`；显式 `previous_response_id` 遇到可分类账号错误不再有专用 hold 分支，会进入同一 fallback 状态机。 +- 本轮修正后，`src/codex/serving/responses.rs` non-streaming 请求前错误与 mid-SSE `error`/`response.failed` 都会进入集中 `UpstreamRecoveryAction` 分类；显式 `previous_response_id` 遇到可分类账号错误不再有专用 hold 分支，会进入同一 fallback 状态机。 - 本轮修正后，`src/codex/serving/dispatch/mod.rs` 的 HTTP SSE streaming 与 WebSocket streaming 请求建立失败也会先尝试 request-level recovery，`previous_response_not_found` 可同账号 strip-and-retry。 +- 本轮修正后，`src/codex/serving/dispatch/fallback.rs` 抽出集中 `UpstreamRecoveryAction` 分类入口，单元测试覆盖 request recovery 优先于 account retry、request recovery 只使用一次后委派给账号 retry、model unsupported 只重试一次；`src/codex/serving/responses.rs`、HTTP SSE streaming 和 WebSocket streaming 请求建立/collect 错误路径已复用该入口。 - 本轮修正后，`src/codex/serving/dispatch/mod.rs` 在 non-streaming、HTTP SSE streaming、WebSocket streaming 的上游 request attempt 外层加入 `retry_upstream_5xx()`：仅对 `CodexClientError::Upstream` 的 5xx 做同账号最多 2 次重试，退避为 1000ms、2000ms，与原版 `withRetry()` 默认策略一致。 - 本轮修正后，`src/codex/serving/dispatch/fallback.rs` 识别 4xx `model not supported`、`model_not_supported`、`model not available`、`model_not_available`，只允许一次备用账号 retry，不标记账号状态，也不驱逐 WebSocket pool。 - 本轮修正后，`src/codex/serving/dispatch/stream.rs` 会把 non-streaming collect 得到的 `response.failed` `model_not_supported`/`model_not_available` 合成为 400 上游错误，使它进入同一 model fallback 状态机。 @@ -427,6 +432,7 @@ Rust 证据： - `src/codex/accounts/cloudflare_challenge.rs:1-120` 实现了 CF path-block tracker，连续次数超过 1 小时会重置，阈值为 3。 - `tests/codex_serving/upstream_fallback.rs:40-1400` 覆盖 429 fallback、streaming 429 fallback、402 标记并 fallback、403 ban、Cloudflare challenge cooldown/cookie 清理、Cloudflare path-block 404 清 cookies 并 fallback、连续 3 次 path-block 后禁用账号、401 token invalid 标记 expired 后 fallback、401 deactivated 标记 banned、无 fallback 时返回 401。 - `tests/codex_serving/responses_websocket.rs` 覆盖 WebSocket 非历史请求先因 429 fallback 到备用账号，再因 401 将备用账号标记 expired 并返回 401，不做同账号 refresh。 +- 本轮新增 `tests/fixtures/responses/http_sse/previous_response_not_found.sse`、`tests/fixtures/responses/http_sse/unanswered_function_call.sse` 与 `tests/codex_serving/upstream_fallback.rs` 覆盖 HTTP SSE non-streaming mid-SSE stale-history 类错误复用同账号 strip-and-retry。 - 本轮新增 `tests/fixtures/responses/http_sse/after_model_unsupported_retry.sse`、`tests/fixtures/responses/http_sse/response_failed_model_unsupported.sse`，并用 `tests/codex_serving/upstream_fallback.rs` 覆盖 HTTP-time model unsupported fallback、non-streaming mid-SSE model unsupported fallback、以及 model unsupported 最多只重试一次。 - 本轮新增 `tests/fixtures/responses/http_sse/after_5xx_retry.sse`、`tests/fixtures/responses/http_sse/stream_after_5xx_retry.sse`，并用 `tests/codex_serving/upstream_fallback.rs` 覆盖 non-streaming 与 streaming 两条路径在同账号连续 2 次 5xx 后第三次成功，且不会切换到备用账号。 - `tests/codex_serving/responses_websocket.rs` 覆盖 `previous_response_id` WS 429 fallback 到备用账号且继续携带同一 `previous_response_id`，并覆盖非历史 WS fallback 和 fallback 账户 refresh。 @@ -446,14 +452,13 @@ Rust 证据： - Cloudflare challenge fallback exhausted status/message 已对齐原版：上游 403 challenge 会作为 retry decision 暴露 502 和 `Upstream blocked the request (Cloudflare challenge)`，无备用账号时返回该 502 语义；`v1_responses_should_return_502_when_cloudflare_challenge_has_no_fallback` 覆盖单账号无 fallback 场景。 - Rust 的 Cloudflare challenge 处理比原版更持久：会把 cooldown 写入数据库，并清理对应账号 cookies。 - 显式 `previous_response_id` 的账号级错误恢复已对齐原版：不降级 HTTP SSE，仍通过 WebSocket 向备用账号重试；若备用账号无法识别该 server-side history，再由 `previous_response_not_found` strip-and-retry 恢复。 +- retry 分类优先级已集中：`UpstreamRecoveryAction` 先处理 implicit/stale history recovery，再处理账号级 retry/fallback，最后返回错误，避免 request recovery 与 account retry 在不同入口各自判断。 未完全对齐： -- recovery 优先级仍不是完整原版状态机。Rust 已实现 implicit resume restore/replay 与 strip-and-retry 优先于账号 fallback，但状态机仍分散在 responses/dispatch 中，没有抽成原版同等的 `RetryAction` 分层。 -- non-streaming mid-SSE failure 仍不是完整原版 catch 状态机。429/402/403、401、model unsupported、400 strip-and-retry、隐式续链 restore 与最终错误外壳已补齐；剩余主要是状态机分层仍分散。 +- recovery 副作用 transition 仍分散在 non-streaming、HTTP SSE streaming、WebSocket streaming 和 compact 各自分支中；本轮已集中分类优先级，但 release、fallback acquire、response formatting 等副作用还没有抽成原版同等的 transition 层。 缺口/后续动作： -- 引入原版级 `UpstreamRecoveryAction` 状态机：implicit resume replay、strip-and-retry、error handler、fallback transition 分层清晰，避免分散在 responses/dispatch 中。 -- 继续补 HTTP SSE mid-SSE failure 的 `previous_response_not_found` 与 unanswered function call 覆盖；当前已有 WS 首帧错误覆盖 streaming/non-streaming。 +- 继续抽出 recovery transition 层：把 release、restore implicit resume、fallback acquire、account exhaustion response 组合成明确的执行单元，进一步贴近原版 `proxy-error-retry-transition.ts` / `proxy-fallback-account-retry.ts`。 ## 9. rate-limit、usage、quota、cookie 持久化 diff --git a/src/codex/serving/dispatch/fallback.rs b/src/codex/serving/dispatch/fallback.rs index f6bf72aeb..f3b24afd1 100644 --- a/src/codex/serving/dispatch/fallback.rs +++ b/src/codex/serving/dispatch/fallback.rs @@ -16,7 +16,7 @@ const DEFAULT_RATE_LIMIT_BACKOFF_SECONDS: u64 = 60; const MAX_RATE_LIMIT_BACKOFF_SECONDS: u64 = 86_400 * 7; const CLOUDFLARE_CHALLENGE_COOLDOWN_SECONDS: u64 = 10; -#[derive(Debug, Clone, Copy)] +#[derive(Debug, Clone, Copy, PartialEq, Eq)] pub(crate) enum UpstreamAccountRetry { ModelUnsupported { status: StatusCode }, RateLimited { retry_after_seconds: u64 }, @@ -32,6 +32,19 @@ pub(crate) enum UpstreamRequestRecovery { StripPreviousResponse { reason: HistoryRecoveryReason }, } +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) enum UpstreamRecoveryAction { + Request(UpstreamRequestRecovery), + Account(UpstreamAccountRetry), + RespondWithError, +} + +#[derive(Debug, Clone, Copy)] +pub(crate) struct UpstreamRecoveryState { + pub(crate) request_recovery_used: bool, + pub(crate) model_unsupported_retry_used: bool, +} + #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub(crate) enum HistoryRecoveryReason { PreviousResponseNotFound, @@ -155,6 +168,20 @@ pub(crate) fn classify_upstream_request_recovery( None } +pub(crate) fn classify_upstream_recovery_action( + error: &CodexClientError, + state: UpstreamRecoveryState, +) -> UpstreamRecoveryAction { + if let Some(recovery) = classify_upstream_request_recovery(error, state.request_recovery_used) { + return UpstreamRecoveryAction::Request(recovery); + } + if let Some(retry) = classify_upstream_account_retry(error, state.model_unsupported_retry_used) + { + return UpstreamRecoveryAction::Account(retry); + } + UpstreamRecoveryAction::RespondWithError +} + pub(crate) fn classify_upstream_account_retry( error: &CodexClientError, model_unsupported_retry_used: bool, @@ -476,3 +503,102 @@ async fn set_account_status( .await .set_status(&account.id, status); } + +#[cfg(test)] +mod tests { + use super::*; + + fn upstream_error(status: StatusCode, code: &str, message: &str) -> CodexClientError { + CodexClientError::Upstream { + status, + retry_after_seconds: None, + body: json!({ + "error": { + "code": code, + "message": message, + } + }) + .to_string(), + } + } + + #[test] + fn classify_upstream_recovery_action_should_prioritize_request_recovery() { + let error = upstream_error( + StatusCode::TOO_MANY_REQUESTS, + "previous_response_not_found", + "Previous response with id resp_missing was not found", + ); + + let action = classify_upstream_recovery_action( + &error, + UpstreamRecoveryState { + request_recovery_used: false, + model_unsupported_retry_used: false, + }, + ); + + assert_eq!( + action, + UpstreamRecoveryAction::Request(UpstreamRequestRecovery::StripPreviousResponse { + reason: HistoryRecoveryReason::PreviousResponseNotFound, + }) + ); + } + + #[test] + fn classify_upstream_recovery_action_should_delegate_after_request_recovery_is_used() { + let error = upstream_error( + StatusCode::TOO_MANY_REQUESTS, + "previous_response_not_found", + "Previous response with id resp_missing was not found", + ); + + let action = classify_upstream_recovery_action( + &error, + UpstreamRecoveryState { + request_recovery_used: true, + model_unsupported_retry_used: false, + }, + ); + + assert_eq!( + action, + UpstreamRecoveryAction::Account(UpstreamAccountRetry::RateLimited { + retry_after_seconds: DEFAULT_RATE_LIMIT_BACKOFF_SECONDS, + }) + ); + } + + #[test] + fn classify_upstream_recovery_action_should_limit_model_retry_to_once() { + let error = upstream_error( + StatusCode::BAD_REQUEST, + "model_not_supported", + "Model gpt-5.5 is not supported on this account plan", + ); + + let first_action = classify_upstream_recovery_action( + &error, + UpstreamRecoveryState { + request_recovery_used: false, + model_unsupported_retry_used: false, + }, + ); + let second_action = classify_upstream_recovery_action( + &error, + UpstreamRecoveryState { + request_recovery_used: false, + model_unsupported_retry_used: true, + }, + ); + + assert_eq!( + first_action, + UpstreamRecoveryAction::Account(UpstreamAccountRetry::ModelUnsupported { + status: StatusCode::BAD_REQUEST, + }) + ); + assert_eq!(second_action, UpstreamRecoveryAction::RespondWithError); + } +} diff --git a/src/codex/serving/dispatch/mod.rs b/src/codex/serving/dispatch/mod.rs index 00cedb53f..e86b473e1 100644 --- a/src/codex/serving/dispatch/mod.rs +++ b/src/codex/serving/dispatch/mod.rs @@ -73,8 +73,8 @@ use crate::codex::serving::http::errors::{ pub(crate) use self::{ fallback::{ - classify_upstream_account_retry, classify_upstream_request_recovery, UpstreamAccountRetry, - UpstreamRequestRecovery, + classify_upstream_account_retry, classify_upstream_recovery_action, UpstreamAccountRetry, + UpstreamRecoveryAction, UpstreamRecoveryState, UpstreamRequestRecovery, }, routing::{no_available_accounts_response, normalize_service_tier_for_upstream}, stream::{completed_response_json, CollectedResponse}, @@ -801,68 +801,74 @@ async fn responses_http_sse_stream( Ok(response) => break response, Err(error) => { deps.account_pool.lock().await.release(&acquired.account.id); - if let Some(recovery) = - classify_upstream_request_recovery(&error, history_recovery_used) - { - apply_upstream_request_recovery_with_deps( - &deps, - &mut request, - recovery, - true, - &log_context, - &mut history_recovery_used, - &mut implicit_resume, - ) - .await; - continue; - } - if let Some(retry) = - classify_upstream_account_retry(&error, model_unsupported_retry_used) - { - if retry.is_model_unsupported() { - model_unsupported_retry_used = true; - } - let fallback = apply_upstream_retry_and_acquire_fallback_with_deps( - &deps, - &acquired.account, - retry, - &request.model, - &mut excluded_account_ids, - request.expects_image_generation(), - ) - .await; - log_codex_upstream_response_with_deps( - &deps, - &log_context, - retry.status(), - EventLevel::Warn, - "v1 responses stream 上游请求将使用备用账户重试", - retry.metadata(true), - ) - .await; - if let Some(fallback) = fallback { - log_context = log_context.with_account(&fallback.account.id); - acquired = fallback; + match classify_upstream_recovery_action( + &error, + UpstreamRecoveryState { + request_recovery_used: history_recovery_used, + model_unsupported_retry_used, + }, + ) { + UpstreamRecoveryAction::Request(recovery) => { + apply_upstream_request_recovery_with_deps( + &deps, + &mut request, + recovery, + true, + &log_context, + &mut history_recovery_used, + &mut implicit_resume, + ) + .await; continue; } - let retry_message = - retry.fallback_response_message(codex_client_error_message(&error)); - let message = fallback_exhausted_message_with_deps(&deps, &retry_message).await; - let error_response = codex_client_error_response_with_status_and_message( - error, - retry.status(), - &message, - ); - log_codex_upstream_response_with_deps( - &deps, - &log_context, - error_response.0, - EventLevel::Error, - "v1 responses stream fallback 已耗尽", - json!({"stream": true}), - ) - .await; - return error_response.into_response(); + UpstreamRecoveryAction::Account(retry) => { + if retry.is_model_unsupported() { + model_unsupported_retry_used = true; + } + let fallback = apply_upstream_retry_and_acquire_fallback_with_deps( + &deps, + &acquired.account, + retry, + &request.model, + &mut excluded_account_ids, + request.expects_image_generation(), + ) + .await; + log_codex_upstream_response_with_deps( + &deps, + &log_context, + retry.status(), + EventLevel::Warn, + "v1 responses stream 上游请求将使用备用账户重试", + retry.metadata(true), + ) + .await; + if let Some(fallback) = fallback { + log_context = log_context.with_account(&fallback.account.id); + acquired = fallback; + continue; + } + let retry_message = + retry.fallback_response_message(codex_client_error_message(&error)); + let message = + fallback_exhausted_message_with_deps(&deps, &retry_message).await; + let error_response = codex_client_error_response_with_status_and_message( + error, + retry.status(), + &message, + ); + log_codex_upstream_response_with_deps( + &deps, + &log_context, + error_response.0, + EventLevel::Error, + "v1 responses stream fallback 已耗尽", + json!({"stream": true}), + ) + .await; + return error_response.into_response(); + } + UpstreamRecoveryAction::RespondWithError => {} } record_failed_request_attempt_with_deps( &deps, @@ -1544,68 +1550,74 @@ async fn responses_websocket_stream( .await; } deps.account_pool.lock().await.release(&acquired.account.id); - if let Some(recovery) = - classify_upstream_request_recovery(&error, history_recovery_used) - { - apply_upstream_request_recovery_with_deps( - &deps, - &mut request, - recovery, - true, - &log_context, - &mut history_recovery_used, - &mut implicit_resume, - ) - .await; - continue; - } - if let Some(retry) = - classify_upstream_account_retry(&error, model_unsupported_retry_used) - { - if retry.is_model_unsupported() { - model_unsupported_retry_used = true; - } - let fallback = apply_upstream_retry_and_acquire_fallback_with_deps( - &deps, - &acquired.account, - retry, - &request.model, - &mut excluded_account_ids, - request.expects_image_generation(), - ) - .await; - log_codex_upstream_response_with_deps( - &deps, - &log_context, - retry.status(), - EventLevel::Warn, - "v1 responses WebSocket 上游请求将使用备用账户重试", - retry.metadata(true), - ) - .await; - if let Some(fallback) = fallback { - log_context = log_context.with_account(&fallback.account.id); - acquired = fallback; + match classify_upstream_recovery_action( + &error, + UpstreamRecoveryState { + request_recovery_used: history_recovery_used, + model_unsupported_retry_used, + }, + ) { + UpstreamRecoveryAction::Request(recovery) => { + apply_upstream_request_recovery_with_deps( + &deps, + &mut request, + recovery, + true, + &log_context, + &mut history_recovery_used, + &mut implicit_resume, + ) + .await; continue; } - let retry_message = - retry.fallback_response_message(codex_client_error_message(&error)); - let message = fallback_exhausted_message_with_deps(&deps, &retry_message).await; - let error_response = codex_client_error_response_with_status_and_message( - error, - retry.status(), - &message, - ); - log_codex_upstream_response_with_deps( - &deps, - &log_context, - error_response.0, - EventLevel::Error, - "v1 responses WebSocket stream fallback 已耗尽", - json!({"stream": true, "transport": "websocket"}), - ) - .await; - return error_response.into_response(); + UpstreamRecoveryAction::Account(retry) => { + if retry.is_model_unsupported() { + model_unsupported_retry_used = true; + } + let fallback = apply_upstream_retry_and_acquire_fallback_with_deps( + &deps, + &acquired.account, + retry, + &request.model, + &mut excluded_account_ids, + request.expects_image_generation(), + ) + .await; + log_codex_upstream_response_with_deps( + &deps, + &log_context, + retry.status(), + EventLevel::Warn, + "v1 responses WebSocket 上游请求将使用备用账户重试", + retry.metadata(true), + ) + .await; + if let Some(fallback) = fallback { + log_context = log_context.with_account(&fallback.account.id); + acquired = fallback; + continue; + } + let retry_message = + retry.fallback_response_message(codex_client_error_message(&error)); + let message = + fallback_exhausted_message_with_deps(&deps, &retry_message).await; + let error_response = codex_client_error_response_with_status_and_message( + error, + retry.status(), + &message, + ); + log_codex_upstream_response_with_deps( + &deps, + &log_context, + error_response.0, + EventLevel::Error, + "v1 responses WebSocket stream fallback 已耗尽", + json!({"stream": true, "transport": "websocket"}), + ) + .await; + return error_response.into_response(); + } + UpstreamRecoveryAction::RespondWithError => {} } record_failed_request_attempt_with_deps( &deps, diff --git a/src/codex/serving/responses.rs b/src/codex/serving/responses.rs index a201add0b..0fb2f6937 100644 --- a/src/codex/serving/responses.rs +++ b/src/codex/serving/responses.rs @@ -17,9 +17,10 @@ use crate::{ codex::models::service::ModelService, codex::serving::dispatch::{ affinity::SessionAffinityRepositoryResult, classify_upstream_account_retry, - classify_upstream_request_recovery, completed_response_json, + classify_upstream_recovery_action, completed_response_json, normalize_service_tier_for_upstream, CodexRequestLogContext, CodexUpstreamService, - CollectedResponse, ImplicitResumeSnapshot, UpstreamRequestRecovery, + CollectedResponse, ImplicitResumeSnapshot, UpstreamRecoveryAction, UpstreamRecoveryState, + UpstreamRequestRecovery, }, codex::serving::http::errors::{ codex_client_error_message, responses_codex_client_error_response, @@ -270,68 +271,74 @@ impl ResponsesService { .await; } let error = failure.upstream_error(); - if let Some(recovery) = - classify_upstream_request_recovery(&error, history_recovery_used) - { - self.apply_request_recovery( - &mut codex_request, - &mut history_recovery_used, - &mut implicit_resume, - &log_context, - recovery, - ) - .await; - continue; - } - if let Some(retry) = classify_upstream_account_retry( + match classify_upstream_recovery_action( &error, - model_unsupported_retry_used, + UpstreamRecoveryState { + request_recovery_used: history_recovery_used, + model_unsupported_retry_used, + }, ) { - if retry.is_model_unsupported() { - model_unsupported_retry_used = true; - } - let fallback = self - .upstream - .apply_retry_and_acquire_fallback( - &acquired.account, - retry, - &codex_request.model, - &mut excluded_account_ids, - codex_request.expects_image_generation(), - ) - .await; - self.upstream - .log_response( + UpstreamRecoveryAction::Request(recovery) => { + self.apply_request_recovery( + &mut codex_request, + &mut history_recovery_used, + &mut implicit_resume, &log_context, - retry.status(), - EventLevel::Warn, - "v1 responses 上游 SSE 失败将使用备用账户重试", - retry.metadata(false), + recovery, ) .await; - if let Some(fallback) = fallback { - log_context = log_context.with_account(&fallback.account.id); - acquired = fallback; continue; } - let retry_message = retry - .fallback_response_message(codex_client_error_message(&error)); - let message = self - .upstream - .fallback_exhausted_message(&retry_message) - .await; - let error_response = - responses_error_response(retry.status(), &message); - self.upstream - .log_response( - &log_context, - error_response.0, - EventLevel::Error, - "v1 responses 上游 SSE fallback 已耗尽", - failure.metadata(false), - ) - .await; - return error_response.into_response(); + UpstreamRecoveryAction::Account(retry) => { + if retry.is_model_unsupported() { + model_unsupported_retry_used = true; + } + let fallback = self + .upstream + .apply_retry_and_acquire_fallback( + &acquired.account, + retry, + &codex_request.model, + &mut excluded_account_ids, + codex_request.expects_image_generation(), + ) + .await; + self.upstream + .log_response( + &log_context, + retry.status(), + EventLevel::Warn, + "v1 responses 上游 SSE 失败将使用备用账户重试", + retry.metadata(false), + ) + .await; + if let Some(fallback) = fallback { + log_context = + log_context.with_account(&fallback.account.id); + acquired = fallback; + continue; + } + let retry_message = retry.fallback_response_message( + codex_client_error_message(&error), + ); + let message = self + .upstream + .fallback_exhausted_message(&retry_message) + .await; + let error_response = + responses_error_response(retry.status(), &message); + self.upstream + .log_response( + &log_context, + error_response.0, + EventLevel::Error, + "v1 responses 上游 SSE fallback 已耗尽", + failure.metadata(false), + ) + .await; + return error_response.into_response(); + } + UpstreamRecoveryAction::RespondWithError => {} } if self .upstream @@ -369,66 +376,71 @@ impl ResponsesService { break (response, collected_response); } Err(error) => { - if let Some(recovery) = - classify_upstream_request_recovery(&error, history_recovery_used) - { - self.apply_request_recovery( - &mut codex_request, - &mut history_recovery_used, - &mut implicit_resume, - &log_context, - recovery, - ) - .await; - continue; - } - if let Some(retry) = - classify_upstream_account_retry(&error, model_unsupported_retry_used) - { - if retry.is_model_unsupported() { - model_unsupported_retry_used = true; - } - let fallback = self - .upstream - .apply_retry_and_acquire_fallback( - &acquired.account, - retry, - &codex_request.model, - &mut excluded_account_ids, - codex_request.expects_image_generation(), - ) - .await; - self.upstream - .log_response( + match classify_upstream_recovery_action( + &error, + UpstreamRecoveryState { + request_recovery_used: history_recovery_used, + model_unsupported_retry_used, + }, + ) { + UpstreamRecoveryAction::Request(recovery) => { + self.apply_request_recovery( + &mut codex_request, + &mut history_recovery_used, + &mut implicit_resume, &log_context, - retry.status(), - EventLevel::Warn, - "v1 responses 上游请求将使用备用账户重试", - retry.metadata(false), + recovery, ) .await; - if let Some(fallback) = fallback { - log_context = log_context.with_account(&fallback.account.id); - acquired = fallback; continue; } - let retry_message = - retry.fallback_response_message(codex_client_error_message(&error)); - let message = self - .upstream - .fallback_exhausted_message(&retry_message) - .await; - let error_response = responses_error_response(retry.status(), &message); - self.upstream - .log_response( - &log_context, - error_response.0, - EventLevel::Error, - "v1 responses 上游请求 fallback 已耗尽", - json!({"stream": false}), - ) - .await; - return error_response.into_response(); + UpstreamRecoveryAction::Account(retry) => { + if retry.is_model_unsupported() { + model_unsupported_retry_used = true; + } + let fallback = self + .upstream + .apply_retry_and_acquire_fallback( + &acquired.account, + retry, + &codex_request.model, + &mut excluded_account_ids, + codex_request.expects_image_generation(), + ) + .await; + self.upstream + .log_response( + &log_context, + retry.status(), + EventLevel::Warn, + "v1 responses 上游请求将使用备用账户重试", + retry.metadata(false), + ) + .await; + if let Some(fallback) = fallback { + log_context = log_context.with_account(&fallback.account.id); + acquired = fallback; + continue; + } + let retry_message = + retry.fallback_response_message(codex_client_error_message(&error)); + let message = self + .upstream + .fallback_exhausted_message(&retry_message) + .await; + let error_response = responses_error_response(retry.status(), &message); + self.upstream + .log_response( + &log_context, + error_response.0, + EventLevel::Error, + "v1 responses 上游请求 fallback 已耗尽", + json!({"stream": false}), + ) + .await; + return error_response.into_response(); + } + UpstreamRecoveryAction::RespondWithError => {} } if self .upstream diff --git a/tests/codex_serving/upstream_fallback.rs b/tests/codex_serving/upstream_fallback.rs index 025271a34..592e92029 100644 --- a/tests/codex_serving/upstream_fallback.rs +++ b/tests/codex_serving/upstream_fallback.rs @@ -33,6 +33,10 @@ const SSE_RESPONSE_FAILED_QUOTA: &str = include_str!("../fixtures/responses/http_sse/response_failed_quota.sse"); const SSE_RESPONSE_FAILED_MODEL_UNSUPPORTED: &str = include_str!("../fixtures/responses/http_sse/response_failed_model_unsupported.sse"); +const SSE_PREVIOUS_RESPONSE_NOT_FOUND: &str = + include_str!("../fixtures/responses/http_sse/previous_response_not_found.sse"); +const SSE_UNANSWERED_FUNCTION_CALL: &str = + include_str!("../fixtures/responses/http_sse/unanswered_function_call.sse"); const AFTER_5XX_RETRY_SSE: &str = include_str!("../fixtures/responses/http_sse/after_5xx_retry.sse"); const STREAM_AFTER_5XX_RETRY_SSE: &str = @@ -651,6 +655,172 @@ async fn v1_responses_should_classify_non_stream_sse_failure_and_retry_next_acco assert_eq!(first_status.0, "quota_exhausted"); } +#[tokio::test] +async fn v1_responses_should_strip_history_after_http_sse_previous_response_not_found() { + let server = MockServer::start().await; + Mock::given(method("POST")) + .and(path("/codex/responses")) + .and(header("authorization", "Bearer access-a")) + .respond_with( + ResponseTemplate::new(200) + .insert_header("content-type", "text/event-stream") + .set_body_string(SSE_PREVIOUS_RESPONSE_NOT_FOUND), + ) + .up_to_n_times(1) + .mount(&server) + .await; + Mock::given(method("POST")) + .and(path("/codex/responses")) + .and(header("authorization", "Bearer access-a")) + .respond_with( + ResponseTemplate::new(200) + .insert_header("content-type", "text/event-stream") + .set_body_string(AFTER_402_SSE), + ) + .mount(&server) + .await; + let imported = build_imported_app_with_accounts( + server.uri(), + &[ImportAccount { + id: "acct_sse_previous_missing", + account_id: "chatgpt-a", + token: "access-a", + refresh_token: "refresh-a", + }], + ) + .await; + + let response = imported + .app + .oneshot( + Request::builder() + .method("POST") + .uri("/v1/responses") + .header( + "authorization", + format!("Bearer {}", imported.client_api_key), + ) + .header("content-type", "application/json") + .body(Body::from( + r#"{"model":"gpt-5.5","input":[],"stream":false,"use_websocket":false,"turnState":"turn-stale"}"#, + )) + .unwrap(), + ) + .await + .unwrap(); + + assert_eq!(response.status(), StatusCode::OK); + let body = response_json(response).await; + assert_eq!(body["id"], "resp_after_402"); + let requests = server.received_requests().await.unwrap(); + let access_a_posts = requests + .iter() + .filter(|request| { + request.method.as_str() == "POST" + && request + .headers + .get("authorization") + .and_then(|value| value.to_str().ok()) + == Some("Bearer access-a") + }) + .collect::>(); + assert_eq!(access_a_posts.len(), 2); + assert_eq!( + access_a_posts[0] + .headers + .get("x-codex-turn-state") + .and_then(|value| value.to_str().ok()), + Some("turn-stale") + ); + assert!(access_a_posts[1] + .headers + .get("x-codex-turn-state") + .is_none()); +} + +#[tokio::test] +async fn v1_responses_should_strip_history_after_http_sse_unanswered_function_call() { + let server = MockServer::start().await; + Mock::given(method("POST")) + .and(path("/codex/responses")) + .and(header("authorization", "Bearer access-a")) + .respond_with( + ResponseTemplate::new(200) + .insert_header("content-type", "text/event-stream") + .set_body_string(SSE_UNANSWERED_FUNCTION_CALL), + ) + .up_to_n_times(1) + .mount(&server) + .await; + Mock::given(method("POST")) + .and(path("/codex/responses")) + .and(header("authorization", "Bearer access-a")) + .respond_with( + ResponseTemplate::new(200) + .insert_header("content-type", "text/event-stream") + .set_body_string(AFTER_402_SSE), + ) + .mount(&server) + .await; + let imported = build_imported_app_with_accounts( + server.uri(), + &[ImportAccount { + id: "acct_sse_unanswered_call", + account_id: "chatgpt-a", + token: "access-a", + refresh_token: "refresh-a", + }], + ) + .await; + + let response = imported + .app + .oneshot( + Request::builder() + .method("POST") + .uri("/v1/responses") + .header( + "authorization", + format!("Bearer {}", imported.client_api_key), + ) + .header("content-type", "application/json") + .body(Body::from( + r#"{"model":"gpt-5.5","input":[],"stream":false,"use_websocket":false,"turnState":"turn-stale"}"#, + )) + .unwrap(), + ) + .await + .unwrap(); + + assert_eq!(response.status(), StatusCode::OK); + let body = response_json(response).await; + assert_eq!(body["id"], "resp_after_402"); + let requests = server.received_requests().await.unwrap(); + let access_a_posts = requests + .iter() + .filter(|request| { + request.method.as_str() == "POST" + && request + .headers + .get("authorization") + .and_then(|value| value.to_str().ok()) + == Some("Bearer access-a") + }) + .collect::>(); + assert_eq!(access_a_posts.len(), 2); + assert_eq!( + access_a_posts[0] + .headers + .get("x-codex-turn-state") + .and_then(|value| value.to_str().ok()), + Some("turn-stale") + ); + assert!(access_a_posts[1] + .headers + .get("x-codex-turn-state") + .is_none()); +} + #[tokio::test] async fn v1_responses_should_retry_next_account_when_model_not_supported() { let server = MockServer::start().await; diff --git a/tests/fixtures/responses/http_sse/previous_response_not_found.sse b/tests/fixtures/responses/http_sse/previous_response_not_found.sse new file mode 100644 index 000000000..4ff486726 --- /dev/null +++ b/tests/fixtures/responses/http_sse/previous_response_not_found.sse @@ -0,0 +1,3 @@ +event: error +data: {"error":{"code":"previous_response_not_found","message":"Previous response with id resp_missing was not found"}} + diff --git a/tests/fixtures/responses/http_sse/unanswered_function_call.sse b/tests/fixtures/responses/http_sse/unanswered_function_call.sse new file mode 100644 index 000000000..c3b8abe78 --- /dev/null +++ b/tests/fixtures/responses/http_sse/unanswered_function_call.sse @@ -0,0 +1,3 @@ +event: error +data: {"error":{"code":"invalid_request","message":"No tool output found for function call call_missing"}} +