feat(settings): configure account affinity and rotation limits

Add relaxed and strict affinity modes, configurable binding retention, and per-request account rotation limits. Freeze policies for each request and preserve continuation ownership and plugin selection behavior.

Expose the controls in a dedicated affinity settings card and cover scheduling, persistence, renewal, and retry boundaries.

Co-authored-by: Codex <noreply@openai.com>
This commit is contained in:
Zyy
2026-10-07 01:30:10 +08:00
co-authored by Codex
parent d922c3af60
commit cb7daffb66
44 changed files with 1613 additions and 119 deletions
@@ -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<RuntimeSettings> 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,
@@ -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)
@@ -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,
@@ -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,
@@ -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,
@@ -273,6 +273,7 @@ fn sdk_settings_contract_matches_host_declarations() {
pub type ModelMappings = BTreeMap<String, String>;
pub type Revision = NonZeroU64;
pub type RotationStrategy = String;
pub type AccountAffinity = String;
pub type PricingOverrides = BTreeMap<String, BTreeMap<String, ModelPriceOverride>>;
#(#types)*
#(#operations)*
@@ -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,
@@ -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(),
@@ -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,
@@ -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::<UpdateRuntimeSettingsRequest>(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}");
}
}
@@ -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<Self> {
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 {
@@ -464,7 +464,7 @@ pub struct ResponseExecutionSession<S: ?Sized> {
/// 已发生的换号次数:选中账号与上一 attempt 不同的路由 attempt 计一次。
/// 首个 attempt 不计;同账号钉选重试(瞬态退避、传输恢复、凭据恢复重放、
/// continuation 精确重连)不消耗。预算耗尽后所有必然换号的重试门关闭,
/// 换号深度由 [`crate::routing::MAX_ACCOUNT_ROTATION_ATTEMPTS`] 封顶。
/// 换号深度由请求冻结的调度策略封顶
account_rotations: u32,
candidate_index: usize,
excluded_accounts: BTreeSet<crate::account::ProviderAccountId>,
@@ -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;
}
@@ -297,6 +297,7 @@ impl ProviderSessionBinding {
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ProviderSessionAlias {
pub session_key: ProviderSessionAffinityKey,
pub root_session_key: Option<ProviderSessionAffinityKey>,
pub follow_only: bool,
}
@@ -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)]
@@ -22,7 +22,10 @@ pub(crate) struct CompiledSettings {
impl CompiledSettings {
pub(crate) fn new(settings: SettingsValues) -> Result<Self, InvalidSettings> {
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,
@@ -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(),
@@ -613,6 +613,42 @@ fn image_stream(image_output_tokens: Option<u64>) -> Vec<Result<GatewayEvent, Pr
]
}
const ROTATION_TEST_ACCOUNTS: [&str; 33] = [
"acct_rotation_0",
"acct_rotation_1",
"acct_rotation_2",
"acct_rotation_3",
"acct_rotation_4",
"acct_rotation_5",
"acct_rotation_6",
"acct_rotation_7",
"acct_rotation_8",
"acct_rotation_9",
"acct_rotation_10",
"acct_rotation_11",
"acct_rotation_12",
"acct_rotation_13",
"acct_rotation_14",
"acct_rotation_15",
"acct_rotation_16",
"acct_rotation_17",
"acct_rotation_18",
"acct_rotation_19",
"acct_rotation_20",
"acct_rotation_21",
"acct_rotation_22",
"acct_rotation_23",
"acct_rotation_24",
"acct_rotation_25",
"acct_rotation_26",
"acct_rotation_27",
"acct_rotation_28",
"acct_rotation_29",
"acct_rotation_30",
"acct_rotation_31",
"acct_rotation_32",
];
fn plan(operation: &Operation) -> 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::<Vec<_>>();
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())
);
}
@@ -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()
);
}
}
@@ -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<String, String>;
pub type Revision = NonZeroU64;
pub type RotationStrategy = String;
pub type AccountAffinity = String;
pub type PricingOverrides = BTreeMap<String, BTreeMap<String, ModelPriceOverride>>;
#[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<RuntimeSettings> 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,
@@ -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,
@@ -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<RuntimeSett
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(|| StoreError::InvalidData {
entity: "runtime settings",
message: "invalid account affinity".to_owned(),
source: None,
})?,
max_account_rotations: to_u32(row.max_account_rotations)?,
openai_session_affinity_ttl_hours: to_u32(row.openai_session_affinity_ttl_hours)?,
responses_max_decompressed_body_bytes: to_u64(row.responses_max_decompressed_body_bytes)?,
account_auto_freeze_enabled: row.account_auto_freeze_enabled,
account_auto_freeze_threshold: to_u32(row.account_auto_freeze_threshold)?,
@@ -33,6 +33,9 @@ pub struct SnapshotRuntimeSettings {
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,
@@ -170,6 +173,9 @@ impl SnapshotStorePort for PgRuntimeSnapshotRepository {
.with_openai_guardian_reserved_concurrency(
data.settings.openai_guardian_reserved_concurrency,
)
.with_openai_account_affinity(data.settings.openai_account_affinity)
.with_max_account_rotations(data.settings.max_account_rotations)
.with_openai_session_affinity_ttl_hours(data.settings.openai_session_affinity_ttl_hours)
.with_smart_scheduling(data.settings.smart_scheduling)
.with_request_profiles(data.settings.request_profiles)
.with_pricing(data.settings.pricing)
@@ -272,6 +278,9 @@ struct SnapshotSettingsRow {
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,
request_location_json: sqlx::types::Json<gateway_core::account::RequestLocation>,
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)?,
},
))
}
@@ -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<String>,
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")
@@ -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;
}
@@ -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<String> = 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);
}
@@ -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<ProviderSessionAffinityKey>,
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<CodexSessionAffinity> {
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<CodexSessionAffinity> {
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<CodexSessionAffinity> {
let mut body = serde_json::from_slice::<Map<String, Value>>(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<String, Value>,
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<String, Value>,
context: &Map<String, Value>,
headers: &[gateway_core::engine::middleware::MiddlewareHeader],
) -> Option<String> {
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<String, Value>,
@@ -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<CodexSessionAffinity> {
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,
@@ -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;
@@ -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::<Vec<_>>();
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<Option<CodexSessionAffinity>, 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<bool, CredentialSelectionError> {
tokio::time::timeout(
SESSION_AFFINITY_TIMEOUT,
@@ -1023,7 +1051,7 @@ impl CodexCredentialSelector {
key,
expected,
account,
CODEX_ROOT_SESSION_TTL,
session_ttl,
),
)
.await
@@ -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))?;
@@ -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)?;
}
@@ -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)?;
}
@@ -44,7 +44,8 @@ pub struct CodexResponsesRequest {
pub client_session_id: Option<String>,
/// 官方元数据中的逻辑会话身份,独立于缓存路由键
pub client_account_session_id: Option<String>,
/// 后代线程只跟随会话账号,不自行迁移绑定
pub client_account_thread_id: Option<String>,
/// 后代线程身份,用于严格模式下只跟随会话账号
pub client_account_follow_only: bool,
/// 客户端 thread ID,仅保留在受控本地上下文
pub client_thread_id: Option<String>,
@@ -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,
@@ -733,6 +733,8 @@ fn apply_protocol_context(request: &mut CodexResponsesRequest, context: &Map<Str
crate::credential::follows_session_with_headers(request.body(), context, &[]);
request.client_account_session_id =
gateway_protocol::openai::codex_account_session_id(request.body(), context);
request.client_account_thread_id =
gateway_protocol::openai::codex_account_thread_id(request.body(), context);
request.client_thread_id = gateway_protocol::openai::codex_thread_id(request.body(), context);
request.client_request_id =
context_string(context, "client_request_id").or_else(|| request.client_request_id.take());
@@ -3,20 +3,21 @@
use super::*;
#[derive(Debug)]
struct ChangeSessionHeader;
struct ChangeTurnMetadata(&'static [u8]);
impl MiddlewarePlan for ChangeSessionHeader {
impl MiddlewarePlan for ChangeTurnMetadata {
fn handle(
&self,
_context: MiddlewareContext,
request: MiddlewareRequest,
next: MiddlewareNext,
) -> BoxFuture<'static, Result<MiddlewareResponse, MiddlewareError>> {
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<FrozenAccountScope>) -> AttemptContext {
relaxed_context_with_ttl(request_id, scope, Duration::from_secs(24 * 3600))
}
fn relaxed_context_with_ttl(
request_id: &str,
scope: Arc<FrozenAccountScope>,
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());
}
@@ -739,9 +739,14 @@ pub(crate) struct MemorySessionAffinity {
Mutex<BTreeMap<(String, String), gateway_core::provider_ports::ProviderSessionBinding>>,
lookups: Mutex<Vec<String>>,
renewal_ttls: Mutex<Vec<Duration>>,
alias_ttls: Mutex<Vec<Duration>>,
}
impl MemorySessionAffinity {
pub(crate) fn alias_ttls(&self) -> Vec<Duration> {
self.alias_ttls.lock().unwrap().clone()
}
pub(crate) fn renewal_ttls(&self) -> Vec<Duration> {
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<bool, ProviderStoreError>> {
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` 与容量失败计数
+2
View File
@@ -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
@@ -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);
@@ -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);
+23 -5
View File
@@ -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 编辑。
保存并发布成功后,新请求使用新值;已鉴权请求沿用原快照,无需重启。调高上限会增加大请求的内存占用,
+16 -9
View File
@@ -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 故障熔断证据
@@ -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
@@ -0,0 +1,45 @@
<script setup lang="ts">
import type { AccountAffinity } from '@/api'
import { BaseCard, BaseForm, BaseFormItem, BaseInput, BasePopover, BaseSegmented } from '@codex-proxy/ui'
import { CircleAlert } from '@lucide/vue'
const props = defineProps<{ disabled: boolean }>()
const accountAffinity = defineModel<AccountAffinity | ''>({ required: true })
const affinityTtlHours = defineModel<string>('ttlHours', { required: true })
const maxAccountRotations = defineModel<string>('maxAccountRotations', { required: true })
const affinityOptions = [{ label: '宽松', value: 'relaxed' }, { label: '严格', value: 'strict' }]
</script>
<template>
<BaseCard title="亲和策略">
<BaseForm class="max-w-6xl sm:grid-cols-2 lg:grid-cols-3">
<BaseFormItem label="账号亲和" description="控制同一会话内的账号绑定方式">
<template #label-extra>
<BasePopover class="-my-1" trigger="hover-click" placement="top-start">
<template #trigger="{ open }">
<button type="button" aria-label="账号亲和说明" :aria-expanded="open" class="inline-flex size-6 shrink-0 cursor-pointer items-center justify-center rounded-cp-sm border-0 bg-transparent p-0 text-cp-text-tertiary outline-none transition-colors hover:text-cp-text focus-visible:ring-2 focus-visible:ring-cp-control-outline motion-reduce:transition-none">
<CircleAlert class="size-3.5" aria-hidden="true" />
</button>
</template>
<div class="max-w-72 space-y-2 px-3 py-2 text-cp-sm leading-relaxed text-cp-text-secondary">
<p>宽松:各线程独立绑定账号,子线程优先沿用主账号,可按模型、可用性和并发容量分流</p>
<p>严格:同一会话共用账号,子线程等待当前账号,并跟随主线程换号</p>
<p>重新选号仍遵循当前调度策略</p>
</div>
</BasePopover>
</template>
<BaseSegmented v-model="accountAffinity" label="账号亲和" :options="affinityOptions" :disabled="props.disabled" class="w-full" />
</BaseFormItem>
<BaseFormItem label="亲和时长" description="账号绑定的保留时间,成功调度后续期">
<BaseInput v-model="affinityTtlHours" aria-label="亲和时长(小时)" type="number" min="1" max="720" step="1" :disabled="props.disabled">
<template #suffix>
小时
</template>
</BaseInput>
</BaseFormItem>
<BaseFormItem label="最大换号次数" description="单请求最多换号次数,0 表示不换号">
<BaseInput v-model="maxAccountRotations" aria-label="最大换号次数" type="number" min="0" max="31" step="1" :disabled="props.disabled" />
</BaseFormItem>
</BaseForm>
</BaseCard>
</template>
@@ -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,
+9
View File
@@ -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"
/>
<AccountAffinityCard
v-model="form.openaiAccountAffinity"
v-model:ttl-hours="openaiSessionAffinityTtlHoursValue"
v-model:max-account-rotations="maxAccountRotationsValue"
:disabled="disabled"
/>
<ConcurrencyPolicyCard
v-model:max-concurrent-per-account="maxConcurrentPerAccountValue"
v-model:request-interval-ms="requestIntervalMsValue"