From 78ba7e412e456453ddb76763740b3ba7490cddf2 Mon Sep 17 00:00:00 2001 From: Zyy <39424122+zyycn@users.noreply.github.com> Date: Mon, 5 Oct 2026 23:41:46 +0800 Subject: [PATCH] fix(codex): align upstream contracts and transport behavior (#352) Co-authored-by: Zyy Co-authored-by: Codex --- backend/Cargo.lock | 6 +- .../gateway-admin/src/ports/provider.rs | 18 +-- .../gateway-admin/tests/use_case/accounts.rs | 25 +++- .../tests/use_case/observability.rs | 76 +++++++++-- .../gateway-admin/tests/use_case/proxies.rs | 4 +- .../gateway-core/src/engine/coordinator.rs | 21 ++-- backend/crates/gateway-core/src/error.rs | 17 +++ .../gateway-core/tests/engine/coordinator.rs | 118 ++++++++++++++++++ backend/crates/gateway-protocol/Cargo.toml | 1 + .../gateway-protocol/src/openai/codex.rs | 8 +- .../gateway-protocol/src/openai/events.rs | 22 ++-- .../gateway-protocol/src/openai/headers.rs | 20 ++- .../crates/gateway-protocol/src/openai/mod.rs | 4 +- .../gateway-protocol/tests/openai/codex.rs | 19 +++ .../gateway-protocol/tests/openai/events.rs | 38 ++++++ .../gateway-protocol/tests/openai/headers.rs | 33 ++++- backend/crates/providers/openai/Cargo.toml | 1 + backend/crates/providers/openai/src/admin.rs | 25 ++-- .../openai/src/credential/token_client.rs | 2 +- .../providers/openai/src/credential/types.rs | 8 ++ .../providers/openai/src/provider/failure.rs | 14 ++- .../providers/openai/src/transport/client.rs | 26 +--- .../openai/src/transport/diagnostics.rs | 12 +- .../src/transport/websocket/handshake.rs | 8 +- .../crates/providers/openai/tests/admin.rs | 38 +++++- .../openai/tests/credential/admin.rs | 19 +++ .../openai/tests/credential/oauth.rs | 4 +- .../openai/tests/credential/token_client.rs | 2 +- .../openai/tests/provider/contract/mod.rs | 100 ++++++++++++--- .../openai/tests/transport/diagnostics.rs | 21 ++++ .../providers/openai/tests/transport/tls.rs | 116 ++++++++++++----- .../providers/openai/tests/transport/usage.rs | 4 +- .../xai/src/transport/request/tools.rs | 5 +- .../xai/tests/transport/canonical.rs | 30 +++++ .../xai/tests/transport/request/mod.rs | 65 ++++++++++ docs/api.md | 15 ++- docs/architecture.md | 1 + 37 files changed, 792 insertions(+), 154 deletions(-) diff --git a/backend/Cargo.lock b/backend/Cargo.lock index e5b3621e8..5bc9950f8 100644 --- a/backend/Cargo.lock +++ b/backend/Cargo.lock @@ -1858,6 +1858,7 @@ dependencies = [ name = "gateway-protocol" version = "0.0.0" dependencies = [ + "httpdate", "memchr", "serde", "serde_json", @@ -1978,9 +1979,9 @@ dependencies = [ [[package]] name = "h2" -version = "0.4.16" +version = "0.4.19" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a9f37a958b41b3b19ee2707c06439c0e9e547e847223eb791ecb0cb821c65e27" +checksum = "ef8e5e5a340588f4452631496976cf8636d4a7ecf600239fdc27615d2530bc16" dependencies = [ "atomic-waker", "bytes", @@ -3234,6 +3235,7 @@ dependencies = [ "gateway-core", "gateway-protocol", "getrandom 0.4.3", + "h2", "hex", "hyper", "hyper-rustls", diff --git a/backend/crates/gateway-admin/src/ports/provider.rs b/backend/crates/gateway-admin/src/ports/provider.rs index 7fa213166..182e038e0 100644 --- a/backend/crates/gateway-admin/src/ports/provider.rs +++ b/backend/crates/gateway-admin/src/ports/provider.rs @@ -147,9 +147,9 @@ pub trait ProviderAdmin: Send + Sync { Err(ProviderAdminError::new(ProviderAdminErrorKind::Unsupported)) } - /// 将原始套餐值投影为展示名称；默认保留未知 Provider 的原始名称 + /// 提供最终套餐展示名称；未覆盖时采用通用的大驼峰格式 fn plan_type_display(&self, plan_type: &str) -> String { - plan_type.to_owned() + default_plan_type_display(plan_type) } /// 账号已经由控制面提交为不可调度状态，释放 Provider 持有的账号级运行时资源 @@ -354,7 +354,7 @@ impl ProviderAdminRegistry { self.plan_type_display(provider_kind, plan_type.as_deref()) } - /// 账号页和 Dashboard 共用的大驼峰套餐展示名称，不修改原始套餐值 + /// 账号页和 Dashboard 共用 Provider 的最终展示名称，不改写官方拼写或原始套餐值 pub(crate) fn plan_type_display( &self, provider_kind: &str, @@ -364,12 +364,10 @@ impl ProviderAdminRegistry { let provider = ProviderKind::new(provider_kind.to_owned()) .ok() .and_then(|kind| self.require(&kind).ok()); - let display = provider.map_or_else( - || plan_type.to_owned(), + Some(provider.map_or_else( + || default_plan_type_display(plan_type), |provider| provider.plan_type_display(plan_type), - ); - // 所有 Provider 及未知套餐统一排版，保留 Plus 后缀的含义 - Some(display.replace('+', " Plus ").to_upper_camel_case()) + )) } /// 返回所有已注册 Provider 的 Dashboard 上游身份画像 @@ -398,3 +396,7 @@ impl ProviderAdminRegistry { self.require(provider_kind)?.calculated_billing(input) } } + +fn default_plan_type_display(plan_type: &str) -> String { + plan_type.replace('+', " Plus ").to_upper_camel_case() +} diff --git a/backend/crates/gateway-admin/tests/use_case/accounts.rs b/backend/crates/gateway-admin/tests/use_case/accounts.rs index 2ee44a126..c97ff0912 100644 --- a/backend/crates/gateway-admin/tests/use_case/accounts.rs +++ b/backend/crates/gateway-admin/tests/use_case/accounts.rs @@ -305,7 +305,12 @@ impl ProviderAdmin for FakeProviderAdmin { } fn plan_type_display(&self, plan_type: &str) -> String { - format!("{} display: {plan_type}", self.kind) + match plan_type { + "prolite" => "ProLite".to_owned(), + "pro" => "Pro".to_owned(), + "promax" => "ProMax".to_owned(), + _ => format!("{} display: {plan_type}", self.kind), + } } async fn account_unavailable(&self, _: &ProviderAccountId) { @@ -1911,7 +1916,7 @@ async fn accounts_list_should_return_complete_directory_semantics() { ), ( Some(" self_serve_business_prolite "), - Some("OpenaiDisplaySelfServeBusinessProlite"), + Some("openai display: self_serve_business_prolite"), ) ); @@ -1931,25 +1936,33 @@ async fn accounts_list_should_return_complete_directory_semantics() { #[tokio::test] async fn accounts_should_fill_missing_plan_from_quota_without_overriding_known_subtypes() { for (stored_plan, quota_plan, expected, expected_display) in [ - (None, Some("free"), Some("free"), Some("OpenaiDisplayFree")), + ( + None, + Some("free"), + Some("free"), + Some("openai display: free"), + ), ( Some(" "), Some("free"), Some("free"), - Some("OpenaiDisplayFree"), + Some("openai display: free"), ), ( Some("unknown"), Some("free"), Some("free"), - Some("OpenaiDisplayFree"), + Some("openai display: free"), ), ( Some("self_serve_business_prolite"), Some("team"), Some("self_serve_business_prolite"), - Some("OpenaiDisplaySelfServeBusinessProlite"), + Some("openai display: self_serve_business_prolite"), ), + (Some("prolite"), None, Some("prolite"), Some("ProLite")), + (Some("pro"), None, Some("pro"), Some("Pro")), + (Some("promax"), None, Some("promax"), Some("ProMax")), (None, None, None, None), ] { let provider = FakeProviderAdmin::new("openai", events()); diff --git a/backend/crates/gateway-admin/tests/use_case/observability.rs b/backend/crates/gateway-admin/tests/use_case/observability.rs index 509ea8dee..d20d78815 100644 --- a/backend/crates/gateway-admin/tests/use_case/observability.rs +++ b/backend/crates/gateway-admin/tests/use_case/observability.rs @@ -17,12 +17,13 @@ use gateway_admin::{ model::{ MutationContext, PageSize, Revision, observability::{ - AccountPoolMetrics, AttemptMetrics, CostCoverage, CurrencyCost, DashboardObservation, - DashboardRuntimeSlots, DiagnosticDimension, DiagnosticObservation, - DiagnosticsObservation, Granularity, HealthStatus, LatencyPercentiles, OpsErrorPage, - OpsErrorQuery, PercentileMilliseconds, RequestMetricPoint, RequestMetrics, TimeRange, - TrendKind, UsageBilling, UsageCalculatedBillingFact, UsageDetail, UsageFilter, - UsageListRecord, UsageOverview, UsagePage, UsageQuery, + AccountPoolMetrics, AttemptMetrics, CostCoverage, CurrencyCost, DashboardAccountUsage, + DashboardObservation, DashboardRuntimeSlots, DiagnosticDimension, + DiagnosticObservation, DiagnosticsObservation, Granularity, HealthStatus, + LatencyPercentiles, OpsErrorPage, OpsErrorQuery, PercentileMilliseconds, + RequestMetricPoint, RequestMetrics, TimeRange, TrendKind, UsageBilling, + UsageCalculatedBillingFact, UsageDetail, UsageFilter, UsageListRecord, UsageOverview, + UsagePage, UsageQuery, }, settings::{ AdminApiKey, AdminApiKeyMutation, ReplaceRuntimeSettings, RotationStrategy, @@ -721,6 +722,65 @@ async fn usage_insights_should_reject_partial_costs_when_billing_stream_fails() ); } +#[tokio::test] +async fn dashboard_should_preserve_provider_plan_names() { + let now = Utc::now(); + let range = observation_range(now); + let store = Arc::new(FixtureObservabilityStore::new(range)); + let plans = [("prolite", "ProLite"), ("pro", "Pro"), ("promax", "ProMax")]; + *store.account_usage.lock().unwrap() = plans + .iter() + .map(|(plan, _)| DashboardAccountUsage { + account_id: format!("account-{plan}"), + provider_kind: "openai".to_owned(), + authentication_kind: "oauth".to_owned(), + name: (*plan).to_owned(), + email: None, + plan_type: Some((*plan).to_owned()), + plan_type_display: None, + request_count: 0, + success_count: 0, + input_tokens: None, + output_tokens: None, + cached_tokens: None, + cache_write_tokens: None, + reasoning_tokens: None, + image_input_tokens: None, + image_output_tokens: None, + image_request_count: 0, + image_request_failed_count: 0, + total_tokens: None, + cost_coverage: CostCoverage::default(), + costs: Vec::new(), + last_used_at: None, + request_buckets: Vec::new(), + quota_used_percent: None, + quota_window: None, + models: Vec::new(), + }) + .collect(); + let provider = + super::accounts::FakeProviderAdmin::new("openai", Arc::new(Mutex::new(Vec::new()))); + let services = super::AdminHarness::new() + .observability(store) + .settings(Arc::new(FixtureSettingsStore { + max_concurrent_per_account: 1, + })) + .provider(provider) + .build() + .await; + let result = services + .observability() + .dashboard_summary(range, TrendKind::Usage) + .await + .unwrap(); + assert_eq!(result.observation.account_usage.len(), plans.len()); + for (account, (raw, display)) in result.observation.account_usage.iter().zip(plans) { + assert_eq!(account.plan_type.as_deref(), Some(raw)); + assert_eq!(account.plan_type_display.as_deref(), Some(display)); + } +} + struct FixtureObservabilityStore { trend: Mutex>, overview: Mutex, @@ -731,6 +791,7 @@ struct FixtureObservabilityStore { summary_observed_at: Mutex>>, slots_observed_at: Mutex>>, usage_records: Mutex>, + account_usage: Mutex>, dashboard_delay: Mutex, dashboard_summary_calls: AtomicUsize, } @@ -752,6 +813,7 @@ impl FixtureObservabilityStore { summary_observed_at: Mutex::new(None), slots_observed_at: Mutex::new(None), usage_records: Mutex::new(Vec::new()), + account_usage: Mutex::new(Vec::new()), dashboard_delay: Mutex::new(StdDuration::ZERO), dashboard_summary_calls: AtomicUsize::new(0), } @@ -824,7 +886,7 @@ impl ObservabilityStore for FixtureObservabilityStore { totals: Default::default(), provider_accounts: AccountPoolMetrics::default(), trend: self.trend.lock().expect("trend").clone(), - account_usage: Vec::new(), + account_usage: self.account_usage.lock().expect("account usage").clone(), recent_requests: Vec::new(), }) } diff --git a/backend/crates/gateway-admin/tests/use_case/proxies.rs b/backend/crates/gateway-admin/tests/use_case/proxies.rs index f09b7a579..8f08ace1e 100644 --- a/backend/crates/gateway-admin/tests/use_case/proxies.rs +++ b/backend/crates/gateway-admin/tests/use_case/proxies.rs @@ -232,8 +232,8 @@ async fn linked_accounts_share_plan_resolution_and_only_read_cached_quota() { assert_eq!( result.items[0].plan_type_display.as_deref(), match expected { - Some("free") => Some("OpenaiDisplayFree"), - Some("plus") => Some("OpenaiDisplayPlus"), + Some("free") => Some("openai display: free"), + Some("plus") => Some("openai display: plus"), _ => None, } ); diff --git a/backend/crates/gateway-core/src/engine/coordinator.rs b/backend/crates/gateway-core/src/engine/coordinator.rs index d809cf68c..0af3e6762 100644 --- a/backend/crates/gateway-core/src/engine/coordinator.rs +++ b/backend/crates/gateway-core/src/engine/coordinator.rs @@ -1510,7 +1510,10 @@ where && !self.delivery_pending && attempt_send_state != UpstreamSendState::Ambiguous => { - Some((AttemptTransport::Fallback, Duration::ZERO)) + Some(( + AttemptTransport::Fallback, + error.retry_after().unwrap_or_default(), + )) } _ => None, }; @@ -1536,7 +1539,10 @@ where .or_default(); if *retries < max_retries.get() { let multiplier = 1_u32.checked_shl(*retries).unwrap_or(u32::MAX); - let delay = initial_delay.saturating_mul(multiplier).min(max_delay); + // 服务器建议优先于本地退避，不能被本地上限缩短或再次指数放大 + let delay = error + .retry_after() + .unwrap_or_else(|| initial_delay.saturating_mul(multiplier).min(max_delay)); *retries = retries.saturating_add(1); Some(delay) } else { @@ -1555,11 +1561,12 @@ where && !self .credential_recovery_attempted_accounts .contains(current.metadata.provider_account_id()); - let retryable = continuation_retry - || same_account_retry - || ordinary_retry - || account_rotation_retry - || transport_recovery.is_some(); + let retryable = !error.retry_is_prohibited() + && (continuation_retry + || same_account_retry + || ordinary_retry + || account_rotation_retry + || transport_recovery.is_some()); let retryable = match self .apply_retry_policy(super::policy::RetryFacts { diff --git a/backend/crates/gateway-core/src/error.rs b/backend/crates/gateway-core/src/error.rs index d587e5a8e..ce3c344f9 100644 --- a/backend/crates/gateway-core/src/error.rs +++ b/backend/crates/gateway-core/src/error.rs @@ -444,6 +444,7 @@ pub struct ProviderError { continuation_recovery_disposition: Option, failure_observation: Option>, replay_safe: bool, + retry_prohibited: bool, pre_delivery_retry: Option>, credential_recovery_required: bool, retry_same_account: bool, @@ -484,6 +485,7 @@ impl ProviderError { continuation_recovery_disposition: None, failure_observation: None, replay_safe: false, + retry_prohibited: false, pre_delivery_retry: None, credential_recovery_required: false, retry_same_account: false, @@ -581,6 +583,19 @@ impl ProviderError { self } + /// 标记必须交付客户端的终止拒绝，即使 payload 未发送也不自动重试 + #[must_use] + pub const fn with_retry_prohibited(mut self) -> Self { + self.retry_prohibited = true; + self + } + + /// 返回 Provider 是否明确禁止本请求的重试与传输回退 + #[must_use] + pub const fn retry_is_prohibited(&self) -> bool { + self.retry_prohibited + } + /// 允许 Core 仅在客户端尚未收到任何事件时执行一次受预算约束的换号恢复 /// /// 该标记不证明上游未执行，也不等价于 [`Self::with_replay_safe`]；它只表达 @@ -880,6 +895,7 @@ impl ProviderError { continuation_recovery_disposition: self.continuation_recovery_disposition, failure_observation: self.failure_observation.clone(), replay_safe: self.replay_safe, + retry_prohibited: self.retry_prohibited, pre_delivery_retry: self.pre_delivery_retry.clone(), credential_recovery_required: self.credential_recovery_required, retry_same_account: self.retry_same_account, @@ -909,6 +925,7 @@ impl fmt::Debug for ProviderError { &self.upstream_request_id().map(|_| ""), ) .field("retry_after", &self.retry_after) + .field("retry_prohibited", &self.retry_prohibited) .field("continuation_failure", &self.continuation_failure) .field( "continuation_recovery_disposition", diff --git a/backend/crates/gateway-core/tests/engine/coordinator.rs b/backend/crates/gateway-core/tests/engine/coordinator.rs index ebf50f2c1..dfb9a79e0 100644 --- a/backend/crates/gateway-core/tests/engine/coordinator.rs +++ b/backend/crates/gateway-core/tests/engine/coordinator.rs @@ -3438,6 +3438,124 @@ fn transient_rejection() -> ProviderError { ) } +#[test] +fn server_retry_advice_should_override_local_backoff_and_transport_fallback() { + let advised_delay = Duration::from_millis(120); + for rejection in [ + transient_rejection().with_transient_retry( + NonZeroU32::MIN, + Duration::from_millis(1), + Duration::from_millis(1), + ), + ProviderError::new(ProviderErrorKind::Transport, UpstreamSendState::NotSent) + .with_pre_delivery_transport_fallback(), + ] { + let operation = generate_operation(); + let route_plan = plan(&operation); + let (coordinator, _, provider) = coordinator(vec![ + Script::Stream { + account_id: "acct_first", + items: vec![Err(rejection.with_retry_after(advised_delay))], + }, + Script::Stream { + account_id: "acct_first", + items: complete_stream(None), + }, + ]); + let mut session = block_on(coordinator.start( + model_request(&operation, SystemTime::now() + Duration::from_secs(30)), + operation, + route_plan, + None, + None, + CancellationToken::new(), + )) + .unwrap(); + let started = std::time::Instant::now(); + let mut collection = Box::pin(session.collect_uncommitted()); + assert!(block_on(async { futures::poll!(&mut collection) }).is_pending()); + assert_eq!(provider.contexts.lock().unwrap().len(), 1); + block_on(collection).unwrap(); + assert!(started.elapsed() >= advised_delay); + let contexts = provider.contexts.lock().unwrap(); + assert_eq!(contexts.len(), 2); + assert_eq!( + contexts[1].required_account(), + Some(&ProviderAccountId::new("acct_first").unwrap()) + ); + } +} + +#[test] +fn zero_server_retry_advice_should_override_nonzero_local_backoff() { + let operation = generate_operation(); + let route_plan = plan(&operation); + let (coordinator, _, _) = coordinator(vec![ + Script::Stream { + account_id: "acct_first", + items: vec![Err(transient_rejection() + .with_transient_retry( + NonZeroU32::MIN, + Duration::from_secs(60), + Duration::from_secs(60), + ) + .with_retry_after(Duration::ZERO))], + }, + Script::Stream { + account_id: "acct_first", + items: complete_stream(None), + }, + ]); + let mut session = block_on(coordinator.start( + model_request(&operation, SystemTime::now() + Duration::from_secs(30)), + operation, + route_plan, + None, + None, + CancellationToken::new(), + )) + .unwrap(); + let mut collection = Box::pin(session.collect_uncommitted()); + assert!(matches!( + block_on(async { futures::poll!(&mut collection) }), + std::task::Poll::Ready(Ok(_)) + )); +} + +#[test] +fn terminal_rejection_should_prevent_not_sent_rotation_and_explicit_transport_recovery() { + for fallback in [false, true] { + let operation = generate_operation(); + let route_plan = plan(&operation); + let mut error = + ProviderError::new(ProviderErrorKind::Unavailable, UpstreamSendState::NotSent) + .with_replay_safe() + .with_retry_prohibited(); + if fallback { + error = error.with_pre_delivery_transport_fallback(); + } + assert!(error.stable_snapshot().retry_is_prohibited()); + let (coordinator, _, provider) = coordinator(vec![Script::Stream { + account_id: "acct_first", + items: vec![Err(error)], + }]); + let mut session = block_on(coordinator.start( + model_request(&operation, SystemTime::now() + Duration::from_secs(30)), + operation, + route_plan, + None, + None, + CancellationToken::new(), + )) + .unwrap(); + assert!(matches!( + block_on(session.collect_uncommitted()), + Err(EngineError::Provider(_)) + )); + assert_eq!(provider.contexts.lock().unwrap().len(), 1); + } +} + #[test] fn transient_retries_back_off_on_the_same_account_then_rotate_with_a_fresh_budget() { let operation = generate_operation(); diff --git a/backend/crates/gateway-protocol/Cargo.toml b/backend/crates/gateway-protocol/Cargo.toml index 3ed2664a9..5af5138fe 100644 --- a/backend/crates/gateway-protocol/Cargo.toml +++ b/backend/crates/gateway-protocol/Cargo.toml @@ -12,6 +12,7 @@ name = "main" path = "tests/main.rs" [dependencies] +httpdate = "1.0.3" memchr = "2.8.3" serde = { version = "1.0.228", features = ["derive"] } serde_json = { version = "1.0.150", features = ["arbitrary_precision", "preserve_order"] } diff --git a/backend/crates/gateway-protocol/src/openai/codex.rs b/backend/crates/gateway-protocol/src/openai/codex.rs index 1fb38ff88..6bb754622 100644 --- a/backend/crates/gateway-protocol/src/openai/codex.rs +++ b/backend/crates/gateway-protocol/src/openai/codex.rs @@ -78,8 +78,12 @@ pub fn codex_responses_request_semantics_with_turn_metadata( let reasoning_effort = body .get("reasoning") .and_then(Value::as_object) - .and_then(|reasoning| non_empty_string(reasoning.get("effort"))) - .map(ToOwned::to_owned); + .and_then(|reasoning| reasoning.get("effort")) + .and_then(|effort| { + non_empty_string(Some(effort)) + .map(ToOwned::to_owned) + .or_else(|| effort.as_u64().map(|value| value.to_string())) + }); let parsed_turn_metadata = turn_metadata .or_else(|| request_turn_metadata(body)) .and_then(|value| serde_json::from_str::(value).ok()); diff --git a/backend/crates/gateway-protocol/src/openai/events.rs b/backend/crates/gateway-protocol/src/openai/events.rs index 90fdb1e5c..bf6aa2f45 100644 --- a/backend/crates/gateway-protocol/src/openai/events.rs +++ b/backend/crates/gateway-protocol/src/openai/events.rs @@ -248,7 +248,9 @@ pub fn retry_after_seconds_from_value(value: &Value) -> Option { .pointer("/response/error") .or_else(|| value.get("error")) .unwrap_or(value); - retry_after_seconds_field(error) + retry_after_seconds_header(error) + .or_else(|| retry_after_seconds_header(value)) + .or_else(|| retry_after_seconds_field(error)) .or_else(|| { error .get("resets_in_seconds") @@ -257,7 +259,6 @@ pub fn retry_after_seconds_from_value(value: &Value) -> Option { }) .or_else(|| retry_after_seconds_from_resets_at(error)) .or_else(|| retry_after_seconds_field(value)) - .or_else(|| retry_after_seconds_header(value)) .or_else(|| retry_after_seconds_from_rate_limit_message(error)) } @@ -280,7 +281,7 @@ fn retry_after_seconds_header(value: &Value) -> Option { .and_then(|headers| { headers.iter().find_map(|(name, value)| { if name.eq_ignore_ascii_case("retry-after") { - json_value_as_positive_u64(value) + json_value_as_retry_after_seconds(value) } else { None } @@ -288,14 +289,13 @@ fn retry_after_seconds_header(value: &Value) -> Option { }) } -fn json_value_as_positive_u64(value: &Value) -> Option { - let seconds = match value { - Value::Number(value) => value.as_u64()?, - Value::String(value) => value.trim().parse::().ok()?, - Value::Array(values) => values.first().and_then(json_value_as_positive_u64)?, - Value::Null | Value::Bool(_) | Value::Object(_) => return None, - }; - (seconds > 0).then_some(seconds) +fn json_value_as_retry_after_seconds(value: &Value) -> Option { + match value { + Value::Number(value) => value.as_u64(), + Value::String(value) => super::headers::parse_retry_after_seconds(value), + Value::Array(values) => values.first().and_then(json_value_as_retry_after_seconds), + Value::Null | Value::Bool(_) | Value::Object(_) => None, + } } /// 标准化的单个限流窗口 diff --git a/backend/crates/gateway-protocol/src/openai/headers.rs b/backend/crates/gateway-protocol/src/openai/headers.rs index 8a89b0229..e9ec350f9 100644 --- a/backend/crates/gateway-protocol/src/openai/headers.rs +++ b/backend/crates/gateway-protocol/src/openai/headers.rs @@ -1,4 +1,22 @@ -//! OpenAI 请求头的 HTTP 传输边界 +//! OpenAI 头部的传输边界与重试时间解析 + +/// 解析 HTTP 和流内错误共用的 Retry-After，保留零延迟并向上取整剩余秒数 +pub fn parse_retry_after_seconds(value: &str) -> Option { + let now = std::time::SystemTime::now(); + let value = value.trim(); + if !value.is_empty() && value.bytes().all(|byte| byte.is_ascii_digit()) { + return value.parse().ok(); + } + let remaining = httpdate::parse_http_date(value) + .ok()? + .duration_since(now) + .unwrap_or_default(); + Some( + remaining + .as_secs() + .saturating_add(u64::from(remaining.subsec_nanos() != 0)), + ) +} /// 判断小写请求头是否属于传输层管理的字段，不得作为业务扩展头透传 /// diff --git a/backend/crates/gateway-protocol/src/openai/mod.rs b/backend/crates/gateway-protocol/src/openai/mod.rs index c207ee6f6..c2df58c2d 100644 --- a/backend/crates/gateway-protocol/src/openai/mod.rs +++ b/backend/crates/gateway-protocol/src/openai/mod.rs @@ -21,4 +21,6 @@ pub use codex::{ CodexResponsesRequestSemantics, codex_responses_request_semantics, codex_responses_request_semantics_with_turn_metadata, codex_session_id, codex_thread_id, }; -pub use headers::{is_transport_managed_request_header, response_header_is_forwardable}; +pub use headers::{ + is_transport_managed_request_header, parse_retry_after_seconds, response_header_is_forwardable, +}; diff --git a/backend/crates/gateway-protocol/tests/openai/codex.rs b/backend/crates/gateway-protocol/tests/openai/codex.rs index 179683125..f69a4c269 100644 --- a/backend/crates/gateway-protocol/tests/openai/codex.rs +++ b/backend/crates/gateway-protocol/tests/openai/codex.rs @@ -5,6 +5,25 @@ use gateway_protocol::openai::{ }; use serde_json::{Map, Value, json}; +#[test] +fn reasoning_effort_should_observe_unsigned_numbers_without_changing_the_body() { + for (effort, expected) in [ + (json!(0), Some("0")), + (json!(64), Some("64")), + (json!(u64::MAX), Some("18446744073709551615")), + (json!(" high "), Some("high")), + (json!(-1), None), + (json!(1.5), None), + (json!(true), None), + (json!(null), None), + ] { + let body = object(json!({"reasoning":{"effort":effort}})); + let semantics = codex_responses_request_semantics(&body, &Map::new()); + assert_eq!(semantics.reasoning_effort.as_deref(), expected); + assert_eq!(body["reasoning"]["effort"], effort); + } +} + #[test] fn codex_session_should_prefer_the_connection_identity_over_body_metadata() { let body = object( diff --git a/backend/crates/gateway-protocol/tests/openai/events.rs b/backend/crates/gateway-protocol/tests/openai/events.rs index fec469526..cb7a6afd4 100644 --- a/backend/crates/gateway-protocol/tests/openai/events.rs +++ b/backend/crates/gateway-protocol/tests/openai/events.rs @@ -464,6 +464,44 @@ fn retry_after_should_reject_malformed_json() { assert_eq!(retry_after_seconds_from_body("not-json"), None); } +#[test] +fn retry_after_should_prioritize_nested_headers_over_other_advice() { + for header in [json!(30), json!("30"), json!(["30"])] { + for wrap_response in [false, true] { + let error = json!({ + "code": "rate_limit_exceeded", "message": "Try again in 2s", + "retry_after_seconds": 10, "headers": {"ReTrY-AfTeR": header} + }); + let body = if wrap_response { + json!({"type":"response.failed", "response":{"error":error}, "headers":{"retry-after":"20"}}) + } else { + json!({"type":"error", "error":error, "headers":{"retry-after":"20"}}) + }; + assert_eq!(retry_after_seconds_from_body(&body.to_string()), Some(30)); + } + } +} + +#[test] +fn retry_after_should_keep_zero_advice_and_fall_back_only_for_invalid_headers() { + for (header, expected) in [ + (json!("0"), 0), + (json!(0), 0), + (json!("Sun, 06 Nov 1994 08:49:37 GMT"), 0), + (json!("invalid"), 2), + (json!({"seconds":30}), 2), + ] { + let body = json!({"type":"response.failed", "response":{"error":{ + "code":"rate_limit_exceeded", "message":"Try again in 2s", + "headers":{"Retry-After":header} + }}}); + assert_eq!( + retry_after_seconds_from_body(&body.to_string()), + Some(expected) + ); + } +} + #[test] fn parse_rate_limit_headers_should_extract_primary_and_secondary_windows() { let headers = vec![ diff --git a/backend/crates/gateway-protocol/tests/openai/headers.rs b/backend/crates/gateway-protocol/tests/openai/headers.rs index a03dfe1df..02fd4108c 100644 --- a/backend/crates/gateway-protocol/tests/openai/headers.rs +++ b/backend/crates/gateway-protocol/tests/openai/headers.rs @@ -1,9 +1,40 @@ //! 验证传输头部分类，以及响应中的连接级和身份字段过滤 use gateway_protocol::openai::{ - is_transport_managed_request_header, response_header_is_forwardable, + is_transport_managed_request_header, parse_retry_after_seconds, response_header_is_forwardable, }; +#[test] +fn retry_after_should_accept_nonnegative_seconds_and_http_dates() { + for (header, expected) in [ + (" 30 ", Some(30)), + ("0", Some(0)), + ("Sun, 06 Nov 1994 08:49:37 GMT", Some(0)), + ("Sunday, 06-Nov-94 08:49:37 GMT", Some(0)), + ("Sun Nov 6 08:49:37 1994", Some(0)), + ("+30", None), + ("-1", None), + ("1.5", None), + ("18446744073709551616", None), + ("", None), + ("later", None), + ] { + assert_eq!(parse_retry_after_seconds(header), expected, "{header}"); + } +} + +#[test] +fn retry_after_http_date_should_not_round_down_the_remaining_delay() { + use std::time::{Duration, SystemTime, UNIX_EPOCH}; + let now = SystemTime::now().duration_since(UNIX_EPOCH).unwrap(); + let retry_at = UNIX_EPOCH + Duration::from_secs(now.as_secs() + 60); + let header = httpdate::fmt_http_date(retry_at); + let seconds = parse_retry_after_seconds(&header).expect("HTTP date"); + let remaining = retry_at.duration_since(SystemTime::now()).unwrap(); + assert!(Duration::from_secs(seconds) >= remaining); + assert!(seconds <= 60); +} + #[test] fn transport_headers_should_include_hop_fields_and_compression() { for name in [ diff --git a/backend/crates/providers/openai/Cargo.toml b/backend/crates/providers/openai/Cargo.toml index 2698f4ac4..86efa7666 100644 --- a/backend/crates/providers/openai/Cargo.toml +++ b/backend/crates/providers/openai/Cargo.toml @@ -27,6 +27,7 @@ hex = "0.4.3" # hyper 受 workspace 的 SOCKS IPv6 修复补丁约束，保留补丁要求的兼容版本。 # 升级前须重复采样 ClientHello，核对算法、扩展集合及排序约束、key_share 与 ALPN。 aws-lc-rs = "=1.18.1" +h2 = "=0.4.19" hyper = { version = "=1.11.1", default-features = false } hyper-rustls = "=0.27.7" hyper-util = { version = "0.1.20", features = ["client-legacy", "http1", "tokio"] } diff --git a/backend/crates/providers/openai/src/admin.rs b/backend/crates/providers/openai/src/admin.rs index 1cbc71cc3..2a3deccf3 100644 --- a/backend/crates/providers/openai/src/admin.rs +++ b/backend/crates/providers/openai/src/admin.rs @@ -207,21 +207,22 @@ impl ProviderAdmin for OpenAiAdminProvider { } fn plan_type_display(&self, plan_type: &str) -> String { - // 与官方 Desktop 的套餐名称映射一致；原始子类型继续由 plan_type 保留 + // Pro 系列按套餐类型区分展示，其他套餐遵循官方 KnownPlan::display_name + // 只映射展示名称，原始套餐值与未知套餐保持不变 match plan_type.to_ascii_lowercase().as_str() { - "free" | "free_workspace" | "guest" => "Free", + "free" => "Free", "go" => "Go", "plus" => "Plus", - "pro" | "prolite" => "Pro", - "team" | "self_serve_business_prolite" | "self_serve_business_usage_based" => { - "Business" - } - "business" - | "ent26" - | "enterprise_cbp_automation" - | "enterprise_cbp_usage_based" - | "enterprise" - | "hc" => "Enterprise", + "pro" => "Pro", + "prolite" => "ProLite", + "promax" => "ProMax", + "team" => "Team", + "self_serve_business_prolite" => "Self Serve Business ProLite", + "self_serve_business_usage_based" => "Self Serve Business Usage Based", + "business" => "Business", + "ent26" | "enterprise" | "hc" => "Enterprise", + "enterprise_cbp_automation" => "Enterprise (Automation)", + "enterprise_cbp_usage_based" => "Enterprise CBP Usage Based", "edu" | "education" => "Edu", "edu_plus" => "Edu Plus", "edu_pro" => "Edu Pro", diff --git a/backend/crates/providers/openai/src/credential/token_client.rs b/backend/crates/providers/openai/src/credential/token_client.rs index ad322ef88..0560e415d 100644 --- a/backend/crates/providers/openai/src/credential/token_client.rs +++ b/backend/crates/providers/openai/src/credential/token_client.rs @@ -23,7 +23,7 @@ pub const OFFICIAL_CODEX_OAUTH_CLIENT_ID: &str = "app_EMoamEEZ73f0CkXaXp7hrann"; /// Codex Desktop 使用的官方 token endpoint pub const OFFICIAL_CODEX_TOKEN_ENDPOINT: &str = "https://auth.openai.com/oauth/token"; /// Codex Desktop loopback callback；管理员复制完整回调 URL 交回固定 complete API -pub const OFFICIAL_CODEX_REDIRECT_URI: &str = "http://localhost:1455/auth/callback"; +pub const OFFICIAL_CODEX_REDIRECT_URI: &str = "http://127.0.0.1:1455/auth/callback"; const PERSONAL_ACCESS_TOKEN_WHOAMI_PATH: &str = "/api/accounts/v1/user-auth-credential/whoami"; /// PAT 验证失败；不保留令牌、响应体或可能包含秘密的底层 HTTP 错误 diff --git a/backend/crates/providers/openai/src/credential/types.rs b/backend/crates/providers/openai/src/credential/types.rs index 84ad38c8e..c79d4eb13 100644 --- a/backend/crates/providers/openai/src/credential/types.rs +++ b/backend/crates/providers/openai/src/credential/types.rs @@ -170,6 +170,7 @@ enum KnownChatgptPlan { Plus, Pro, ProLite, + ProMax, Team, #[serde(rename = "self_serve_business_prolite")] SelfServeBusinessProLite, @@ -185,6 +186,10 @@ enum KnownChatgptPlan { Enterprise, #[serde(alias = "education")] Edu, + #[serde(rename = "edu_plus")] + EduPlus, + #[serde(rename = "edu_pro")] + EduPro, } impl KnownChatgptPlan { @@ -195,6 +200,7 @@ impl KnownChatgptPlan { Self::Plus => "plus", Self::Pro => "pro", Self::ProLite => "prolite", + Self::ProMax => "promax", Self::Team => "team", Self::SelfServeBusinessProLite => "self_serve_business_prolite", Self::SelfServeBusinessUsageBased => "self_serve_business_usage_based", @@ -204,6 +210,8 @@ impl KnownChatgptPlan { Self::EnterpriseCbpUsageBased => "enterprise_cbp_usage_based", Self::Enterprise => "enterprise", Self::Edu => "edu", + Self::EduPlus => "edu_plus", + Self::EduPro => "edu_pro", } } } diff --git a/backend/crates/providers/openai/src/provider/failure.rs b/backend/crates/providers/openai/src/provider/failure.rs index 0a430f8ab..84dc9baf8 100644 --- a/backend/crates/providers/openai/src/provider/failure.rs +++ b/backend/crates/providers/openai/src/provider/failure.rs @@ -545,6 +545,10 @@ pub(super) fn apply_websocket_recovery_policy( failure: &mut MappedProviderFailure, context: WebSocketRecoveryContext<'_>, ) { + // Flex 拒绝即使发生在 WS opening 阶段也必须直达客户端，不能转换成传输回退 + if failure.error.retry_is_prohibited() { + return; + } // 明确账号拒绝走已有换号路径，容量拒绝走请求内退避；两者都不消耗 WS 传输预算 if failure.error.replay_is_safe() && (failure.account_failure.is_some() @@ -1351,6 +1355,9 @@ pub(super) fn map_upstream_failure( provider_error_kind(category) }; let mut error = provider_error(error_kind, send_state); + if category == CodexFailureCategory::FlexUnavailable { + error = error.with_retry_prohibited(); + } error = error.with_raw_upstream_error(RawUpstreamError::new(failure.raw_body.clone())); if let Some(message) = failure.client_message.as_ref() { error = error.with_client_visible_upstream_error(ClientVisibleUpstreamError::new( @@ -1399,10 +1406,7 @@ pub(super) fn map_upstream_failure( let max_delay = Duration::from_secs(8); error = error.with_transient_retry( NonZeroU32::new(3).unwrap_or(NonZeroU32::MIN), - failure - .retry_after_seconds - .map_or(Duration::from_millis(500), Duration::from_secs) - .min(max_delay), + Duration::from_millis(500), max_delay, ); } @@ -1474,6 +1478,7 @@ pub(super) const fn provider_error_kind(category: CodexFailureCategory) -> Provi CodexFailureCategory::QuotaExhausted => ProviderErrorKind::QuotaExhausted, CodexFailureCategory::CloudflareChallenge | CodexFailureCategory::CloudflarePathBlocked + | CodexFailureCategory::FlexUnavailable | CodexFailureCategory::Unavailable => ProviderErrorKind::Unavailable, CodexFailureCategory::CapacityUnavailable => ProviderErrorKind::UpstreamCapacityUnavailable, CodexFailureCategory::InvalidRequest => ProviderErrorKind::InvalidRequest, @@ -1528,6 +1533,7 @@ pub(super) fn account_failure( | CodexFailureCategory::PermissionDenied | CodexFailureCategory::Timeout | CodexFailureCategory::CapacityUnavailable + | CodexFailureCategory::FlexUnavailable | CodexFailureCategory::Unavailable | CodexFailureCategory::Transport => None, } diff --git a/backend/crates/providers/openai/src/transport/client.rs b/backend/crates/providers/openai/src/transport/client.rs index 7d3243318..706713d2e 100644 --- a/backend/crates/providers/openai/src/transport/client.rs +++ b/backend/crates/providers/openai/src/transport/client.rs @@ -10,7 +10,6 @@ use std::{ use crate::transport::profile::CodexWireProfileState; use bytes::Bytes; -use chrono::{DateTime, NaiveDateTime, Utc}; use futures::{Stream, StreamExt}; use gateway_core::engine::middleware::MiddlewareHeader; use gateway_protocol::openai::{ @@ -882,33 +881,10 @@ pub(super) fn retry_after_seconds(headers: &HeaderMap, body: Option<&str>) -> Op headers .get(RETRY_AFTER) .and_then(|value| value.to_str().ok()) - .and_then(parse_retry_after) + .and_then(gateway_protocol::openai::parse_retry_after_seconds) .or_else(|| body.and_then(retry_after_seconds_from_body)) } -/// 解析 RFC 7231 `Retry-After` 值：delay-seconds 或 HTTP-date，统一换算为剩余秒数 -pub(super) fn parse_retry_after(value: &str) -> Option { - let value = value.trim(); - if let Ok(seconds) = value.parse::() { - return (seconds > 0).then_some(seconds); - } - let remaining = parse_http_date(value)?.signed_duration_since(Utc::now()); - u64::try_from(remaining.num_seconds()) - .ok() - .filter(|seconds| *seconds > 0) -} - -/// IMF-fixdate 优先；RFC 7231 要求接收方兼容过时的 RFC 850 与 asctime 形式 -fn parse_http_date(value: &str) -> Option> { - if let Ok(date) = DateTime::parse_from_rfc2822(value) { - return Some(date.with_timezone(&Utc)); - } - ["%A, %d-%b-%y %H:%M:%S GMT", "%a %b %e %H:%M:%S %Y"] - .iter() - .find_map(|format| NaiveDateTime::parse_from_str(value, format).ok()) - .map(|date| date.and_utc()) -} - pub(super) fn truncate_for_error(body: &str) -> String { body.chars().take(200).collect() } diff --git a/backend/crates/providers/openai/src/transport/diagnostics.rs b/backend/crates/providers/openai/src/transport/diagnostics.rs index 2460670b7..49954179e 100644 --- a/backend/crates/providers/openai/src/transport/diagnostics.rs +++ b/backend/crates/providers/openai/src/transport/diagnostics.rs @@ -35,6 +35,7 @@ const PERSISTABLE_UPSTREAM_CODES: &[&str] = &[ "billing_limit", "cyber_policy", "deactivated_workspace", + "flex_unavailable", "identity_verification_required", "insufficient_quota", "invalid_api_key", @@ -104,6 +105,8 @@ pub enum CodexFailureCategory { Timeout, /// 模型容量暂时不足，只影响当前请求，不证明账号健康异常 CapacityUnavailable, + /// Flex 容量不足是当前请求的终止错误，不冷却账号或自动重放 + FlexUnavailable, Unavailable, Transport, } @@ -231,6 +234,9 @@ impl CodexUpstreamFailure { /// 返回该拒绝是否允许换号重放 #[must_use] pub const fn replay_is_safe(&self) -> bool { + if matches!(self.category, CodexFailureCategory::FlexUnavailable) { + return false; + } match self.send_phase { CodexUpstreamSendPhase::BeforePayload => true, CodexUpstreamSendPhase::Ambiguous => false, @@ -431,6 +437,10 @@ fn classify_upstream_failure( let message = fields.message.to_ascii_lowercase(); let body = body.to_ascii_lowercase(); + if code == "flex_unavailable" { + return CodexFailureCategory::FlexUnavailable; + } + // 容量拒绝可能带 400/429/503；仅用结构化错误字段识别，不能扫描任意正文 if is_capacity_error( fields.code.as_deref(), @@ -441,7 +451,7 @@ fn classify_upstream_failure( } // 与官方 Codex HTTP/WS 路径一致：429 只有结构化 - // `error.type=usage_limit_reached` 才能确认额度窗口耗尽；其余 429 都是临时限流 + // `error.type=usage_limit_reached` 才能确认额度窗口耗尽；排除容量错误后才按临时限流处理 // SSE `response.failed` 的结构化字段形态不同，保留其 code/type 语义单独分类 if matches!(source, UpstreamFailureSource::HttpResponse) && status == Some(StatusCode::TOO_MANY_REQUESTS) diff --git a/backend/crates/providers/openai/src/transport/websocket/handshake.rs b/backend/crates/providers/openai/src/transport/websocket/handshake.rs index 3d609c147..a010c5944 100644 --- a/backend/crates/providers/openai/src/transport/websocket/handshake.rs +++ b/backend/crates/providers/openai/src/transport/websocket/handshake.rs @@ -21,10 +21,8 @@ use crate::{ responses::CodexResponsesRequest, websocket::websocket_response_create_payload_text, }, transport::{ - client::{CodexClientVisibleUpstreamResponse, parse_retry_after}, - diagnostics::CodexUpstreamSendPhase, - endpoints::CODEX_RESPONSES_PATH, - response_meta, tls, + client::CodexClientVisibleUpstreamResponse, diagnostics::CodexUpstreamSendPhase, + endpoints::CODEX_RESPONSES_PATH, response_meta, tls, }, }; @@ -364,7 +362,7 @@ fn websocket_opening_error(response: &WsResponse>>) -> CodexWebSo .headers() .get("retry-after") .and_then(|value| value.to_str().ok()) - .and_then(parse_retry_after) + .and_then(gateway_protocol::openai::parse_retry_after_seconds) .or_else(|| events::retry_after_seconds_from_body(&body)); CodexWebSocketExchangeError::upstream( status_code, diff --git a/backend/crates/providers/openai/tests/admin.rs b/backend/crates/providers/openai/tests/admin.rs index 3d8e451d7..43772bd0a 100644 --- a/backend/crates/providers/openai/tests/admin.rs +++ b/backend/crates/providers/openai/tests/admin.rs @@ -66,6 +66,42 @@ const COMPLETED_SESSION_SSE: &str = concat!( "data: {\"type\":\"response.completed\",\"response\":{\"id\":\"resp_initialized_session\",\"model\":\"gpt-5.4\",\"status\":\"completed\",\"output\":[],\"usage\":{\"input_tokens\":1,\"output_tokens\":1,\"total_tokens\":2}}}\n\n" ); +#[tokio::test] +async fn plan_display_should_distinguish_pro_tiers_and_preserve_other_official_names() { + let config = valid_config(); + let bundle = provider_openai::initialize(config.config.clone(), provider_ports()) + .await + .unwrap(); + let admin = bundle.admin_provider(); + for (raw, display) in [ + ("prolite", "ProLite"), + ("pro", "Pro"), + ("promax", "ProMax"), + ("free", "Free"), + ("go", "Go"), + ("plus", "Plus"), + ("team", "Team"), + ("self_serve_business_prolite", "Self Serve Business ProLite"), + ( + "self_serve_business_usage_based", + "Self Serve Business Usage Based", + ), + ("business", "Business"), + ("ent26", "Enterprise"), + ("enterprise", "Enterprise"), + ("hc", "Enterprise"), + ("enterprise_cbp_automation", "Enterprise (Automation)"), + ("enterprise_cbp_usage_based", "Enterprise CBP Usage Based"), + ("edu", "Edu"), + ("education", "Edu"), + ("edu_plus", "Edu Plus"), + ("edu_pro", "Edu Pro"), + ("future_plan", "future_plan"), + ] { + assert_eq!(admin.plan_type_display(raw), display); + } +} + #[tokio::test] async fn account_capabilities_distinguish_oauth_from_api_key_and_unknown_credentials() { let config = valid_config(); @@ -645,7 +681,7 @@ async fn openai_admin_provider_persists_the_full_pending_envelope_and_binds_owne request_id: "request-complete".to_owned(), }, flow_id: started.flow_id, - callback_url: "http://localhost:1455/auth/callback?code=unused&state=unused".to_owned(), + callback_url: "http://127.0.0.1:1455/auth/callback?code=unused&state=unused".to_owned(), }) .await .expect_err("wrong owner"); diff --git a/backend/crates/providers/openai/tests/credential/admin.rs b/backend/crates/providers/openai/tests/credential/admin.rs index 78eb91a13..008c0c356 100644 --- a/backend/crates/providers/openai/tests/credential/admin.rs +++ b/backend/crates/providers/openai/tests/credential/admin.rs @@ -720,6 +720,25 @@ async fn oauth_import_uses_official_plan_alias_projection_and_user_id_fallback() assert_eq!(account.plan_type(), Some("enterprise")); } +#[tokio::test] +async fn oauth_import_should_preserve_pro_max_and_future_plan_claims() { + let service = CodexCredentialAdminService::new( + Arc::new(UnusedRefresher), + Arc::new(TestLeaseCoordinator::default()), + runtime_policy(), + ); + for plan in ["prolite", "pro", "promax", "future_plan"] { + let access_token = test_jwt(serde_json::json!({"https://api.openai.com/auth":{ + "chatgpt_user_id":"plan-user", "chatgpt_plan_type":plan + }})); + let prepared = service + .prepare_import_document(serde_json::json!({"accessToken":access_token})) + .await + .unwrap(); + assert_eq!(prepared.accounts()[0].account.plan_type(), Some(plan)); + } +} + #[tokio::test] async fn oauth_import_uses_id_token_then_access_token_for_missing_claims() { let id_token = test_jwt(serde_json::json!({ diff --git a/backend/crates/providers/openai/tests/credential/oauth.rs b/backend/crates/providers/openai/tests/credential/oauth.rs index d4ddf6f24..f56bfc838 100644 --- a/backend/crates/providers/openai/tests/credential/oauth.rs +++ b/backend/crates/providers/openai/tests/credential/oauth.rs @@ -194,7 +194,7 @@ async fn complete( owner_ref: "test-owner".to_owned(), flow_id: started.flow_id, callback_url: SecretString::from(format!( - "http://localhost:1455/auth/callback?code=code-from-browser&state={state}" + "http://127.0.0.1:1455/auth/callback?code=code-from-browser&state={state}" )), }) .await?; @@ -299,7 +299,7 @@ async fn authorize_url_matches_the_official_desktop_parameter_contract() { ); assert_eq!( parameters.get("redirect_uri").map(String::as_str), - Some("http://localhost:1455/auth/callback") + Some("http://127.0.0.1:1455/auth/callback") ); assert_eq!( parameters.get("scope").map(String::as_str), diff --git a/backend/crates/providers/openai/tests/credential/token_client.rs b/backend/crates/providers/openai/tests/credential/token_client.rs index 198fa800a..fe23fb365 100644 --- a/backend/crates/providers/openai/tests/credential/token_client.rs +++ b/backend/crates/providers/openai/tests/credential/token_client.rs @@ -328,7 +328,7 @@ async fn authorization_code_exchange_should_require_bounded_oidc_token_set_and_p assert!(body.contains("client_id=test-public-client")); assert!(body.contains("code=authorization+code")); assert!(body.contains("code_verifier=pkce-verifier-secret")); - assert!(body.contains("redirect_uri=http%3A%2F%2Flocalhost%3A1455%2Fauth%2Fcallback")); + assert!(body.contains("redirect_uri=http%3A%2F%2F127.0.0.1%3A1455%2Fauth%2Fcallback")); } #[tokio::test] diff --git a/backend/crates/providers/openai/tests/provider/contract/mod.rs b/backend/crates/providers/openai/tests/provider/contract/mod.rs index 424af1eac..c3414e9cd 100644 --- a/backend/crates/providers/openai/tests/provider/contract/mod.rs +++ b/backend/crates/providers/openai/tests/provider/contract/mod.rs @@ -8025,10 +8025,11 @@ async fn capacity_http_and_websocket_opening_rejections_use_bounded_business_ret assert_eq!(error.kind(), ProviderErrorKind::UpstreamCapacityUnavailable); assert_eq!(error.upstream_status(), Some(status)); assert!(error.replay_is_safe()); + assert_eq!(error.retry_after(), Some(Duration::from_secs(129600))); assert!( matches!(error.pre_delivery_retry(), Some(PreDeliveryRetry::SameAccountTransientRetry { max_retries, initial_delay, max_delay, - }) if max_retries.get() == 3 && initial_delay == Duration::from_secs(8) && max_delay == Duration::from_secs(8)) + }) if max_retries.get() == 3 && initial_delay == Duration::from_millis(500) && max_delay == Duration::from_secs(8)) ); assert!(provider_openai::openai_failure_affects_account_score( &error @@ -8052,6 +8053,68 @@ async fn capacity_http_and_websocket_opening_rejections_use_bounded_business_ret } } +#[tokio::test] +async fn flex_http_and_websocket_opening_rejections_should_not_retry_or_cool_down_accounts() { + use gateway_core::provider_ports::ProviderCooldownPort as _; + for use_websocket in [false, true] { + let store = Arc::new(MemoryAccountStore::default()); + let account_id = "acct_provider_contract"; + create_account(&store, account_id).await; + let cooldowns = Arc::new(MemoryCooldownPort::new()); + let server = MockServer::start().await; + let body = json!({"error":{"type":"resource_unavailable","code":"flex_unavailable","message":"Flex capacity unavailable."}}); + Mock::given(method(if use_websocket { "GET" } else { "POST" })) + .and(path("/codex/responses")) + .respond_with( + ResponseTemplate::new(429) + .insert_header("retry-after", "300") + .set_body_json(&body), + ) + .expect(1) + .mount(&server) + .await; + let operation = if use_websocket { + generate_operation() + } else { + http_generate_operation() + }; + let (provider, _) = + provider_with_capacity_tracking(&store, server.uri(), Arc::clone(&cooldowns)); + let mut stream = provider + .execute( + planned_request("openai", operation), + context("req_flex", CancellationToken::new()), + ) + .await + .unwrap(); + let error = loop { + match stream.next().await { + Some(Ok(_)) => {} + Some(Err(error)) => break error, + None => panic!("expected Flex failure"), + } + }; + assert_eq!(error.kind(), ProviderErrorKind::Unavailable); + assert_eq!( + error.upstream_code().map(|code| code.as_str()), + Some("flex_unavailable") + ); + assert!(!error.replay_is_safe()); + assert!(error.pre_delivery_retry().is_none()); + assert!(error.retry_is_prohibited()); + assert!(!provider_openai::openai_failure_affects_account_score( + &error + )); + let response = error.client_visible_upstream_response().unwrap(); + assert_eq!(response.status(), 429); + assert_eq!(response.body().as_ref(), body.to_string().as_bytes()); + let account = store.account(account_id).unwrap(); + assert_eq!(account.credential_state(), CredentialState::Ready); + assert_eq!(account.quota().access(), QuotaAccessState::Unknown); + assert!(cooldowns.read(account.id()).await.unwrap().is_none()); + } +} + fn capacity_websocket_operation() -> Operation { let payload = ProtocolPayload::json_object( "openai", @@ -8302,6 +8365,12 @@ async fn capacity_feedback_in_stream_only_counts_explicit_overload() { ProviderErrorKind::UpstreamCapacityUnavailable, true, ), + ( + "flex_unavailable", + "Flex capacity unavailable.", + ProviderErrorKind::Unavailable, + false, + ), ( "invalid_prompt", "Invalid prompt: we've limited access to this content for safety reasons.", @@ -9589,20 +9658,21 @@ fn endpoint_observation_should_read_models_without_rewriting_or_requiring_a_cata #[test] fn request_observation_preserves_the_raw_reasoning_effort() { let store = Arc::new(MemoryAccountStore::default()); - let payload = ProtocolPayload::json_object( - "openai", - Map::from_iter([("reasoning".to_owned(), json!({"effort": "future-value"}))]), - ) - .expect("protocol payload"); - let operation = Operation::Generate(GenerateRequest::from_protocol_payload(payload)); - - let client_key_id = ClientApiKeyId::new("key_openai_observation").expect("client key"); - let observation = provider(&store).request_observation(&operation, &client_key_id); - - assert_eq!( - observation.reasoning_effort.as_deref(), - Some("future-value") - ); + for (effort, expected) in [ + (json!("future-value"), "future-value"), + (json!(64), "64"), + (json!(0), "0"), + ] { + let payload = ProtocolPayload::json_object( + "openai", + Map::from_iter([("reasoning".to_owned(), json!({"effort":effort}))]), + ) + .unwrap(); + let operation = Operation::Generate(GenerateRequest::from_protocol_payload(payload)); + let client_key_id = ClientApiKeyId::new("key_openai_observation").unwrap(); + let observation = provider(&store).request_observation(&operation, &client_key_id); + assert_eq!(observation.reasoning_effort.as_deref(), Some(expected)); + } } #[test] diff --git a/backend/crates/providers/openai/tests/transport/diagnostics.rs b/backend/crates/providers/openai/tests/transport/diagnostics.rs index 503a76f4f..5f087732d 100644 --- a/backend/crates/providers/openai/tests/transport/diagnostics.rs +++ b/backend/crates/providers/openai/tests/transport/diagnostics.rs @@ -22,6 +22,27 @@ fn classify(status: u16, body: &str) -> CodexFailureCategory { .category() } +#[test] +fn flex_capacity_rejection_should_be_terminal_even_before_payload_send() { + for phase in [ + CodexUpstreamSendPhase::BeforePayload, + CodexUpstreamSendPhase::AfterPayload, + ] { + let failure = CodexUpstreamFailure::from_response( + reqwest::StatusCode::TOO_MANY_REQUESTS, + r#"{"error":{"type":"resource_unavailable","code":"flex_unavailable","message":"Flex capacity unavailable."}}"#, + Some(300), + &CodexUpstreamDiagnostics::default(), + None, + &[], + &[], + phase, + ); + assert_eq!(failure.category(), CodexFailureCategory::FlexUnavailable); + assert!(!failure.replay_is_safe()); + } +} + #[test] fn capacity_rejections_are_request_scoped_across_http_statuses() { for status in [400, 429, 503] { diff --git a/backend/crates/providers/openai/tests/transport/tls.rs b/backend/crates/providers/openai/tests/transport/tls.rs index 226284a0b..fd7da18a0 100644 --- a/backend/crates/providers/openai/tests/transport/tls.rs +++ b/backend/crates/providers/openai/tests/transport/tls.rs @@ -6,10 +6,15 @@ use bytes::{Buf as _, Bytes}; use super::*; -// 2026-09-19 从 Desktop 26.915.31945 / Core 0.155.0-alpha.9.2 各重复抓取三次 +// 基线来自官方 Codex 823ea830c0 的客户端源码，各路径重复采样五次 +// HTTP 另与本机 Core 0.160.0 的采样核对，源码客户端与本项目还通过真实 models 请求验证 // HTTP/native-tls 的扩展顺序固定；WebSocket/rustls 会随机化扩展顺序和临时密钥 #[derive(Debug, PartialEq, Eq)] struct ClientHello { + legacy_version: u16, + supported_versions: Vec, + point_formats: Vec, + psk_modes: Vec, cipher_suites: Vec, extensions: Vec, groups: Vec, @@ -19,11 +24,11 @@ struct ClientHello { } #[test] -fn http_client_hello_should_match_official_native_tls_transport() { - const CASE_ENV: &str = "CODEX_PROXY_TEST_NATIVE_TLS_CLIENT_HELLO"; - const CASE_COMPLETED: &str = "native-tls-client-hello-case-completed"; +fn http_client_hellos_should_match_official_tls_transports() { + const CASE_ENV: &str = "CODEX_PROXY_TEST_HTTP_CLIENT_HELLO"; + const CASE_COMPLETED: &str = "http-tls-client-hello-case-completed"; - if std::env::var_os(CASE_ENV).is_some() { + if let Ok(case) = std::env::var(CASE_ENV) { let runtime = tokio::runtime::Builder::new_current_thread() .enable_all() .build() @@ -46,7 +51,13 @@ fn http_client_hello_should_match_official_native_tls_transport() { ); hello }); - assert_eq!(hello, official_http_hello()); + if case == "custom-ca" { + let mut normalized = hello; + normalized.extensions.sort_unstable(); + assert_eq!(normalized, official_http_rustls_hello()); + } else { + assert_eq!(hello, official_http_hello()); + } println!("\n{CASE_COMPLETED}"); return; } @@ -54,30 +65,38 @@ fn http_client_hello_should_match_official_native_tls_transport() { // Cargo 运行测试时可能注入 SSL_CERT_FILE，生产代码会把它视作自定义 CA 并切到 rustls； // 因此在清理相关环境变量的子进程里验证默认路径 let current_exe = std::env::current_exe().expect("current test binary path"); - let output = Command::new(current_exe) - .arg("--exact") - .arg("transport::tls::http_client_hello_should_match_official_native_tls_transport") - .arg("--nocapture") - .env(CASE_ENV, "1") - .env_remove(provider_openai::transport::tls::CODEX_CA_CERT_ENV) - .env_remove(provider_openai::transport::tls::SSL_CERT_FILE_ENV) - .output() - .expect("run isolated native TLS ClientHello case"); - assert!( - output.status.success(), - "isolated native TLS ClientHello case failed\nstdout:\n{}\nstderr:\n{}", - String::from_utf8_lossy(&output.stdout), - String::from_utf8_lossy(&output.stderr) - ); - let stdout = String::from_utf8_lossy(&output.stdout); - assert_eq!( - stdout - .lines() - .filter(|line| *line == CASE_COMPLETED) - .count(), - 1, - "isolated native TLS ClientHello case did not complete exactly once\nstdout:\n{stdout}" - ); + let directory = tempfile::tempdir().unwrap(); + let ca_path = directory.path().join("ca.pem"); + std::fs::write(&ca_path, TEST_CA_PEM).unwrap(); + for case in ["native", "custom-ca"] { + let mut command = Command::new(¤t_exe); + command + .arg("--exact") + .arg("transport::tls::http_client_hellos_should_match_official_tls_transports") + .arg("--nocapture") + .env(CASE_ENV, case) + .env_remove(provider_openai::transport::tls::CODEX_CA_CERT_ENV) + .env_remove(provider_openai::transport::tls::SSL_CERT_FILE_ENV); + if case == "custom-ca" { + command.env(provider_openai::transport::tls::CODEX_CA_CERT_ENV, &ca_path); + } + let output = command.output().expect("run isolated TLS ClientHello case"); + assert!( + output.status.success(), + "isolated {case} TLS ClientHello case failed\nstdout:\n{}\nstderr:\n{}", + String::from_utf8_lossy(&output.stdout), + String::from_utf8_lossy(&output.stderr) + ); + let stdout = String::from_utf8_lossy(&output.stdout); + assert_eq!( + stdout + .lines() + .filter(|line| *line == CASE_COMPLETED) + .count(), + 1, + "isolated {case} TLS ClientHello case did not complete exactly once\nstdout:\n{stdout}" + ); + } } #[tokio::test] @@ -107,6 +126,10 @@ async fn websocket_client_hello_should_match_official_rustls_transport() { fn official_http_hello() -> ClientHello { ClientHello { + legacy_version: 771, + supported_versions: vec![772, 771], + point_formats: vec![0], + psk_modes: vec![1], cipher_suites: vec![ 4866, 4867, 4865, 49196, 49200, 159, 52393, 52392, 52394, 49195, 49199, 158, 49188, 49192, 107, 49187, 49191, 103, 49162, 49172, 57, 49161, 49171, 51, 157, 156, 61, 60, @@ -125,6 +148,10 @@ fn official_http_hello() -> ClientHello { fn official_websocket_hello() -> ClientHello { ClientHello { + legacy_version: 771, + supported_versions: vec![772, 771], + point_formats: vec![0], + psk_modes: vec![1], cipher_suites: vec![ 4866, 4865, 4867, 49196, 49195, 52393, 49200, 49199, 52392, 255, ], @@ -138,6 +165,14 @@ fn official_websocket_hello() -> ClientHello { } } +fn official_http_rustls_hello() -> ClientHello { + let mut hello = official_websocket_hello(); + hello.extensions.push(16); + hello.extensions.sort_unstable(); + hello.alpn = vec!["h2".to_owned(), "http/1.1".to_owned()]; + hello +} + async fn read_client_hello(listener: TcpListener) -> ClientHello { let (mut stream, _) = listener.accept().await.unwrap(); let mut header = [0; 5]; @@ -151,7 +186,9 @@ async fn read_client_hello(listener: TcpListener) -> ClientHello { let mut hello = Bytes::from(record); assert_eq!(hello.get_u8(), 1, "ClientHello message"); - hello.advance(3 + 2 + 32); // 消息长度、旧版协议版本与随机数 + hello.advance(3); // 消息长度 + let legacy_version = hello.get_u16(); + hello.advance(32); // 随机数 let session_id_len = usize::from(hello.get_u8()); hello.advance(session_id_len); let cipher_suites = u16_values(take_vector(&mut hello)); @@ -187,6 +224,10 @@ async fn read_client_hello(listener: TcpListener) -> ClientHello { } } ClientHello { + legacy_version, + supported_versions: u16_values(values[&43].slice(1..)), + point_formats: values[&11].slice(1..).to_vec(), + psk_modes: values[&45].slice(1..).to_vec(), cipher_suites, extensions: extension_order, groups, @@ -208,3 +249,16 @@ fn u16_values(mut bytes: Bytes) -> Vec { } values } + +// 仅用于触发自定义 CA 路径，采集器在 ClientHello 后终止握手 +const TEST_CA_PEM: &str = r"-----BEGIN CERTIFICATE----- +MIIBZTCCARegAwIBAgIUbyYrPhPgUSvZ6mOSWi/YkCn0nKIwBQYDK2VwMCgxJjAk +BgNVBAMMHUNvZGV4IFRMUyBmaW5nZXJwcmludCB0ZXN0IENBMB4XDTI2MTAwNTE1 +MDY0N1oXDTM2MTAwMjE1MDY0N1owKDEmMCQGA1UEAwwdQ29kZXggVExTIGZpbmdl +cnByaW50IHRlc3QgQ0EwKjAFBgMrZXADIQB2lY3pLKckV8KmxZiVZcmw70+RriQE +oAoUPx49BMSE9KNTMFEwHQYDVR0OBBYEFD8ham5c8YWHeYP7rsqbE/+8Sm1NMB8G +A1UdIwQYMBaAFD8ham5c8YWHeYP7rsqbE/+8Sm1NMA8GA1UdEwEB/wQFMAMBAf8w +BQYDK2VwA0EA15f5R32nl9ke0xc5sNM67R4tUEnxUiG2tUSGYinnFqFgEacMdK37 +T/qdob9dovtvUOziymxLs80cMCsGpbFiCw== +-----END CERTIFICATE----- +"; diff --git a/backend/crates/providers/openai/tests/transport/usage.rs b/backend/crates/providers/openai/tests/transport/usage.rs index d862235ac..1d9284df4 100644 --- a/backend/crates/providers/openai/tests/transport/usage.rs +++ b/backend/crates/providers/openai/tests/transport/usage.rs @@ -646,7 +646,7 @@ async fn retry_after_http_date_should_be_converted_to_remaining_seconds() { } #[tokio::test] -async fn retry_after_http_date_in_the_past_should_be_ignored() { +async fn retry_after_http_date_in_the_past_should_preserve_zero_delay() { let server = MockServer::start().await; Mock::given(method("GET")) .and(path("/api/codex/usage")) @@ -674,7 +674,7 @@ async fn retry_after_http_date_in_the_past_should_be_ignored() { else { panic!("expected an upstream rate-limit error"); }; - assert_eq!(retry_after_seconds, None); + assert_eq!(retry_after_seconds, Some(0)); } #[tokio::test] diff --git a/backend/crates/providers/xai/src/transport/request/tools.rs b/backend/crates/providers/xai/src/transport/request/tools.rs index 75343651f..e30e2e3ab 100644 --- a/backend/crates/providers/xai/src/transport/request/tools.rs +++ b/backend/crates/providers/xai/src/transport/request/tools.rs @@ -265,7 +265,10 @@ impl ToolNormalizer { } let mut converted = Vec::new(); for child in children { - if child.pointer("/type").and_then(Value::as_str) != Some("function") { + if !matches!( + child.get("type").and_then(Value::as_str), + Some("function" | "custom") + ) { return Err(GrokRequestEncodeError::InvalidRequestNormalization); } converted.extend(self.normalize_tool(child, name, client_search, force)?); diff --git a/backend/crates/providers/xai/tests/transport/canonical.rs b/backend/crates/providers/xai/tests/transport/canonical.rs index c8d93b0a6..9ae7025b0 100644 --- a/backend/crates/providers/xai/tests/transport/canonical.rs +++ b/backend/crates/providers/xai/tests/transport/canonical.rs @@ -693,6 +693,36 @@ fn decoder_should_restore_namespace_custom_search_and_apply_patch_wire_events() ); } +#[test] +fn decoder_should_restore_namespaced_custom_tool_identity_and_raw_input() { + use serde_json::json; + let request = tool_request(json!({"model":"client", "input":"run", + "tools":[{"type":"namespace","name":"functions","tools":[{"type":"custom","name":"exec"}]}] + })); + let item = json!({"type":"function_call","id":"fc_exec","call_id":"call_exec","name":"functions__exec","arguments":"{\"input\":\"echo hello\"}"}); + let body = [ + json!({"type":"response.created","response":{"id":"resp_exec"}}), + json!({"type":"response.output_item.added","output_index":0,"item":item}), + json!({"type":"response.output_item.done","output_index":0,"item":item}), + json!({"type":"response.completed","response":{"id":"resp_exec","status":"completed","output":[item]}}), + ].iter().map(|event| format!("event: {}\ndata: {event}\n\n", event["type"].as_str().unwrap())).collect::(); + let events = GrokCanonicalDecoder::for_request("grok-4.5", &request) + .push(body.as_bytes()) + .unwrap(); + let wire = wire_events(&events); + let done = wire + .iter() + .find(|event| event.event_type() == Some("response.output_item.done")) + .unwrap(); + let item = &done.data()["item"]; + assert_eq!(item["type"], "custom_tool_call"); + assert_eq!(item["namespace"], "functions"); + assert_eq!(item["name"], "exec"); + assert_eq!(item["id"], "ctc_exec"); + assert_eq!(item["call_id"], "call_exec"); + assert_eq!(item["input"], "echo hello"); +} + #[test] fn decoder_should_restore_custom_apply_patch_arguments_as_raw_input() { let patch = concat!( diff --git a/backend/crates/providers/xai/tests/transport/request/mod.rs b/backend/crates/providers/xai/tests/transport/request/mod.rs index 2860962fa..223658c84 100644 --- a/backend/crates/providers/xai/tests/transport/request/mod.rs +++ b/backend/crates/providers/xai/tests/transport/request/mod.rs @@ -19,6 +19,71 @@ fn client_key() -> ClientApiKeyId { ClientApiKeyId::new("key_xai_request_test").expect("client key id") } +#[test] +fn namespaced_custom_tools_should_preserve_grammar_choice_and_history() { + let grammar = "start: /.+/"; + let request = raw_request(json!({ + "model":"client", + "tools":[{"type":"namespace","name":"functions","tools":[ + {"type":"function","name":"lookup","parameters":{"type":"object"}}, + {"type":"custom","name":"exec","format":{"type":"grammar","syntax":"lark","definition":grammar}} + ]}], + "tool_choice":{"type":"custom","namespace":"functions","name":"exec"}, + "input":[ + {"type":"custom_tool_call","namespace":"functions","name":"exec","id":"ctc_exec","call_id":"call_exec","input":"echo hello"}, + {"type":"custom_tool_call_output","call_id":"call_exec","output":"hello"} + ] + })); + let encoded = GrokResponsesRequest::encode(&request, "grok-4.5", &client_key()).unwrap(); + let body = Value::Object(encoded.body().clone()); + assert_eq!(body["tools"][0]["name"], "functions__lookup"); + assert_eq!(body["tools"][1]["name"], "functions__exec"); + assert!( + body["tools"][1]["description"] + .as_str() + .unwrap() + .contains(grammar) + ); + assert_eq!(body["tools"][1]["parameters"]["required"], json!(["input"])); + assert_eq!( + body["tool_choice"], + json!({"type":"function","name":"functions__exec"}) + ); + assert_eq!(body["input"][0]["name"], "functions__exec"); + assert_eq!(body["input"][0]["id"], "fc_exec"); + assert_eq!(body["input"][0]["call_id"], "call_exec"); + assert_eq!( + body["input"][0]["arguments"], + json!("{\"input\":\"echo hello\"}") + ); + assert_eq!(body["input"][1]["call_id"], "call_exec"); +} + +#[test] +fn additional_tools_should_accept_namespaced_custom_tools() { + let request = raw_request(json!({ + "model":"client", "input":[{"type":"additional_tools","tools":[ + {"type":"namespace","name":"functions","tools":[{"type":"custom","name":"exec"}]} + ]}] + })); + let encoded = GrokResponsesRequest::encode(&request, "grok-4.5", &client_key()).unwrap(); + assert_eq!(encoded.body()["tools"][0]["name"], "functions__exec"); + assert_eq!(encoded.body()["input"], json!([])); +} + +#[test] +fn namespace_should_still_reject_non_function_or_custom_children() { + for kind in ["namespace", "web_search", "tool_search", "future_tool"] { + let request = raw_request( + json!({"tools":[{"type":"namespace","name":"functions","tools":[{"type":kind,"name":"invalid"}]}]}), + ); + assert!(matches!( + GrokResponsesRequest::encode(&request, "grok-4.5", &client_key()), + Err(GrokRequestEncodeError::InvalidRequestField { field: "tools" }) + )); + } +} + #[test] fn encoder_should_preserve_raw_images_hosted_tools_and_unknown_fields() { let request = raw_request(json!({ diff --git a/docs/api.md b/docs/api.md index dd97f6de9..1346c9762 100644 --- a/docs/api.md +++ b/docs/api.md @@ -399,8 +399,10 @@ Responses wire 之间的协议转换层，转换只在 xAI Provider 内完成。 全局精确映射，未命中时模型名原样交给候选 Provider；分组只限定账号集合，不参与模型改名 OpenAI 明确返回 `server_is_overloaded`、`slow_down` 或模型容量不足错误时，代理在允许安全重放且 -尚未交付输出的前提下，先做最多 3 次同账号指数退避，再通过现有调度换号。默认间隔从 500ms 开始， -上游 `Retry-After` 参与退避计算，单次等待不超过 8 秒；重试同时受请求总尝试次数和截止时间约束。 +尚未交付输出的前提下，先做最多 3 次同账号重试，再通过现有调度换号。没有有效服务器建议时， +按 500ms 起步的指数退避等待，单次最多 8 秒；有效 `Retry-After` 优先，接受秒数、HTTP 日期和零延迟， +不受本地退避上限截断。流内失败事件使用相同解析规则，WS 转 HTTP 回退也遵守服务器建议； +等待受请求截止时间和取消约束，重试受请求总尝试次数约束。 `server_is_overloaded`、`slow_down` 等可计分的结构化错误按已发送的失败尝试计入 Smart 账号 健康分。已确认容量拒绝的平滑权重为 0.4，其他可计分失败与成功样本保持 0.2。 失败率使用账号级平滑与时间衰减，影响后续普通选路，已有可用账号的 @@ -415,6 +417,8 @@ SSE/WS 的 `response.failed` 保留原消息、响应 ID 与其他业务字段 客户端决定如何恢复，代理不因此重放已提交的请求。 明确额度耗尽触发账号隔离与安全换号， 包括 WebSocket 握手返回的 429；不会因其长 `Retry-After` 而转入同账号传输恢复等待。 +`flex_unavailable` 是当前请求的终止拒绝，保留原始错误，不自动重试、切换传输或冷却账号 + OpenAI 选号阶段确认本次可选账号全部额度耗尽时，HTTP 返回 `429`，WebSocket 错误帧返回 `status: 429`，两者的 `error.type` 与 `error.code` 均为 `usage_limit_reached`，提示客户端停止 本轮自动重试，等待额度恢复或补充可用账号。空账号池、认证失效和临时容量不足不按额度耗尽处理； @@ -598,9 +602,10 @@ OpenAI 点数随现有额度刷新和正常请求的额度信息同步，不与 字符，允许换行和制表符，保存时去除首尾空白，空字符串清空备注。备注独立于上游身份，导入时未显式提供备注、 重新授权、凭据刷新及批量调度更新均保留已有备注 -账号视图和 Dashboard 账号概览中的 `planType` 保留原始套餐值；`planTypeDisplay` 由后端先按 Provider 解析名称， -再统一为大驼峰格式，前端直接展示该字段，例如 `Free`、`SuperGrokPro`、`EduPlus`。 -OpenAI 的 `self_serve_business_prolite` 等 Team 套餐显示为 `Business`；新套餐也使用相同格式。 +账号视图和 Dashboard 账号概览中的 `planType` 保留原始套餐值；`planTypeDisplay` 使用 Provider 提供的最终名称， +前端直接展示。OpenAI 的 `prolite`、`pro`、`promax` 分别显示为 `ProLite`、`Pro`、`ProMax`； +其余已知套餐使用官方名称，未知 OpenAI 套餐保留原值。 +展示名称保留空格和括号，未提供专属名称规则的 Provider 使用通用大驼峰格式。 OpenAI 主动额度刷新和正常响应携带的明确套餐会同步到账号，支持升级与降级； 空值或 `unknown` 不覆盖已有套餐，同族泛化值（如 `team`）保留已知的具体套餐子类型。 账号套餐为空或 `unknown` 时，后端优先用已保存的上游额度响应 diff --git a/docs/architecture.md b/docs/architecture.md index 6a5b8804c..f7f559b84 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -369,6 +369,7 @@ Client Key 鉴权完成后，API adapter 从有界请求头识别 Codex Desktop/ 重试与提交边界。具体错误合同见 [数据面接口](api.md#3-openai-数据面与模型目录) - Provider 可将明确容量拒绝标记为有界同账号退避，Core 在既有安全重放边界内执行，按账号维护请求内 预算，耗尽后复用普通换号路径。该退避消耗总路由预算，与 WS 传输恢复、OAuth 刷新及账号额度冷却分开 +- Provider 显式标记的终止拒绝禁止自动重试与传输回退，优先于未发送状态和安全重放证明 - 跨 Provider 只在账号范围和能力都允许，且请求尚未到达上游或已被证明可安全重放时发生 - 可恢复观测写入失败不能替换已经确定的客户端协议结果