diff --git a/backend/crates/gateway-admin/src/model/settings.rs b/backend/crates/gateway-admin/src/model/settings.rs index 1b20732cc..00e33c6d4 100644 --- a/backend/crates/gateway-admin/src/model/settings.rs +++ b/backend/crates/gateway-admin/src/model/settings.rs @@ -39,6 +39,9 @@ pub struct RuntimeSettings { pub max_waiting_per_account: u32, pub concurrency_wait_timeout_seconds: u32, pub openai_guardian_reserved_concurrency: u32, + pub openai_account_affinity: gateway_core::account::AccountAffinity, + pub max_account_rotations: u32, + pub openai_session_affinity_ttl_hours: u32, pub responses_max_decompressed_body_bytes: u64, pub smart_scheduling: gateway_core::account::SmartSchedulingConfig, pub rotation_strategy: RotationStrategy, @@ -79,6 +82,9 @@ pub struct ReplaceRuntimeSettings { pub max_waiting_per_account: u32, pub concurrency_wait_timeout_seconds: u32, pub openai_guardian_reserved_concurrency: u32, + pub openai_account_affinity: gateway_core::account::AccountAffinity, + pub max_account_rotations: u32, + pub openai_session_affinity_ttl_hours: u32, pub responses_max_decompressed_body_bytes: u64, pub smart_scheduling: gateway_core::account::SmartSchedulingConfig, pub rotation_strategy: RotationStrategy, @@ -185,6 +191,9 @@ impl From for ReplaceRuntimeSettings { max_waiting_per_account: settings.max_waiting_per_account, concurrency_wait_timeout_seconds: settings.concurrency_wait_timeout_seconds, openai_guardian_reserved_concurrency: settings.openai_guardian_reserved_concurrency, + openai_account_affinity: settings.openai_account_affinity, + max_account_rotations: settings.max_account_rotations, + openai_session_affinity_ttl_hours: settings.openai_session_affinity_ttl_hours, responses_max_decompressed_body_bytes: settings.responses_max_decompressed_body_bytes, smart_scheduling: settings.smart_scheduling, rotation_strategy: settings.rotation_strategy, diff --git a/backend/crates/gateway-admin/src/use_case/settings.rs b/backend/crates/gateway-admin/src/use_case/settings.rs index dfc76ecc1..82d0257c8 100644 --- a/backend/crates/gateway-admin/src/use_case/settings.rs +++ b/backend/crates/gateway-admin/src/use_case/settings.rs @@ -108,6 +108,9 @@ fn validate_settings(command: &ReplaceRuntimeSettings) -> Result<(), AdminError> && isize::try_from(command.responses_max_decompressed_body_bytes).is_ok() && command.refresh_margin_seconds > 0 && command.refresh_concurrency > 0 + && command.max_account_rotations <= gateway_core::account::MAX_ACCOUNT_ROTATIONS + && (1..=gateway_core::account::MAX_SESSION_AFFINITY_TTL_HOURS) + .contains(&command.openai_session_affinity_ttl_hours) && command.max_waiting_per_key <= 1_000 && command.max_waiting_per_account <= 1_000 && (1..=120).contains(&command.concurrency_wait_timeout_seconds) diff --git a/backend/crates/gateway-admin/tests/use_case/accounts.rs b/backend/crates/gateway-admin/tests/use_case/accounts.rs index c97ff0912..1f1cc3404 100644 --- a/backend/crates/gateway-admin/tests/use_case/accounts.rs +++ b/backend/crates/gateway-admin/tests/use_case/accounts.rs @@ -1122,6 +1122,9 @@ impl SettingsStore for StaticSettingsStore { max_waiting_per_account: 0, concurrency_wait_timeout_seconds: 30, openai_guardian_reserved_concurrency: 0, + openai_account_affinity: gateway_core::account::AccountAffinity::Relaxed, + max_account_rotations: 3, + openai_session_affinity_ttl_hours: 24, responses_max_decompressed_body_bytes: 64 * 1024 * 1024, smart_scheduling: gateway_core::account::SmartSchedulingConfig::default(), rotation_strategy: RotationStrategy::Smart, diff --git a/backend/crates/gateway-admin/tests/use_case/freeze_recovery.rs b/backend/crates/gateway-admin/tests/use_case/freeze_recovery.rs index d063be995..0c3ca290c 100644 --- a/backend/crates/gateway-admin/tests/use_case/freeze_recovery.rs +++ b/backend/crates/gateway-admin/tests/use_case/freeze_recovery.rs @@ -38,6 +38,9 @@ fn runtime_settings(enabled: bool, probe_enabled: bool, adaptive: bool) -> Runti max_waiting_per_account: 0, concurrency_wait_timeout_seconds: 30, openai_guardian_reserved_concurrency: 0, + openai_account_affinity: gateway_core::account::AccountAffinity::Relaxed, + max_account_rotations: 3, + openai_session_affinity_ttl_hours: 24, responses_max_decompressed_body_bytes: 64 * 1024 * 1024, smart_scheduling: gateway_core::account::SmartSchedulingConfig::default(), rotation_strategy: gateway_admin::model::settings::RotationStrategy::Smart, diff --git a/backend/crates/gateway-admin/tests/use_case/observability.rs b/backend/crates/gateway-admin/tests/use_case/observability.rs index d20d78815..0b85c32a0 100644 --- a/backend/crates/gateway-admin/tests/use_case/observability.rs +++ b/backend/crates/gateway-admin/tests/use_case/observability.rs @@ -1001,6 +1001,9 @@ impl SettingsStore for FixtureSettingsStore { max_waiting_per_account: 0, concurrency_wait_timeout_seconds: 30, openai_guardian_reserved_concurrency: 0, + openai_account_affinity: gateway_core::account::AccountAffinity::Relaxed, + max_account_rotations: 3, + openai_session_affinity_ttl_hours: 24, responses_max_decompressed_body_bytes: 64 * 1024 * 1024, smart_scheduling: gateway_core::account::SmartSchedulingConfig::default(), rotation_strategy: RotationStrategy::Smart, diff --git a/backend/crates/gateway-admin/tests/use_case/service_contract.rs b/backend/crates/gateway-admin/tests/use_case/service_contract.rs index 4548426ce..073997dd3 100644 --- a/backend/crates/gateway-admin/tests/use_case/service_contract.rs +++ b/backend/crates/gateway-admin/tests/use_case/service_contract.rs @@ -273,6 +273,7 @@ fn sdk_settings_contract_matches_host_declarations() { pub type ModelMappings = BTreeMap; pub type Revision = NonZeroU64; pub type RotationStrategy = String; + pub type AccountAffinity = String; pub type PricingOverrides = BTreeMap>; #(#types)* #(#operations)* diff --git a/backend/crates/gateway-admin/tests/use_case/settings.rs b/backend/crates/gateway-admin/tests/use_case/settings.rs index bb45bb25d..1f6af84dd 100644 --- a/backend/crates/gateway-admin/tests/use_case/settings.rs +++ b/backend/crates/gateway-admin/tests/use_case/settings.rs @@ -288,6 +288,9 @@ async fn settings_should_reject_zero_refresh_margin_before_store_call() { max_waiting_per_account: 0, concurrency_wait_timeout_seconds: 30, openai_guardian_reserved_concurrency: 0, + openai_account_affinity: gateway_core::account::AccountAffinity::Relaxed, + max_account_rotations: 3, + openai_session_affinity_ttl_hours: 24, responses_max_decompressed_body_bytes: 64 * 1024 * 1024, smart_scheduling: gateway_core::account::SmartSchedulingConfig::default(), rotation_strategy: RotationStrategy::Smart, diff --git a/backend/crates/gateway-api/src/admin/settings.rs b/backend/crates/gateway-api/src/admin/settings.rs index 1527217bd..384b60613 100644 --- a/backend/crates/gateway-api/src/admin/settings.rs +++ b/backend/crates/gateway-api/src/admin/settings.rs @@ -55,6 +55,9 @@ pub struct RuntimeSettingsView { pub max_waiting_per_account: u32, pub concurrency_wait_timeout_seconds: u32, pub openai_guardian_reserved_concurrency: u32, + pub openai_account_affinity: gateway_core::account::AccountAffinity, + pub max_account_rotations: u32, + pub openai_session_affinity_ttl_hours: u32, pub responses_max_decompressed_body_bytes: u64, pub smart_scheduling: gateway_core::account::SmartSchedulingConfig, pub rotation_strategy: String, @@ -101,6 +104,9 @@ pub struct UpdateRuntimeSettingsRequest { pub max_waiting_per_account: u32, pub concurrency_wait_timeout_seconds: u32, pub openai_guardian_reserved_concurrency: u32, + pub openai_account_affinity: gateway_core::account::AccountAffinity, + pub max_account_rotations: u32, + pub openai_session_affinity_ttl_hours: u32, pub responses_max_decompressed_body_bytes: u64, pub smart_scheduling: gateway_core::account::SmartSchedulingConfig, pub rotation_strategy: String, @@ -128,6 +134,14 @@ impl UpdateRuntimeSettingsRequest { .validate() .map_err(|_| WireValidationError::new("requestLocation"))?; validate_model_mappings(&self.model_mappings)?; + if !(1..=gateway_core::account::MAX_SESSION_AFFINITY_TTL_HOURS) + .contains(&self.openai_session_affinity_ttl_hours) + { + return Err(WireValidationError::new("openaiSessionAffinityTtlHours")); + } + if self.max_account_rotations > gateway_core::account::MAX_ACCOUNT_ROTATIONS { + return Err(WireValidationError::new("maxAccountRotations")); + } for (value, field) in [ (self.max_waiting_per_key, "maxWaitingPerKey"), (self.max_waiting_per_account, "maxWaitingPerAccount"), @@ -237,6 +251,9 @@ impl UpdateRuntimeSettingsRequest { max_waiting_per_account: self.max_waiting_per_account, concurrency_wait_timeout_seconds: self.concurrency_wait_timeout_seconds, openai_guardian_reserved_concurrency: self.openai_guardian_reserved_concurrency, + openai_account_affinity: self.openai_account_affinity, + max_account_rotations: self.max_account_rotations, + openai_session_affinity_ttl_hours: self.openai_session_affinity_ttl_hours, responses_max_decompressed_body_bytes: self.responses_max_decompressed_body_bytes, smart_scheduling: self.smart_scheduling, rotation_strategy: RotationStrategy::parse(&self.rotation_strategy) @@ -287,6 +304,9 @@ impl From<(RuntimeSettings, crate::time::TimePresenter)> for RuntimeSettingsView max_waiting_per_account: settings.max_waiting_per_account, concurrency_wait_timeout_seconds: settings.concurrency_wait_timeout_seconds, openai_guardian_reserved_concurrency: settings.openai_guardian_reserved_concurrency, + openai_account_affinity: settings.openai_account_affinity, + max_account_rotations: settings.max_account_rotations, + openai_session_affinity_ttl_hours: settings.openai_session_affinity_ttl_hours, responses_max_decompressed_body_bytes: settings.responses_max_decompressed_body_bytes, smart_scheduling: settings.smart_scheduling, smart_scheduling_defaults: gateway_core::account::SmartSchedulingConfig::default(), diff --git a/backend/crates/gateway-api/tests/admin/mod.rs b/backend/crates/gateway-api/tests/admin/mod.rs index e454f44d6..ed4cb7421 100644 --- a/backend/crates/gateway-api/tests/admin/mod.rs +++ b/backend/crates/gateway-api/tests/admin/mod.rs @@ -605,6 +605,9 @@ impl SettingsStore for MemorySettingsStore { max_waiting_per_account: command.max_waiting_per_account, concurrency_wait_timeout_seconds: command.concurrency_wait_timeout_seconds, openai_guardian_reserved_concurrency: command.openai_guardian_reserved_concurrency, + openai_account_affinity: command.openai_account_affinity, + max_account_rotations: command.max_account_rotations, + openai_session_affinity_ttl_hours: command.openai_session_affinity_ttl_hours, responses_max_decompressed_body_bytes: command.responses_max_decompressed_body_bytes, smart_scheduling: command.smart_scheduling, rotation_strategy: command.rotation_strategy, @@ -1614,6 +1617,9 @@ fn test_runtime_settings() -> RuntimeSettings { max_waiting_per_account: 0, concurrency_wait_timeout_seconds: 30, openai_guardian_reserved_concurrency: 0, + openai_account_affinity: gateway_core::account::AccountAffinity::Relaxed, + max_account_rotations: 3, + openai_session_affinity_ttl_hours: 24, responses_max_decompressed_body_bytes: 64 * 1024 * 1024, smart_scheduling: gateway_core::account::SmartSchedulingConfig::default(), rotation_strategy: RotationStrategy::Smart, diff --git a/backend/crates/gateway-api/tests/admin/settings.rs b/backend/crates/gateway-api/tests/admin/settings.rs index 770b0f1ec..249911f63 100644 --- a/backend/crates/gateway-api/tests/admin/settings.rs +++ b/backend/crates/gateway-api/tests/admin/settings.rs @@ -53,6 +53,9 @@ fn update_body() -> Value { "maxWaitingPerKey": 0, "maxWaitingPerAccount": 0, "openaiGuardianReservedConcurrency": 0, + "openaiAccountAffinity": "relaxed", + "maxAccountRotations": 3, + "openaiSessionAffinityTtlHours": 24, "concurrencyWaitTimeoutSeconds": 30, "responsesMaxDecompressedBodyBytes": 67108864, "rotationStrategy": "round_robin", @@ -224,6 +227,9 @@ fn settings_response_should_cover_the_full_runtime_settings_contract() { max_waiting_per_account: 0, concurrency_wait_timeout_seconds: 30, openai_guardian_reserved_concurrency: 0, + openai_account_affinity: gateway_core::account::AccountAffinity::Relaxed, + max_account_rotations: 3, + openai_session_affinity_ttl_hours: 24, responses_max_decompressed_body_bytes: 64 * 1024 * 1024, smart_scheduling: gateway_core::account::SmartSchedulingConfig::default(), rotation_strategy: RotationStrategy::RoundRobin, @@ -273,6 +279,9 @@ fn settings_response_should_cover_the_full_runtime_settings_contract() { "maxWaitingPerKey": 0, "maxWaitingPerAccount": 0, "openaiGuardianReservedConcurrency": 0, + "openaiAccountAffinity": "relaxed", + "maxAccountRotations": 3, + "openaiSessionAffinityTtlHours": 24, "concurrencyWaitTimeoutSeconds": 30, "responsesMaxDecompressedBodyBytes": 67108864, "rotationStrategy": "round_robin", @@ -343,6 +352,9 @@ fn settings_request_and_response_fields_should_stay_in_lockstep() { max_waiting_per_account: 0, concurrency_wait_timeout_seconds: 30, openai_guardian_reserved_concurrency: 0, + openai_account_affinity: gateway_core::account::AccountAffinity::Relaxed, + max_account_rotations: 3, + openai_session_affinity_ttl_hours: 24, responses_max_decompressed_body_bytes: 64 * 1024 * 1024, smart_scheduling: request.smart_scheduling, rotation_strategy: RotationStrategy::parse(&request.rotation_strategy) @@ -1329,3 +1341,76 @@ async fn guardian_reservation_round_trips_and_rejects_invalid_values() { .remove("openaiGuardianReservedConcurrency"); assert!(serde_json::from_value::(omitted).is_err()); } + +#[tokio::test] +async fn account_affinity_and_rotation_budget_round_trip_and_reject_invalid_values() { + for (mode, rotations, ttl) in [("relaxed", 0, 1), ("strict", 31, 720)] { + let fixture = AdminTestFixture::new().await; + fixture.auth.insert_session("valid-session"); + let mut body = update_body(); + body["openaiAccountAffinity"] = json!(mode); + body["maxAccountRotations"] = json!(rotations); + body["openaiSessionAffinityTtlHours"] = json!(ttl); + let response = app(fixture.state()) + .oneshot(request( + Method::POST, + "/api/admin/settings/update", + Some(body), + )) + .await + .unwrap(); + assert_eq!(response.status(), StatusCode::OK); + let response = response_json(response).await; + assert_eq!(response["data"]["openaiAccountAffinity"], mode); + assert_eq!(response["data"]["maxAccountRotations"], rotations); + assert_eq!(response["data"]["openaiSessionAffinityTtlHours"], ttl); + } + // 类型错误沿用 AdminJson 的 422，合法类型越界由字段校验返回 400 + for (field, value, expected_status) in [ + ( + "openaiAccountAffinity", + json!("unknown"), + StatusCode::UNPROCESSABLE_ENTITY, + ), + ( + "maxAccountRotations", + json!(-1), + StatusCode::UNPROCESSABLE_ENTITY, + ), + ("maxAccountRotations", json!(32), StatusCode::BAD_REQUEST), + ( + "maxAccountRotations", + json!(1.5), + StatusCode::UNPROCESSABLE_ENTITY, + ), + ( + "openaiSessionAffinityTtlHours", + json!(0), + StatusCode::BAD_REQUEST, + ), + ( + "openaiSessionAffinityTtlHours", + json!(721), + StatusCode::BAD_REQUEST, + ), + ( + "openaiSessionAffinityTtlHours", + json!(1.5), + StatusCode::UNPROCESSABLE_ENTITY, + ), + ] { + let fixture = AdminTestFixture::new().await; + fixture.auth.insert_session("valid-session"); + let mut body = update_body(); + body[field] = value; + let response = app(fixture.state()) + .oneshot(request( + Method::POST, + "/api/admin/settings/update", + Some(body), + )) + .await + .unwrap(); + assert_eq!(response.status(), expected_status, "{field}"); + } +} diff --git a/backend/crates/gateway-core/src/account/selection.rs b/backend/crates/gateway-core/src/account/selection.rs index 9601ff27f..00b83cc93 100644 --- a/backend/crates/gateway-core/src/account/selection.rs +++ b/backend/crates/gateway-core/src/account/selection.rs @@ -46,6 +46,40 @@ impl RotationStrategy { } } +/// 单请求换号预算不能超过路由尝试总数减一 +pub const MAX_ACCOUNT_ROTATIONS: u32 = 31; + +/// 会话绑定与轮次关联共用的保留时长上限，单位小时 +pub const MAX_SESSION_AFFINITY_TTL_HOURS: u32 = 720; + +/// OpenAI 会话账号亲和；具体线程身份由 Provider 解释 +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, serde::Serialize, serde::Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum AccountAffinity { + #[default] + Relaxed, + Strict, +} + +impl AccountAffinity { + #[must_use] + pub const fn as_str(self) -> &'static str { + match self { + Self::Relaxed => "relaxed", + Self::Strict => "strict", + } + } + + #[must_use] + pub fn parse(value: &str) -> Option { + match value { + "relaxed" => Some(Self::Relaxed), + "strict" => Some(Self::Strict), + _ => None, + } + } +} + /// 从 `runtime_settings` 冻结到一次请求计划的账号调度策略 #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub struct AccountSelectionPolicy { @@ -55,6 +89,9 @@ pub struct AccountSelectionPolicy { request_interval: Duration, queue_policy: ConcurrencyQueuePolicy, openai_guardian_reserved_concurrency: u32, + openai_account_affinity: AccountAffinity, + max_account_rotations: u32, + openai_session_affinity_ttl: Duration, } impl AccountSelectionPolicy { @@ -70,6 +107,9 @@ impl AccountSelectionPolicy { max_concurrent_per_account: max_concurrent_per_account.into(), request_interval, openai_guardian_reserved_concurrency: 0, + openai_account_affinity: AccountAffinity::Relaxed, + max_account_rotations: 3, + openai_session_affinity_ttl: Duration::from_secs(24 * 3600), queue_policy: ConcurrencyQueuePolicy { max_waiting: 0, timeout: Duration::ZERO, @@ -77,6 +117,39 @@ impl AccountSelectionPolicy { } } + #[must_use] + pub const fn with_openai_account_affinity(mut self, affinity: AccountAffinity) -> Self { + self.openai_account_affinity = affinity; + self + } + + #[must_use] + pub const fn openai_account_affinity(self) -> AccountAffinity { + self.openai_account_affinity + } + + #[must_use] + pub const fn with_max_account_rotations(mut self, rotations: u32) -> Self { + self.max_account_rotations = rotations; + self + } + + #[must_use] + pub const fn max_account_rotations(self) -> u32 { + self.max_account_rotations + } + + #[must_use] + pub const fn with_openai_session_affinity_ttl(mut self, ttl: Duration) -> Self { + self.openai_session_affinity_ttl = ttl; + self + } + + #[must_use] + pub const fn openai_session_affinity_ttl(self) -> Duration { + self.openai_session_affinity_ttl + } + /// 只传递冻结的运行设置，Guardian 分类与预留策略由 OpenAI Provider 解释 #[must_use] pub const fn with_openai_guardian_reserved_concurrency(mut self, reserved: u32) -> Self { diff --git a/backend/crates/gateway-core/src/engine/coordinator.rs b/backend/crates/gateway-core/src/engine/coordinator.rs index aeb199a30..1c56882b0 100644 --- a/backend/crates/gateway-core/src/engine/coordinator.rs +++ b/backend/crates/gateway-core/src/engine/coordinator.rs @@ -464,7 +464,7 @@ pub struct ResponseExecutionSession { /// 已发生的换号次数：选中账号与上一 attempt 不同的路由 attempt 计一次。 /// 首个 attempt 不计；同账号钉选重试（瞬态退避、传输恢复、凭据恢复重放、 /// continuation 精确重连）不消耗。预算耗尽后所有必然换号的重试门关闭， - /// 换号深度由 [`crate::routing::MAX_ACCOUNT_ROTATION_ATTEMPTS`] 封顶。 + /// 换号深度由请求冻结的调度策略封顶 account_rotations: u32, candidate_index: usize, excluded_accounts: BTreeSet, @@ -1129,7 +1129,7 @@ where ) // 跨 Provider 候选推进必然换号（账号行按 Provider 隔离）， // 预算耗尽后不再推进，交回容量类失败的原有终态语义。 - && self.account_rotations < crate::routing::MAX_ACCOUNT_ROTATION_ATTEMPTS + && self.account_rotations < self.plan.account_selection_policy().max_account_rotations() && self.advance_provider_candidate(); let retryable = self .apply_retry_policy(super::policy::RetryFacts { @@ -1605,7 +1605,8 @@ where && transient_retry.is_none() && transport_recovery.is_none() && (ordinary_retry || account_rotation_retry) - && self.account_rotations < crate::routing::MAX_ACCOUNT_ROTATION_ATTEMPTS; + && self.account_rotations + < self.plan.account_selection_policy().max_account_rotations(); let retryable = !error.retry_is_prohibited() && (continuation_retry || same_account_retry @@ -1848,7 +1849,8 @@ where // 两个排除臂都必然换号；预算耗尽后不再排除当前账号做跨账号续写重放， // 落回不可重试路径以原始上游错误终态。 ContinuationAttempt::ReplayOwner | ContinuationAttempt::ReplayAny - if self.account_rotations >= crate::routing::MAX_ACCOUNT_ROTATION_ATTEMPTS => + if self.account_rotations + >= self.plan.account_selection_policy().max_account_rotations() => { return false; } @@ -1899,9 +1901,8 @@ where { return false; } - // Native 期间所有 attempt 都在 pin 账号上，到这里预算必未耗尽； - // 与其余换号路径保持同一预算门，防止状态机演化后破坏换号上限。 - if self.account_rotations >= crate::routing::MAX_ACCOUNT_ROTATION_ATTEMPTS { + // Native 期间仍可能配置为禁止换号，跨账号重放共用请求冻结的预算 + if self.account_rotations >= self.plan.account_selection_policy().max_account_rotations() { return false; } diff --git a/backend/crates/gateway-core/src/provider_ports.rs b/backend/crates/gateway-core/src/provider_ports.rs index 5d8d19421..50f063af2 100644 --- a/backend/crates/gateway-core/src/provider_ports.rs +++ b/backend/crates/gateway-core/src/provider_ports.rs @@ -297,6 +297,7 @@ impl ProviderSessionBinding { #[derive(Debug, Clone, PartialEq, Eq)] pub struct ProviderSessionAlias { pub session_key: ProviderSessionAffinityKey, + pub root_session_key: Option, pub follow_only: bool, } diff --git a/backend/crates/gateway-core/src/routing/mod.rs b/backend/crates/gateway-core/src/routing/mod.rs index cc608d54e..2118b334d 100644 --- a/backend/crates/gateway-core/src/routing/mod.rs +++ b/backend/crates/gateway-core/src/routing/mod.rs @@ -24,11 +24,7 @@ use crate::account::{AccountSelectionPolicy, FastMode}; use crate::operation::{CapabilityRequirements, Feature, OperationKind}; use crate::validation::{IdentifierError, RoutingError, validate_text}; -const MAX_REQUEST_ATTEMPTS: u32 = 32; - -/// 单请求内允许的换号次数上限，选中账号与上一 attempt 不同的路由 attempt 记一次 -/// 初始账号和同账号重试不计入换号次数，所有尝试仍受总路由预算限制 -pub(crate) const MAX_ACCOUNT_ROTATION_ATTEMPTS: u32 = 3; +const MAX_REQUEST_ATTEMPTS: u32 = crate::account::MAX_ACCOUNT_ROTATIONS + 1; /// 客户端请求中的模型名称 #[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash, serde::Serialize)] diff --git a/backend/crates/gateway-core/src/settings/compiled.rs b/backend/crates/gateway-core/src/settings/compiled.rs index b126fc71f..7aecf512b 100644 --- a/backend/crates/gateway-core/src/settings/compiled.rs +++ b/backend/crates/gateway-core/src/settings/compiled.rs @@ -22,7 +22,10 @@ pub(crate) struct CompiledSettings { impl CompiledSettings { pub(crate) fn new(settings: SettingsValues) -> Result { - if settings.max_waiting_per_key > 1_000 + if !(1..=crate::account::MAX_SESSION_AFFINITY_TTL_HOURS) + .contains(&settings.openai_session_affinity_ttl_hours) + || settings.max_account_rotations > crate::account::MAX_ACCOUNT_ROTATIONS + || settings.max_waiting_per_key > 1_000 || settings.max_waiting_per_account > 1_000 || !(1..=120).contains(&settings.concurrency_wait_timeout_seconds) { @@ -58,6 +61,11 @@ impl CompiledSettings { Duration::from_millis(settings.request_interval_ms), ) .with_openai_guardian_reserved_concurrency(settings.openai_guardian_reserved_concurrency) + .with_openai_account_affinity(settings.openai_account_affinity) + .with_max_account_rotations(settings.max_account_rotations) + .with_openai_session_affinity_ttl(Duration::from_secs( + u64::from(settings.openai_session_affinity_ttl_hours) * 3600, + )) .with_smart_scheduling(settings.smart_scheduling) .with_queue(ConcurrencyQueuePolicy { max_waiting: settings.max_waiting_per_account, diff --git a/backend/crates/gateway-core/src/settings/values.rs b/backend/crates/gateway-core/src/settings/values.rs index 0e6c90e33..b9aa05158 100644 --- a/backend/crates/gateway-core/src/settings/values.rs +++ b/backend/crates/gateway-core/src/settings/values.rs @@ -17,6 +17,9 @@ pub struct SettingsValues { pub(crate) max_waiting_per_account: u32, pub(crate) concurrency_wait_timeout_seconds: u32, pub(crate) openai_guardian_reserved_concurrency: u32, + pub(crate) openai_account_affinity: crate::account::AccountAffinity, + pub(crate) max_account_rotations: u32, + pub(crate) openai_session_affinity_ttl_hours: u32, pub(crate) responses_max_decompressed_body_bytes: u64, pub(crate) request_interval_ms: u64, pub(crate) smart_scheduling: crate::account::SmartSchedulingConfig, @@ -27,6 +30,27 @@ pub struct SettingsValues { } impl SettingsValues { + #[must_use] + pub const fn with_openai_session_affinity_ttl_hours(mut self, hours: u32) -> Self { + self.openai_session_affinity_ttl_hours = hours; + self + } + + #[must_use] + pub const fn with_openai_account_affinity( + mut self, + affinity: crate::account::AccountAffinity, + ) -> Self { + self.openai_account_affinity = affinity; + self + } + + #[must_use] + pub const fn with_max_account_rotations(mut self, rotations: u32) -> Self { + self.max_account_rotations = rotations; + self + } + #[must_use] pub const fn with_openai_guardian_reserved_concurrency(mut self, reserved: u32) -> Self { self.openai_guardian_reserved_concurrency = reserved; @@ -114,6 +138,9 @@ impl SettingsValues { max_waiting_per_account: 0, concurrency_wait_timeout_seconds: 30, openai_guardian_reserved_concurrency: 0, + openai_account_affinity: crate::account::AccountAffinity::Relaxed, + max_account_rotations: 3, + openai_session_affinity_ttl_hours: 24, responses_max_decompressed_body_bytes: 64 * 1024 * 1024, request_interval_ms, smart_scheduling: crate::account::SmartSchedulingConfig::default(), diff --git a/backend/crates/gateway-core/tests/engine/coordinator.rs b/backend/crates/gateway-core/tests/engine/coordinator.rs index 6050fca28..c6eed28ef 100644 --- a/backend/crates/gateway-core/tests/engine/coordinator.rs +++ b/backend/crates/gateway-core/tests/engine/coordinator.rs @@ -613,6 +613,42 @@ fn image_stream(image_output_tokens: Option) -> Vec RoutingPlan { plan_with_policy( operation, @@ -687,6 +723,7 @@ fn plan_with_profiles( "acct_wrong", ] .into_iter() + .chain(ROTATION_TEST_ACCOUNTS) .map(|id| { ( ProviderAccountId::new(id).expect("account"), @@ -706,6 +743,17 @@ fn plan_with_profiles( None, ) .with_smart_scheduling(account_selection_policy.smart_scheduling()) + .with_max_account_rotations(account_selection_policy.max_account_rotations()) + .with_openai_account_affinity(account_selection_policy.openai_account_affinity()) + .with_openai_session_affinity_ttl_hours( + u32::try_from( + account_selection_policy + .openai_session_affinity_ttl() + .as_secs() + / 3600, + ) + .unwrap(), + ) .with_request_location(request_location, true), vec![provider.clone()], vec![ProviderModel::new( @@ -5635,3 +5683,138 @@ fn prepare_failure_can_forbid_provider_fallback_without_affecting_ordinary_empty assert_eq!(xai.contexts.lock().unwrap().len(), usize::from(!prohibited)); } } + +#[test] +fn configured_rotation_budget_limits_actual_account_changes() { + for budget in [0, 1, 31] { + let operation = generate_operation(); + let route_plan = plan_with_policy( + &operation, + AccountSelectionPolicy::new( + RotationStrategy::Smart, + NonZeroU32::new(2).unwrap(), + Duration::ZERO, + ) + .with_max_account_rotations(budget) + .with_openai_session_affinity_ttl(Duration::from_secs(168 * 3600)), + ); + let (coordinator, store, provider) = coordinator( + ROTATION_TEST_ACCOUNTS + .iter() + .map(|id| Script::Stream { + account_id: id, + items: vec![Err(ProviderError::new( + ProviderErrorKind::RateLimited, + UpstreamSendState::Sent, + ) + .with_status(429) + .with_replay_safe())], + }) + .collect(), + ); + 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!(block_on(session.collect_uncommitted()).is_err()); + let contexts = provider.contexts.lock().unwrap(); + assert_eq!(contexts.len(), (budget + 1) as usize); + assert!(contexts.iter().all(|context| { + context.account_selection_policy().max_account_rotations() == budget + && context + .account_selection_policy() + .openai_session_affinity_ttl() + == Duration::from_secs(168 * 3600) + })); + assert_eq!( + store.state.lock().unwrap().finalizations[0].attempt_count, + budget + 1 + ); + } +} + +#[test] +fn larger_rotation_budget_reaches_a_healthy_account_after_four_failures() { + let operation = generate_operation(); + let route_plan = plan_with_policy( + &operation, + AccountSelectionPolicy::new( + RotationStrategy::Smart, + NonZeroU32::new(2).unwrap(), + Duration::ZERO, + ) + .with_max_account_rotations(4), + ); + let mut scripts = ROTATION_TEST_ACCOUNTS[..4] + .iter() + .map(|id| Script::Stream { + account_id: id, + items: vec![Err(ProviderError::new( + ProviderErrorKind::RateLimited, + UpstreamSendState::Sent, + ) + .with_replay_safe())], + }) + .collect::>(); + scripts.push(Script::Stream { + account_id: ROTATION_TEST_ACCOUNTS[4], + items: complete_stream(None), + }); + let (coordinator, _, provider) = coordinator(scripts); + let mut session = block_on(coordinator.start( + model_request(&operation, SystemTime::now() + Duration::from_secs(30)), + operation, + route_plan, + None, + None, + CancellationToken::new(), + )) + .unwrap(); + block_on(session.collect_uncommitted()).unwrap(); + assert_eq!(provider.contexts.lock().unwrap().len(), 5); +} + +#[test] +fn zero_rotation_budget_still_allows_retrying_the_same_account() { + let operation = generate_operation(); + let route_plan = plan_with_policy( + &operation, + AccountSelectionPolicy::new( + RotationStrategy::Smart, + NonZeroU32::new(2).unwrap(), + Duration::ZERO, + ) + .with_max_account_rotations(0), + ); + let (coordinator, _, provider) = coordinator(vec![ + Script::Stream { + account_id: "acct_first", + items: vec![Err(transient_rejection())], + }, + 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(); + block_on(session.collect_uncommitted()).unwrap(); + let contexts = provider.contexts.lock().unwrap(); + assert_eq!(contexts.len(), 2); + assert_eq!( + contexts[1].required_account(), + Some(&ProviderAccountId::new("acct_first").unwrap()) + ); +} diff --git a/backend/crates/gateway-core/tests/settings/mod.rs b/backend/crates/gateway-core/tests/settings/mod.rs index e362dd32c..3dc945919 100644 --- a/backend/crates/gateway-core/tests/settings/mod.rs +++ b/backend/crates/gateway-core/tests/settings/mod.rs @@ -194,3 +194,65 @@ fn changing_only_limits_reuses_the_resolved_account_scope() { assert_eq!(resolved.limits(), RateLimits::unlimited()); assert_eq!(resolved.defaults().limits.max_concurrency, 3); } + +#[test] +fn affinity_and_rotation_settings_are_validated_without_mutating_the_snapshot() { + use gateway_core::account::AccountAffinity; + let original = snapshot(1, "host"); + let facts = serde_json::to_value(original.settings()).unwrap(); + assert_eq!(facts["openai_account_affinity"], "relaxed"); + assert_eq!(facts["max_account_rotations"], 3); + for rotations in [0, 31] { + let changed = original + .with_settings( + &original + .settings() + .clone() + .with_openai_account_affinity(AccountAffinity::Strict) + .with_max_account_rotations(rotations), + ) + .unwrap(); + let changed = serde_json::to_value(changed.settings()).unwrap(); + assert_eq!(changed["max_account_rotations"], rotations); + assert_eq!(changed["openai_account_affinity"], "strict"); + } + assert!( + original + .with_settings(&original.settings().clone().with_max_account_rotations(32)) + .is_err() + ); + assert_eq!(serde_json::to_value(original.settings()).unwrap(), facts); +} + +#[test] +fn session_affinity_ttl_rejects_unbounded_or_empty_retention() { + let original = snapshot(1, "host"); + assert_eq!( + serde_json::to_value(original.settings()).unwrap()["openai_session_affinity_ttl_hours"], + 24 + ); + for hours in [1, 168, 720] { + assert!( + original + .with_settings( + &original + .settings() + .clone() + .with_openai_session_affinity_ttl_hours(hours) + ) + .is_ok() + ); + } + for hours in [0, 721, u32::MAX] { + assert!( + original + .with_settings( + &original + .settings() + .clone() + .with_openai_session_affinity_ttl_hours(hours) + ) + .is_err() + ); + } +} diff --git a/backend/crates/gateway-plugin/sdk/src/call/services/settings.rs b/backend/crates/gateway-plugin/sdk/src/call/services/settings.rs index 39b3edb02..ff52ba1a9 100644 --- a/backend/crates/gateway-plugin/sdk/src/call/services/settings.rs +++ b/backend/crates/gateway-plugin/sdk/src/call/services/settings.rs @@ -1,7 +1,6 @@ //! 宿主设置服务的操作标识与请求、响应数据合同 //! //! 从宿主设置类型与 service/settings.rs 生成；更新命令见 SDK 维护说明 - use super::Operation; use serde::{Deserialize, Serialize}; use std::{ @@ -14,6 +13,7 @@ pub type ProviderRequestProfileUpdates = pub type ModelMappings = BTreeMap; pub type Revision = NonZeroU64; pub type RotationStrategy = String; +pub type AccountAffinity = String; pub type PricingOverrides = BTreeMap>; #[derive(Clone, Serialize, Deserialize)] #[serde(deny_unknown_fields)] @@ -31,6 +31,9 @@ pub struct RuntimeSettings { pub max_waiting_per_account: u32, pub concurrency_wait_timeout_seconds: u32, pub openai_guardian_reserved_concurrency: u32, + pub openai_account_affinity: AccountAffinity, + pub max_account_rotations: u32, + pub openai_session_affinity_ttl_hours: u32, pub responses_max_decompressed_body_bytes: u64, pub smart_scheduling: SmartSchedulingConfig, pub rotation_strategy: RotationStrategy, @@ -69,6 +72,9 @@ pub struct ReplaceRuntimeSettings { pub max_waiting_per_account: u32, pub concurrency_wait_timeout_seconds: u32, pub openai_guardian_reserved_concurrency: u32, + pub openai_account_affinity: AccountAffinity, + pub max_account_rotations: u32, + pub openai_session_affinity_ttl_hours: u32, pub responses_max_decompressed_body_bytes: u64, pub smart_scheduling: SmartSchedulingConfig, pub rotation_strategy: RotationStrategy, @@ -264,6 +270,9 @@ impl From for ReplaceRuntimeSettings { max_waiting_per_account: settings.max_waiting_per_account, concurrency_wait_timeout_seconds: settings.concurrency_wait_timeout_seconds, openai_guardian_reserved_concurrency: settings.openai_guardian_reserved_concurrency, + openai_account_affinity: settings.openai_account_affinity, + max_account_rotations: settings.max_account_rotations, + openai_session_affinity_ttl_hours: settings.openai_session_affinity_ttl_hours, responses_max_decompressed_body_bytes: settings.responses_max_decompressed_body_bytes, smart_scheduling: settings.smart_scheduling, rotation_strategy: settings.rotation_strategy, diff --git a/backend/crates/gateway-store/src/admin_adapter.rs b/backend/crates/gateway-store/src/admin_adapter.rs index 1d2582d15..a44b8d0a0 100644 --- a/backend/crates/gateway-store/src/admin_adapter.rs +++ b/backend/crates/gateway-store/src/admin_adapter.rs @@ -94,6 +94,9 @@ impl SettingsStore for AdminSettingsStoreAdapter { max_waiting_per_account: command.max_waiting_per_account, concurrency_wait_timeout_seconds: command.concurrency_wait_timeout_seconds, openai_guardian_reserved_concurrency: command.openai_guardian_reserved_concurrency, + openai_account_affinity: command.openai_account_affinity, + max_account_rotations: command.max_account_rotations, + openai_session_affinity_ttl_hours: command.openai_session_affinity_ttl_hours, responses_max_decompressed_body_bytes: command .responses_max_decompressed_body_bytes, smart_scheduling: command.smart_scheduling, @@ -133,6 +136,9 @@ impl SettingsStore for AdminSettingsStoreAdapter { "max_waiting_per_account".to_owned(), "concurrency_wait_timeout_seconds".to_owned(), "openai_guardian_reserved_concurrency".to_owned(), + "openai_account_affinity".to_owned(), + "max_account_rotations".to_owned(), + "openai_session_affinity_ttl_hours".to_owned(), "responses_max_decompressed_body_bytes".to_owned(), "rotation_strategy".to_owned(), "smart_scheduling_json".to_owned(), @@ -241,6 +247,9 @@ pub(crate) fn admin_runtime_settings( max_waiting_per_account: settings.max_waiting_per_account, concurrency_wait_timeout_seconds: settings.concurrency_wait_timeout_seconds, openai_guardian_reserved_concurrency: settings.openai_guardian_reserved_concurrency, + openai_account_affinity: settings.openai_account_affinity, + max_account_rotations: settings.max_account_rotations, + openai_session_affinity_ttl_hours: settings.openai_session_affinity_ttl_hours, responses_max_decompressed_body_bytes: settings.responses_max_decompressed_body_bytes, smart_scheduling: settings.smart_scheduling, rotation_strategy, diff --git a/backend/crates/gateway-store/src/postgres/runtime_settings.rs b/backend/crates/gateway-store/src/postgres/runtime_settings.rs index 2b0b27707..e1ebd673f 100644 --- a/backend/crates/gateway-store/src/postgres/runtime_settings.rs +++ b/backend/crates/gateway-store/src/postgres/runtime_settings.rs @@ -31,6 +31,9 @@ pub struct RuntimeSettings { pub max_waiting_per_account: u32, pub concurrency_wait_timeout_seconds: u32, pub openai_guardian_reserved_concurrency: u32, + pub openai_account_affinity: gateway_core::account::AccountAffinity, + pub max_account_rotations: u32, + pub openai_session_affinity_ttl_hours: u32, pub responses_max_decompressed_body_bytes: u64, pub smart_scheduling: gateway_core::account::SmartSchedulingConfig, pub rotation_strategy: String, @@ -134,6 +137,9 @@ pub struct RuntimeSettingsUpdate { pub max_waiting_per_account: u32, pub concurrency_wait_timeout_seconds: u32, pub openai_guardian_reserved_concurrency: u32, + pub openai_account_affinity: gateway_core::account::AccountAffinity, + pub max_account_rotations: u32, + pub openai_session_affinity_ttl_hours: u32, pub responses_max_decompressed_body_bytes: u64, pub smart_scheduling: gateway_core::account::SmartSchedulingConfig, pub rotation_strategy: String, @@ -195,6 +201,9 @@ impl RuntimeSettingsUpdate { || !valid_probe_model(self.account_warmup_model.as_deref()) || (self.account_warmup_enabled && self.account_warmup_model.is_none()) || RotationStrategy::parse(&self.rotation_strategy).is_none() + || self.max_account_rotations > gateway_core::account::MAX_ACCOUNT_ROTATIONS + || !(1..=gateway_core::account::MAX_SESSION_AFFINITY_TTL_HOURS) + .contains(&self.openai_session_affinity_ttl_hours) || self.request_profile_updates.len() > 256 || self .request_profile_updates @@ -265,7 +274,7 @@ pub(crate) async fn load_runtime_settings_from_pool(pool: &PgPool) -> StoreResul refresh_concurrency, max_concurrent_per_account, request_interval_ms, rotation_strategy, smart_scheduling_json, model_mappings_json, usage_retention_days, ops_event_retention_days, audit_retention_days, min_codex_desktop_version, - min_codex_cli_version, updated_at, responses_max_decompressed_body_bytes, max_waiting_per_key, max_waiting_per_account, concurrency_wait_timeout_seconds, openai_guardian_reserved_concurrency, + min_codex_cli_version, updated_at, responses_max_decompressed_body_bytes, max_waiting_per_key, max_waiting_per_account, concurrency_wait_timeout_seconds, openai_guardian_reserved_concurrency, openai_account_affinity, max_account_rotations, openai_session_affinity_ttl_hours, account_auto_freeze_enabled, account_auto_freeze_threshold, account_auto_freeze_window_seconds, account_auto_freeze_duration_seconds, account_auto_freeze_probe_enabled, account_auto_freeze_probe_model, @@ -443,7 +452,7 @@ pub(crate) async fn load_runtime_settings_in_transaction( refresh_concurrency, max_concurrent_per_account, request_interval_ms, rotation_strategy, smart_scheduling_json, model_mappings_json, usage_retention_days, ops_event_retention_days, audit_retention_days, min_codex_desktop_version, - min_codex_cli_version, updated_at, responses_max_decompressed_body_bytes, max_waiting_per_key, max_waiting_per_account, concurrency_wait_timeout_seconds, openai_guardian_reserved_concurrency, + min_codex_cli_version, updated_at, responses_max_decompressed_body_bytes, max_waiting_per_key, max_waiting_per_account, concurrency_wait_timeout_seconds, openai_guardian_reserved_concurrency, openai_account_affinity, max_account_rotations, openai_session_affinity_ttl_hours, account_auto_freeze_enabled, account_auto_freeze_threshold, account_auto_freeze_window_seconds, account_auto_freeze_duration_seconds, account_auto_freeze_probe_enabled, account_auto_freeze_probe_model, @@ -520,6 +529,9 @@ pub(crate) async fn update_runtime_settings_in_transaction( account_warmup_model = $29, smart_scheduling_json = $30, openai_guardian_reserved_concurrency = $31, + openai_account_affinity = $32, + max_account_rotations = $33, + openai_session_affinity_ttl_hours = $34, updated_at = now() where id = 1 returning config_revision", @@ -567,6 +579,9 @@ pub(crate) async fn update_runtime_settings_in_transaction( .bind(update.account_warmup_model.as_deref()) .bind(sqlx::types::Json(update.smart_scheduling)) .bind(i64::from(update.openai_guardian_reserved_concurrency)) + .bind(update.openai_account_affinity.as_str()) + .bind(i64::from(update.max_account_rotations)) + .bind(i64::from(update.openai_session_affinity_ttl_hours)) .fetch_optional(&mut **transaction) .await .map_err(|source| postgres_unavailable("update runtime settings in transaction", source))? @@ -642,6 +657,9 @@ struct RuntimeSettingsRow { max_waiting_per_account: i64, concurrency_wait_timeout_seconds: i64, openai_guardian_reserved_concurrency: i64, + openai_account_affinity: String, + max_account_rotations: i64, + openai_session_affinity_ttl_hours: i64, responses_max_decompressed_body_bytes: i64, account_auto_freeze_enabled: bool, account_auto_freeze_threshold: i64, @@ -696,6 +714,16 @@ fn runtime_settings_from_row(row: RuntimeSettingsRow) -> StoreResult, request_location_enabled: bool, responses_max_decompressed_body_bytes: i64, @@ -283,7 +292,7 @@ async fn load_settings( transaction: &mut Transaction<'_, Postgres>, ) -> StoreResult<(Revision, SnapshotRuntimeSettings)> { let row = sqlx::query_as::<_, SnapshotSettingsRow>( - "select config_revision, refresh_margin_seconds, refresh_concurrency, max_concurrent_per_account, request_interval_ms, rotation_strategy, smart_scheduling_json, model_mappings_json, min_codex_desktop_version, min_codex_cli_version, max_waiting_per_key, max_waiting_per_account, concurrency_wait_timeout_seconds, openai_guardian_reserved_concurrency, request_location_json, request_location_enabled, responses_max_decompressed_body_bytes, provider_request_profiles_json, pricing_overrides_json, pricing_synced_json from runtime_settings where id = 1", + "select config_revision, refresh_margin_seconds, refresh_concurrency, max_concurrent_per_account, request_interval_ms, rotation_strategy, smart_scheduling_json, model_mappings_json, min_codex_desktop_version, min_codex_cli_version, max_waiting_per_key, max_waiting_per_account, concurrency_wait_timeout_seconds, openai_guardian_reserved_concurrency, openai_account_affinity, max_account_rotations, openai_session_affinity_ttl_hours, request_location_json, request_location_enabled, responses_max_decompressed_body_bytes, provider_request_profiles_json, pricing_overrides_json, pricing_synced_json from runtime_settings where id = 1", ) .fetch_optional(&mut **transaction) .await @@ -323,6 +332,12 @@ async fn load_settings( max_waiting_per_account: to_u32(row.max_waiting_per_account)?, concurrency_wait_timeout_seconds: to_u32(row.concurrency_wait_timeout_seconds)?, openai_guardian_reserved_concurrency: to_u32(row.openai_guardian_reserved_concurrency)?, + openai_account_affinity: gateway_core::account::AccountAffinity::parse( + &row.openai_account_affinity, + ) + .ok_or_else(|| invalid("invalid account affinity"))?, + max_account_rotations: to_u32(row.max_account_rotations)?, + openai_session_affinity_ttl_hours: to_u32(row.openai_session_affinity_ttl_hours)?, }, )) } diff --git a/backend/crates/gateway-store/src/redis/provider_session_affinity.rs b/backend/crates/gateway-store/src/redis/provider_session_affinity.rs index 1f8ae486f..42d67ad21 100644 --- a/backend/crates/gateway-store/src/redis/provider_session_affinity.rs +++ b/backend/crates/gateway-store/src/redis/provider_session_affinity.rs @@ -14,7 +14,8 @@ use crate::StoreResult; use super::{namespace, resource_fingerprint}; -const MAX_SESSION_AFFINITY_TTL: Duration = Duration::from_secs(24 * 60 * 60); +const MAX_SESSION_AFFINITY_TTL: Duration = + Duration::from_secs(gateway_core::account::MAX_SESSION_AFFINITY_TTL_HOURS as u64 * 3600); // 比较完整记录而非账号 ID；冲突时既不改绑定，也不续期 const COMPARE_AND_BIND_SCRIPT: &str = r#" @@ -55,6 +56,8 @@ impl BindingRecord { #[serde(rename_all = "camelCase", deny_unknown_fields)] struct AliasRecord { session_key: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + root_session_key: Option, follow_only: bool, } @@ -167,6 +170,10 @@ impl ProviderSessionAffinityPort for RedisProviderSessionAffinityRepository { Ok(gateway_core::provider_ports::ProviderSessionAlias { session_key: ProviderSessionAffinityKey::try_new(record.session_key)?, follow_only: record.follow_only, + root_session_key: record + .root_session_key + .map(ProviderSessionAffinityKey::try_new) + .transpose()?, }) }) .transpose() @@ -184,6 +191,10 @@ impl ProviderSessionAffinityPort for RedisProviderSessionAffinityRepository { let record = serde_json::to_string(&AliasRecord { session_key: session.session_key.expose_to_store().to_owned(), follow_only: session.follow_only, + root_session_key: session + .root_session_key + .as_ref() + .map(|key| key.expose_to_store().to_owned()), }) .map_err(|_| provider_invalid("encode session alias"))?; redis::Script::new("local current = redis.call('GET', KEYS[1]); if not current or current == ARGV[1] then redis.call('PSETEX', KEYS[1], tonumber(ARGV[3]), ARGV[2]); return 1 end; return 0") diff --git a/backend/crates/gateway-store/tests/postgres/runtime_settings.rs b/backend/crates/gateway-store/tests/postgres/runtime_settings.rs index df0f255ac..4b94dea27 100644 --- a/backend/crates/gateway-store/tests/postgres/runtime_settings.rs +++ b/backend/crates/gateway-store/tests/postgres/runtime_settings.rs @@ -22,6 +22,9 @@ fn settings_with_margin(refresh_margin_seconds: u64) -> RuntimeSettingsUpdate { max_waiting_per_account: 0, concurrency_wait_timeout_seconds: 30, openai_guardian_reserved_concurrency: 0, + openai_account_affinity: gateway_core::account::AccountAffinity::Relaxed, + max_account_rotations: 3, + openai_session_affinity_ttl_hours: 24, responses_max_decompressed_body_bytes: 64 * 1024 * 1024, smart_scheduling: gateway_core::account::SmartSchedulingConfig::default(), rotation_strategy: "smart".to_owned(), @@ -1092,3 +1095,73 @@ async fn warmup_cursor_resolves_dst_and_deduplicates_across_timezones() { assert!(repository.claim_warmup_slot(zone, missing).await.is_err()); database.close().await; } + +#[tokio::test] +async fn account_affinity_upgrade_defaults_and_updates_reach_the_snapshot() { + use gateway_core::account::AccountAffinity; + use gateway_store::postgres::{PgRuntimeSnapshotRepository, RuntimeSnapshotRepository}; + let Some(database) = TestDatabase::create_through("account_affinity_upgrade", 23).await else { + return; + }; + sqlx::query("update runtime_settings set rotation_strategy = 'sticky' where id = 1") + .execute(&database.pool) + .await + .unwrap(); + super::TEST_MIGRATOR.run(&database.pool).await.unwrap(); + let repository = PgRuntimeSettingsRepository::new(database.pool.clone()); + let before = repository.load_runtime_settings().await.unwrap(); + assert_eq!(before.rotation_strategy, "sticky"); + assert_eq!(before.openai_account_affinity, AccountAffinity::Relaxed); + assert_eq!(before.max_account_rotations, 3); + assert_eq!(before.openai_session_affinity_ttl_hours, 24); + for (mode, budget, ttl) in [ + (AccountAffinity::Strict, 31, 168), + (AccountAffinity::Relaxed, 0, 720), + ] { + let mut update = settings_with_margin(3600); + update.openai_account_affinity = mode; + update.max_account_rotations = budget; + update.openai_session_affinity_ttl_hours = ttl; + repository.update_runtime_settings(update).await.unwrap(); + let loaded = repository.load_runtime_settings().await.unwrap(); + assert_eq!( + (loaded.openai_account_affinity, loaded.max_account_rotations), + (mode, budget) + ); + let snapshot = PgRuntimeSnapshotRepository::new(database.pool.clone()) + .load_runtime_snapshot() + .await + .unwrap(); + assert_eq!( + ( + snapshot.settings.openai_account_affinity, + snapshot.settings.max_account_rotations + ), + (mode, budget) + ); + assert!(snapshot.config_revision > before.config_revision); + assert_eq!(loaded.openai_session_affinity_ttl_hours, ttl); + assert_eq!(snapshot.settings.openai_session_affinity_ttl_hours, ttl); + } + let mut invalid = settings_with_margin(3600); + invalid.max_account_rotations = 32; + assert!(repository.update_runtime_settings(invalid).await.is_err()); + for sql in [ + "update runtime_settings set openai_session_affinity_ttl_hours = 0 where id = 1", + "update runtime_settings set openai_session_affinity_ttl_hours = 721 where id = 1", + "update runtime_settings set max_account_rotations = 32 where id = 1", + "update runtime_settings set max_account_rotations = -1 where id = 1", + "update runtime_settings set openai_account_affinity = 'unknown' where id = 1", + ] { + assert!(sqlx::query(sql).execute(&database.pool).await.is_err()); + } + assert_eq!( + repository + .load_runtime_settings() + .await + .unwrap() + .max_account_rotations, + 0 + ); + database.close().await; +} diff --git a/backend/crates/gateway-store/tests/redis/provider_session_affinity.rs b/backend/crates/gateway-store/tests/redis/provider_session_affinity.rs index 61558ad45..ad9d5ee4a 100644 --- a/backend/crates/gateway-store/tests/redis/provider_session_affinity.rs +++ b/backend/crates/gateway-store/tests/redis/provider_session_affinity.rs @@ -190,10 +190,12 @@ async fn turn_alias_cannot_be_reassigned_and_follows_session_migration() { let b = ProviderAccountId::new("acct_b").unwrap(); let ttl = Duration::from_secs(60); let alias = gateway_core::provider_ports::ProviderSessionAlias { + root_session_key: None, session_key: session.clone(), follow_only: true, }; let other_alias = gateway_core::provider_ports::ProviderSessionAlias { + root_session_key: None, session_key: other, follow_only: true, }; @@ -209,6 +211,7 @@ async fn turn_alias_cannot_be_reassigned_and_follows_session_migration() { .unwrap() ); let different_role = gateway_core::provider_ports::ProviderSessionAlias { + root_session_key: None, session_key: session.clone(), follow_only: false, }; @@ -238,3 +241,98 @@ async fn turn_alias_cannot_be_reassigned_and_follows_session_migration() { &b ); } + +#[tokio::test] +async fn child_turn_alias_preserves_root_identity_without_allowing_reassignment() { + let Some((repo, _, _)) = affinity_repository().await else { + return; + }; + let provider = ProviderKind::new("openai").unwrap(); + let turn = ProviderSessionAffinityKey::try_new("child-turn").unwrap(); + let alias = gateway_core::provider_ports::ProviderSessionAlias { + session_key: ProviderSessionAffinityKey::try_new("child-session").unwrap(), + root_session_key: Some(ProviderSessionAffinityKey::try_new("root-session").unwrap()), + follow_only: false, + }; + let ttl = Duration::from_secs(60); + assert!( + repo.bind_alias(&provider, &turn, &alias, ttl) + .await + .unwrap() + ); + assert_eq!( + repo.load_alias(&provider, &turn).await.unwrap(), + Some(alias.clone()) + ); + let changed = gateway_core::provider_ports::ProviderSessionAlias { + root_session_key: Some(ProviderSessionAffinityKey::try_new("other-root").unwrap()), + ..alias + }; + assert!( + !repo + .bind_alias(&provider, &turn, &changed, ttl) + .await + .unwrap() + ); +} + +#[tokio::test] +async fn binding_and_turn_alias_can_renew_beyond_one_day_and_shorten_without_scanning() { + let Some((repo, mut connection, namespace)) = affinity_repository().await else { + return; + }; + let provider = ProviderKind::new("openai").unwrap(); + let session = ProviderSessionAffinityKey::try_new("long-session").unwrap(); + let turn = ProviderSessionAffinityKey::try_new("long-turn").unwrap(); + let account = ProviderAccountId::new("acct_long").unwrap(); + let alias = gateway_core::provider_ports::ProviderSessionAlias { + session_key: session.clone(), + root_session_key: None, + follow_only: false, + }; + let mut binding = None; + for hours in [24, 168, 720, 1] { + let ttl = Duration::from_secs(hours * 3600); + binding = repo + .compare_and_bind(&provider, &session, binding.as_ref(), &account, ttl) + .await + .unwrap(); + assert!(binding.is_some()); + assert!( + repo.bind_alias(&provider, &turn, &alias, ttl) + .await + .unwrap() + ); + let keys: Vec = redis::cmd("KEYS") + .arg(format!("{namespace}:*")) + .query_async(&mut connection) + .await + .unwrap(); + assert_eq!(keys.len(), 2); + for key in keys { + let remaining: u64 = redis::cmd("PTTL") + .arg(key) + .query_async(&mut connection) + .await + .unwrap(); + let expected = hours * 3600 * 1000; + assert!( + (expected - 5000..=expected).contains(&remaining), + "{remaining} vs {expected}" + ); + } + } + for ttl in [Duration::ZERO, Duration::from_secs(721 * 3600)] { + assert!( + repo.compare_and_bind(&provider, &session, binding.as_ref(), &account, ttl) + .await + .is_err() + ); + assert!( + repo.bind_alias(&provider, &turn, &alias, ttl) + .await + .is_err() + ); + } + assert_eq!(repo.load(&provider, &session).await.unwrap(), binding); +} diff --git a/backend/crates/providers/openai/src/credential/affinity.rs b/backend/crates/providers/openai/src/credential/affinity.rs index 9b8fab82d..664c3b405 100644 --- a/backend/crates/providers/openai/src/credential/affinity.rs +++ b/backend/crates/providers/openai/src/credential/affinity.rs @@ -1,10 +1,9 @@ //! 逻辑会话账号绑定与线程传输隔离键的单向派生 -use std::time::Duration; - +use gateway_core::account::AccountAffinity; use gateway_core::operation::RawJsonPayload; use gateway_core::policy::ClientApiKeyId; -use gateway_core::provider_ports::ProviderSessionAffinityKey; +use gateway_core::provider_ports::{ProviderSessionAffinityKey, ProviderSessionAlias}; use serde_json::{Map, Value}; use sha2::{Digest, Sha256}; @@ -12,11 +11,12 @@ use crate::transport::protocol::responses::CodexResponsesRequest; use crate::transport::request::derive_conversation_anchor; const AFFINITY_KEY_HASH_LENGTH: usize = 12; -pub(crate) const CODEX_ROOT_SESSION_TTL: Duration = Duration::from_secs(24 * 60 * 60); /// 一次请求派生出的账号亲和键及其结构化日志上下文 pub(crate) struct CodexSessionAffinity { key: ProviderSessionAffinityKey, + root_key: Option, + turn_alias: Option<(ProviderSessionAffinityKey, ProviderSessionAlias)>, key_hash: String, anchor_source: &'static str, anchor: String, @@ -26,19 +26,76 @@ pub(crate) struct CodexSessionAffinity { impl CodexSessionAffinity { pub(crate) fn from_turn_alias( - alias: gateway_core::provider_ports::ProviderSessionAlias, + turn: ProviderSessionAffinityKey, + alias: ProviderSessionAlias, + mode: AccountAffinity, ) -> Self { - let key = alias.session_key; + let strict_root = alias + .root_session_key + .as_ref() + .filter(|_| mode == AccountAffinity::Strict); + // 模式切换对新请求生效，旧轮次的子线程身份在严格模式下也归入根绑定 + let key = strict_root.unwrap_or(&alias.session_key).clone(); Self { key_hash: short_key_hash(&key), key, + root_key: if strict_root.is_some() { + None + } else { + alias.root_session_key.clone() + }, anchor_source: "turn-session", anchor: String::new(), session_id: None, - follow_only: alias.follow_only, + follow_only: strict_root.is_some() || alias.follow_only, + turn_alias: Some((turn, alias)), } } + pub(crate) fn turn_alias(&self) -> Option<&ProviderSessionAffinityKey> { + self.turn_alias.as_ref().map(|(key, _)| key) + } + + pub(crate) fn alias_record(&self) -> ProviderSessionAlias { + // 当前请求可加严跟随约束，续期仍保留轮次最初登记的身份，不能改写关联 + self.turn_alias + .as_ref() + .map(|(_, alias)| alias.clone()) + .unwrap_or_else(|| ProviderSessionAlias { + session_key: self.key.clone(), + follow_only: self.follow_only, + root_session_key: self.root_key.clone(), + }) + } + + pub(crate) const fn root_key(&self) -> Option<&ProviderSessionAffinityKey> { + self.root_key.as_ref() + } + + fn with_policy( + mut self, + mode: AccountAffinity, + thread: Option<&str>, + follow_only: bool, + ) -> Self { + self.follow_only = follow_only; + if mode == AccountAffinity::Relaxed + && let Some(thread) = non_empty(thread) + && self.session_id.as_deref() != Some(thread) + && let Some(key) = opaque_affinity_key( + "child-thread", + &format!("{}\0{thread}", self.key.expose_to_store()), + ) + { + self.root_key = Some(self.key); + self.key = key; + self.key_hash = short_key_hash(&self.key); + self.follow_only = false; + } + // 无法识别子线程时保留跟随约束，避免子请求改写根绑定 + self + } + pub(crate) const fn follow_only(&self) -> bool { self.follow_only } @@ -107,6 +164,7 @@ pub(crate) fn derive_previous_response_id_hash( pub(crate) fn derive_codex_session_affinity( request: &CodexResponsesRequest, client_api_key_id: &ClientApiKeyId, + mode: AccountAffinity, ) -> Option { let session_id = non_empty(request.client_account_session_id.as_deref())?; session_affinity( @@ -115,7 +173,13 @@ pub(crate) fn derive_codex_session_affinity( Some(session_id.to_owned()), client_api_key_id, ) - .map(|affinity| affinity.with_follow_only(request.client_account_follow_only)) + .map(|affinity| { + affinity.with_policy( + mode, + request.client_account_thread_id.as_deref(), + request.client_account_follow_only, + ) + }) } /// 原生 JSON 端点与 Responses 共用绑定命名空间，Search id 是逻辑会话身份 @@ -123,8 +187,9 @@ pub(crate) fn derive_codex_endpoint_session_affinity( payload: &RawJsonPayload, client_api_key_id: &ClientApiKeyId, body_session_field: &str, + mode: AccountAffinity, ) -> Option { - derive_endpoint_affinity_with_headers(payload, client_api_key_id, body_session_field, &[]) + derive_endpoint_affinity_with_headers(payload, client_api_key_id, body_session_field, &[], mode) } pub(crate) fn derive_endpoint_affinity_with_headers( @@ -132,6 +197,7 @@ pub(crate) fn derive_endpoint_affinity_with_headers( client_api_key_id: &ClientApiKeyId, body_session_field: &str, headers: &[gateway_core::engine::middleware::MiddlewareHeader], + mode: AccountAffinity, ) -> Option { let mut body = serde_json::from_slice::>(payload.body()).unwrap_or_default(); if let Some(session) = body.get(body_session_field).cloned() { @@ -145,11 +211,11 @@ pub(crate) fn derive_endpoint_affinity_with_headers( client_api_key_id, ) .map(|affinity| { - affinity.with_follow_only(follows_session_with_headers( - &body, - payload.context(), - headers, - )) + affinity.with_policy( + mode, + account_thread_with_headers(&body, payload.context(), headers).as_deref(), + follows_session_with_headers(&body, payload.context(), headers), + ) }) } @@ -159,21 +225,11 @@ pub(crate) fn follows_session_with_headers( context: &Map, headers: &[gateway_core::engine::middleware::MiddlewareHeader], ) -> bool { - let header = |name: &str| { - headers - .iter() - .find(|h| h.name().eq_ignore_ascii_case(name)) - .and_then(|h| std::str::from_utf8(h.value()).ok()) - }; - let metadata = header("x-codex-turn-metadata"); - let thread = metadata - .and_then(gateway_protocol::openai::turn_metadata_thread_id) - .or_else(|| { - header("thread-id") - .filter(|value| !value.trim().is_empty()) - .map(str::to_owned) - }) - .or_else(|| gateway_protocol::openai::codex_account_thread_id(body, context)); + let metadata = headers + .iter() + .find(|h| h.name().eq_ignore_ascii_case("x-codex-turn-metadata")) + .and_then(|h| std::str::from_utf8(h.value()).ok()); + let thread = account_thread_with_headers(body, context, headers); if let Some((session, thread)) = account_session_with_headers(body, context, headers).zip(thread) { @@ -187,6 +243,27 @@ pub(crate) fn follows_session_with_headers( .is_some() } +pub(crate) fn account_thread_with_headers( + body: &Map, + context: &Map, + headers: &[gateway_core::engine::middleware::MiddlewareHeader], +) -> Option { + let header = |name: &str| { + headers + .iter() + .find(|h| h.name().eq_ignore_ascii_case(name)) + .and_then(|h| std::str::from_utf8(h.value()).ok()) + }; + header("x-codex-turn-metadata") + .and_then(gateway_protocol::openai::turn_metadata_thread_id) + .or_else(|| { + header("thread-id") + .and_then(|value| non_empty(Some(value))) + .map(str::to_owned) + }) + .or_else(|| gateway_protocol::openai::codex_account_thread_id(body, context)) +} + /// 中间件最终请求头按实际传输的覆盖语义参与身份复验 pub(crate) fn account_session_with_headers( body: &Map, @@ -220,6 +297,7 @@ pub(crate) fn derive_live_session_affinity( request: &gateway_core::operation::ProviderHttpRequest, headers: &[gateway_core::engine::middleware::MiddlewareHeader], client_api_key_id: &ClientApiKeyId, + mode: AccountAffinity, ) -> Option { let mut effective = headers.to_vec(); effective.extend( @@ -244,11 +322,11 @@ pub(crate) fn derive_live_session_affinity( client_api_key_id, ) .map(|affinity| { - affinity.with_follow_only(follows_session_with_headers( - &Map::new(), - &Map::new(), - &effective, - )) + affinity.with_policy( + mode, + account_thread_with_headers(&Map::new(), &Map::new(), &effective).as_deref(), + follows_session_with_headers(&Map::new(), &Map::new(), &effective), + ) }) } @@ -281,6 +359,8 @@ fn session_affinity( let key_hash = short_key_hash(&key); Some(CodexSessionAffinity { key, + root_key: None, + turn_alias: None, key_hash, anchor_source, anchor, diff --git a/backend/crates/providers/openai/src/credential/mod.rs b/backend/crates/providers/openai/src/credential/mod.rs index 435044416..b8df5b4dc 100644 --- a/backend/crates/providers/openai/src/credential/mod.rs +++ b/backend/crates/providers/openai/src/credential/mod.rs @@ -17,11 +17,12 @@ pub mod token_client; mod types; pub(crate) use affinity::{ - CodexSessionAffinity, account_session_with_headers, derive_codex_cyber_policy_session_key, - derive_codex_endpoint_session_affinity, derive_codex_session_affinity, - derive_codex_transport_key, derive_endpoint_affinity_with_headers, - derive_live_session_affinity, derive_previous_response_id_hash, derive_turn_alias, - follows_session_with_headers, turn_id_with_headers, + CodexSessionAffinity, account_session_with_headers, account_thread_with_headers, + derive_codex_cyber_policy_session_key, derive_codex_endpoint_session_affinity, + derive_codex_session_affinity, derive_codex_transport_key, + derive_endpoint_affinity_with_headers, derive_live_session_affinity, + derive_previous_response_id_hash, derive_turn_alias, follows_session_with_headers, + turn_id_with_headers, }; pub(crate) use oauth::oauth_owner_ref; pub(crate) use types::parse_access_token_expiration; diff --git a/backend/crates/providers/openai/src/credential/selector.rs b/backend/crates/providers/openai/src/credential/selector.rs index ff2827e9b..ac792f094 100644 --- a/backend/crates/providers/openai/src/credential/selector.rs +++ b/backend/crates/providers/openai/src/credential/selector.rs @@ -26,7 +26,7 @@ use secrecy::ExposeSecret; use thiserror::Error; use url::Url; -use super::affinity::{CODEX_ROOT_SESSION_TTL, CodexSessionAffinity}; +use super::affinity::CodexSessionAffinity; use super::cookie::CodexCookiePolicy; use super::quota::CodexCredentialQuotaService; use super::refresh::refresh_recovery_deadline; @@ -342,6 +342,7 @@ impl CodexCredentialSelector { lease: &mut CodexCredentialLease, session_affinity: Option<&CodexSessionAffinity>, cyber_policy_session_key: Option<&ProviderSessionAffinityKey>, + session_ttl: Duration, ) -> Result<(), CredentialSelectionError> { let selected_account = lease.account.id().clone(); if let Some(affinity) = session_affinity @@ -353,7 +354,12 @@ impl CodexCredentialSelector { .as_ref() .is_some_and(|binding| binding.account_id() != &selected_account) || !self - .admit_session(affinity.key(), current.as_ref(), &selected_account) + .admit_session( + affinity.key(), + current.as_ref(), + &selected_account, + session_ttl, + ) .await? { return Err(CredentialSelectionError::SessionBound(Box::new( @@ -634,16 +640,28 @@ impl CodexCredentialSelector { AccountCandidate { account, signals } }) .collect::>(); - let mut affinity = - binding - .as_ref() - .map_or_else(AffinitySelection::default, |binding| { - affinity_selection_for_bound_account( - binding.account_id().clone(), - &candidates, - SystemTime::now(), - ) - }); + let inherited = if binding.is_none() && !diagnostic { + if let Some(root) = request + .session_affinity_observation + .and_then(CodexSessionAffinity::root_key) + { + self.lookup_session_affinity(root).await? + } else { + None + } + } else { + None + }; + let mut affinity = binding.as_ref().or(inherited.as_ref()).map_or_else( + AffinitySelection::default, + |binding| { + affinity_selection_for_bound_account( + binding.account_id().clone(), + &candidates, + SystemTime::now(), + ) + }, + ); let cyber_policy_scope = self .prepare_cyber_policy_scope(cyber_policy_session_key) .await; @@ -654,7 +672,7 @@ impl CodexCredentialSelector { { excluded.extend(state.excluded_accounts().iter().cloned()); } - // 继续遵守既有重试排除与固定账号规则；换号时迁移共享会话绑定 + // 继续遵守既有重试排除与固定账号规则；只迁移本请求所属绑定 if let Some(required) = pinned_account.as_ref() { excluded.extend( candidates @@ -883,7 +901,15 @@ impl CodexCredentialSelector { ProviderLeaseAcquisition::Acquired(guard) => { if let Some(key) = binding_key && !self - .admit_session(key, binding.as_ref(), account.id()) + .admit_session( + key, + binding.as_ref(), + account.id(), + request + .attempt + .account_selection_policy() + .openai_session_affinity_ttl(), + ) .await? { drop(guard); @@ -959,17 +985,15 @@ impl CodexCredentialSelector { &self, turn: &ProviderSessionAffinityKey, session: &CodexSessionAffinity, + session_ttl: Duration, ) -> Result<(), CredentialSelectionError> { let applied = tokio::time::timeout( SESSION_AFFINITY_TIMEOUT, self.session_affinity.bind_alias( &self.provider_kind, turn, - &gateway_core::provider_ports::ProviderSessionAlias { - session_key: session.key().clone(), - follow_only: session.follow_only(), - }, - CODEX_ROOT_SESSION_TTL, + &session.alias_record(), + session_ttl, ), ) .await @@ -986,6 +1010,7 @@ impl CodexCredentialSelector { pub(crate) async fn session_for_turn( &self, turn: &ProviderSessionAffinityKey, + mode: gateway_core::account::AccountAffinity, ) -> Result, CredentialSelectionError> { tokio::time::timeout( SESSION_AFFINITY_TIMEOUT, @@ -994,7 +1019,9 @@ impl CodexCredentialSelector { .await .map_err(binding_store_error)? .map_err(binding_store_error) - .map(|key| key.map(CodexSessionAffinity::from_turn_alias)) + .map(|alias| { + alias.map(|alias| CodexSessionAffinity::from_turn_alias(turn.clone(), alias, mode)) + }) } async fn lookup_session_affinity( @@ -1015,6 +1042,7 @@ impl CodexCredentialSelector { key: &ProviderSessionAffinityKey, expected: Option<&ProviderSessionBinding>, account: &ProviderAccountId, + session_ttl: Duration, ) -> Result { tokio::time::timeout( SESSION_AFFINITY_TIMEOUT, @@ -1023,7 +1051,7 @@ impl CodexCredentialSelector { key, expected, account, - CODEX_ROOT_SESSION_TTL, + session_ttl, ), ) .await diff --git a/backend/crates/providers/openai/src/provider/execution.rs b/backend/crates/providers/openai/src/provider/execution.rs index d6af23208..8bc35628d 100644 --- a/backend/crates/providers/openai/src/provider/execution.rs +++ b/backend/crates/providers/openai/src/provider/execution.rs @@ -63,6 +63,7 @@ impl CodexProvider { context.client_api_key_ref(), "session_id", headers, + context.account_selection_policy().openai_account_affinity(), ); let turn = headers .iter() @@ -78,7 +79,10 @@ impl CodexProvider { turn.and_then(|turn| derive_turn_alias(turn, context.client_api_key_ref())) { self.selector - .session_for_turn(&alias) + .session_for_turn( + &alias, + context.account_selection_policy().openai_account_affinity(), + ) .await .map_err(map_selection_error)? } else { @@ -87,7 +91,9 @@ impl CodexProvider { if explicit .as_ref() .zip(inferred.as_ref()) - .is_some_and(|(explicit, inferred)| explicit.key() != inferred.key()) + .is_some_and(|(explicit, inferred)| { + explicit.key() != inferred.key() && Some(explicit.key()) != inferred.root_key() + }) { return Err(provider_error( ProviderErrorKind::InvalidRequest, @@ -97,9 +103,12 @@ impl CodexProvider { } let follow_only = inferred .as_ref() - .is_some_and(CodexSessionAffinity::follow_only); - Ok(explicit - .or(inferred) + .is_some_and(CodexSessionAffinity::follow_only) + || explicit + .as_ref() + .is_some_and(CodexSessionAffinity::follow_only); + Ok(inferred + .or(explicit) .map(|affinity| affinity.with_follow_only(follow_only))) } @@ -125,6 +134,7 @@ impl CodexProvider { search.payload(), context.client_api_key_ref(), "id", + context.account_selection_policy().openai_account_affinity(), ); let response_origin = self.search_url.clone(); self.execute_raw_json_endpoint( @@ -211,12 +221,20 @@ impl CodexProvider { context.client_api_key_ref(), "id", &middleware_headers, + context.account_selection_policy().openai_account_affinity(), ), _ => None, }; if !context.is_diagnostic_required_account() { self.selector - .validate_translated_selection(&mut lease, affinity.as_ref(), None) + .validate_translated_selection( + &mut lease, + affinity.as_ref(), + None, + context + .account_selection_policy() + .openai_session_affinity_ttl(), + ) .await .map_err(map_selection_error)?; } @@ -267,6 +285,21 @@ impl CodexProvider { )); } } + if !context.is_diagnostic_required_account() + && let Some(affinity) = affinity.as_ref() + && let Some(turn) = affinity.turn_alias() + { + self.selector + .remember_turn( + turn, + affinity, + context + .account_selection_policy() + .openai_session_affinity_ttl(), + ) + .await + .map_err(map_selection_error)?; + } let allows_account_state_mutation = lease.allows_account_state_mutation(); let provider_kind = ProviderKind::new(PROVIDER_NAME) .map_err(|_| provider_error(ProviderErrorKind::Protocol, UpstreamSendState::NotSent))?; diff --git a/backend/crates/providers/openai/src/provider/live.rs b/backend/crates/providers/openai/src/provider/live.rs index 16017290d..02acfd786 100644 --- a/backend/crates/providers/openai/src/provider/live.rs +++ b/backend/crates/providers/openai/src/provider/live.rs @@ -243,8 +243,12 @@ impl CodexProvider { UpstreamSendState::NotSent, ) })?; - let session_affinity = - derive_live_session_affinity(&request, &[], context.client_api_key_ref()); + let session_affinity = derive_live_session_affinity( + &request, + &[], + context.client_api_key_ref(), + context.account_selection_policy().openai_account_affinity(), + ); let selection_started_at = Instant::now(); let lease = self .selector @@ -317,10 +321,18 @@ impl CodexProvider { &request, &middleware_headers, context.client_api_key_ref(), + context.account_selection_policy().openai_account_affinity(), ); if !context.is_diagnostic_required_account() { self.selector - .validate_translated_selection(&mut lease, session_affinity.as_ref(), None) + .validate_translated_selection( + &mut lease, + session_affinity.as_ref(), + None, + context + .account_selection_policy() + .openai_session_affinity_ttl(), + ) .await .map_err(map_selection_error)?; } diff --git a/backend/crates/providers/openai/src/provider/mod.rs b/backend/crates/providers/openai/src/provider/mod.rs index bb34992e6..f857155c8 100644 --- a/backend/crates/providers/openai/src/provider/mod.rs +++ b/backend/crates/providers/openai/src/provider/mod.rs @@ -213,8 +213,11 @@ impl CodexProvider { clear_request_turn_state(&mut upstream); } } - let session_affinity = - derive_codex_session_affinity(&upstream, context.client_api_key_ref()); + let session_affinity = derive_codex_session_affinity( + &upstream, + context.client_api_key_ref(), + context.account_selection_policy().openai_account_affinity(), + ); let cyber_policy_session_key = derive_codex_cyber_policy_session_key(&upstream, context.client_api_key_ref()); PreparedGenerateRequest { @@ -382,11 +385,13 @@ impl Provider for CodexProvider { request.payload(), client_api_key_id, "id", + gateway_core::account::AccountAffinity::Strict, ), Operation::GenerateImage(request) => derive_codex_endpoint_session_affinity( request.payload(), client_api_key_id, "session_id", + gateway_core::account::AccountAffinity::Strict, ), _ => None, }; @@ -414,8 +419,12 @@ impl Provider for CodexProvider { let reasoning_effort = semantics.reasoning_effort.clone(); let previous_response_id = encoded.previous_response_id(); let continuation = ContinuationRequestObservation { - affinity_hash: derive_codex_session_affinity(&encoded, client_api_key_id) - .map(|affinity| affinity.persistence_hash().to_owned()), + affinity_hash: derive_codex_session_affinity( + &encoded, + client_api_key_id, + gateway_core::account::AccountAffinity::Strict, + ) + .map(|affinity| affinity.persistence_hash().to_owned()), previous_response_id_hash: previous_response_id.map(|response_id| { derive_previous_response_id_hash(response_id, client_api_key_id) }), @@ -714,6 +723,11 @@ impl CodexProvider { generate.protocol_payload().context(), &middleware_headers, ); + upstream.client_account_thread_id = crate::credential::account_thread_with_headers( + generate.protocol_payload().body(), + generate.protocol_payload().context(), + &middleware_headers, + ); let processed = self.prepare_generate_request(&generate, upstream, &context); let mut upstream_request = processed.upstream; let session_transport_key = @@ -728,6 +742,9 @@ impl CodexProvider { &mut lease, session_affinity.as_ref(), cyber_policy_session_key.as_ref(), + context + .account_selection_policy() + .openai_session_affinity_ttl(), ) .await .map_err(map_selection_error)?; @@ -742,7 +759,13 @@ impl CodexProvider { .and_then(|turn| derive_turn_alias(&turn, context.client_api_key_ref())) { self.selector - .remember_turn(&turn, affinity) + .remember_turn( + &turn, + affinity, + context + .account_selection_policy() + .openai_session_affinity_ttl(), + ) .await .map_err(map_selection_error)?; } diff --git a/backend/crates/providers/openai/src/transport/protocol/responses.rs b/backend/crates/providers/openai/src/transport/protocol/responses.rs index 5a8fbb9a0..3811e652d 100644 --- a/backend/crates/providers/openai/src/transport/protocol/responses.rs +++ b/backend/crates/providers/openai/src/transport/protocol/responses.rs @@ -44,7 +44,8 @@ pub struct CodexResponsesRequest { pub client_session_id: Option, /// 官方元数据中的逻辑会话身份，独立于缓存路由键 pub client_account_session_id: Option, - /// 后代线程只跟随会话账号，不自行迁移绑定 + pub client_account_thread_id: Option, + /// 后代线程身份，用于严格模式下只跟随会话账号 pub client_account_follow_only: bool, /// 客户端 thread ID，仅保留在受控本地上下文 pub client_thread_id: Option, @@ -505,6 +506,7 @@ impl CodexResponsesRequest { client_conversation_id: None, client_session_id: None, client_account_session_id: None, + client_account_thread_id: None, client_account_follow_only: false, client_thread_id: None, client_request_id: None, diff --git a/backend/crates/providers/openai/src/transport/request.rs b/backend/crates/providers/openai/src/transport/request.rs index 6d0cad009..d62b833ec 100644 --- a/backend/crates/providers/openai/src/transport/request.rs +++ b/backend/crates/providers/openai/src/transport/request.rs @@ -733,6 +733,8 @@ fn apply_protocol_context(request: &mut CodexResponsesRequest, context: &Map BoxFuture<'static, Result> { + let metadata = self.0; Box::pin(async move { let (protocol, mut headers, body) = request.into_parts(); headers.push(MiddlewareHeader::new( "x-codex-turn-metadata", - Bytes::from_static(br#"{"session_id":"session-b"}"#), + Bytes::from_static(metadata), )); next.run(MiddlewareRequest::new(protocol, headers, body)) .await @@ -64,7 +65,7 @@ async fn final_middleware_session_header_cannot_use_another_sessions_account() { request("session-a"), context_with_middleware( "req_final_identity", - Arc::new(ChangeSessionHeader), + Arc::new(ChangeTurnMetadata(br#"{"session_id":"session-b"}"#)), FastMode::Default, ), ) @@ -270,9 +271,35 @@ async fn descendant_images_wait_for_the_owner_even_when_ordinary_queues_are_disa .lock() .unwrap() .insert(ProviderAccountId::new("acct_subagent_a").unwrap()); - for kind in [ImageRequestKind::Generation, ImageRequestKind::Edit] { + for (kind, turn, explicit_child) in [ + (ImageRequestKind::Generation, "child-turn", false), + (ImageRequestKind::Edit, "child-turn", false), + (ImageRequestKind::Generation, "root-turn", true), + (ImageRequestKind::Edit, "root-turn", true), + ] { + let image = if explicit_child { + Operation::GenerateImage(ImageRequest::from_raw_json( + kind, + RawJsonPayload::new( + "openai", + Bytes::from_static( + br#"{"prompt":"a square","model":"gpt-image-1","session_id":"root"}"#, + ), + ) + .unwrap() + .with_context(Map::from_iter([ + ("image_turn_id".into(), json!(turn)), + ( + "turn_metadata".into(), + json!(json!({"session_id":"root","thread_id":"child"}).to_string()), + ), + ])), + )) + } else { + image_for_turn(kind, turn) + }; let mut pending = Box::pin(provider.clone().execute( - planned_provider_endpoint_request("openai", image_for_turn(kind, "child-turn")), + planned_provider_endpoint_request("openai", image), context("req_child_image_wait", CancellationToken::new()), )); assert!( @@ -557,3 +584,434 @@ async fn old_native_continuation_cannot_restore_the_pre_migration_account() { ); assert!(server.received_requests().await.unwrap().is_empty()); } + +fn relaxed_context(request_id: &str, scope: Arc) -> AttemptContext { + relaxed_context_with_ttl(request_id, scope, Duration::from_secs(24 * 3600)) +} + +fn relaxed_context_with_ttl( + request_id: &str, + scope: Arc, + ttl: Duration, +) -> AttemptContext { + AttemptContext::new( + RequestAttemptContext::new( + ModelRequestId::new(request_id).unwrap(), + ClientApiKeyId::new("key_openai_contract").unwrap(), + ), + NonZeroU32::new(1).unwrap(), + SystemTime::now() + Duration::from_secs(5), + gateway_core::account::AccountSelectionPolicy::new( + gateway_core::account::RotationStrategy::Smart, + NonZeroU32::new(2).unwrap(), + Duration::ZERO, + ) + .with_openai_session_affinity_ttl(ttl), + AccountAttemptContext::new(BTreeSet::new(), None, None).with_account_scope(scope), + None, + CancellationToken::new(), + ) +} + +#[tokio::test] +async fn relaxed_children_keep_independent_bindings_across_busy_accounts_and_root_migration() { + let store = Arc::new(MemoryAccountStore::default()); + create_account(&store, "acct_subagent_a").await; + let affinity = Arc::new(MemorySessionAffinity::default()); + let leases = Arc::new(TestLeaseCoordinator::default()); + let server = MockServer::start().await; + let provider = provider_with_affinity_and_base_url_and_leases( + &store, + affinity.clone(), + server.uri(), + leases.clone(), + ); + let select = |thread: &'static str| { + let provider = provider.clone(); + async move { + provider + .execute( + planned_request( + "openai", + Operation::Generate(generate_with_session_context( + "root", + Some(thread), + None, + )), + ), + relaxed_context("req_relaxed", contract_account_scope()), + ) + .await + .unwrap() + } + }; + let root = select("root").await; + create_account(&store, "acct_subagent_b").await; + store.set_scheduling("acct_subagent_b", None, AccountWeight::new(100).unwrap()); + let sibling = select("sibling").await; + assert_eq!( + sibling.metadata().provider_account_id().as_str(), + "acct_subagent_a", + "new child inherits root preference over weight" + ); + drop(sibling); + leases + .busy_accounts + .lock() + .unwrap() + .insert(ProviderAccountId::new("acct_subagent_a").unwrap()); + let child = select("child").await; + assert_eq!( + child.metadata().provider_account_id().as_str(), + "acct_subagent_b" + ); + drop(child); + leases.busy_accounts.lock().unwrap().clear(); + for (thread, expected) in [ + ("root", "acct_subagent_a"), + ("sibling", "acct_subagent_a"), + ("child", "acct_subagent_b"), + ] { + assert_eq!( + select(thread) + .await + .metadata() + .provider_account_id() + .as_str(), + expected + ); + } + leases + .busy_accounts + .lock() + .unwrap() + .insert(ProviderAccountId::new("acct_subagent_a").unwrap()); + assert_eq!( + select("root") + .await + .metadata() + .provider_account_id() + .as_str(), + "acct_subagent_b" + ); + leases.busy_accounts.lock().unwrap().clear(); + assert_eq!( + select("sibling") + .await + .metadata() + .provider_account_id() + .as_str(), + "acct_subagent_a", + "root migration cannot move an existing child" + ); + assert_eq!( + select("new-sibling") + .await + .metadata() + .provider_account_id() + .as_str(), + "acct_subagent_b" + ); + assert_eq!( + root.metadata().provider_account_id().as_str(), + "acct_subagent_a", + "the admitted root lease stays on its original account" + ); + assert_eq!(affinity.binding_count(), 4); +} + +#[tokio::test] +async fn relaxed_child_can_claim_without_a_root_and_model_access_can_split_accounts() { + use gateway_core::account::{AccountModelAccess, AccountModelAccessMode}; + let store = Arc::new(MemoryAccountStore::default()); + create_account(&store, "acct_subagent_a").await; + let affinity = Arc::new(MemorySessionAffinity::default()); + let provider = provider_with_affinity(&store, affinity.clone()); + let select = |thread, model: &'static str, scope| { + provider.clone().execute( + planned_request_for_model( + "openai", + Operation::Generate(generate_with_session_context("root", Some(thread), None)), + model, + ), + relaxed_context("req_relaxed_model", scope), + ) + }; + drop( + select("early-child", "gpt-5.4", contract_account_scope()) + .await + .unwrap(), + ); + assert_eq!( + affinity.binding_count(), + 1, + "early child must not claim the root binding" + ); + drop( + select("root", "gpt-5.4", contract_account_scope()) + .await + .unwrap(), + ); + create_account(&store, "acct_subagent_b").await; + let directory = RuntimeAccountDirectory::new( + [ + ("acct_subagent_a", "gpt-5.4"), + ("acct_subagent_b", "gpt-5.5"), + ] + .into_iter() + .map(|(id, model)| { + ( + ProviderAccountId::new(id).unwrap(), + RuntimeAccount::new(ProviderKind::new("openai").unwrap(), BTreeSet::new()) + .with_model_access( + AccountModelAccess::new( + AccountModelAccessMode::Allowlist, + vec![model.to_owned()], + ) + .unwrap(), + ), + ) + }) + .collect(), + ); + let scope = Arc::new(FrozenAccountScope::new( + Arc::new(directory), + ClientRoutingScope::all_accounts(), + )); + let child = select("model-child", "gpt-5.5", scope.clone()) + .await + .unwrap(); + assert_eq!( + child.metadata().provider_account_id().as_str(), + "acct_subagent_b" + ); + drop(child); + assert_eq!( + select("root", "gpt-5.4", scope) + .await + .unwrap() + .metadata() + .provider_account_id() + .as_str(), + "acct_subagent_a" + ); + assert_eq!(affinity.binding_count(), 3); +} + +#[tokio::test] +async fn relaxed_image_turn_follows_child_binding_and_accepts_matching_root_identity() { + let store = Arc::new(MemoryAccountStore::default()); + create_account(&store, "acct_subagent_a").await; + let affinity = Arc::new(MemorySessionAffinity::default()); + let leases = Arc::new(TestLeaseCoordinator::default()); + let server = MockServer::start().await; + let provider = provider_with_affinity_and_base_url_and_leases( + &store, + affinity, + server.uri(), + leases.clone(), + ); + drop( + provider + .clone() + .execute( + planned_request("openai", turn_request("root", "root-turn")), + relaxed_context("req_relaxed_root", contract_account_scope()), + ) + .await + .unwrap(), + ); + create_account(&store, "acct_subagent_b").await; + leases + .busy_accounts + .lock() + .unwrap() + .insert(ProviderAccountId::new("acct_subagent_a").unwrap()); + let child = Operation::Generate(GenerateRequest::from_protocol_payload( + ProtocolPayload::json_object("openai", json!({"model":"gpt-5.4","input":"child","client_metadata":{"session_id":"root","thread_id":"child","turn_id":"child-turn"}}).as_object().unwrap().clone()).unwrap() + )); + drop( + provider + .clone() + .execute( + planned_request("openai", child), + relaxed_context("req_relaxed_child", contract_account_scope()), + ) + .await + .unwrap(), + ); + leases.busy_accounts.lock().unwrap().clear(); + for session in [None, Some("root")] { + let mut body = json!({"prompt":"a square", "model":"gpt-image-1"}); + if let Some(session) = session { + body["session_id"] = json!(session); + } + let image = Operation::GenerateImage(ImageRequest::from_raw_json( + ImageRequestKind::Generation, + RawJsonPayload::new("openai", Bytes::from(serde_json::to_vec(&body).unwrap())) + .unwrap() + .with_context(Map::from_iter([( + "image_turn_id".into(), + json!("child-turn"), + )])), + )); + let stream = provider + .clone() + .execute( + planned_provider_endpoint_request("openai", image), + relaxed_context("req_relaxed_image", contract_account_scope()), + ) + .await + .unwrap(); + assert_eq!( + stream.metadata().provider_account_id().as_str(), + "acct_subagent_b" + ); + } + // 新请求切到严格模式后，旧子线程轮次必须跟随根绑定；再切回宽松仍保留子绑定 + for strict in [true, false] { + let attempt = if strict { + context("req_strict_old_turn", CancellationToken::new()) + } else { + relaxed_context("req_relaxed_old_turn", contract_account_scope()) + }; + let stream = provider + .clone() + .execute( + planned_provider_endpoint_request( + "openai", + image_for_turn(ImageRequestKind::Generation, "child-turn"), + ), + attempt, + ) + .await + .unwrap(); + assert_eq!( + stream.metadata().provider_account_id().as_str(), + if strict { + "acct_subagent_a" + } else { + "acct_subagent_b" + } + ); + } +} + +#[tokio::test] +async fn configured_ttl_renews_binding_and_image_alias_only_at_successful_admission() { + let store = Arc::new(MemoryAccountStore::default()); + create_account(&store, "acct_subagent_a").await; + let affinity = Arc::new(MemorySessionAffinity::default()); + let server = MockServer::start().await; + let provider = provider_with_affinity_and_base_url(&store, affinity.clone(), server.uri()); + let week = Duration::from_secs(7 * 24 * 3600); + let hour = Duration::from_secs(3600); + for ttl in [week, hour] { + let stream = provider + .clone() + .execute( + planned_request("openai", turn_request("root", "ttl-turn")), + relaxed_context_with_ttl("req_ttl_root", contract_account_scope(), ttl), + ) + .await + .unwrap(); + let before = affinity.renewal_ttls(); + assert_eq!(before.last(), Some(&ttl)); + assert_eq!(affinity.alias_ttls().last(), Some(&ttl)); + drop(stream); + assert_eq!( + affinity.renewal_ttls(), + before, + "dropping the response cannot renew the binding" + ); + } + drop( + provider + .execute( + planned_provider_endpoint_request( + "openai", + image_for_turn(ImageRequestKind::Generation, "ttl-turn"), + ), + relaxed_context_with_ttl("req_ttl_image", contract_account_scope(), week), + ) + .await + .unwrap(), + ); + assert_eq!(affinity.renewal_ttls(), vec![week, hour, week]); + assert_eq!(affinity.alias_ttls(), vec![week, hour, week]); +} + +#[tokio::test] +async fn relaxed_final_thread_rewrite_cannot_use_another_threads_account() { + let store = Arc::new(MemoryAccountStore::default()); + create_account(&store, "acct_subagent_a").await; + let affinity = Arc::new(MemorySessionAffinity::default()); + let server = MockServer::start().await; + let provider = provider_with_affinity_and_base_url(&store, affinity, server.uri()); + let request = |thread| { + planned_request( + "openai", + Operation::Generate(generate_with_session_context("root", Some(thread), None)), + ) + }; + drop( + provider + .clone() + .execute( + request("root"), + relaxed_context("req_seed_root", contract_account_scope()), + ) + .await + .unwrap(), + ); + create_account(&store, "acct_subagent_b").await; + let a = store.account("acct_subagent_a").unwrap(); + store.set_enabled(a.id(), false).await.unwrap(); + drop( + provider + .clone() + .execute( + request("child"), + relaxed_context("req_seed_child", contract_account_scope()), + ) + .await + .unwrap(), + ); + store.set_enabled(a.id(), true).await.unwrap(); + let plan = FrozenMiddlewarePlan::new( + Arc::new(ChangeTurnMetadata( + br#"{"session_id":"root","thread_id":"child"}"#, + )), + ExtensionSetReference::new( + ExtensionSetId::new("rewrite-thread".into()).unwrap(), + Arc::new(TestExtensionLease), + ), + ); + let attempt = AttemptContext::new( + RequestAttemptContext::new( + ModelRequestId::new("req_final_thread").unwrap(), + ClientApiKeyId::new("key_openai_contract").unwrap(), + ) + .with_middleware( + Some(plan), + Arc::from([]), + "/v1/responses".into(), + ClientTransport::HttpSse, + ), + NonZeroU32::MIN, + SystemTime::now() + Duration::from_secs(5), + account_policy() + .with_openai_account_affinity(gateway_core::account::AccountAffinity::Relaxed), + AccountAttemptContext::new(BTreeSet::new(), None, None) + .with_account_scope(contract_account_scope()), + None, + CancellationToken::new(), + ); + let error = provider + .execute(request("root"), attempt) + .await + .err() + .expect("thread owner mismatch"); + assert!(error.retry_is_prohibited()); + assert!(server.received_requests().await.unwrap().is_empty()); +} diff --git a/backend/crates/providers/openai/tests/support.rs b/backend/crates/providers/openai/tests/support.rs index 7512351f3..67c46dba2 100644 --- a/backend/crates/providers/openai/tests/support.rs +++ b/backend/crates/providers/openai/tests/support.rs @@ -739,9 +739,14 @@ pub(crate) struct MemorySessionAffinity { Mutex>, lookups: Mutex>, renewal_ttls: Mutex>, + alias_ttls: Mutex>, } impl MemorySessionAffinity { + pub(crate) fn alias_ttls(&self) -> Vec { + self.alias_ttls.lock().unwrap().clone() + } + pub(crate) fn renewal_ttls(&self) -> Vec { self.renewal_ttls.lock().expect("affinity TTL lock").clone() } @@ -877,7 +882,7 @@ impl ProviderSessionAffinityPort for MemorySessionAffinity { provider: &'a ProviderKind, alias: &'a ProviderSessionAffinityKey, session: &'a gateway_core::provider_ports::ProviderSessionAlias, - _: Duration, + ttl: Duration, ) -> BoxFuture<'a, Result> { Box::pin(async move { let mut aliases = self.aliases.lock().unwrap(); @@ -887,7 +892,11 @@ impl ProviderSessionAffinityPort for MemorySessionAffinity { alias.expose_to_store().to_owned(), )) .or_insert_with(|| session.clone()); - Ok(current == session) + let applied = current == session; + if applied { + self.alias_ttls.lock().unwrap().push(ttl); + } + Ok(applied) }) } } @@ -1085,6 +1094,7 @@ pub(crate) fn account_policy() -> gateway_core::account::AccountSelectionPolicy NonZeroU32::new(2).expect("nonzero concurrency"), Duration::from_millis(10), ) + .with_openai_account_affinity(gateway_core::account::AccountAffinity::Strict) } /// 内存 `ProviderCooldownPort`：实现 `read`/`put_if_later` 与容量失败计数 diff --git a/backend/migrations/.frozen-sha256 b/backend/migrations/.frozen-sha256 index 1831d5822..6ba463e2b 100644 --- a/backend/migrations/.frozen-sha256 +++ b/backend/migrations/.frozen-sha256 @@ -21,3 +21,5 @@ e8c1339166ac02fb2d4a077424ee5f64f6dc20eb4d3f8ec4fcb09c0f2a0e61f6 0020_plugin_tr 22faf2646168275150b1786c0f463f9ff2ca52463a272479e4fcc6f46c8cfa37 0021_runtime_scheduling.sql 77a140ddfb9dfae9b7d7349e5bfc9c8a7e7927bae6e77c1a627b34fcbb714bdb 0022_refresh_margin_and_group_fast_mode.sql 68fe28522870903f9a9c9772e13edda29b9b0f1798d350b1025c273b05d25370 0023_error_details.sql +11cb65ee4858c532360d37fa161ffa3e7dad44e81791c24b949154485d255804 0024_account_affinity.sql +e71acda93c9a315bff7bb95a599cabac029c627d5d1b6f4488547f568ad22415 0025_session_affinity_ttl.sql diff --git a/backend/migrations/0024_account_affinity.sql b/backend/migrations/0024_account_affinity.sql new file mode 100644 index 000000000..63dfb22ab --- /dev/null +++ b/backend/migrations/0024_account_affinity.sql @@ -0,0 +1,5 @@ +alter table runtime_settings + add column openai_account_affinity text not null default 'relaxed' + check (openai_account_affinity in ('relaxed', 'strict')), + add column max_account_rotations bigint not null default 3 + check (max_account_rotations between 0 and 31); diff --git a/backend/migrations/0025_session_affinity_ttl.sql b/backend/migrations/0025_session_affinity_ttl.sql new file mode 100644 index 000000000..96b2ef5ac --- /dev/null +++ b/backend/migrations/0025_session_affinity_ttl.sql @@ -0,0 +1,3 @@ +alter table runtime_settings + add column openai_session_affinity_ttl_hours bigint not null default 24 + check (openai_session_affinity_ttl_hours between 1 and 720); diff --git a/docs/api.md b/docs/api.md index 9a7bf0b61..da306f791 100644 --- a/docs/api.md +++ b/docs/api.md @@ -1233,11 +1233,11 @@ HTTP 请求头及新建 WS 的握手提示按当时的最终出站档位构造 `response.create`；空闲连接不占名额，内部重试不重复占用。 修改 Key 策略对既有 WebSocket 连接的下一次请求同样生效，已开始的请求保持原有快照 -运行设置可以分别启用 Key 与账号的有界排队。Key 并发满时按 Key 等待；OpenAI 根线程按配置等待当前账号， -或沿既有调度策略重选账号并迁移会话绑定。内置调度中的后代线程只等待当前会话账号，不自行换号； -根线程迁移后，后代线程跟随新账号。没有绑定的后代线程等待根线程首次认领 +运行设置可以分别启用 Key 与账号的有界排队。Key 并发满时按 Key 等待；OpenAI 宽松亲和下各线程独立调度， +新子线程优先沿用根账号，必要时可分流；已有子绑定不随根账号迁移。严格亲和下后代线程只等待当前会话账号， +根线程迁移后跟随新账号，没有绑定时等待根线程首次认领 -后代线程始终排队，沿用 `concurrencyWaitTimeoutSeconds` 和请求截止时间； +严格亲和下的后代线程始终排队，沿用 `concurrencyWaitTimeoutSeconds` 和请求截止时间； `maxWaitingPerAccount` 为 0 时使用每队列 1,000 人上限，否则沿用配置的上限。 可识别来源的 Search、Images、Live 创建请求采用同样规则，身份与轮次关联见[会话绑定](architecture.md#6-路由账号范围与-continuation)。 插件显式选号保留原有行为。普通请求关闭账号排队且没有可用容量时返回 `503` / `account_capacity_unavailable`， @@ -1290,6 +1290,9 @@ maxConcurrentPerAccount maxWaitingPerKey maxWaitingPerAccount openaiGuardianReservedConcurrency +openaiAccountAffinity +openaiSessionAffinityTtlHours +maxAccountRotations concurrencyWaitTimeoutSeconds responsesMaxDecompressedBodyBytes requestIntervalMs @@ -1333,7 +1336,7 @@ accountWarmupModel `maxWaitingPerKey` 与 `maxWaitingPerAccount` 是全局统一的普通排队容量，取值 0～1,000，默认 0（关闭）； 每个 Key、每个账号各自独立计数，没有单对象覆盖字段。执行并发为 5、最大排队数为 5 时， 该对象最多容纳 5 个执行请求与 5 个等待请求。Key 并发为 0（不限）时跳过 Key 排队。 -OpenAI 后代线程的会话账号等待不随普通账号排队关闭，见 [Client Key 等待规则](#7-client-key)。 +OpenAI 严格亲和下后代线程的会话账号等待不随普通账号排队关闭，见 [Client Key 等待规则](#7-client-key)。 `concurrencyWaitTimeoutSeconds` 取值 1～120，默认 30，从首次入队开始计时，密钥与账号两层共享该等待时限； 切换账号或内部重试不重新计时；若插件设置了请求总时限，等待也计入该时限。排队超时不用于中断已开始的上游生成。 设置更新请求须包含这三个字段，新请求使用更新后的快照 @@ -1345,6 +1348,21 @@ Guardian 以 `subagent_kind` 或 `client_metadata.x-openai-subagent` 值 `guardi 仍受总等待容量与等待时限约束。不限并发的账号和关闭排队时的其余行为不变。 设置更新请求须包含该字段 +`openaiAccountAffinity` 控制 OpenAI 账号亲和，默认 `relaxed`（宽松），也可设置为 `strict`（严格）。 +宽松模式允许子代理按模型权限、可用性与容量独立分流，各线程保留自己的账号绑定；严格模式让同一会话共用账号， +子代理等待当前账号。需要选号时仍使用 `rotationStrategy` 指定的调度策略。 +插件显式选号及原生续写的状态归属约束仍然生效，缺少可识别线程身份的子请求仍按跟随方式处理 + +`openaiSessionAffinityTtlHours` 是账号绑定及其会话关联的滑动保留时长，单位小时，默认 24，取值 1～720。 +各类请求统一在发送前成功准入时按请求冻结值续期。 +响应完成不回写或续期。调整时长不扫描已有 Redis 键，已有记录保留原到期时间，下一次成功准入时使用新值； +真正过期或丢失后按缺失绑定处理，存储读取错误不会被当作过期 + +`maxAccountRotations` 是单请求最大换号次数，默认 3，取值 0～31；0 表示不换号。 +首次选号、同账号重试和选号时过滤不可用候选不计入换号次数，总路由尝试仍最多 32 次。 +提高该值允许请求尝试更多账号，但不放宽安全重放或交付后的重试限制。 +设置更新请求须包含账号亲和、亲和时长与最大换号次数，保存后对新请求生效，执行中请求及其重试沿用冻结值 + `responsesMaxDecompressedBodyBytes` 是压缩 Responses HTTP 请求的解压输出上限，单位字节，默认 67108864（64 MiB）。必须为正整数，且可表示为进程平台的 `isize`；管理端以整数 MiB 编辑。 保存并发布成功后，新请求使用新值；已鉴权请求沿用原快照，无需重启。调高上限会增加大请求的内存占用， diff --git a/docs/architecture.md b/docs/architecture.md index 821fdf384..634aee431 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -373,9 +373,9 @@ Client Key 鉴权完成后，API adapter 从有界请求头识别 Codex Desktop/ 重试与提交边界。具体错误合同见 [数据面接口](api.md#3-openai-数据面与模型目录) - Provider 可将明确容量拒绝标记为有界同账号退避，Core 在既有安全重放边界内执行，按账号维护请求内 预算，耗尽后复用普通换号路径。该退避消耗总路由预算，与 WS 传输恢复、OAuth 刷新及账号额度冷却分开 -- 换号有独立的请求内预算：选中账号与上一 attempt 不同的路由 attempt 至多 3 次，跨 Provider 候选 - 推进与 continuation 排除重放同样受限；预算耗尽后不再换号，以最后一个原始上游错误终态。同账号重试 - 与总路由预算不受影响 +- 换号预算从运行设置冻结到请求计划，选中账号与上一 attempt 不同时计数；首次选号、同账号重试和 + 选号阶段过滤候选不消耗预算。跨 Provider 候选推进与 continuation 排除重放同样受限，预算耗尽后 + 以最后一个原始上游错误终态；总路由尝试仍最多 32 次 - Provider 显式标记的终止拒绝禁止自动重试与传输回退，优先于未发送状态和安全重放证明 - 跨 Provider 只在账号范围和能力都允许，且请求尚未到达上游或已被证明可安全重放时发生 - 可恢复观测写入失败不能替换已经确定的客户端协议结果 @@ -610,16 +610,23 @@ Continuation 仍受原请求的 Client Key、账号范围、Provider 和发送/ - xAI 使用客户端提交的完整历史作为重放输入 - scope 外账号、跨 Key 复用或不明确发送结果均 fail closed -OpenAI 按 Client Key 隔离逻辑会话绑定，根线程、后代线程及可关联的 Search、Images、Live 创建请求读取同一个当前账号。 -Provider 解释官方逻辑 `session_id` 与 `thread_id`，不将缓存路由键或请求内容推断为账号身份。 -内置调度中，根线程保留既有重试、换号与权重策略；后代线程没有绑定时等待根线程认领，账号不可用时只排队， -根线程迁移后重新读取绑定。插件显式选号沿用原有裁决，native continuation 仍受独立的状态 owner 约束 +OpenAI Provider 解释官方逻辑 `session_id` 与 `thread_id`，按 Client Key 隔离账号绑定， +不将缓存路由键或请求内容推断为账号身份。账号亲和模式随请求策略冻结： + +- 宽松：根线程与各子线程独立绑定，新子线程优先使用根账号，模型权限、可用性或容量不允许时可独立选号； + 根绑定只提供初始偏好，子线程认领、迁移和续写校验只操作自己的绑定，根迁移也不改写已有子绑定 +- 严格：根线程及后代共享当前账号，后代没有绑定时等待根线程认领，账号不可用时只排队，根迁移后重读共享绑定 + +无法识别线程的后代请求保留跟随约束。可关联的 Search、Images、Live 创建请求使用所属线程的绑定。 +需要选号时统一使用配置的调度策略；插件显式选号沿用原有裁决，native continuation 仍受独立的状态 owner 与安全重放边界约束 Store 只保存带版本的绑定事实，Provider 在取得租约后、发送前原子认领或迁移；快照冲突释放租约并重新选择。 响应完成不写回账号，迁移前已准入的请求可以完成，不能把绑定改回旧账号。绑定存储不可用时不绕过绑定发送 官方独立 Images 仅携带 `turn_id`，通过已观测 Responses 轮次关联到会话，关联同时保留根线程或后代线程的调度权限。 -轮次关联只保存不透明会话键，不复制账号；绑定和轮次关联保留 24 小时，均按 Client Key 隔离。 +轮次关联保存不透明线程键及可选的根会话键，不复制账号；显式根身份只用于校验轮次归属。 +绑定和轮次关联按同一请求策略中的可配置时长保留，均按 Client Key 隔离。 +成功准入按冻结时长续期，已关联的 Images 同时续期轮次关联；更新配置不扫描旧键，响应完成不续期。 未知轮次不推测会话归属，已有轮次不能改指其他会话或更改调度权限 ### 并发等待 @@ -632,7 +639,7 @@ Core 的 `concurrency` 拥有中立的有界等待位置、优先级与同级 FI 新请求不能越过同级或更高优先级的等待者；队首短周期重读容量，取消/完成等待通过 Drop 回收位置并唤醒后继。 Provider 等待重试仍在同一次未发送准备阶段重新读取账号资格、容量、目录和冷却，不增加 attempt； Key 的 RPM 在成功准入时才计数，金额限制在入队前及成功准入后检查。 -普通账号容量等待只包含并发与本地请求间隔；后代线程还可等待根线程恢复或迁移绑定，但等待不使权限、额度或冷却检查失效。 +普通账号容量等待只包含并发与本地请求间隔；严格亲和下的后代线程还可等待根线程恢复或迁移绑定，但等待不使权限、额度或冷却检查失效。 密钥准入、账号选择与后续重试共享请求级等待预算，从首次入队开始计时；显式设置请求总时限时也受其约束。 切换等待层、账号或 Provider 不重置等待时限；没有发生排队时不启动该计时，也不据此中断已开始的上游生成。 队列满/超时属于本地容量拒绝，不作为上游限流反馈或 Provider 故障熔断证据 diff --git a/frontend/src/api/modules/settings/index.ts b/frontend/src/api/modules/settings/index.ts index c756c2e91..fea626987 100644 --- a/frontend/src/api/modules/settings/index.ts +++ b/frontend/src/api/modules/settings/index.ts @@ -13,6 +13,8 @@ export interface RequestLocation { timezone: string } +export type AccountAffinity = 'relaxed' | 'strict' + export type RotationStrategy = 'smart' | 'quota_reset_priority' | 'round_robin' | 'sticky' export interface SmartSchedulingConfig { @@ -40,6 +42,9 @@ export interface RuntimeSettings { refreshConcurrency: number maxConcurrentPerAccount: number openaiGuardianReservedConcurrency: number + openaiAccountAffinity: AccountAffinity + maxAccountRotations: number + openaiSessionAffinityTtlHours: number requestIntervalMs: number maxWaitingPerKey: number maxWaitingPerAccount: number diff --git a/frontend/src/views/settings/components/AccountAffinityCard.vue b/frontend/src/views/settings/components/AccountAffinityCard.vue new file mode 100644 index 000000000..faa9819a8 --- /dev/null +++ b/frontend/src/views/settings/components/AccountAffinityCard.vue @@ -0,0 +1,45 @@ + + + diff --git a/frontend/src/views/settings/composables/useSettingsForm.ts b/frontend/src/views/settings/composables/useSettingsForm.ts index 5772e034e..7ab66c9e6 100644 --- a/frontend/src/views/settings/composables/useSettingsForm.ts +++ b/frontend/src/views/settings/composables/useSettingsForm.ts @@ -1,5 +1,5 @@ import type { rotationOptions } from '../constants' -import type { SmartSchedulingConfig } from '@/api' +import type { AccountAffinity, SmartSchedulingConfig } from '@/api' import type { ProviderRequestProfiles, ProviderRequestProfileUpdates } from '@/api/modules/settings/profiles' import { toast } from '@codex-proxy/ui' import { cloneDeep, isEqual } from 'es-toolkit' @@ -32,6 +32,9 @@ export function useSettingsForm() { refreshConcurrency: null as number | null, maxConcurrentPerAccount: null as number | null, openaiGuardianReservedConcurrency: null as number | null, + openaiAccountAffinity: '' as AccountAffinity | '', + maxAccountRotations: null as number | null, + openaiSessionAffinityTtlHours: null as number | null, requestIntervalMs: null as number | null, maxWaitingPerKey: null as number | null, maxWaitingPerAccount: null as number | null, @@ -73,7 +76,7 @@ export function useSettingsForm() { mappings.value = initial.mappings } - function numericModel(key: 'refreshMarginSeconds' | 'refreshConcurrency' | 'maxConcurrentPerAccount' | 'openaiGuardianReservedConcurrency' | 'requestIntervalMs' | 'maxWaitingPerKey' | 'maxWaitingPerAccount' | 'concurrencyWaitTimeoutSeconds' | 'responsesMaxDecompressedBodyMiB' | 'accountAutoFreezeThreshold' | 'accountAutoFreezeWindowSeconds' | 'accountAutoFreezeDurationSeconds') { + function numericModel(key: 'refreshMarginSeconds' | 'refreshConcurrency' | 'maxConcurrentPerAccount' | 'openaiGuardianReservedConcurrency' | 'maxAccountRotations' | 'openaiSessionAffinityTtlHours' | 'requestIntervalMs' | 'maxWaitingPerKey' | 'maxWaitingPerAccount' | 'concurrencyWaitTimeoutSeconds' | 'responsesMaxDecompressedBodyMiB' | 'accountAutoFreezeThreshold' | 'accountAutoFreezeWindowSeconds' | 'accountAutoFreezeDurationSeconds') { return computed({ get: () => (form[key] === null ? '' : String(form[key])), set: (value: string) => { @@ -91,6 +94,8 @@ export function useSettingsForm() { const refreshConcurrencyValue = numericModel('refreshConcurrency') const maxConcurrentPerAccountValue = numericModel('maxConcurrentPerAccount') const openaiGuardianReservedConcurrencyValue = numericModel('openaiGuardianReservedConcurrency') + const maxAccountRotationsValue = numericModel('maxAccountRotations') + const openaiSessionAffinityTtlHoursValue = numericModel('openaiSessionAffinityTtlHours') const requestIntervalMsValue = numericModel('requestIntervalMs') const maxWaitingPerKeyValue = numericModel('maxWaitingPerKey') const maxWaitingPerAccountValue = numericModel('maxWaitingPerAccount') @@ -116,6 +121,9 @@ export function useSettingsForm() { form.refreshConcurrency = data.refreshConcurrency form.maxConcurrentPerAccount = data.maxConcurrentPerAccount form.openaiGuardianReservedConcurrency = data.openaiGuardianReservedConcurrency + form.openaiAccountAffinity = data.openaiAccountAffinity + form.maxAccountRotations = data.maxAccountRotations + form.openaiSessionAffinityTtlHours = data.openaiSessionAffinityTtlHours form.requestIntervalMs = data.requestIntervalMs form.maxWaitingPerKey = data.maxWaitingPerKey form.maxWaitingPerAccount = data.maxWaitingPerAccount @@ -201,11 +209,19 @@ export function useSettingsForm() { const savedSettings = saved.value if (saving.value || loading.value || !savedSettings) return - const { refreshMarginSeconds, refreshConcurrency, maxConcurrentPerAccount, openaiGuardianReservedConcurrency, requestIntervalMs, rotationStrategy, maxWaitingPerKey, maxWaitingPerAccount, concurrencyWaitTimeoutSeconds, responsesMaxDecompressedBodyMiB, accountAutoFreezeThreshold, accountAutoFreezeWindowSeconds, accountAutoFreezeDurationSeconds } = form - if (refreshMarginSeconds === null || refreshConcurrency === null || maxConcurrentPerAccount === null || openaiGuardianReservedConcurrency === null || requestIntervalMs === null || !rotationStrategy || maxWaitingPerKey === null || maxWaitingPerAccount === null || concurrencyWaitTimeoutSeconds === null) { + const { refreshMarginSeconds, refreshConcurrency, maxConcurrentPerAccount, openaiGuardianReservedConcurrency, openaiAccountAffinity, maxAccountRotations, openaiSessionAffinityTtlHours, requestIntervalMs, rotationStrategy, maxWaitingPerKey, maxWaitingPerAccount, concurrencyWaitTimeoutSeconds, responsesMaxDecompressedBodyMiB, accountAutoFreezeThreshold, accountAutoFreezeWindowSeconds, accountAutoFreezeDurationSeconds } = form + if (!openaiAccountAffinity || maxAccountRotations === null || openaiSessionAffinityTtlHours === null || refreshMarginSeconds === null || refreshConcurrency === null || maxConcurrentPerAccount === null || openaiGuardianReservedConcurrency === null || requestIntervalMs === null || !rotationStrategy || maxWaitingPerKey === null || maxWaitingPerAccount === null || concurrencyWaitTimeoutSeconds === null) { toast.warning('请完整填写并发、队列、凭据刷新参数和调度策略') return } + if (!Number.isInteger(openaiSessionAffinityTtlHours) || openaiSessionAffinityTtlHours < 1 || openaiSessionAffinityTtlHours > 720) { + toast.warning('亲和时长应为 1～720 小时的整数') + return + } + if (!Number.isInteger(maxAccountRotations) || maxAccountRotations < 0 || maxAccountRotations > 31) { + toast.warning('最大换号次数应为 0～31 的整数，0 表示不换号') + return + } if (!Number.isInteger(maxConcurrentPerAccount) || maxConcurrentPerAccount < 0 || maxConcurrentPerAccount > 4294967295) { toast.warning('默认账号并发上限应为 0～4294967295 的整数，0 表示不限制') return @@ -281,6 +297,9 @@ export function useSettingsForm() { refreshConcurrency, maxConcurrentPerAccount, openaiGuardianReservedConcurrency, + openaiAccountAffinity, + maxAccountRotations, + openaiSessionAffinityTtlHours, requestIntervalMs, maxWaitingPerKey, maxWaitingPerAccount, @@ -330,6 +349,8 @@ export function useSettingsForm() { refreshConcurrencyValue, maxConcurrentPerAccountValue, openaiGuardianReservedConcurrencyValue, + maxAccountRotationsValue, + openaiSessionAffinityTtlHoursValue, requestIntervalMsValue, maxWaitingPerKeyValue, maxWaitingPerAccountValue, diff --git a/frontend/src/views/settings/index.vue b/frontend/src/views/settings/index.vue index b6b69c5a9..b8090ac56 100644 --- a/frontend/src/views/settings/index.vue +++ b/frontend/src/views/settings/index.vue @@ -5,6 +5,7 @@ import { Save, Undo2 } from '@lucide/vue' import { computed, reactive, shallowRef, watch } from 'vue' import { useRoute, useRouter } from 'vue-router' +import AccountAffinityCard from './components/AccountAffinityCard.vue' import AccountAutoFreezeCard from './components/AccountAutoFreezeCard.vue' import AccountWarmupCard from './components/AccountWarmupCard.vue' import SettingsBackupSection from './components/backup/SettingsBackupSection.vue' @@ -61,6 +62,8 @@ const { refreshConcurrencyValue, maxConcurrentPerAccountValue, openaiGuardianReservedConcurrencyValue, + maxAccountRotationsValue, + openaiSessionAffinityTtlHoursValue, requestIntervalMsValue, maxWaitingPerKeyValue, maxWaitingPerAccountValue, @@ -151,6 +154,12 @@ watch(section, (value) => { :disabled="disabled" :options="rotationOptions" /> +