From b7cb250f841d4152d05f16d62bf9a9545c97408a Mon Sep 17 00:00:00 2001 From: xcosmosbox <2162381070@qq.com> Date: Tue, 1 Sep 2026 12:46:19 +0800 Subject: [PATCH 1/4] feat(apikey): add hybrid rotation (aggregate-first) strategy --- apps/src/app/apikeys/page.tsx | 1 + apps/src/app/models/page.tsx | 1 + apps/src/app/platform-mode/page-sections.tsx | 1 + apps/src/components/modals/api-key-modal.tsx | 15 ++- apps/src/lib/i18n/messages/en.ts | 5 +- apps/src/lib/i18n/messages/ko.ts | 5 +- apps/src/lib/i18n/messages/ru.ts | 5 +- crates/service/src/apikey/apikey_create.rs | 2 + crates/service/src/apikey/apikey_profile.rs | 7 + .../src/apikey/apikey_profile_tests.rs | 18 +++ .../service/src/apikey/apikey_update_model.rs | 2 + crates/service/src/codex_profile_tests.rs | 10 ++ crates/service/src/gateway/mod.rs | 1 + .../src/gateway/upstream/executor/mod.rs | 4 + .../gateway/upstream/executor/mod_tests.rs | 7 + .../upstream/protocol/aggregate_api.rs | 60 ++++++++- crates/service/src/gateway/upstream/proxy.rs | 124 ++++++++++++++---- 17 files changed, 224 insertions(+), 44 deletions(-) diff --git a/apps/src/app/apikeys/page.tsx b/apps/src/app/apikeys/page.tsx index f043a79fc..495ed5e3d 100644 --- a/apps/src/app/apikeys/page.tsx +++ b/apps/src/app/apikeys/page.tsx @@ -83,6 +83,7 @@ const ROTATION_STRATEGY_LABELS: Record = { account_rotation: "账号轮转", aggregate_api_rotation: "聚合API轮转", hybrid_rotation: "混合轮转（账号优先）", + hybrid_aggregate_first_rotation: "混合轮转（聚合优先）", }; function userCanOwnApiKey(user: AppUser): boolean { diff --git a/apps/src/app/models/page.tsx b/apps/src/app/models/page.tsx index b467372cb..a8b785eaa 100644 --- a/apps/src/app/models/page.tsx +++ b/apps/src/app/models/page.tsx @@ -104,6 +104,7 @@ const ROTATION_STRATEGY_LABELS: Record = { account_rotation: "OpenAI 账号池", aggregate_api_rotation: "聚合 API", hybrid_rotation: "混合路由", + hybrid_aggregate_first_rotation: "混合路由", }; function routeStrategyLabel( diff --git a/apps/src/app/platform-mode/page-sections.tsx b/apps/src/app/platform-mode/page-sections.tsx index 929aa0604..9c5085b3f 100644 --- a/apps/src/app/platform-mode/page-sections.tsx +++ b/apps/src/app/platform-mode/page-sections.tsx @@ -41,6 +41,7 @@ const ROTATION_STRATEGY_LABELS: Record = { account_rotation: "OpenAI 账号池", aggregate_api_rotation: "聚合 API", hybrid_rotation: "混合路由", + hybrid_aggregate_first_rotation: "混合路由", }; function rotationStrategyLabel(strategy: string, t: Translate): string { diff --git a/apps/src/components/modals/api-key-modal.tsx b/apps/src/components/modals/api-key-modal.tsx index 9b063ccd1..d89481177 100644 --- a/apps/src/components/modals/api-key-modal.tsx +++ b/apps/src/components/modals/api-key-modal.tsx @@ -78,6 +78,7 @@ const ROTATION_STRATEGY_LABELS: Record = { account_rotation: "账号轮转", aggregate_api_rotation: "聚合API轮转", hybrid_rotation: "混合轮转（账号优先）", + hybrid_aggregate_first_rotation: "混合轮转（聚合优先）", }; const ACCOUNT_PLAN_FILTER_LABELS: Record = { @@ -172,7 +173,8 @@ export function ApiKeyModal({ const memberOwnershipEnabled = isAdminMode && showMemberOwnership; const usesAccountPlanFilter = rotationStrategy === "account_rotation" || - rotationStrategy === "hybrid_rotation"; + rotationStrategy === "hybrid_rotation" || + rotationStrategy === "hybrid_aggregate_first_rotation"; const billableUsers = useMemo( () => appUsers.filter((user) => userCanOwnApiKey(user)), [appUsers], @@ -494,13 +496,16 @@ export function ApiKeyModal({ {t("混合轮转（账号优先）")} - - - + + {t("混合轮转（聚合优先）")} + + + +

{t( - "账号轮转只走账号池；聚合API轮转只走聚合API；混合轮转先走账号池，账号耗尽后使用聚合API兜底。", + "账号轮转只走账号池；聚合API轮转只走聚合API；混合轮转（账号优先）先走账号池，账号耗尽后使用聚合API兜底；混合轮转（聚合优先）先走聚合API，聚合不可用时回落账号池。", )}

diff --git a/apps/src/lib/i18n/messages/en.ts b/apps/src/lib/i18n/messages/en.ts index a4d6eefd1..9c5e53904 100644 --- a/apps/src/lib/i18n/messages/en.ts +++ b/apps/src/lib/i18n/messages/en.ts @@ -288,6 +288,7 @@ export const EN_MESSAGES: MessageCatalog = { 账号轮转: "Account rotation", 聚合API轮转: "Aggregate API rotation", "混合轮转（账号优先）": "Hybrid rotation (accounts first)", + "混合轮转（聚合优先）": "Hybrid rotation (aggregate APIs first)", "总使用 Token": "Total token usage", 最近调用: "Last called", 从未调用: "Never called", @@ -762,8 +763,8 @@ export const EN_MESSAGES: MessageCatalog = { "Configure gateway credentials. You can bind a specific model, reasoning tier, or custom upstream.", "密钥名称 (可选)": "Key name (optional)", "例如：主机房 / 测试": "e.g. Main room / Test", - "账号轮转只走账号池；聚合API轮转只走聚合API；混合轮转先走账号池，账号耗尽后使用聚合API兜底。": - "Account rotation uses only the account pool; aggregate API rotation uses only aggregate APIs; hybrid rotation tries accounts first and falls back to aggregate APIs when accounts are exhausted.", + "账号轮转只走账号池；聚合API轮转只走聚合API；混合轮转（账号优先）先走账号池，账号耗尽后使用聚合API兜底；混合轮转（聚合优先）先走聚合API，聚合不可用时回落账号池。": + "Account rotation uses only the account pool; aggregate API rotation uses only aggregate APIs; hybrid rotation (accounts first) tries accounts first and falls back to aggregate APIs when accounts are exhausted; hybrid rotation (aggregate APIs first) tries aggregate APIs first and falls back to the account pool when they are unavailable.", "仅对账号轮转和混合轮转生效，可限制这把平台密钥只从指定账号计划类型中选路由账号。": "Applies only to account rotation and hybrid rotation. Limits this API key to routing through accounts with the selected plan type.", "默认按路径通配：": "Path matching by default:", diff --git a/apps/src/lib/i18n/messages/ko.ts b/apps/src/lib/i18n/messages/ko.ts index 393e99924..b5c65ea44 100644 --- a/apps/src/lib/i18n/messages/ko.ts +++ b/apps/src/lib/i18n/messages/ko.ts @@ -287,6 +287,7 @@ export const KO_MESSAGES: MessageCatalog = { 账号轮转: "계정 로테이션", 聚合API轮转: "집계 API 로테이션", "混合轮转（账号优先）": "하이브리드 로테이션(계정 우선)", + "混合轮转（聚合优先）": "하이브리드 로테이션(집계 API 우선)", 协议: "프로토콜", 轮转策略: "로테이션 전략", 绑定模型: "바인딩 모델", @@ -1190,8 +1191,8 @@ export const KO_MESSAGES: MessageCatalog = { "원하는 색상 테마를 선택해 작업 분위기에 맞추세요", 在渐变版本和默认版本之间切换: "그라데이션 버전과 기본 버전 사이 전환", 暂无请求日志: "요청 로그가 없습니다", - "账号轮转只走账号池；聚合API轮转只走聚合API；混合轮转先走账号池，账号耗尽后使用聚合API兜底。": - "계정 로테이션은 계정 풀만 사용하고, 집계 API 로테이션은 집계 API만 사용합니다. 하이브리드는 계정을 먼저 사용한 뒤 계정이 소진되면 집계 API로 폴백합니다.", + "账号轮转只走账号池；聚合API轮转只走聚合API；混合轮转（账号优先）先走账号池，账号耗尽后使用聚合API兜底；混合轮转（聚合优先）先走聚合API，聚合不可用时回落账号池。": + "계정 로테이션은 계정 풀만 사용하고, 집계 API 로테이션은 집계 API만 사용합니다. 하이브리드(계정 우선)는 계정을 먼저 사용한 뒤 계정이 소진되면 집계 API로 폴백하고, 하이브리드(집계 API 우선)는 집계 API를 먼저 사용한 뒤 사용할 수 없을 때 계정 풀로 폴백합니다.", "仅对账号轮转和混合轮转生效，可限制这把平台密钥只从指定账号计划类型中选路由账号。": "계정 로테이션과 하이브리드 로테이션에만 적용되며, 이 API 키가 지정한 계정 플랜 유형에서만 라우팅 계정을 선택하도록 제한합니다.", 账号选路策略: "계정 라우팅 전략", diff --git a/apps/src/lib/i18n/messages/ru.ts b/apps/src/lib/i18n/messages/ru.ts index 7d17be371..94dc660a2 100644 --- a/apps/src/lib/i18n/messages/ru.ts +++ b/apps/src/lib/i18n/messages/ru.ts @@ -272,6 +272,7 @@ export const RU_MESSAGES: MessageCatalog = { 账号轮转: "Ротация аккаунтов", 聚合API轮转: "Ротация агрегированных API", "混合轮转（账号优先）": "Гибридная ротация (сначала аккаунты)", + "混合轮转（聚合优先）": "Гибридная ротация (сначала агрегированные API)", 协议: "Протокол", 轮转策略: "Стратегия ротации", 绑定模型: "Связанная модель", @@ -1146,8 +1147,8 @@ export const RU_MESSAGES: MessageCatalog = { 在渐变版本和默认版本之间切换: "Переключение между градиентной и стандартной версией", 暂无请求日志: "Нет логов запросов", - "账号轮转只走账号池；聚合API轮转只走聚合API；混合轮转先走账号池，账号耗尽后使用聚合API兜底。": - "Ротация аккаунтов использует только пул аккаунтов; ротация API использует только API; гибрид сначала пробует аккаунты и переключается на API при исчерпании аккаунтов.", + "账号轮转只走账号池；聚合API轮转只走聚合API；混合轮转（账号优先）先走账号池，账号耗尽后使用聚合API兜底；混合轮转（聚合优先）先走聚合API，聚合不可用时回落账号池。": + "Ротация аккаунтов использует только пул аккаунтов; ротация агрегированных API использует только API; гибридная ротация (сначала аккаунты) сначала пробует аккаунты и переключается на API при исчерпании аккаунтов; гибридная ротация (сначала API) сначала пробует API и переключается на пул аккаунтов при их недоступности.", "仅对账号轮转和混合轮转生效，可限制这把平台密钥只从指定账号计划类型中选路由账号。": "Применяется только к ротации аккаунтов и гибридной ротации и ограничивает маршрутизацию этим API-ключом аккаунтами выбранного типа плана.", 账号选路策略: "Стратегия маршрутизации аккаунтов", diff --git a/crates/service/src/apikey/apikey_create.rs b/crates/service/src/apikey/apikey_create.rs index fc0fba97f..59c0a2960 100644 --- a/crates/service/src/apikey/apikey_create.rs +++ b/crates/service/src/apikey/apikey_create.rs @@ -112,6 +112,7 @@ pub(crate) fn create_api_key( }; let account_plan_filter = if rotation_strategy == crate::apikey_profile::ROTATION_ACCOUNT || rotation_strategy == crate::apikey_profile::ROTATION_HYBRID + || rotation_strategy == crate::apikey_profile::ROTATION_HYBRID_AGGREGATE_FIRST { crate::account_plan::normalize_account_plan_filter(account_plan_filter)? } else { @@ -119,6 +120,7 @@ pub(crate) fn create_api_key( }; let account_group_filter = if rotation_strategy == crate::apikey_profile::ROTATION_ACCOUNT || rotation_strategy == crate::apikey_profile::ROTATION_HYBRID + || rotation_strategy == crate::apikey_profile::ROTATION_HYBRID_AGGREGATE_FIRST { crate::account_group::normalize_account_group_filter(account_group_filter) } else { diff --git a/crates/service/src/apikey/apikey_profile.rs b/crates/service/src/apikey/apikey_profile.rs index efa12d515..8c3c404e1 100644 --- a/crates/service/src/apikey/apikey_profile.rs +++ b/crates/service/src/apikey/apikey_profile.rs @@ -7,6 +7,7 @@ pub(crate) const AUTH_X_API_KEY: &str = "x_api_key"; pub(crate) const ROTATION_ACCOUNT: &str = "account_rotation"; pub(crate) const ROTATION_AGGREGATE_API: &str = "aggregate_api_rotation"; pub(crate) const ROTATION_HYBRID: &str = "hybrid_rotation"; +pub(crate) const ROTATION_HYBRID_AGGREGATE_FIRST: &str = "hybrid_aggregate_first_rotation"; /// 函数 `normalize_key` /// @@ -174,6 +175,12 @@ pub(crate) fn normalize_rotation_strategy(value: Option) -> Result Ok(ROTATION_HYBRID.to_string()), + "hybrid_aggregate_first" + | "hybrid_aggregate_first_rotation" + | "mixed_aggregate_first" + | "aggregate_first" + | "聚合优先" + | "聚合优先账号兜底" => Ok(ROTATION_HYBRID_AGGREGATE_FIRST.to_string()), other => Err(format!("unsupported rotation strategy: {other}")), }, None => Ok(ROTATION_ACCOUNT.to_string()), diff --git a/crates/service/src/apikey/apikey_profile_tests.rs b/crates/service/src/apikey/apikey_profile_tests.rs index b2953fbec..80911a6ce 100644 --- a/crates/service/src/apikey/apikey_profile_tests.rs +++ b/crates/service/src/apikey/apikey_profile_tests.rs @@ -3,6 +3,7 @@ use super::{ is_gemini_generate_content_request_path, normalize_protocol_type, normalize_rotation_strategy, resolve_gateway_protocol_type, PROTOCOL_ANTHROPIC_NATIVE, PROTOCOL_GEMINI_NATIVE, PROTOCOL_OPENAI_COMPAT, ROTATION_ACCOUNT, ROTATION_AGGREGATE_API, ROTATION_HYBRID, + ROTATION_HYBRID_AGGREGATE_FIRST, }; #[test] @@ -22,6 +23,23 @@ fn normalize_rotation_strategy_accepts_hybrid_aliases() { } } +#[test] +fn normalize_rotation_strategy_accepts_hybrid_aggregate_first_aliases() { + for value in [ + "hybrid_aggregate_first_rotation", + "hybrid_aggregate_first", + "mixed_aggregate_first", + "aggregate_first", + "聚合优先", + "聚合优先账号兜底", + ] { + assert_eq!( + normalize_rotation_strategy(Some(value.to_string())).as_deref(), + Ok(ROTATION_HYBRID_AGGREGATE_FIRST) + ); + } +} + #[test] fn normalize_rotation_strategy_keeps_existing_values() { assert_eq!( diff --git a/crates/service/src/apikey/apikey_update_model.rs b/crates/service/src/apikey/apikey_update_model.rs index d92d671f5..62c843830 100644 --- a/crates/service/src/apikey/apikey_update_model.rs +++ b/crates/service/src/apikey/apikey_update_model.rs @@ -73,6 +73,7 @@ pub(crate) fn update_api_key_model( let normalized_account_plan_filter = if normalized_rotation_strategy == crate::apikey_profile::ROTATION_ACCOUNT || normalized_rotation_strategy == crate::apikey_profile::ROTATION_HYBRID + || normalized_rotation_strategy == crate::apikey_profile::ROTATION_HYBRID_AGGREGATE_FIRST { crate::account_plan::normalize_account_plan_filter(account_plan_filter)? } else { @@ -105,6 +106,7 @@ pub(crate) fn update_api_key_model( Some(ROTATION_AGGREGATE_API) => Some(None), Some(crate::apikey_profile::ROTATION_ACCOUNT) | Some(crate::apikey_profile::ROTATION_HYBRID) + | Some(crate::apikey_profile::ROTATION_HYBRID_AGGREGATE_FIRST) if update_account_group_filter => { Some(crate::account_group::normalize_account_group_filter( diff --git a/crates/service/src/codex_profile_tests.rs b/crates/service/src/codex_profile_tests.rs index 5bd7e1708..d79d8be5b 100644 --- a/crates/service/src/codex_profile_tests.rs +++ b/crates/service/src/codex_profile_tests.rs @@ -312,6 +312,12 @@ fn rotation_strategy_selects_catalog_ownership() { ), crate::codex_model_catalog::GatewayCatalogPolicy::Managed ); + assert_eq!( + crate::codex_model_catalog::gateway_catalog_policy_for_rotation_strategy( + crate::apikey_profile::ROTATION_HYBRID_AGGREGATE_FIRST + ), + crate::codex_model_catalog::GatewayCatalogPolicy::Managed + ); } #[test] @@ -320,6 +326,10 @@ fn api_key_candidates_expose_route_and_catalog_ownership() { (crate::apikey_profile::ROTATION_ACCOUNT, "official"), (crate::apikey_profile::ROTATION_AGGREGATE_API, "managed"), (crate::apikey_profile::ROTATION_HYBRID, "managed"), + ( + crate::apikey_profile::ROTATION_HYBRID_AGGREGATE_FIRST, + "managed", + ), ] { let candidate = api_key_candidate(ApiKeyCodexProfileCandidate { id: format!("key-{rotation_strategy}"), diff --git a/crates/service/src/gateway/mod.rs b/crates/service/src/gateway/mod.rs index 4300c505c..53255ea39 100644 --- a/crates/service/src/gateway/mod.rs +++ b/crates/service/src/gateway/mod.rs @@ -1062,6 +1062,7 @@ pub(crate) fn gateway_supports_official_responses_websocket( } if api_key.rotation_strategy == crate::apikey_profile::ROTATION_AGGREGATE_API || api_key.rotation_strategy == crate::apikey_profile::ROTATION_HYBRID + || api_key.rotation_strategy == crate::apikey_profile::ROTATION_HYBRID_AGGREGATE_FIRST { return false; } diff --git a/crates/service/src/gateway/upstream/executor/mod.rs b/crates/service/src/gateway/upstream/executor/mod.rs index 975fc618e..118ab8f72 100644 --- a/crates/service/src/gateway/upstream/executor/mod.rs +++ b/crates/service/src/gateway/upstream/executor/mod.rs @@ -20,6 +20,7 @@ pub(super) enum GatewayUpstreamRouteKind { AccountRotation, AggregateApi, HybridAccountFirst, + HybridAggregateFirst, } #[derive(Debug, Clone, Copy, PartialEq, Eq)] @@ -47,6 +48,9 @@ pub(super) fn resolve_gateway_upstream_execution_plan( let route_kind = match rotation_strategy { crate::apikey_profile::ROTATION_AGGREGATE_API => GatewayUpstreamRouteKind::AggregateApi, crate::apikey_profile::ROTATION_HYBRID => GatewayUpstreamRouteKind::HybridAccountFirst, + crate::apikey_profile::ROTATION_HYBRID_AGGREGATE_FIRST => { + GatewayUpstreamRouteKind::HybridAggregateFirst + } _ => GatewayUpstreamRouteKind::AccountRotation, }; GatewayUpstreamExecutionPlan { diff --git a/crates/service/src/gateway/upstream/executor/mod_tests.rs b/crates/service/src/gateway/upstream/executor/mod_tests.rs index 1090c9c27..9cbd13db7 100644 --- a/crates/service/src/gateway/upstream/executor/mod_tests.rs +++ b/crates/service/src/gateway/upstream/executor/mod_tests.rs @@ -42,6 +42,13 @@ fn protocol_and_rotation_map_to_execution_plan() { route_kind: GatewayUpstreamRouteKind::HybridAccountFirst, } ); + assert_eq!( + resolve_gateway_upstream_execution_plan("openai_compat", "hybrid_aggregate_first_rotation"), + GatewayUpstreamExecutionPlan { + executor_kind: GatewayUpstreamExecutorKind::CodexResponses, + route_kind: GatewayUpstreamRouteKind::HybridAggregateFirst, + } + ); assert_eq!( resolve_gateway_upstream_execution_plan("gemini_native", "aggregate_api_rotation"), GatewayUpstreamExecutionPlan { diff --git a/crates/service/src/gateway/upstream/protocol/aggregate_api.rs b/crates/service/src/gateway/upstream/protocol/aggregate_api.rs index bc0601791..fd391fd40 100644 --- a/crates/service/src/gateway/upstream/protocol/aggregate_api.rs +++ b/crates/service/src/gateway/upstream/protocol/aggregate_api.rs @@ -1026,6 +1026,24 @@ pub(crate) fn resolve_aggregate_api_rotation_candidates( /// /// # 返回 /// 返回函数执行结果 +/// 聚合路径失败时的处理策略。 +pub(in super::super) enum AggregateFailurePolicy { + /// 失败时直接向客户端响应错误（默认行为）。 + RespondError, + /// 失败且请求尚未消费时，把请求归还给调用方， + /// 由调用方继续后续路由（聚合优先混合轮转回落账号池）。 + ReleaseRequest, +} + +/// 聚合代理一次调用的最终结果。 +pub(in super::super) enum AggregateAttemptOutcome { + /// 请求已经响应完毕（成功桥接或已向客户端返回错误）。 + Responded, + /// 请求未被消费且聚合路径失败（仅 `ReleaseRequest` 策略出现）， + /// 调用方可继续后续路由。 + RequestReleased { request: Request, error: String }, +} + pub(in super::super) struct AggregateProxyRequest<'a> { pub request: Request, pub storage: &'a Storage, @@ -1053,11 +1071,12 @@ pub(in super::super) struct AggregateProxyRequest<'a> { pub aggregate_api_candidates: Vec, pub request_deadline: Option, pub started_at: Instant, + pub failure_policy: AggregateFailurePolicy, } pub(in super::super) fn proxy_aggregate_request( params: AggregateProxyRequest<'_>, -) -> Result<(), String> { +) -> Result { let AggregateProxyRequest { request, storage, @@ -1085,6 +1104,7 @@ pub(in super::super) fn proxy_aggregate_request( aggregate_api_candidates, request_deadline, started_at, + failure_policy, } = params; let estimated_input_tokens = super::super::super::request_log::estimate_input_tokens_from_body(body.as_ref()); @@ -1099,9 +1119,16 @@ pub(in super::super) fn proxy_aggregate_request( Some(message.as_str()), started_at.elapsed().as_millis(), ); + if matches!(failure_policy, AggregateFailurePolicy::ReleaseRequest) { + // 聚合优先混合轮转：无聚合候选时归还请求，由调用方回落账号池。 + return Ok(AggregateAttemptOutcome::RequestReleased { + request, + error: message, + }); + } let request = request; respond_error(request, 404, message.as_str(), Some(trace_id)); - return Ok(()); + return Ok(AggregateAttemptOutcome::Responded); } let mut request = Some(request); @@ -1270,7 +1297,7 @@ pub(in super::super) fn proxy_aggregate_request( Some(started_at.elapsed().as_millis()), ); respond_error(request, 504, message.as_str(), Some(trace_id)); - return Ok(()); + return Ok(AggregateAttemptOutcome::Responded); } let mut url = base_upstream_url.clone(); @@ -1519,7 +1546,7 @@ pub(in super::super) fn proxy_aggregate_request( } if succeeded { - return Ok(()); + return Ok(AggregateAttemptOutcome::Responded); } if candidate_idx + 1 < total_candidates { @@ -1530,6 +1557,29 @@ pub(in super::super) fn proxy_aggregate_request( let message = last_attempt_error.unwrap_or_else(|| "aggregate api upstream response failed".to_string()); let status_code = last_failure_status; + if matches!(failure_policy, AggregateFailurePolicy::ReleaseRequest) { + if let Some(released_request) = request.take() { + // 聚合优先混合轮转：聚合候选全部失败且请求尚未消费， + // 保留失败日志后将请求归还调用方，由其回落账号池。 + super::super::super::record_gateway_request_outcome( + path, + status_code, + Some("aggregate_api"), + ); + super::super::super::trace_log::log_request_final( + trace_id, + status_code, + Some(key_id), + last_attempt_url.as_deref(), + Some(message.as_str()), + started_at.elapsed().as_millis(), + ); + return Ok(AggregateAttemptOutcome::RequestReleased { + request: released_request, + error: message, + }); + } + } let request = request.take().ok_or_else(|| { "aggregate api request already consumed before failure response".to_string() })?; @@ -1583,7 +1633,7 @@ pub(in super::super) fn proxy_aggregate_request( Some(started_at.elapsed().as_millis()), ); respond_error(request, status_code, message.as_str(), Some(trace_id)); - Ok(()) + Ok(AggregateAttemptOutcome::Responded) } fn aggregate_api_secrets_by_candidate_id( diff --git a/crates/service/src/gateway/upstream/proxy.rs b/crates/service/src/gateway/upstream/proxy.rs index 906abfc6a..b2143d6db 100644 --- a/crates/service/src/gateway/upstream/proxy.rs +++ b/crates/service/src/gateway/upstream/proxy.rs @@ -14,6 +14,7 @@ use super::proxy_pipeline::candidate_executor::{ use super::proxy_pipeline::execution_context::GatewayUpstreamExecutionContext; use super::proxy_pipeline::request_gate::acquire_request_gate; use super::proxy_pipeline::request_setup::prepare_request_setup; +use super::protocol::aggregate_api::{AggregateAttemptOutcome, AggregateFailurePolicy}; use super::proxy_pipeline::response_finalize::respond_terminal; use super::support::precheck::{prepare_candidates_for_proxy, CandidatePrecheckResult}; @@ -100,6 +101,10 @@ fn should_try_provider_executor_aggregate_route( && !has_enabled_default_account_pool_route(model) }) } + // 聚合优先混合轮转：只要聚合路由可用就先走聚合，聚合耗尽后回落账号池 + GatewayUpstreamRouteKind::HybridAggregateFirst => { + configured_model.is_none_or(has_enabled_aggregate_api_route) + } GatewayUpstreamRouteKind::AccountRotation => false, } } @@ -113,6 +118,23 @@ fn is_hybrid_account_first_route( ) } +fn is_hybrid_aggregate_first_route( + execution_plan: super::executor::GatewayUpstreamExecutionPlan, +) -> bool { + matches!( + execution_plan.route_kind, + GatewayUpstreamRouteKind::HybridAggregateFirst + ) +} + +/// 混合轮转（无论账号优先还是聚合优先）直连聚合时使用透传路径与透传请求体。 +fn is_hybrid_passthrough_route( + execution_plan: super::executor::GatewayUpstreamExecutionPlan, +) -> bool { + is_hybrid_account_first_route(execution_plan) + || is_hybrid_aggregate_first_route(execution_plan) +} + fn respond_when_account_candidates_empty( execution_plan: super::executor::GatewayUpstreamExecutionPlan, configured_model: Option<&ManagedModelV2>, @@ -152,6 +174,7 @@ fn route_kind_label(value: GatewayUpstreamRouteKind) -> &'static str { GatewayUpstreamRouteKind::AccountRotation => "account_rotation", GatewayUpstreamRouteKind::AggregateApi => "aggregate_api", GatewayUpstreamRouteKind::HybridAccountFirst => "hybrid_account_first", + GatewayUpstreamRouteKind::HybridAggregateFirst => "hybrid_aggregate_first", } } @@ -215,6 +238,10 @@ fn validate_model_route( has_enabled_default_account_pool_route(&managed_model) || has_enabled_aggregate_api_route(&managed_model) } + GatewayUpstreamRouteKind::HybridAggregateFirst => { + has_enabled_default_account_pool_route(&managed_model) + || has_enabled_aggregate_api_route(&managed_model) + } }; if !route_enabled { return Err((503, format!("model_unavailable: {model}"))); @@ -557,7 +584,8 @@ fn proxy_with_aggregate_candidates( request_deadline: Option, started_at: Instant, aggregate_api_candidates: Vec, -) -> Result<(), String> { + failure_policy: AggregateFailurePolicy, +) -> Result { let mut aggregate_api_candidates = aggregate_api_candidates; super::protocol::aggregate_api::apply_gateway_route_strategy_to_aggregate_candidates( &mut aggregate_api_candidates, @@ -602,6 +630,7 @@ fn proxy_with_aggregate_candidates( aggregate_api_candidates, request_deadline, started_at, + failure_policy, }, ) } @@ -780,12 +809,20 @@ pub(in super::super) fn proxy_validated_request( } }; + // 聚合优先混合轮转：聚合路径失败且请求未被消费时，需要把请求交还给账号路径继续， + // 因此这里使用可变绑定。 + let mut request = request; if should_try_provider_executor_aggregate_route(execution_plan, configured_model.as_ref()) { - let (aggregate_path, aggregate_body) = if is_hybrid_account_first_route(execution_plan) { + let (aggregate_path, aggregate_body) = if is_hybrid_passthrough_route(execution_plan) { (passthrough_path.as_str(), &passthrough_body) } else { (path.as_str(), &body) }; + let aggregate_failure_policy = if is_hybrid_aggregate_first_route(execution_plan) { + AggregateFailurePolicy::ReleaseRequest + } else { + AggregateFailurePolicy::RespondError + }; match resolve_aggregate_candidates_for_route( &storage, protocol_type.as_str(), @@ -793,7 +830,7 @@ pub(in super::super) fn proxy_validated_request( model_for_log.as_deref(), ) { Ok(aggregate_api_candidates) => { - return proxy_with_aggregate_candidates( + match proxy_with_aggregate_candidates( request, &storage, trace_id.as_str(), @@ -818,31 +855,56 @@ pub(in super::super) fn proxy_validated_request( request_deadline, started_at, aggregate_api_candidates, - ); + aggregate_failure_policy, + ) { + Ok(AggregateAttemptOutcome::Responded) => return Ok(()), + Ok(AggregateAttemptOutcome::RequestReleased { + request: released_request, + error, + }) => { + // 聚合优先混合轮转：聚合候选全部失败，回落账号池。 + log::debug!( + "event=gateway_hybrid_aggregate_first_fallback_to_account trace_id={} err={}", + trace_id, + error + ); + request = released_request; + } + Err(err) => return Err(err), + } } Err(err) => { - return respond_aggregate_route_error( - request, - &storage, - trace_id.as_str(), - key_id.as_str(), - original_path.as_str(), - aggregate_path, - request_method.as_str(), - super::super::ResponseAdapter::Passthrough, - service_tier_for_log.as_deref(), - effective_service_tier_for_log.as_deref(), - service_tier_source_for_log.as_deref(), - gateway_mode_for_log.as_deref(), - client_model_for_log.as_deref(), - model_for_log.as_deref(), - model_source_for_log.as_deref(), - client_reasoning_for_log.as_deref(), - reasoning_for_log.as_deref(), - reasoning_source_for_log.as_deref(), - started_at, - err, - ); + if is_hybrid_aggregate_first_route(execution_plan) { + // 聚合优先混合轮转：聚合候选解析失败，回落账号池。 + log::debug!( + "event=gateway_hybrid_aggregate_first_fallback_to_account trace_id={} err={}", + trace_id, + err + ); + } else { + return respond_aggregate_route_error( + request, + &storage, + trace_id.as_str(), + key_id.as_str(), + original_path.as_str(), + aggregate_path, + request_method.as_str(), + super::super::ResponseAdapter::Passthrough, + service_tier_for_log.as_deref(), + effective_service_tier_for_log.as_deref(), + service_tier_source_for_log.as_deref(), + gateway_mode_for_log.as_deref(), + client_model_for_log.as_deref(), + model_for_log.as_deref(), + model_source_for_log.as_deref(), + client_reasoning_for_log.as_deref(), + reasoning_for_log.as_deref(), + reasoning_source_for_log.as_deref(), + started_at, + err, + ); + } } } } @@ -877,6 +939,7 @@ pub(in super::super) fn proxy_validated_request( model_for_log.as_deref(), ) { Ok(aggregate_api_candidates) => { + // 账号优先混合轮转的聚合兜底是最后一步：失败直接响应错误。 return proxy_with_aggregate_candidates( request, &storage, @@ -902,7 +965,9 @@ pub(in super::super) fn proxy_validated_request( request_deadline, started_at, aggregate_api_candidates, - ); + AggregateFailurePolicy::RespondError, + ) + .map(|_| ()); } Err(err) => { return respond_hybrid_route_error( @@ -1078,6 +1143,7 @@ pub(in super::super) fn proxy_validated_request( model_for_log.as_deref(), ) { Ok(aggregate_api_candidates) => { + // 账号优先混合轮转的聚合兜底是最后一步：失败直接响应错误。 return proxy_with_aggregate_candidates( request, &storage, @@ -1103,7 +1169,9 @@ pub(in super::super) fn proxy_validated_request( request_deadline, started_at, aggregate_api_candidates, - ); + AggregateFailurePolicy::RespondError, + ) + .map(|_| ()); } Err(err) => { return respond_hybrid_route_error( From 718aba37393e877cd3d5bd0bae8dbb6cf8f6506f Mon Sep 17 00:00:00 2001 From: xcosmosbox <2162381070@qq.com> Date: Tue, 1 Sep 2026 12:58:58 +0800 Subject: [PATCH 2/4] ci: trigger workflow From 14bba017fe2bded2428f418f7cbac0f8b9914940 Mon Sep 17 00:00:00 2001 From: xcosmosbox <2162381070@qq.com> Date: Tue, 1 Sep 2026 13:05:47 +0800 Subject: [PATCH 3/4] style: fix rustfmt formatting for hybrid aggregate-first strategy --- crates/service/src/apikey/apikey_update_model.rs | 3 ++- crates/service/src/gateway/upstream/proxy.rs | 5 ++--- 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/crates/service/src/apikey/apikey_update_model.rs b/crates/service/src/apikey/apikey_update_model.rs index 62c843830..68347c6d3 100644 --- a/crates/service/src/apikey/apikey_update_model.rs +++ b/crates/service/src/apikey/apikey_update_model.rs @@ -73,7 +73,8 @@ pub(crate) fn update_api_key_model( let normalized_account_plan_filter = if normalized_rotation_strategy == crate::apikey_profile::ROTATION_ACCOUNT || normalized_rotation_strategy == crate::apikey_profile::ROTATION_HYBRID - || normalized_rotation_strategy == crate::apikey_profile::ROTATION_HYBRID_AGGREGATE_FIRST + || normalized_rotation_strategy + == crate::apikey_profile::ROTATION_HYBRID_AGGREGATE_FIRST { crate::account_plan::normalize_account_plan_filter(account_plan_filter)? } else { diff --git a/crates/service/src/gateway/upstream/proxy.rs b/crates/service/src/gateway/upstream/proxy.rs index b2143d6db..b118e0dfe 100644 --- a/crates/service/src/gateway/upstream/proxy.rs +++ b/crates/service/src/gateway/upstream/proxy.rs @@ -8,13 +8,13 @@ use super::super::local_validation::LocalValidationResult; use super::executor::{ resolve_gateway_upstream_execution_plan, GatewayUpstreamExecutorKind, GatewayUpstreamRouteKind, }; +use super::protocol::aggregate_api::{AggregateAttemptOutcome, AggregateFailurePolicy}; use super::proxy_pipeline::candidate_executor::{ execute_candidate_sequence, CandidateExecutionResult, CandidateExecutorParams, }; use super::proxy_pipeline::execution_context::GatewayUpstreamExecutionContext; use super::proxy_pipeline::request_gate::acquire_request_gate; use super::proxy_pipeline::request_setup::prepare_request_setup; -use super::protocol::aggregate_api::{AggregateAttemptOutcome, AggregateFailurePolicy}; use super::proxy_pipeline::response_finalize::respond_terminal; use super::support::precheck::{prepare_candidates_for_proxy, CandidatePrecheckResult}; @@ -131,8 +131,7 @@ fn is_hybrid_aggregate_first_route( fn is_hybrid_passthrough_route( execution_plan: super::executor::GatewayUpstreamExecutionPlan, ) -> bool { - is_hybrid_account_first_route(execution_plan) - || is_hybrid_aggregate_first_route(execution_plan) + is_hybrid_account_first_route(execution_plan) || is_hybrid_aggregate_first_route(execution_plan) } fn respond_when_account_candidates_empty( From f4516cfcdc9a2a906349b3f8f09ed4871857ff91 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=AB=98=20=E5=AE=8F=E9=A1=BA?= <18272669457@163.com> Date: Wed, 2 Sep 2026 19:26:20 +0800 Subject: [PATCH 4/4] fix(gateway): constrain aggregate-first fallback --- .../upstream/protocol/aggregate_api.rs | 30 +- crates/service/src/gateway/upstream/proxy.rs | 16 +- .../src/gateway/upstream/proxy_tests.rs | 38 ++- .../tests/gateway_logs/hybrid_routing.rs | 280 +++++++++++++++++- 4 files changed, 338 insertions(+), 26 deletions(-) diff --git a/crates/service/src/gateway/upstream/protocol/aggregate_api.rs b/crates/service/src/gateway/upstream/protocol/aggregate_api.rs index fd391fd40..af2d3cb9d 100644 --- a/crates/service/src/gateway/upstream/protocol/aggregate_api.rs +++ b/crates/service/src/gateway/upstream/protocol/aggregate_api.rs @@ -1110,6 +1110,13 @@ pub(in super::super) fn proxy_aggregate_request( super::super::super::request_log::estimate_input_tokens_from_body(body.as_ref()); if aggregate_api_candidates.is_empty() { let message = "aggregate api not found".to_string(); + if matches!(failure_policy, AggregateFailurePolicy::ReleaseRequest) { + // 聚合优先混合轮转：无聚合候选时归还请求，由调用方回落账号池。 + return Ok(AggregateAttemptOutcome::RequestReleased { + request, + error: message, + }); + } super::super::super::record_gateway_request_outcome(path, 404, Some("aggregate_api")); super::super::super::trace_log::log_request_final( trace_id, @@ -1119,13 +1126,6 @@ pub(in super::super) fn proxy_aggregate_request( Some(message.as_str()), started_at.elapsed().as_millis(), ); - if matches!(failure_policy, AggregateFailurePolicy::ReleaseRequest) { - // 聚合优先混合轮转：无聚合候选时归还请求，由调用方回落账号池。 - return Ok(AggregateAttemptOutcome::RequestReleased { - request, - error: message, - }); - } let request = request; respond_error(request, 404, message.as_str(), Some(trace_id)); return Ok(AggregateAttemptOutcome::Responded); @@ -1560,20 +1560,8 @@ pub(in super::super) fn proxy_aggregate_request( if matches!(failure_policy, AggregateFailurePolicy::ReleaseRequest) { if let Some(released_request) = request.take() { // 聚合优先混合轮转：聚合候选全部失败且请求尚未消费， - // 保留失败日志后将请求归还调用方，由其回落账号池。 - super::super::super::record_gateway_request_outcome( - path, - status_code, - Some("aggregate_api"), - ); - super::super::super::trace_log::log_request_final( - trace_id, - status_code, - Some(key_id), - last_attempt_url.as_deref(), - Some(message.as_str()), - started_at.elapsed().as_millis(), - ); + // 将请求归还调用方，由其回落账号池。这里不能提前记录请求终态， + // 否则账号兜底完成后会重复计数并覆盖同一 trace 的最终结果。 return Ok(AggregateAttemptOutcome::RequestReleased { request: released_request, error: message, diff --git a/crates/service/src/gateway/upstream/proxy.rs b/crates/service/src/gateway/upstream/proxy.rs index b118e0dfe..6b22a1550 100644 --- a/crates/service/src/gateway/upstream/proxy.rs +++ b/crates/service/src/gateway/upstream/proxy.rs @@ -127,6 +127,14 @@ fn is_hybrid_aggregate_first_route( ) } +fn should_fallback_to_account_after_aggregate_exhaustion( + execution_plan: super::executor::GatewayUpstreamExecutionPlan, + configured_model: Option<&ManagedModelV2>, +) -> bool { + is_hybrid_aggregate_first_route(execution_plan) + && configured_model.is_none_or(has_enabled_default_account_pool_route) +} + /// 混合轮转（无论账号优先还是聚合优先）直连聚合时使用透传路径与透传请求体。 fn is_hybrid_passthrough_route( execution_plan: super::executor::GatewayUpstreamExecutionPlan, @@ -817,7 +825,11 @@ pub(in super::super) fn proxy_validated_request( } else { (path.as_str(), &body) }; - let aggregate_failure_policy = if is_hybrid_aggregate_first_route(execution_plan) { + let fallback_to_account = should_fallback_to_account_after_aggregate_exhaustion( + execution_plan, + configured_model.as_ref(), + ); + let aggregate_failure_policy = if fallback_to_account { AggregateFailurePolicy::ReleaseRequest } else { AggregateFailurePolicy::RespondError @@ -873,7 +885,7 @@ pub(in super::super) fn proxy_validated_request( } } Err(err) => { - if is_hybrid_aggregate_first_route(execution_plan) { + if fallback_to_account { // 聚合优先混合轮转：聚合候选解析失败，回落账号池。 log::debug!( "event=gateway_hybrid_aggregate_first_fallback_to_account trace_id={} err={}", diff --git a/crates/service/src/gateway/upstream/proxy_tests.rs b/crates/service/src/gateway/upstream/proxy_tests.rs index 155e4141f..70782b378 100644 --- a/crates/service/src/gateway/upstream/proxy_tests.rs +++ b/crates/service/src/gateway/upstream/proxy_tests.rs @@ -2,7 +2,8 @@ use super::{ exhausted_gateway_error_for_log, has_enabled_aggregate_api_route, has_enabled_default_account_pool_route, hybrid_route_error_message, provider_upstream_hint, request_deadline_for_path, resolve_aggregate_candidates_for_route, resolve_upstream_is_stream, - respond_when_account_candidates_empty, should_fallback_to_aggregate_after_account_exhaustion, + respond_when_account_candidates_empty, should_fallback_to_account_after_aggregate_exhaustion, + should_fallback_to_aggregate_after_account_exhaustion, should_try_provider_executor_aggregate_route, validate_model_route, }; use crate::gateway::upstream::executor::{ @@ -426,6 +427,41 @@ fn hybrid_dual_route_keeps_account_first_and_aggregate_fallback() { )); } +#[test] +fn hybrid_aggregate_first_fallback_requires_default_account_pool_route() { + let hybrid = execution_plan(GatewayUpstreamRouteKind::HybridAggregateFirst); + let aggregate_only = model_with_routes(&[("aggregate_api", "agg-test")]); + let account_only = model_with_routes(&[("account_pool", "default")]); + let dual_route = + model_with_routes(&[("account_pool", "default"), ("aggregate_api", "agg-test")]); + let non_default_pool = + model_with_routes(&[("account_pool", "secondary"), ("aggregate_api", "agg-test")]); + + assert!(should_try_provider_executor_aggregate_route( + hybrid, + Some(&aggregate_only), + )); + assert!(!should_fallback_to_account_after_aggregate_exhaustion( + hybrid, + Some(&aggregate_only), + )); + assert!(!should_try_provider_executor_aggregate_route( + hybrid, + Some(&account_only), + )); + assert!(should_fallback_to_account_after_aggregate_exhaustion( + hybrid, + Some(&dual_route), + )); + assert!(!should_fallback_to_account_after_aggregate_exhaustion( + hybrid, + Some(&non_default_pool), + )); + assert!(should_fallback_to_account_after_aggregate_exhaustion( + hybrid, None, + )); +} + #[test] fn hybrid_route_error_mentions_both_pools() { let message = hybrid_route_error_message( diff --git a/crates/service/tests/gateway_logs/hybrid_routing.rs b/crates/service/tests/gateway_logs/hybrid_routing.rs index c6eb2852a..1d85fd8a5 100644 --- a/crates/service/tests/gateway_logs/hybrid_routing.rs +++ b/crates/service/tests/gateway_logs/hybrid_routing.rs @@ -139,8 +139,38 @@ fn insert_hybrid_key(storage: &Storage, key_id: &str, platform_key: &str, now: i .expect("insert hybrid API key"); } +fn insert_hybrid_aggregate_first_key( + storage: &Storage, + key_id: &str, + platform_key: &str, + now: i64, +) { + storage + .insert_api_key(&ApiKey { + id: key_id.to_string(), + name: Some(key_id.to_string()), + model_slug: Some(MODEL.to_string()), + reasoning_effort: None, + service_tier: None, + rotation_strategy: "hybrid_aggregate_first_rotation".to_string(), + aggregate_api_id: None, + account_plan_filter: None, + aggregate_api_url: None, + client_type: "codex".to_string(), + protocol_type: "openai_compat".to_string(), + auth_scheme: "authorization_bearer".to_string(), + upstream_base_url: None, + static_headers_json: None, + key_hash: hash_platform_key_for_test(platform_key), + status: "active".to_string(), + created_at: now, + last_used_at: None, + }) + .expect("insert aggregate-first hybrid API key"); +} + #[test] -fn hybrid_aggregate_only_skips_active_account_and_uses_aggregate_api() { +fn hybrid_aggregate_first_aggregate_only_skips_active_account_and_uses_aggregate_api() { let _lock = test_env_guard(); let dir = new_test_dir("codexmanager-hybrid-aggregate-only"); let db_path: PathBuf = dir.join("codexmanager.db"); @@ -166,7 +196,7 @@ fn hybrid_aggregate_only_skips_active_account_and_uses_aggregate_api() { insert_active_account(&storage, "acc_hybrid_aggregate_only", now); insert_aggregate_api(&storage, aggregate_id, &aggregate_addr, "/responses", now); replace_with_aggregate_only_route(&storage, aggregate_id); - insert_hybrid_key(&storage, key_id, platform_key, now); + insert_hybrid_aggregate_first_key(&storage, key_id, platform_key, now); let server = codexmanager_service::start_one_shot_server().expect("start server"); let request = serde_json::json!({ @@ -214,6 +244,252 @@ fn hybrid_aggregate_only_skips_active_account_and_uses_aggregate_api() { assert_eq!(log.actual_source_id.as_deref(), Some(aggregate_id)); } +#[test] +fn hybrid_aggregate_first_aggregate_only_failure_never_uses_account_pool() { + let _lock = test_env_guard(); + let dir = new_test_dir("codexmanager-hybrid-aggregate-first-aggregate-only-failure"); + let db_path: PathBuf = dir.join("codexmanager.db"); + let _db_guard = EnvGuard::set("CODEXMANAGER_DB_PATH", db_path.to_string_lossy().as_ref()); + + let (local_addr, local_rx, local_join) = start_mock_upstream_sequence_lenient( + vec![(200, response_json("resp_local_must_not_run"))], + Duration::from_secs(2), + ); + let local_base = format!("http://{local_addr}/backend-api/codex"); + let _upstream_guard = EnvGuard::set("CODEXMANAGER_UPSTREAM_BASE_URL", &local_base); + let aggregate_failure = r#"{"error":{"message":"aggregate unavailable"}}"#.to_string(); + let (aggregate_addr, aggregate_rx, aggregate_join) = start_mock_upstream_sequence_lenient( + vec![ + (503, aggregate_failure.clone()), + (503, aggregate_failure.clone()), + (503, aggregate_failure.clone()), + (503, aggregate_failure), + ], + Duration::from_secs(2), + ); + + let storage = Storage::open(&db_path).expect("open db"); + storage.init().expect("init db"); + let now = now_ts(); + let aggregate_id = "agg_hybrid_aggregate_first_only_failure"; + let key_id = "gk_hybrid_aggregate_first_only_failure"; + let platform_key = "pk_hybrid_aggregate_first_only_failure"; + insert_active_account(&storage, "acc_hybrid_aggregate_first_only_failure", now); + insert_aggregate_api(&storage, aggregate_id, &aggregate_addr, "/responses", now); + replace_with_aggregate_only_route(&storage, aggregate_id); + insert_hybrid_aggregate_first_key(&storage, key_id, platform_key, now); + + let server = codexmanager_service::start_one_shot_server().expect("start server"); + let request = serde_json::json!({ + "model": MODEL, + "input": "hello", + "stream": false + }); + let request = serde_json::to_string(&request).expect("serialize request"); + let (status, response_body) = post_http_raw( + &server.addr, + "/v1/responses", + &request, + &[ + ("Content-Type", "application/json"), + ("Authorization", &format!("Bearer {platform_key}")), + ], + ); + server.join(); + local_join.join().expect("join local upstream"); + aggregate_join.join().expect("join aggregate upstream"); + + assert_eq!(status, 502, "gateway response: {response_body}"); + assert_eq!( + local_rx.try_iter().count(), + 0, + "aggregate-only model must not bypass its route and use an account" + ); + assert_eq!(aggregate_rx.try_iter().count(), 4, "aggregate retry count"); +} + +#[test] +fn hybrid_aggregate_first_dual_route_falls_back_once_and_preserves_tools() { + let _lock = test_env_guard(); + let dir = new_test_dir("codexmanager-hybrid-aggregate-first-dual-fallback"); + let db_path: PathBuf = dir.join("codexmanager.db"); + let _db_guard = EnvGuard::set("CODEXMANAGER_DB_PATH", db_path.to_string_lossy().as_ref()); + + let (local_addr, local_rx, local_join) = start_mock_upstream_sequence_lenient( + vec![(200, response_json("resp_account_fallback"))], + Duration::from_secs(2), + ); + let local_base = format!("http://{local_addr}/backend-api/codex"); + let _upstream_guard = EnvGuard::set("CODEXMANAGER_UPSTREAM_BASE_URL", &local_base); + let aggregate_failure = r#"{"error":{"message":"aggregate unavailable"}}"#.to_string(); + let (aggregate_addr, aggregate_rx, aggregate_join) = start_mock_upstream_sequence_lenient( + vec![ + (503, aggregate_failure.clone()), + (503, aggregate_failure.clone()), + (503, aggregate_failure.clone()), + (503, aggregate_failure), + ], + Duration::from_secs(2), + ); + + let storage = Storage::open(&db_path).expect("open db"); + storage.init().expect("init db"); + let now = now_ts(); + let aggregate_id = "agg_hybrid_aggregate_first_dual_fallback"; + let key_id = "gk_hybrid_aggregate_first_dual_fallback"; + let platform_key = "pk_hybrid_aggregate_first_dual_fallback"; + insert_active_account(&storage, "acc_hybrid_aggregate_first_dual_fallback", now); + insert_aggregate_api(&storage, aggregate_id, &aggregate_addr, "/responses", now); + seed_dual_routes(&storage, aggregate_id); + insert_hybrid_aggregate_first_key(&storage, key_id, platform_key, now); + + let server = codexmanager_service::start_one_shot_server().expect("start server"); + let request = serde_json::json!({ + "model": MODEL, + "input": "use the tool", + "stream": false, + "tools": [{ + "type": "function", + "name": "get_answer", + "description": "Return an answer", + "parameters": { + "type": "object", + "properties": { "question": { "type": "string" } } + } + }] + }); + let request = serde_json::to_string(&request).expect("serialize request"); + let (status, response_body) = post_http_raw( + &server.addr, + "/v1/responses", + &request, + &[ + ("Content-Type", "application/json"), + ("Authorization", &format!("Bearer {platform_key}")), + ], + ); + server.join(); + local_join.join().expect("join local upstream"); + aggregate_join.join().expect("join aggregate upstream"); + + assert_eq!(status, 200, "gateway response: {response_body}"); + assert!(response_body.contains("resp_account_fallback")); + assert_eq!(aggregate_rx.try_iter().count(), 4, "aggregate retry count"); + let local_requests = local_rx.try_iter().collect::>(); + assert_eq!(local_requests.len(), 1, "account fallback request count"); + let local_body: serde_json::Value = + serde_json::from_slice(&decode_upstream_request_body(&local_requests[0])) + .expect("parse account fallback body"); + assert_eq!(local_body["tools"][0]["name"], "get_answer"); + + let logs = storage + .list_request_logs(Some(&format!("key:={key_id}")), 10) + .expect("list request logs") + .into_iter() + .filter(|item| item.request_path == "/v1/responses") + .collect::>(); + assert_eq!(logs.len(), 1, "fallback must persist one final request log"); + assert_eq!(logs[0].status_code, Some(200)); + assert_eq!( + logs[0].actual_source_kind.as_deref(), + Some("openai_account") + ); +} + +#[test] +fn hybrid_aggregate_first_dual_route_streams_from_account_after_aggregate_failure() { + let _lock = test_env_guard(); + let dir = new_test_dir("codexmanager-hybrid-aggregate-first-stream-fallback"); + let db_path: PathBuf = dir.join("codexmanager.db"); + let _db_guard = EnvGuard::set("CODEXMANAGER_DB_PATH", db_path.to_string_lossy().as_ref()); + + let account_sse = concat!( + "data: {\"type\":\"response.output_text.delta\",\"delta\":\"account fallback ok\"}\n\n", + "data: {\"type\":\"response.completed\",\"response\":{\"id\":\"resp_hybrid_aggregate_first_stream\",\"model\":\"gpt-hybrid-route-test\",\"usage\":{\"input_tokens\":3,\"output_tokens\":2,\"total_tokens\":5}}}\n\n", + "data: [DONE]\n\n" + ); + let (local_addr, local_rx, local_join) = + start_mock_upstream_sequence_lenient_with_content_types( + vec![( + 200, + account_sse.to_string(), + "text/event-stream".to_string(), + )], + Duration::from_secs(2), + ); + let local_base = format!("http://{local_addr}/backend-api/codex"); + let _upstream_guard = EnvGuard::set("CODEXMANAGER_UPSTREAM_BASE_URL", &local_base); + let aggregate_failure = r#"{"error":{"message":"aggregate unavailable"}}"#.to_string(); + let (aggregate_addr, aggregate_rx, aggregate_join) = start_mock_upstream_sequence_lenient( + vec![ + (503, aggregate_failure.clone()), + (503, aggregate_failure.clone()), + (503, aggregate_failure.clone()), + (503, aggregate_failure), + ], + Duration::from_secs(2), + ); + + let storage = Storage::open(&db_path).expect("open db"); + storage.init().expect("init db"); + let now = now_ts(); + let aggregate_id = "agg_hybrid_aggregate_first_stream_fallback"; + let key_id = "gk_hybrid_aggregate_first_stream_fallback"; + let platform_key = "pk_hybrid_aggregate_first_stream_fallback"; + insert_active_account(&storage, "acc_hybrid_aggregate_first_stream_fallback", now); + insert_aggregate_api(&storage, aggregate_id, &aggregate_addr, "/responses", now); + seed_dual_routes(&storage, aggregate_id); + insert_hybrid_aggregate_first_key(&storage, key_id, platform_key, now); + + let server = codexmanager_service::start_one_shot_server().expect("start server"); + let request = serde_json::json!({ + "model": MODEL, + "input": "hello", + "stream": true + }); + let request = serde_json::to_string(&request).expect("serialize request"); + let (status, response_body) = post_http_raw( + &server.addr, + "/v1/responses", + &request, + &[ + ("Content-Type", "application/json"), + ("Authorization", &format!("Bearer {platform_key}")), + ], + ); + server.join(); + local_join.join().expect("join local upstream"); + aggregate_join.join().expect("join aggregate upstream"); + + assert_eq!(status, 200, "gateway response: {response_body}"); + assert!(response_body.contains("account fallback ok")); + assert!(response_body.contains("resp_hybrid_aggregate_first_stream")); + assert_eq!(aggregate_rx.try_iter().count(), 4, "aggregate retry count"); + let local_requests = local_rx.try_iter().collect::>(); + assert_eq!(local_requests.len(), 1, "account fallback request count"); + let local_body: serde_json::Value = + serde_json::from_slice(&decode_upstream_request_body(&local_requests[0])) + .expect("parse account fallback body"); + assert_eq!(local_body["stream"], true); + + let logs = storage + .list_request_logs(Some(&format!("key:={key_id}")), 10) + .expect("list request logs") + .into_iter() + .filter(|item| item.request_path == "/v1/responses") + .collect::>(); + assert_eq!( + logs.len(), + 1, + "stream fallback must persist one final request log" + ); + assert_eq!(logs[0].status_code, Some(200)); + assert_eq!( + logs[0].actual_source_kind.as_deref(), + Some("openai_account") + ); +} + #[test] fn hybrid_aggregate_only_streams_chat_completions_tool_calls_from_aggregate_api() { let _lock = test_env_guard();