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

fix(codex): align upstream contracts and transport behavior (#352)

Browse Source 
Co-authored-by: Zyy <zyycn@users.noreply.github.com>
Co-authored-by: Codex <noreply@openai.com>
This commit is contained in:
Zyy
2026-10-05 23:41:46 +08:00
committed by GitHub
co-authored by Zyy Codex
parent f8439eb4b9
commit 78ba7e412e
37 changed files with 792 additions and 154 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/Cargo.lock
Generated
+4 -2
View File
@@ -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",
backend/crates/gateway-admin/src/ports/provider.rs
+10 -8
View File
@@ -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()
}
backend/crates/gateway-admin/tests/use_case/accounts.rs
+19 -6
View File
@@ -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());
backend/crates/gateway-admin/tests/use_case/observability.rs
+69 -7
View File
@@ -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<Vec<RequestMetricPoint>>,
overview: Mutex<UsageOverview>,
@@ -731,6 +791,7 @@ struct FixtureObservabilityStore {
summary_observed_at: Mutex<Option<DateTime<Utc>>>,
slots_observed_at: Mutex<Option<DateTime<Utc>>>,
usage_records: Mutex<Vec<UsageListRecord>>,
account_usage: Mutex<Vec<DashboardAccountUsage>>,
dashboard_delay: Mutex<StdDuration>,
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(),
})
}
backend/crates/gateway-admin/tests/use_case/proxies.rs
+2 -2
View File
@@ -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,
}
);
backend/crates/gateway-core/src/engine/coordinator.rs
+14 -7
View File
@@ -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 {
backend/crates/gateway-core/src/error.rs
+17
View File
@@ -444,6 +444,7 @@ pub struct ProviderError {
continuation_recovery_disposition: Option<ContinuationRecoveryDisposition>,
failure_observation: Option<Box<ProviderErrorFailureObservation>>,
replay_safe: bool,
retry_prohibited: bool,
pre_delivery_retry: Option<Box<PreDeliveryRetry>>,
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(|_| "<classified-safe>"),
)
.field("retry_after", &self.retry_after)
.field("retry_prohibited", &self.retry_prohibited)
.field("continuation_failure", &self.continuation_failure)
.field(
"continuation_recovery_disposition",
backend/crates/gateway-core/tests/engine/coordinator.rs
+118
View File
@@ -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();
backend/crates/gateway-protocol/Cargo.toml
+1
View File
@@ -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"] }
backend/crates/gateway-protocol/src/openai/codex.rs
+6 -2
View File
@@ -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>(value).ok());
backend/crates/gateway-protocol/src/openai/events.rs
+11 -11
View File
@@ -248,7 +248,9 @@ pub fn retry_after_seconds_from_value(value: &Value) -> Option<u64> {
.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<u64> {
})
.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<u64> {
.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<u64> {
})
}
fn json_value_as_positive_u64(value: &Value) -> Option<u64> {
let seconds = match value {
Value::Number(value) => value.as_u64()?,
Value::String(value) => value.trim().parse::<u64>().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<u64> {
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,
}
}
/// 标准化的单个限流窗口
backend/crates/gateway-protocol/src/openai/headers.rs
+19 -1
View File
@@ -1,4 +1,22 @@
//! OpenAI 请求头的 HTTP 传输边界
//! OpenAI 头部的传输边界与重试时间解析
/// 解析 HTTP 和流内错误共用的 Retry-After，保留零延迟并向上取整剩余秒数
pub fn parse_retry_after_seconds(value: &str) -> Option<u64> {
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)),
)
}
/// 判断小写请求头是否属于传输层管理的字段，不得作为业务扩展头透传
///
backend/crates/gateway-protocol/src/openai/mod.rs
+3 -1
View File
@@ -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,
};
backend/crates/gateway-protocol/tests/openai/codex.rs
+19
View File
@@ -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(
backend/crates/gateway-protocol/tests/openai/events.rs
+38
View File
@@ -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![
backend/crates/gateway-protocol/tests/openai/headers.rs
+32 -1
View File
@@ -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 [
backend/crates/providers/openai/Cargo.toml
+1
View File
@@ -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"] }
backend/crates/providers/openai/src/admin.rs
+13 -12
View File
@@ -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",
backend/crates/providers/openai/src/credential/token_client.rs
+1 -1
View File
@@ -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 错误
backend/crates/providers/openai/src/credential/types.rs
+8
View File
@@ -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",
}
}
}
backend/crates/providers/openai/src/provider/failure.rs
+10 -4
View File
@@ -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,
}
backend/crates/providers/openai/src/transport/client.rs
+1 -25
View File
@@ -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<u64> {
let value = value.trim();
if let Ok(seconds) = value.parse::<u64>() {
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<DateTime<Utc>> {
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()
}
backend/crates/providers/openai/src/transport/diagnostics.rs
+11 -1
View File
@@ -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)
backend/crates/providers/openai/src/transport/websocket/handshake.rs
+3 -5
View File
@@ -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<Option<Vec<u8>>>) -> 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,
backend/crates/providers/openai/tests/admin.rs
+37 -1
View File
@@ -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");
backend/crates/providers/openai/tests/credential/admin.rs
+19
View File
@@ -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!({
backend/crates/providers/openai/tests/credential/oauth.rs
+2 -2
View File
@@ -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),
backend/crates/providers/openai/tests/credential/token_client.rs
+1 -1
View File
@@ -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]
backend/crates/providers/openai/tests/provider/contract/mod.rs
+85 -15
View File
@@ -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]
backend/crates/providers/openai/tests/transport/diagnostics.rs
+21
View File
@@ -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] {
backend/crates/providers/openai/tests/transport/tls.rs
+85 -31
View File
@@ -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<u16>,
point_formats: Vec<u8>,
psk_modes: Vec<u8>,
cipher_suites: Vec<u16>,
extensions: Vec<u16>,
groups: Vec<u16>,
@@ -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(&current_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<u16> {
}
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-----
";
backend/crates/providers/openai/tests/transport/usage.rs
+2 -2
View File
@@ -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]
backend/crates/providers/xai/src/transport/request/tools.rs
+4 -1
View File
@@ -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)?);
backend/crates/providers/xai/tests/transport/canonical.rs
+30
View File
@@ -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::<String>();
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!(
backend/crates/providers/xai/tests/transport/request/mod.rs
+65
View File
@@ -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!({
docs/api.md
+10 -5
View File
@@ -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` 时，后端优先用已保存的上游额度响应
docs/architecture.md
+1
View File
@@ -369,6 +369,7 @@ Client Key 鉴权完成后，API adapter 从有界请求头识别 Codex Desktop/
重试与提交边界。具体错误合同见 [数据面接口](api.md#3-openai-数据面与模型目录)
- Provider 可将明确容量拒绝标记为有界同账号退避，Core 在既有安全重放边界内执行，按账号维护请求内
预算，耗尽后复用普通换号路径。该退避消耗总路由预算，与 WS 传输恢复、OAuth 刷新及账号额度冷却分开
- Provider 显式标记的终止拒绝禁止自动重试与传输回退，优先于未发送状态和安全重放证明
- 跨 Provider 只在账号范围和能力都允许，且请求尚未到达上游或已被证明可安全重放时发生
- 可恢复观测写入失败不能替换已经确定的客户端协议结果