diff --git a/backend/crates/gateway-admin/src/model/settings.rs b/backend/crates/gateway-admin/src/model/settings.rs index caa73bfbd..316aedb3d 100644 --- a/backend/crates/gateway-admin/src/model/settings.rs +++ b/backend/crates/gateway-admin/src/model/settings.rs @@ -38,6 +38,7 @@ pub struct RuntimeSettings { pub max_waiting_per_key: u32, pub max_waiting_per_account: u32, pub concurrency_wait_timeout_seconds: u32, + pub openai_guardian_reserved_concurrency: u32, pub responses_max_decompressed_body_bytes: u64, pub smart_scheduling: gateway_core::account::SmartSchedulingConfig, pub rotation_strategy: RotationStrategy, @@ -77,6 +78,7 @@ pub struct ReplaceRuntimeSettings { pub max_waiting_per_key: u32, pub max_waiting_per_account: u32, pub concurrency_wait_timeout_seconds: u32, + pub openai_guardian_reserved_concurrency: u32, pub responses_max_decompressed_body_bytes: u64, pub smart_scheduling: gateway_core::account::SmartSchedulingConfig, pub rotation_strategy: RotationStrategy, @@ -182,6 +184,7 @@ impl From for ReplaceRuntimeSettings { max_waiting_per_key: settings.max_waiting_per_key, 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, responses_max_decompressed_body_bytes: settings.responses_max_decompressed_body_bytes, smart_scheduling: settings.smart_scheduling, rotation_strategy: settings.rotation_strategy, diff --git a/backend/crates/gateway-admin/tests/use_case/accounts.rs b/backend/crates/gateway-admin/tests/use_case/accounts.rs index c716fc2c0..73780025a 100644 --- a/backend/crates/gateway-admin/tests/use_case/accounts.rs +++ b/backend/crates/gateway-admin/tests/use_case/accounts.rs @@ -1114,6 +1114,7 @@ impl SettingsStore for StaticSettingsStore { max_waiting_per_key: 0, max_waiting_per_account: 0, concurrency_wait_timeout_seconds: 30, + openai_guardian_reserved_concurrency: 0, responses_max_decompressed_body_bytes: 64 * 1024 * 1024, smart_scheduling: gateway_core::account::SmartSchedulingConfig::default(), rotation_strategy: RotationStrategy::Smart, diff --git a/backend/crates/gateway-admin/tests/use_case/freeze_recovery.rs b/backend/crates/gateway-admin/tests/use_case/freeze_recovery.rs index 8058daae2..b01f0a775 100644 --- a/backend/crates/gateway-admin/tests/use_case/freeze_recovery.rs +++ b/backend/crates/gateway-admin/tests/use_case/freeze_recovery.rs @@ -37,6 +37,7 @@ fn runtime_settings(enabled: bool, probe_enabled: bool, adaptive: bool) -> Runti max_waiting_per_key: 0, max_waiting_per_account: 0, concurrency_wait_timeout_seconds: 30, + openai_guardian_reserved_concurrency: 0, responses_max_decompressed_body_bytes: 64 * 1024 * 1024, smart_scheduling: gateway_core::account::SmartSchedulingConfig::default(), rotation_strategy: gateway_admin::model::settings::RotationStrategy::Smart, diff --git a/backend/crates/gateway-admin/tests/use_case/observability.rs b/backend/crates/gateway-admin/tests/use_case/observability.rs index 960f60585..bee566672 100644 --- a/backend/crates/gateway-admin/tests/use_case/observability.rs +++ b/backend/crates/gateway-admin/tests/use_case/observability.rs @@ -936,6 +936,7 @@ impl SettingsStore for FixtureSettingsStore { max_waiting_per_key: 0, max_waiting_per_account: 0, concurrency_wait_timeout_seconds: 30, + openai_guardian_reserved_concurrency: 0, responses_max_decompressed_body_bytes: 64 * 1024 * 1024, smart_scheduling: gateway_core::account::SmartSchedulingConfig::default(), rotation_strategy: RotationStrategy::Smart, diff --git a/backend/crates/gateway-admin/tests/use_case/settings.rs b/backend/crates/gateway-admin/tests/use_case/settings.rs index ddd8ba2db..f0149681f 100644 --- a/backend/crates/gateway-admin/tests/use_case/settings.rs +++ b/backend/crates/gateway-admin/tests/use_case/settings.rs @@ -285,6 +285,7 @@ async fn settings_should_reject_zero_refresh_margin_before_store_call() { max_waiting_per_key: 0, max_waiting_per_account: 0, concurrency_wait_timeout_seconds: 30, + openai_guardian_reserved_concurrency: 0, responses_max_decompressed_body_bytes: 64 * 1024 * 1024, smart_scheduling: gateway_core::account::SmartSchedulingConfig::default(), rotation_strategy: RotationStrategy::Smart, diff --git a/backend/crates/gateway-api/src/admin/settings.rs b/backend/crates/gateway-api/src/admin/settings.rs index 3f3e474ed..bd5a1ee95 100644 --- a/backend/crates/gateway-api/src/admin/settings.rs +++ b/backend/crates/gateway-api/src/admin/settings.rs @@ -54,6 +54,7 @@ pub struct RuntimeSettingsView { pub max_waiting_per_key: u32, pub max_waiting_per_account: u32, pub concurrency_wait_timeout_seconds: u32, + pub openai_guardian_reserved_concurrency: u32, pub responses_max_decompressed_body_bytes: u64, pub smart_scheduling: gateway_core::account::SmartSchedulingConfig, pub rotation_strategy: String, @@ -99,6 +100,7 @@ pub struct UpdateRuntimeSettingsRequest { pub max_waiting_per_key: u32, pub max_waiting_per_account: u32, pub concurrency_wait_timeout_seconds: u32, + pub openai_guardian_reserved_concurrency: u32, pub responses_max_decompressed_body_bytes: u64, pub smart_scheduling: gateway_core::account::SmartSchedulingConfig, pub rotation_strategy: String, @@ -234,6 +236,7 @@ impl UpdateRuntimeSettingsRequest { max_waiting_per_key: self.max_waiting_per_key, 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, responses_max_decompressed_body_bytes: self.responses_max_decompressed_body_bytes, smart_scheduling: self.smart_scheduling, rotation_strategy: RotationStrategy::parse(&self.rotation_strategy) @@ -283,6 +286,7 @@ impl From<(RuntimeSettings, crate::time::TimePresenter)> for RuntimeSettingsView max_waiting_per_key: settings.max_waiting_per_key, 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, responses_max_decompressed_body_bytes: settings.responses_max_decompressed_body_bytes, smart_scheduling: settings.smart_scheduling, smart_scheduling_defaults: gateway_core::account::SmartSchedulingConfig::default(), diff --git a/backend/crates/gateway-api/tests/admin/mod.rs b/backend/crates/gateway-api/tests/admin/mod.rs index 221e4ea52..7b6e6a575 100644 --- a/backend/crates/gateway-api/tests/admin/mod.rs +++ b/backend/crates/gateway-api/tests/admin/mod.rs @@ -570,6 +570,7 @@ impl SettingsStore for MemorySettingsStore { max_waiting_per_key: command.max_waiting_per_key, 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, responses_max_decompressed_body_bytes: command.responses_max_decompressed_body_bytes, smart_scheduling: command.smart_scheduling, rotation_strategy: command.rotation_strategy, @@ -1578,6 +1579,7 @@ fn test_runtime_settings() -> RuntimeSettings { max_waiting_per_key: 0, max_waiting_per_account: 0, concurrency_wait_timeout_seconds: 30, + openai_guardian_reserved_concurrency: 0, responses_max_decompressed_body_bytes: 64 * 1024 * 1024, smart_scheduling: gateway_core::account::SmartSchedulingConfig::default(), rotation_strategy: RotationStrategy::Smart, diff --git a/backend/crates/gateway-api/tests/admin/settings.rs b/backend/crates/gateway-api/tests/admin/settings.rs index 88dd8bdb3..5a4f3dae8 100644 --- a/backend/crates/gateway-api/tests/admin/settings.rs +++ b/backend/crates/gateway-api/tests/admin/settings.rs @@ -50,6 +50,7 @@ fn update_body() -> Value { "requestIntervalMs": 25, "maxWaitingPerKey": 0, "maxWaitingPerAccount": 0, + "openaiGuardianReservedConcurrency": 0, "concurrencyWaitTimeoutSeconds": 30, "responsesMaxDecompressedBodyBytes": 67108864, "rotationStrategy": "round_robin", @@ -220,6 +221,7 @@ fn settings_response_should_cover_the_full_runtime_settings_contract() { max_waiting_per_key: 0, max_waiting_per_account: 0, concurrency_wait_timeout_seconds: 30, + openai_guardian_reserved_concurrency: 0, responses_max_decompressed_body_bytes: 64 * 1024 * 1024, smart_scheduling: gateway_core::account::SmartSchedulingConfig::default(), rotation_strategy: RotationStrategy::RoundRobin, @@ -268,6 +270,7 @@ fn settings_response_should_cover_the_full_runtime_settings_contract() { "requestIntervalMs": 25, "maxWaitingPerKey": 0, "maxWaitingPerAccount": 0, + "openaiGuardianReservedConcurrency": 0, "concurrencyWaitTimeoutSeconds": 30, "responsesMaxDecompressedBodyBytes": 67108864, "rotationStrategy": "round_robin", @@ -337,6 +340,7 @@ fn settings_request_and_response_fields_should_stay_in_lockstep() { max_waiting_per_key: 0, max_waiting_per_account: 0, concurrency_wait_timeout_seconds: 30, + openai_guardian_reserved_concurrency: 0, responses_max_decompressed_body_bytes: 64 * 1024 * 1024, smart_scheduling: request.smart_scheduling, rotation_strategy: RotationStrategy::parse(&request.rotation_strategy) @@ -1276,3 +1280,50 @@ async fn settings_update_rejects_a_stale_version_without_replacing_the_saved_val 9999 ); } + +#[tokio::test] +async fn guardian_reservation_round_trips_and_rejects_invalid_values() { + let fixture = AdminTestFixture::new().await; + fixture.auth.insert_session("valid-session"); + let response = app(fixture.state()) + .oneshot(request(Method::GET, "/api/admin/settings", None)) + .await + .unwrap(); + let mut revision = response_json(response).await["data"]["configRevision"].clone(); + for reserved in [1_u32, u32::MAX, 0] { + let mut body = update_body(); + body["configRevision"] = revision; + body["openaiGuardianReservedConcurrency"] = json!(reserved); + let response = app(fixture.state()) + .oneshot(request( + Method::POST, + "/api/admin/settings/update", + Some(body), + )) + .await + .unwrap(); + assert_eq!(response.status(), StatusCode::OK); + assert_eq!( + response_json(response).await["data"]["openaiGuardianReservedConcurrency"], + reserved + ); + let response = app(fixture.state()) + .oneshot(request(Method::GET, "/api/admin/settings", None)) + .await + .unwrap(); + let data = response_json(response).await["data"].clone(); + revision = data["configRevision"].clone(); + assert_eq!(data["openaiGuardianReservedConcurrency"], reserved); + } + for invalid in [json!(-1), json!(1.5), json!(4294967296_u64)] { + let mut body = update_body(); + body["openaiGuardianReservedConcurrency"] = invalid; + assert!(serde_json::from_value::(body).is_err()); + } + let mut omitted = update_body(); + omitted + .as_object_mut() + .unwrap() + .remove("openaiGuardianReservedConcurrency"); + assert!(serde_json::from_value::(omitted).is_err()); +} diff --git a/backend/crates/gateway-core/src/account/model.rs b/backend/crates/gateway-core/src/account/model.rs index ad2d33b46..9c6f61272 100644 --- a/backend/crates/gateway-core/src/account/model.rs +++ b/backend/crates/gateway-core/src/account/model.rs @@ -90,6 +90,18 @@ impl AccountConcurrency { Self::Limited(limit) => Some(limit), } } + + /// 扣除当前请求不可占用的预留名额；不限并发不受影响，有限上限至少保留 1 个名额。 + #[must_use] + pub const fn excluding_reserved(self, reserved: u32) -> Self { + match self { + Self::Unlimited => Self::Unlimited, + Self::Limited(limit) => match NonZeroU32::new(limit.get().saturating_sub(reserved)) { + Some(remaining) => Self::Limited(remaining), + None => Self::Limited(NonZeroU32::MIN), + }, + } + } } impl From for AccountConcurrency { diff --git a/backend/crates/gateway-core/src/account/selection.rs b/backend/crates/gateway-core/src/account/selection.rs index 666fbfabc..78480c88d 100644 --- a/backend/crates/gateway-core/src/account/selection.rs +++ b/backend/crates/gateway-core/src/account/selection.rs @@ -54,6 +54,7 @@ pub struct AccountSelectionPolicy { max_concurrent_per_account: AccountConcurrency, request_interval: Duration, queue_policy: ConcurrencyQueuePolicy, + openai_guardian_reserved_concurrency: u32, } impl AccountSelectionPolicy { @@ -68,6 +69,7 @@ impl AccountSelectionPolicy { smart_scheduling: SmartSchedulingConfig::default(), max_concurrent_per_account: max_concurrent_per_account.into(), request_interval, + openai_guardian_reserved_concurrency: 0, queue_policy: ConcurrencyQueuePolicy { max_waiting: 0, timeout: Duration::ZERO, @@ -75,6 +77,18 @@ impl AccountSelectionPolicy { } } + /// 只传递冻结的运行设置，Guardian 分类与预留策略由 OpenAI Provider 解释。 + #[must_use] + pub const fn with_openai_guardian_reserved_concurrency(mut self, reserved: u32) -> Self { + self.openai_guardian_reserved_concurrency = reserved; + self + } + + #[must_use] + pub const fn openai_guardian_reserved_concurrency(self) -> u32 { + self.openai_guardian_reserved_concurrency + } + #[must_use] pub const fn with_queue(mut self, policy: ConcurrencyQueuePolicy) -> Self { self.queue_policy = policy; @@ -433,6 +447,18 @@ pub struct AccountSelectionContext { pub round_robin_cursor: u64, pub eligibility: AccountEligibilityPolicy, pub account_scope: Option>, + /// 本请求不可占用的每账号预留并发名额，由 Provider 按请求类别决定；0 表示不预留。 + pub reserved_concurrency: u32, +} + +impl AccountSelectionContext { + /// 本请求在该账号上可使用的并发上限；资格判断、租约与策略投影必须共用这一口径。 + #[must_use] + pub fn concurrency_limit(&self, account: &ProviderAccount) -> AccountConcurrency { + account + .effective_concurrency(self.policy.max_concurrent_per_account()) + .excluding_reserved(self.reserved_concurrency) + } } /// 选择账号时是否执行本地调度资格投影。 @@ -568,13 +594,8 @@ impl AccountSelector { ) }) .try_fold((0_u64, 0_u64), |(used, total), candidate| { - let capacity = u64::from( - candidate - .account - .effective_concurrency(context.policy.max_concurrent_per_account()) - .limit()? - .get(), - ); + let capacity = + u64::from(context.concurrency_limit(&candidate.account).limit()?.get()); Some(( used.saturating_add(u64::from(candidate.signals.in_flight)), total.saturating_add(capacity), @@ -789,9 +810,8 @@ impl AccountSelector { if context.excluded_accounts.contains(candidate.account.id()) { return Some(AccountSchedulingBlocker::Excluded); } - if candidate - .account - .effective_concurrency(context.policy.max_concurrent_per_account()) + if context + .concurrency_limit(&candidate.account) .limit() .is_some_and(|limit| candidate.signals.in_flight >= limit.get()) { diff --git a/backend/crates/gateway-core/src/concurrency.rs b/backend/crates/gateway-core/src/concurrency.rs index 005a5a4c4..48b680a63 100644 --- a/backend/crates/gateway-core/src/concurrency.rs +++ b/backend/crates/gateway-core/src/concurrency.rs @@ -43,8 +43,21 @@ pub enum QueueRejection { Timeout, } +/// 同一队列内的等待类别：高优先级排在已有高优先级等待者之后、全部普通等待者之前。 +#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)] +pub enum WaitPriority { + #[default] + Normal, + High, +} + +struct Waiter { + waker: Arc, + priority: WaitPriority, +} + struct QueueState { - queues: HashMap>>, + queues: HashMap>, total: usize, } @@ -75,6 +88,20 @@ impl ConcurrencyWaitQueue { .is_some_and(|queue| !queue.is_empty()) } + /// 是否存在排在该类别请求之前的等待者；普通请求让位于全部等待者，高优先级只让位于同类。 + fn has_waiters_ahead_of(&self, key: &K, priority: WaitPriority) -> bool { + self.state + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .queues + .get(key) + .is_some_and(|queue| { + queue + .iter() + .any(|waiter| priority == WaitPriority::Normal || waiter.priority == priority) + }) + } + pub fn enqueue( &self, keys: &[K], @@ -91,6 +118,18 @@ impl ConcurrencyWaitQueue { keys: &[K], max_waiting: u32, deadline: Instant, + priority: impl FnMut(&K, usize) -> f64, + ) -> Result, QueueRejection> { + self.enqueue_ranked(keys, max_waiting, deadline, WaitPriority::Normal, priority) + } + + /// 高优先级等待者不受单队列 `max_waiting` 限制，但仍要求排队已开启并受全局容量约束。 + fn enqueue_ranked( + &self, + keys: &[K], + max_waiting: u32, + deadline: Instant, + waiter_priority: WaitPriority, mut priority: impl FnMut(&K, usize) -> f64, ) -> Result, QueueRejection> { if Instant::now() >= deadline { @@ -107,17 +146,30 @@ impl ConcurrencyWaitQueue { .iter() .filter_map(|key| { let len = state.queues.get(key).map_or(0, VecDeque::len); - (len < max_waiting as usize).then(|| (key, priority(key, len))) + let admitted = max_waiting > 0 + && (waiter_priority == WaitPriority::High || len < max_waiting as usize); + admitted.then(|| (key, priority(key, len))) }) .min_by(|(_, left), (_, right)| right.total_cmp(left)) .map(|(key, _)| key.clone()) .ok_or(QueueRejection::Full)?; let waker = Arc::new(AtomicWaker::new()); - state - .queues - .entry(key.clone()) - .or_default() - .push_back(Arc::clone(&waker)); + let queue = state.queues.entry(key.clone()).or_default(); + let position = match waiter_priority { + WaitPriority::Normal => queue.len(), + WaitPriority::High => queue + .iter() + .position(|waiter| waiter.priority == WaitPriority::Normal) + .unwrap_or(queue.len()), + }; + // 插到队首时原队首失去位置，它的有界重查会自然让出；新队首随后自行轮询到位。 + queue.insert( + position, + Waiter { + waker: Arc::clone(&waker), + priority: waiter_priority, + }, + ); state.total += 1; Ok(WaitTicket { state: Arc::clone(&self.state), @@ -142,19 +194,20 @@ impl WaitTicket { &self.key } + fn is_head(&self) -> bool { + self.state + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .queues + .get(&self.key) + .and_then(VecDeque::front) + .is_some_and(|head| Arc::ptr_eq(&head.waker, &self.waker)) + } + pub async fn turn(&self) -> Result<(), QueueRejection> { let ready = poll_fn(|cx| { self.waker.register(cx.waker()); - let state = self - .state - .lock() - .unwrap_or_else(std::sync::PoisonError::into_inner); - if state - .queues - .get(&self.key) - .and_then(VecDeque::front) - .is_some_and(|head| Arc::ptr_eq(head, &self.waker)) - { + if self.is_head() { Poll::Ready(()) } else { Poll::Pending @@ -188,11 +241,14 @@ impl Drop for WaitTicket { let Some(queue) = state.queues.get_mut(&self.key) else { return; }; - let Some(index) = queue.iter().position(|item| Arc::ptr_eq(item, &self.waker)) else { + let Some(index) = queue + .iter() + .position(|item| Arc::ptr_eq(&item.waker, &self.waker)) + else { return; }; queue.remove(index); - let next = queue.front().cloned(); + let next = queue.front().map(|waiter| Arc::clone(&waiter.waker)); if queue.is_empty() { state.queues.remove(&self.key); } @@ -212,6 +268,7 @@ pub struct CapacityWait<'a, K: Eq + Hash> { policy: ConcurrencyQueuePolicy, request_deadline: SystemTime, budget: &'a ConcurrencyWaitBudget, + priority: WaitPriority, started_at: Option, ticket: Option>, } @@ -229,17 +286,26 @@ impl<'a, K: Clone + Eq + Hash> CapacityWait<'a, K> { policy, request_deadline, budget, + priority: WaitPriority::Normal, started_at: None, ticket: None, } } + #[must_use] + pub const fn with_priority(mut self, priority: WaitPriority) -> Self { + self.priority = priority; + self + } + #[must_use] pub fn can_try(&self, key: &K) -> bool { - self.ticket - .as_ref() - .is_some_and(|ticket| ticket.key() == key) - || !self.queue.has_waiters(key) + if let Some(ticket) = self.ticket.as_ref().filter(|ticket| ticket.key() == key) { + // 重读容量期间可能被高优先级请求插队，持有 ticket 不代表仍有选号资格。 + ticket.is_head() + } else { + !self.queue.has_waiters_ahead_of(key, self.priority) + } } #[must_use] @@ -271,10 +337,11 @@ impl<'a, K: Clone + Eq + Hash> CapacityWait<'a, K> { self.ticket = None; } if self.ticket.is_none() { - self.ticket = Some(self.queue.enqueue_with_priority( + self.ticket = Some(self.queue.enqueue_ranked( keys, self.policy.max_waiting, deadline, + self.priority, priority, )?); } diff --git a/backend/crates/gateway-core/src/diagnostics/selection.rs b/backend/crates/gateway-core/src/diagnostics/selection.rs index 7d805818a..9226c3096 100644 --- a/backend/crates/gateway-core/src/diagnostics/selection.rs +++ b/backend/crates/gateway-core/src/diagnostics/selection.rs @@ -63,6 +63,7 @@ impl TraceContext { "preferredAccountId": context.preferred_account.as_ref().map(|id| id.as_str()), "preferredResult": selection.as_ref().map(|s| format!("{:?}", s.preferred())), "roundRobinCursor": context.round_robin_cursor, + "reservedConcurrency": context.reserved_concurrency, "smartScoreTolerance": smart.then(|| context.policy.smart_scheduling().score_tolerance()), "candidateCount": candidates.len(), "omittedCandidates": candidates.len().saturating_sub(observations.len()), diff --git a/backend/crates/gateway-core/src/engine/policy.rs b/backend/crates/gateway-core/src/engine/policy.rs index 30911043c..f7385e781 100644 --- a/backend/crates/gateway-core/src/engine/policy.rs +++ b/backend/crates/gateway-core/src/engine/policy.rs @@ -427,17 +427,13 @@ impl RequestPolicyContext { if eligible.is_empty() { return Ok(None); } - let policy = context.policy; let projected = eligible .iter() .map(|candidate| AccountScheduleCandidate { account_id: candidate.account.id().clone(), weight: candidate.account.weight().get(), in_flight: candidate.signals.in_flight, - maximum_concurrency: candidate - .account - .effective_concurrency(policy.max_concurrent_per_account()) - .get(), + maximum_concurrency: context.concurrency_limit(&candidate.account).get(), last_started_at: candidate.signals.last_started_at, quota_reset_at: candidate.signals.quota_reset_at, quota_remaining_rank: candidate.signals.quota_remaining_rank, diff --git a/backend/crates/gateway-core/src/settings/compiled.rs b/backend/crates/gateway-core/src/settings/compiled.rs index 7ff6a0c79..32907b868 100644 --- a/backend/crates/gateway-core/src/settings/compiled.rs +++ b/backend/crates/gateway-core/src/settings/compiled.rs @@ -55,6 +55,7 @@ impl CompiledSettings { AccountConcurrency::new(settings.max_concurrent_per_account), Duration::from_millis(settings.request_interval_ms), ) + .with_openai_guardian_reserved_concurrency(settings.openai_guardian_reserved_concurrency) .with_smart_scheduling(settings.smart_scheduling) .with_queue(ConcurrencyQueuePolicy { max_waiting: settings.max_waiting_per_account, diff --git a/backend/crates/gateway-core/src/settings/values.rs b/backend/crates/gateway-core/src/settings/values.rs index c72951caf..0b28b4356 100644 --- a/backend/crates/gateway-core/src/settings/values.rs +++ b/backend/crates/gateway-core/src/settings/values.rs @@ -14,6 +14,7 @@ pub struct SettingsValues { pub(crate) max_waiting_per_key: u32, pub(crate) max_waiting_per_account: u32, pub(crate) concurrency_wait_timeout_seconds: u32, + pub(crate) openai_guardian_reserved_concurrency: u32, pub(crate) responses_max_decompressed_body_bytes: u64, pub(crate) request_interval_ms: u64, pub(crate) smart_scheduling: crate::account::SmartSchedulingConfig, @@ -24,6 +25,12 @@ pub struct SettingsValues { } impl SettingsValues { + #[must_use] + pub const fn with_openai_guardian_reserved_concurrency(mut self, reserved: u32) -> Self { + self.openai_guardian_reserved_concurrency = reserved; + self + } + #[must_use] pub fn request_profiles(&self) -> &BTreeMap { &self.request_profiles @@ -104,6 +111,7 @@ impl SettingsValues { max_waiting_per_key: 0, max_waiting_per_account: 0, concurrency_wait_timeout_seconds: 30, + openai_guardian_reserved_concurrency: 0, responses_max_decompressed_body_bytes: 64 * 1024 * 1024, request_interval_ms, smart_scheduling: crate::account::SmartSchedulingConfig::default(), diff --git a/backend/crates/gateway-core/tests/account/mod.rs b/backend/crates/gateway-core/tests/account/mod.rs index ec72805a2..6d2458101 100644 --- a/backend/crates/gateway-core/tests/account/mod.rs +++ b/backend/crates/gateway-core/tests/account/mod.rs @@ -13,12 +13,13 @@ use std::time::{Duration, SystemTime}; use serde_json::{Map, Value}; use gateway_core::account::{ - AccountAttemptFeedback, AccountCandidate, AccountConcurrencyLimit, AccountEligibilityPolicy, - AccountFeedbackStats, AccountQuotaSignals, AccountRuntimeSignals, AccountSchedulingBlocker, - AccountSelectionContext, AccountSelectionPolicy, AccountSelector, AccountStatus, AccountWeight, - CredentialCasUpdate, CredentialRevision, CredentialState, OpaqueProviderData, - PlaintextCredential, PreferredAccountSelection, ProviderAccount, ProviderAccountId, - ProviderAccountIdentity, ProviderAccountUpdate, QuotaEvidence, QuotaState, RotationStrategy, + AccountAttemptFeedback, AccountCandidate, AccountConcurrency, AccountConcurrencyLimit, + AccountEligibilityPolicy, AccountFeedbackStats, AccountQuotaSignals, AccountRuntimeSignals, + AccountSchedulingBlocker, AccountSelectionContext, AccountSelectionPolicy, AccountSelector, + AccountStatus, AccountWeight, CredentialCasUpdate, CredentialRevision, CredentialState, + OpaqueProviderData, PlaintextCredential, PreferredAccountSelection, ProviderAccount, + ProviderAccountId, ProviderAccountIdentity, ProviderAccountUpdate, QuotaEvidence, QuotaState, + RotationStrategy, }; use gateway_core::routing::{ ClientRoutingScope, FrozenAccountScope, ProviderKind, RuntimeAccount, RuntimeAccountDirectory, @@ -109,6 +110,7 @@ fn context(strategy: RotationStrategy) -> AccountSelectionContext { round_robin_cursor: 0, eligibility: AccountEligibilityPolicy::Enforce, account_scope: None, + reserved_concurrency: 0, } } @@ -1093,3 +1095,42 @@ fn due_probe_freeze_blocks_scheduling_and_remains_visible_as_rate_limited() { .expect("available account"); assert_eq!(selected.candidate().account.id().as_str(), "acct_ready"); } + +#[test] +fn reserved_concurrency_only_blocks_requests_that_cannot_use_the_reserved_slots() { + let candidates = [candidate_with_concurrency("acct_reserved", 4, 5)]; + let mut normal = context(RotationStrategy::Smart); + normal.reserved_concurrency = 1; + assert_eq!(normal.concurrency_limit(&candidates[0].account).get(), 4); + assert!(AccountSelector.select(&candidates, &normal).is_none()); + let capacity = AccountSelector + .capacity_snapshot(&candidates, &normal) + .unwrap(); + assert_eq!((capacity.used_slots(), capacity.total_slots()), (4, 4)); + // 预留造成的暂满仍是可等待的容量约束，不能被当成不可用账号。 + assert_eq!( + AccountSelector.wait_candidates(&candidates, &normal).len(), + 1 + ); + + let prioritized = context(RotationStrategy::Smart); + let capacity = AccountSelector + .capacity_snapshot(&candidates, &prioritized) + .unwrap(); + assert_eq!((capacity.used_slots(), capacity.total_slots()), (4, 5)); + let selected = AccountSelector + .select(&candidates, &prioritized) + .expect("reserved slot available"); + assert_eq!(selected.candidate().account.id().as_str(), "acct_reserved"); + let full = [candidate_with_concurrency("acct_reserved", 5, 5)]; + assert!(AccountSelector.select(&full, &prioritized).is_none()); + + // 预留不能让普通请求失去全部名额，也不改变不限并发。 + let mut oversized = context(RotationStrategy::Smart); + oversized.reserved_concurrency = 10; + assert_eq!(oversized.concurrency_limit(&candidates[0].account).get(), 1); + assert_eq!( + AccountConcurrency::Unlimited.excluding_reserved(2), + AccountConcurrency::Unlimited + ); +} diff --git a/backend/crates/gateway-core/tests/concurrency.rs b/backend/crates/gateway-core/tests/concurrency.rs index 6731f6abf..87b02d254 100644 --- a/backend/crates/gateway-core/tests/concurrency.rs +++ b/backend/crates/gateway-core/tests/concurrency.rs @@ -3,7 +3,7 @@ use std::time::{Duration, Instant, SystemTime}; use futures::{FutureExt, executor::block_on}; use gateway_core::concurrency::{ CapacityWait, ConcurrencyQueuePolicy, ConcurrencyWaitBudget, ConcurrencyWaitQueue, - QueueRejection, + QueueRejection, WaitPriority, }; #[test] @@ -231,3 +231,97 @@ fn changed_scores_keep_existing_position_until_account_becomes_ineligible() { assert!(!queue.has_waiters(&"b")); }); } + +#[test] +fn high_priority_waiters_go_ahead_of_normal_waiters_and_keep_fifo_among_themselves() { + block_on(async { + let queue = ConcurrencyWaitQueue::default(); + let deadline = Instant::now() + Duration::from_secs(2); + let policy = ConcurrencyQueuePolicy { + max_waiting: 2, + timeout: Duration::from_secs(2), + }; + let request_deadline = SystemTime::now() + Duration::from_secs(2); + let first_normal = queue.enqueue(&["a"], 2, deadline).unwrap(); + let second_normal = queue.enqueue(&["a"], 2, deadline).unwrap(); + + let normal_budget = ConcurrencyWaitBudget::default(); + let normal = CapacityWait::new(&queue, policy, request_deadline, &normal_budget); + let first_budget = ConcurrencyWaitBudget::default(); + let mut first_high = CapacityWait::new(&queue, policy, request_deadline, &first_budget) + .with_priority(WaitPriority::High); + // 只有普通等待者时，高优先级请求可以直接尝试租约，普通请求仍须排队。 + assert!(first_high.can_try(&"a")); + assert!(!normal.can_try(&"a")); + + // 普通等待者已占满单队列上限，高优先级仍可入队并直接成为队首。 + first_high.wait(&["a"]).await.unwrap(); + let second_budget = ConcurrencyWaitBudget::default(); + let mut second_high = CapacityWait::new(&queue, policy, request_deadline, &second_budget) + .with_priority(WaitPriority::High); + assert!(!second_high.can_try(&"a")); + assert!(second_high.wait(&["a"]).now_or_never().is_none()); + assert!(first_normal.turn().now_or_never().is_none()); + + drop(first_high); + second_high.wait(&["a"]).await.unwrap(); + assert!(first_normal.turn().now_or_never().is_none()); + drop(second_high); + first_normal.turn().await.unwrap(); + drop(first_normal); + second_normal.turn().await.unwrap(); + }); +} + +#[test] +fn displaced_normal_head_yields_its_lease_attempt_until_high_priority_waiters_leave() { + block_on(async { + let queue = ConcurrencyWaitQueue::default(); + let policy = ConcurrencyQueuePolicy { + max_waiting: 2, + timeout: Duration::from_secs(2), + }; + let deadline = SystemTime::now() + Duration::from_secs(2); + let normal_budget = ConcurrencyWaitBudget::default(); + let high_budget = ConcurrencyWaitBudget::default(); + let mut normal = CapacityWait::new(&queue, policy, deadline, &normal_budget); + normal.wait(&["account"]).await.unwrap(); + assert!(normal.can_try(&"account")); + + // 普通队首已经开始重读容量，尚未取得租约时被高优先级请求插队。 + let mut high = CapacityWait::new(&queue, policy, deadline, &high_budget) + .with_priority(WaitPriority::High); + high.wait(&["account"]).await.unwrap(); + assert!(!normal.can_try(&"account")); + assert!(high.can_try(&"account")); + assert!(normal.wait(&["account"]).now_or_never().is_none()); + + // 审批取消也必须恢复原队首的资格，不能丢失其等待位置。 + drop(high); + normal.wait(&["account"]).await.unwrap(); + assert!(normal.can_try(&"account")); + drop(normal); + assert!(!queue.has_waiters(&"account")); + }); +} + +#[test] +fn high_priority_waiters_still_require_queueing_to_be_enabled() { + block_on(async { + let queue = ConcurrencyWaitQueue::<&str>::default(); + let policy = ConcurrencyQueuePolicy { + max_waiting: 0, + timeout: Duration::from_secs(2), + }; + let budget = ConcurrencyWaitBudget::default(); + let mut waiting = CapacityWait::new( + &queue, + policy, + SystemTime::now() + Duration::from_secs(2), + &budget, + ) + .with_priority(WaitPriority::High); + assert_eq!(waiting.wait(&["a"]).await, Err(QueueRejection::Full)); + assert!(!queue.has_waiters(&"a")); + }); +} diff --git a/backend/crates/gateway-core/tests/engine/execution.rs b/backend/crates/gateway-core/tests/engine/execution.rs index f3162d2b5..88660a7f5 100644 --- a/backend/crates/gateway-core/tests/engine/execution.rs +++ b/backend/crates/gateway-core/tests/engine/execution.rs @@ -1921,6 +1921,7 @@ impl Provider for PolicySelectingProvider { round_robin_cursor: 0, eligibility: AccountEligibilityPolicy::Enforce, account_scope: attempt.account_scope().cloned(), + reserved_concurrency: 0, }; match attempt .select_account( diff --git a/backend/crates/gateway-core/tests/engine/policy.rs b/backend/crates/gateway-core/tests/engine/policy.rs index a113c5ef4..298206835 100644 --- a/backend/crates/gateway-core/tests/engine/policy.rs +++ b/backend/crates/gateway-core/tests/engine/policy.rs @@ -116,6 +116,7 @@ fn context() -> AccountSelectionContext { round_robin_cursor: 0, eligibility: AccountEligibilityPolicy::Enforce, account_scope: None, + reserved_concurrency: 0, } } diff --git a/backend/crates/gateway-core/tests/routing/snapshot.rs b/backend/crates/gateway-core/tests/routing/snapshot.rs index d21884925..97666dc23 100644 --- a/backend/crates/gateway-core/tests/routing/snapshot.rs +++ b/backend/crates/gateway-core/tests/routing/snapshot.rs @@ -364,7 +364,7 @@ fn compiler_accepts_unlimited_default_account_concurrency() { } #[test] -fn compiled_plans_keep_their_smart_config_after_a_new_snapshot_is_built() { +fn compiled_plans_keep_their_scheduling_config_after_a_new_snapshot_is_built() { use gateway_core::account::SmartSchedulingConfig; let configs = [ SmartSchedulingConfig::default(), @@ -376,7 +376,8 @@ fn compiled_plans_keep_their_smart_config_after_a_new_snapshot_is_built() { revision(index as u64 + 1), revision(index as u64 + 1), SettingsValues::new(3, 0, "smart", BTreeMap::new(), None, None) - .with_smart_scheduling(config), + .with_smart_scheduling(config) + .with_openai_guardian_reserved_concurrency(index as u32), vec![], vec![], vec![SnapshotProviderAccountFacts::new( @@ -401,7 +402,12 @@ fn compiled_plans_keep_their_smart_config_after_a_new_snapshot_is_built() { .unwrap(), ); } - for (plan, config) in plans.iter().zip(configs) { + for (index, (plan, config)) in plans.iter().zip(configs).enumerate() { + assert_eq!( + plan.account_selection_policy() + .openai_guardian_reserved_concurrency(), + index as u32 + ); assert_eq!(plan.account_selection_policy().smart_scheduling(), config); } } diff --git a/backend/crates/gateway-plugin/runtime/tests/generation/policy/mod.rs b/backend/crates/gateway-plugin/runtime/tests/generation/policy/mod.rs index 30fbb3a95..8873101ff 100644 --- a/backend/crates/gateway-plugin/runtime/tests/generation/policy/mod.rs +++ b/backend/crates/gateway-plugin/runtime/tests/generation/policy/mod.rs @@ -557,6 +557,7 @@ fn selection_context() -> AccountSelectionContext { round_robin_cursor: 0, eligibility: AccountEligibilityPolicy::Enforce, account_scope: None, + reserved_concurrency: 0, } } diff --git a/backend/crates/gateway-plugin/runtime/tests/rpc/session.rs b/backend/crates/gateway-plugin/runtime/tests/rpc/session.rs index 75ee1a397..7c8bd2a34 100644 --- a/backend/crates/gateway-plugin/runtime/tests/rpc/session.rs +++ b/backend/crates/gateway-plugin/runtime/tests/rpc/session.rs @@ -518,7 +518,7 @@ async fn concurrent_calls_keep_their_wire_ids_monotonic() { tokio::time::sleep(Duration::from_millis(20)).await; callbacks.release(); for call in calls { - assert!(call.await.unwrap().is_ok()); + call.await.unwrap().unwrap(); } session.shutdown(Duration::from_secs(1)).await; } diff --git a/backend/crates/gateway-plugin/sdk/src/call/services/settings.rs b/backend/crates/gateway-plugin/sdk/src/call/services/settings.rs index 6a5fecfb2..14d282235 100644 --- a/backend/crates/gateway-plugin/sdk/src/call/services/settings.rs +++ b/backend/crates/gateway-plugin/sdk/src/call/services/settings.rs @@ -27,6 +27,7 @@ pub struct RuntimeSettings { pub max_waiting_per_key: u32, pub max_waiting_per_account: u32, pub concurrency_wait_timeout_seconds: u32, + pub openai_guardian_reserved_concurrency: u32, pub responses_max_decompressed_body_bytes: u64, pub smart_scheduling: SmartSchedulingConfig, pub rotation_strategy: RotationStrategy, @@ -64,6 +65,7 @@ pub struct ReplaceRuntimeSettings { pub max_waiting_per_key: u32, pub max_waiting_per_account: u32, pub concurrency_wait_timeout_seconds: u32, + pub openai_guardian_reserved_concurrency: u32, pub responses_max_decompressed_body_bytes: u64, pub smart_scheduling: SmartSchedulingConfig, pub rotation_strategy: RotationStrategy, @@ -258,6 +260,7 @@ impl From for ReplaceRuntimeSettings { max_waiting_per_key: settings.max_waiting_per_key, 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, responses_max_decompressed_body_bytes: settings.responses_max_decompressed_body_bytes, smart_scheduling: settings.smart_scheduling, rotation_strategy: settings.rotation_strategy, diff --git a/backend/crates/gateway-store/src/admin_adapter.rs b/backend/crates/gateway-store/src/admin_adapter.rs index 595c23509..200202d6d 100644 --- a/backend/crates/gateway-store/src/admin_adapter.rs +++ b/backend/crates/gateway-store/src/admin_adapter.rs @@ -93,6 +93,7 @@ impl SettingsStore for AdminSettingsStoreAdapter { max_waiting_per_key: command.max_waiting_per_key, 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, responses_max_decompressed_body_bytes: command .responses_max_decompressed_body_bytes, smart_scheduling: command.smart_scheduling, @@ -131,6 +132,7 @@ impl SettingsStore for AdminSettingsStoreAdapter { "max_waiting_per_key".to_owned(), "max_waiting_per_account".to_owned(), "concurrency_wait_timeout_seconds".to_owned(), + "openai_guardian_reserved_concurrency".to_owned(), "responses_max_decompressed_body_bytes".to_owned(), "rotation_strategy".to_owned(), "smart_scheduling_json".to_owned(), @@ -238,6 +240,7 @@ pub(crate) fn admin_runtime_settings( max_waiting_per_key: settings.max_waiting_per_key, 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, responses_max_decompressed_body_bytes: settings.responses_max_decompressed_body_bytes, smart_scheduling: settings.smart_scheduling, rotation_strategy, diff --git a/backend/crates/gateway-store/src/postgres/runtime_settings.rs b/backend/crates/gateway-store/src/postgres/runtime_settings.rs index 0e15f9815..2e568dd00 100644 --- a/backend/crates/gateway-store/src/postgres/runtime_settings.rs +++ b/backend/crates/gateway-store/src/postgres/runtime_settings.rs @@ -30,6 +30,7 @@ pub struct RuntimeSettings { pub max_waiting_per_key: u32, pub max_waiting_per_account: u32, pub concurrency_wait_timeout_seconds: u32, + pub openai_guardian_reserved_concurrency: u32, pub responses_max_decompressed_body_bytes: u64, pub smart_scheduling: gateway_core::account::SmartSchedulingConfig, pub rotation_strategy: String, @@ -132,6 +133,7 @@ pub struct RuntimeSettingsUpdate { pub max_waiting_per_key: u32, pub max_waiting_per_account: u32, pub concurrency_wait_timeout_seconds: u32, + pub openai_guardian_reserved_concurrency: u32, pub responses_max_decompressed_body_bytes: u64, pub smart_scheduling: gateway_core::account::SmartSchedulingConfig, pub rotation_strategy: String, @@ -262,7 +264,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, + 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, 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, @@ -442,7 +444,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, + 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, 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, @@ -517,6 +519,7 @@ pub(crate) async fn update_runtime_settings_in_transaction( account_warmup_schedule_time = $28, account_warmup_model = $29, smart_scheduling_json = $30, + openai_guardian_reserved_concurrency = $31, updated_at = now() where id = 1 returning config_revision", @@ -563,6 +566,7 @@ pub(crate) async fn update_runtime_settings_in_transaction( .bind(&update.account_warmup_schedule_time) .bind(update.account_warmup_model.as_deref()) .bind(sqlx::types::Json(update.smart_scheduling)) + .bind(i64::from(update.openai_guardian_reserved_concurrency)) .fetch_optional(&mut **transaction) .await .map_err(|_| postgres_unavailable("update runtime settings in transaction"))? @@ -635,6 +639,7 @@ struct RuntimeSettingsRow { max_waiting_per_key: i64, max_waiting_per_account: i64, concurrency_wait_timeout_seconds: i64, + openai_guardian_reserved_concurrency: i64, responses_max_decompressed_body_bytes: i64, account_auto_freeze_enabled: bool, account_auto_freeze_threshold: i64, @@ -688,6 +693,7 @@ fn runtime_settings_from_row(row: RuntimeSettingsRow) -> StoreResult, request_location_enabled: bool, responses_max_decompressed_body_bytes: i64, @@ -275,7 +280,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, 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, 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 @@ -313,6 +318,7 @@ async fn load_settings( max_waiting_per_key: to_u32(row.max_waiting_per_key)?, 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)?, }, )) } diff --git a/backend/crates/gateway-store/tests/postgres/runtime_settings.rs b/backend/crates/gateway-store/tests/postgres/runtime_settings.rs index 8f5433233..31db79e8d 100644 --- a/backend/crates/gateway-store/tests/postgres/runtime_settings.rs +++ b/backend/crates/gateway-store/tests/postgres/runtime_settings.rs @@ -19,6 +19,7 @@ fn settings_with_margin(refresh_margin_seconds: u64) -> RuntimeSettingsUpdate { max_waiting_per_key: 0, max_waiting_per_account: 0, concurrency_wait_timeout_seconds: 30, + openai_guardian_reserved_concurrency: 0, responses_max_decompressed_body_bytes: 64 * 1024 * 1024, smart_scheduling: gateway_core::account::SmartSchedulingConfig::default(), rotation_strategy: "smart".to_owned(), @@ -361,23 +362,26 @@ async fn concurrency_queue_settings_round_trip_into_the_runtime_snapshot() { ( before.max_waiting_per_key, before.max_waiting_per_account, - before.concurrency_wait_timeout_seconds + before.concurrency_wait_timeout_seconds, + before.openai_guardian_reserved_concurrency, ), - (0, 0, 30) + (0, 0, 30, 0) ); let mut update = settings_with_margin(3600); update.max_waiting_per_key = 5; update.max_waiting_per_account = 7; update.concurrency_wait_timeout_seconds = 12; + update.openai_guardian_reserved_concurrency = 1; repository.update_runtime_settings(update).await.unwrap(); let settings = repository.load_runtime_settings().await.unwrap(); assert_eq!( ( settings.max_waiting_per_key, settings.max_waiting_per_account, - settings.concurrency_wait_timeout_seconds + settings.concurrency_wait_timeout_seconds, + settings.openai_guardian_reserved_concurrency, ), - (5, 7, 12) + (5, 7, 12, 1) ); let snapshot = PgRuntimeSnapshotRepository::new(database.pool.clone()) .load_runtime_snapshot() @@ -387,9 +391,10 @@ async fn concurrency_queue_settings_round_trip_into_the_runtime_snapshot() { ( snapshot.settings.max_waiting_per_key, snapshot.settings.max_waiting_per_account, - snapshot.settings.concurrency_wait_timeout_seconds + snapshot.settings.concurrency_wait_timeout_seconds, + snapshot.settings.openai_guardian_reserved_concurrency, ), - (5, 7, 12) + (5, 7, 12, 1) ); assert!(snapshot.config_revision > before.config_revision); database.close().await; diff --git a/backend/crates/providers/openai/src/credential/selector.rs b/backend/crates/providers/openai/src/credential/selector.rs index e5a3120b3..f9ce8a3f7 100644 --- a/backend/crates/providers/openai/src/credential/selector.rs +++ b/backend/crates/providers/openai/src/credential/selector.rs @@ -11,7 +11,7 @@ use gateway_core::account::{ AccountSelector, AccountStatus, CredentialState, PreferredAccountSelection, ProviderAccount, ProviderAccountId, QuotaEvidence, }; -use gateway_core::concurrency::{CapacityWait, ConcurrencyWaitQueue, QueueRejection}; +use gateway_core::concurrency::{CapacityWait, ConcurrencyWaitQueue, QueueRejection, WaitPriority}; use gateway_core::engine::{AttemptContext, ContinuationAttempt, policy::AccountPolicyError}; use gateway_core::provider_ports::{ ProviderLeaseAcquisition, ProviderLeaseGuard, ProviderLeasePort, ProviderLeaseRequest, @@ -106,6 +106,8 @@ struct CredentialSelectionInput<'a> { attempt: &'a AttemptContext, session_affinity_key: Option<&'a ProviderSessionAffinityKey>, session_affinity_observation: Option<&'a CodexSessionAffinity>, + /// Codex Guardian 自动审批请求；仅在配置了预留名额时获得预留与队列优先。 + guardian: bool, } #[derive(Clone)] @@ -299,6 +301,7 @@ impl CodexCredentialSelector { attempt: request.attempt, session_affinity_key: request.session_affinity_key, session_affinity_observation: None, + guardian: false, }; self.select_inner(&input, None, Some(request.upstream_model)) .await @@ -310,6 +313,7 @@ impl CodexCredentialSelector { cyber_policy_session_key: Option<&ProviderSessionAffinityKey>, session_affinity_observation: Option<&CodexSessionAffinity>, requires_websocket: bool, + guardian: bool, ) -> Result { let input = CredentialSelectionInput { requires_websocket, @@ -317,6 +321,7 @@ impl CodexCredentialSelector { attempt: request.attempt, session_affinity_key: request.session_affinity_key, session_affinity_observation, + guardian, }; self.select_inner( &input, @@ -376,6 +381,7 @@ impl CodexCredentialSelector { attempt: request.attempt, session_affinity_key: request.session_affinity.map(CodexSessionAffinity::key), session_affinity_observation: request.session_affinity, + guardian: false, }; self.select_inner(&input, None, None).await } @@ -387,12 +393,24 @@ impl CodexCredentialSelector { upstream_model: Option<&str>, ) -> Result { let queue_policy = request.attempt.account_selection_policy().queue_policy(); + // 预留名额只对普通请求生效；Guardian 可用满全部名额，并在账号队列中排在普通请求之前。 + let reserve = request + .attempt + .account_selection_policy() + .openai_guardian_reserved_concurrency(); + let prioritized = request.guardian && reserve > 0; + let reserved_concurrency = if prioritized { 0 } else { reserve }; let mut waiting = CapacityWait::new( &self.waiting, queue_policy, request.attempt.deadline(), request.attempt.concurrency_wait_budget(), - ); + ) + .with_priority(if prioritized { + WaitPriority::High + } else { + WaitPriority::Normal + }); let continuation_account = match request.attempt.continuation_attempt() { ContinuationAttempt::Native => request .attempt @@ -604,6 +622,7 @@ impl CodexCredentialSelector { AccountEligibilityPolicy::Enforce }, account_scope: request.attempt.account_scope().cloned(), + reserved_concurrency, }; let wait_context = AccountSelectionContext { excluded_accounts: base_excluded.clone(), @@ -724,6 +743,11 @@ impl CodexCredentialSelector { } Err(error) => return Err(error.into()), }; + // 凭据重读和插件选号都可能挂起，取得租约前再次让位于新队首。 + if !waiting.can_try(account.id()) { + excluded.insert(account.id().clone()); + continue; + } let allows_account_state_mutation = !diagnostic || account.enabled(); match self .leases @@ -732,7 +756,7 @@ impl CodexCredentialSelector { self.provider_kind.clone(), account.id().clone(), account.revision(), - account.effective_concurrency(policy.max_concurrent_per_account()), + context.concurrency_limit(&account), policy.request_interval(), request.attempt.deadline(), ), diff --git a/backend/crates/providers/openai/src/provider/mod.rs b/backend/crates/providers/openai/src/provider/mod.rs index e63138403..8ce3f355f 100644 --- a/backend/crates/providers/openai/src/provider/mod.rs +++ b/backend/crates/providers/openai/src/provider/mod.rs @@ -182,13 +182,11 @@ impl CodexProvider { fn prepare_generate_request( &self, generate: &GenerateRequest, - upstream_model: &UpstreamModelId, + mut upstream: CodexResponsesRequest, context: &AttemptContext, - ) -> Result { + ) -> PreparedGenerateRequest { let previous_session = decode_openai_session_state(generate); let continuation_requested = generate.native_continuation_requested(); - let mut upstream = encode_generate_request(generate, upstream_model.as_str(), None) - .map_err(map_request_error)?; if let Some(conversation_id) = previous_session .as_ref() .and_then(|state| state.conversation_id.as_ref()) @@ -215,13 +213,13 @@ impl CodexProvider { derive_codex_session_affinity(&upstream, context.client_api_key_ref()); let cyber_policy_session_key = derive_codex_cyber_policy_session_key(&upstream, context.client_api_key_ref()); - Ok(PreparedGenerateRequest { + PreparedGenerateRequest { upstream, previous_session, continuation_requested, session_affinity, cyber_policy_session_key, - }) + } } fn client_for_request( @@ -517,10 +515,18 @@ impl Provider for CodexProvider { } // 其他协议必须先取得真实账号，再按固定 attempt 阶段调用转换器；选号前不能 // 把未知正文当成 OpenAI wire 解释会话、亲和或传输字段。 - let preselection = (adapter.is_none() - && generate.protocol_payload().protocol() == PROVIDER_NAME) - .then(|| self.prepare_generate_request(generate, upstream_model, &context)) - .transpose()?; + let upstream = (generate.protocol_payload().protocol() == PROVIDER_NAME) + .then(|| encode_generate_request(generate, upstream_model.as_str(), None)) + .transpose() + .map_err(map_request_error)?; + // 固定调度约束：Guardian 分类先于选号，不能随原生/适配器发送路径改变。 + // 复用编码器的权威 metadata 解析，但只为原生路径准备会话与传输状态。 + let guardian = upstream + .as_ref() + .is_some_and(CodexResponsesRequest::is_guardian); + let preselection = upstream + .filter(|_| adapter.is_none()) + .map(|upstream| self.prepare_generate_request(generate, upstream, &context)); let (selection_session_affinity, selection_cyber_policy_key, requires_websocket) = preselection.map_or((None, None, false), |prepared| { let requires_websocket = @@ -552,6 +558,7 @@ impl Provider for CodexProvider { selection_cyber_policy_key.as_ref(), selection_session_affinity.as_ref(), requires_websocket, + guardian, ) .await .map_err(map_selection_error) @@ -648,7 +655,9 @@ impl CodexProvider { )); } validate_openai_reasoning(generate.protocol_payload().body())?; - let processed = self.prepare_generate_request(&generate, &upstream_model, &context)?; + let upstream = encode_generate_request(&generate, upstream_model.as_str(), None) + .map_err(map_request_error)?; + let processed = self.prepare_generate_request(&generate, upstream, &context); let mut upstream_request = processed.upstream; let previous_session = processed.previous_session; let continuation_requested = processed.continuation_requested; diff --git a/backend/crates/providers/openai/src/transport/protocol/responses.rs b/backend/crates/providers/openai/src/transport/protocol/responses.rs index b6daed375..87e96e0da 100644 --- a/backend/crates/providers/openai/src/transport/protocol/responses.rs +++ b/backend/crates/providers/openai/src/transport/protocol/responses.rs @@ -15,6 +15,8 @@ pub(crate) const X_CODEX_TURN_STATE_CLIENT_METADATA_KEY: &str = "x-codex-turn-st /// 本地生成 history unavailable 错误时使用的官方提示文本。 pub(crate) const PREVIOUS_RESPONSE_NOT_FOUND_MESSAGE: &str = "Previous response was not found. Retrying the full request."; +/// Codex 自动审批（Guardian）请求声明的子代理类型。 +const GUARDIAN_SUBAGENT_KIND: &str = "guardian"; /// Codex Responses 上游请求体。 /// /// 发往上游的 Responses 请求。`body` 持有客户端原始 JSON object，逐字段（含顺序、 @@ -694,6 +696,11 @@ impl CodexResponsesRequest { }) } + /// Codex 执行命令前的 Guardian 自动审批请求；客户端以 `guardian` 子代理类型声明。 + pub fn is_guardian(&self) -> bool { + self.subagent_kind().as_deref() == Some(GUARDIAN_SUBAGENT_KIND) + } + /// 设置 / 合并 client metadata。 pub fn set_client_metadata(&mut self, client_metadata: Option) { match client_metadata { diff --git a/backend/crates/providers/openai/tests/credential/quota/scheduling.rs b/backend/crates/providers/openai/tests/credential/quota/scheduling.rs index 0ac7f7693..9ff120d67 100644 --- a/backend/crates/providers/openai/tests/credential/quota/scheduling.rs +++ b/backend/crates/providers/openai/tests/credential/quota/scheduling.rs @@ -106,6 +106,7 @@ async fn old_weekly_quota_should_outweigh_one_in_flight_title_request() { round_robin_cursor: cursor, eligibility: AccountEligibilityPolicy::Enforce, account_scope: None, + reserved_concurrency: 0, }; let selected = AccountSelector .select(&candidates, &context) diff --git a/backend/crates/providers/openai/tests/provider/contract/capacity.rs b/backend/crates/providers/openai/tests/provider/contract/capacity.rs index 9ebe5ba8c..63a248506 100644 --- a/backend/crates/providers/openai/tests/provider/contract/capacity.rs +++ b/backend/crates/providers/openai/tests/provider/contract/capacity.rs @@ -1,4 +1,5 @@ use futures::FutureExt; +use gateway_core::account::AccountRuntimeSignals; use gateway_core::concurrency::ConcurrencyQueuePolicy; use super::*; @@ -129,3 +130,139 @@ async fn queued_session_sends_only_after_capacity_recovers_while_a_new_child_can ); server.verify().await; } + +fn unqueued_context(request_id: &str, reserved: u32) -> 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), + account_policy().with_openai_guardian_reserved_concurrency(reserved), + AccountAttemptContext::new(BTreeSet::new(), None, None) + .with_account_scope(contract_account_scope()), + None, + CancellationToken::new(), + ) +} + +fn subagent_operation(subagent: Option<&str>) -> Operation { + let mut body = Map::from_iter([ + ("model".to_owned(), json!("gpt-5.4")), + ("input".to_owned(), json!("guardian reservation test")), + ]); + if let Some(subagent) = subagent { + body.insert( + "client_metadata".to_owned(), + json!({"x-openai-subagent": subagent}), + ); + } + Operation::Generate(GenerateRequest::from_protocol_payload( + ProtocolPayload::json_object("openai", body) + .unwrap() + .with_context(Map::from_iter([("use_websocket".to_owned(), json!(false))])), + )) +} + +#[tokio::test] +async fn guardian_requests_can_use_the_reserved_slot_that_normal_requests_cannot() { + let store = Arc::new(MemoryAccountStore::default()); + create_account(&store, "acct_guardian").await; + let account = ProviderAccountId::new("acct_guardian").unwrap(); + let leases = Arc::new(TestLeaseCoordinator::default()); + let server = MockServer::start().await; + Mock::given(method("POST")) + .and(path("/codex/responses")) + .respond_with( + ResponseTemplate::new(200) + .insert_header("content-type", "text/event-stream") + .set_body_string(CAPTURE_COMPLETED_SSE), + ) + .expect(3) + .mount(&server) + .await; + // 默认账号上限为 2、预留 1：普通请求只能使用 1 个名额，Guardian 可以用满 2 个。 + let provider = provider_with_affinity_and_base_url_and_leases( + &store, + Arc::new(MemorySessionAffinity::default()), + server.uri(), + leases.clone(), + ); + + let mut normal = provider + .clone() + .execute( + planned_request("openai", subagent_operation(Some("collab_spawn"))), + unqueued_context("req_guardian_normal_idle", 1), + ) + .await + .unwrap(); + while let Some(event) = normal.next().await { + event.unwrap(); + } + drop(normal); + let limits = |leases: &TestLeaseCoordinator| { + leases + .requests + .lock() + .unwrap() + .iter() + .map(|request| request.max_concurrent().get()) + .collect::>() + }; + assert_eq!(limits(&leases), [1]); + + leases.signals.lock().unwrap().insert( + account.clone(), + AccountRuntimeSignals { + in_flight: 1, + last_started_at: None, + quota_reset_at: None, + quota_remaining_rank: None, + cooldown: None, + failure_rate_basis_points: None, + first_output_latency_ms: None, + }, + ); + assert!( + provider + .clone() + .execute( + planned_request("openai", subagent_operation(None)), + unqueued_context("req_guardian_normal_busy", 1), + ) + .await + .is_err() + ); + assert_eq!(limits(&leases), [1]); + + let mut guardian = provider + .clone() + .execute( + planned_request("openai", subagent_operation(Some("guardian"))), + unqueued_context("req_guardian_reserved", 1), + ) + .await + .unwrap(); + assert_eq!(guardian.metadata().provider_account_id(), &account); + while let Some(event) = guardian.next().await { + event.unwrap(); + } + assert_eq!(limits(&leases), [1, 2]); + drop(guardian); + + // 同一个 Provider 的新请求读取关闭后的策略，不需要重新初始化选择器。 + let mut unreserved = provider + .execute( + planned_request("openai", subagent_operation(None)), + unqueued_context("req_guardian_reservation_disabled", 0), + ) + .await + .unwrap(); + while let Some(event) = unreserved.next().await { + event.unwrap(); + } + assert_eq!(limits(&leases), [1, 2, 2]); + server.verify().await; +} diff --git a/backend/crates/providers/openai/tests/provider/contract/mod.rs b/backend/crates/providers/openai/tests/provider/contract/mod.rs index 24a92ed54..3903a4219 100644 --- a/backend/crates/providers/openai/tests/provider/contract/mod.rs +++ b/backend/crates/providers/openai/tests/provider/contract/mod.rs @@ -1339,6 +1339,7 @@ fn contract_account_scope() -> Arc { "acct_continuation_prefetch", "acct_disabled_scheduling", "acct_first_event_latency", + "acct_guardian", "acct_header_new", "acct_header_old", "acct_header_same", diff --git a/backend/crates/providers/openai/tests/provider/contract/upstream_adapter.rs b/backend/crates/providers/openai/tests/provider/contract/upstream_adapter.rs index d6cef5deb..895dac33a 100644 --- a/backend/crates/providers/openai/tests/provider/contract/upstream_adapter.rs +++ b/backend/crates/providers/openai/tests/provider/contract/upstream_adapter.rs @@ -1,4 +1,5 @@ use super::*; +use gateway_core::account::AccountRuntimeSignals; use gateway_core::engine::upstream_adapter::{ UpstreamAccountConnection, UpstreamAdapter, UpstreamAdapterInvocation, UpstreamAdapterPlan, }; @@ -257,6 +258,108 @@ async fn selected_adapter_connection( receiver.await.unwrap() } +#[tokio::test] +async fn guardian_reservation_survives_upstream_adapters_and_metadata_precedence() { + // 上限 2、已有 1 个在途请求：预留启用时只有 Guardian 能继续取得租约。 + for (subagent, turn_kind, reserved, allowed) in [ + (Some("guardian"), None, 1, true), + (None, Some("guardian"), 1, true), + (Some("collab_spawn"), Some("guardian"), 1, true), + (Some("guardian"), Some("collab_spawn"), 1, false), + (Some("collab_spawn"), None, 1, false), + (None, None, 1, false), + (None, None, 0, true), + ] { + let store = Arc::new(MemoryAccountStore::default()); + create_account(&store, "acct_provider_contract").await; + let leases = Arc::new(TestLeaseCoordinator::default()); + leases.signals.lock().unwrap().insert( + ProviderAccountId::new("acct_provider_contract").unwrap(), + AccountRuntimeSignals { + in_flight: 1, + last_started_at: None, + quota_reset_at: None, + quota_remaining_rank: None, + cooldown: None, + failure_rate_basis_points: None, + first_output_latency_ms: None, + }, + ); + let server = MockServer::start().await; + let provider = provider_with_affinity_and_base_url_and_leases( + &store, + Arc::new(MemorySessionAffinity::default()), + server.uri(), + leases.clone(), + ); + let (selected, receiver) = oneshot::channel(); + let plan = gateway_core::engine::upstream_adapter::FrozenUpstreamAdapterPlan::new( + Arc::new(ConnectionProbe { + selected: Arc::new(Mutex::new(Some(selected))), + }), + ExtensionSetReference::new( + ExtensionSetId::new("guardian-adapter-test".to_owned()).unwrap(), + Arc::new(TestExtensionLease), + ), + ); + let context = AttemptContext::new( + RequestAttemptContext::new( + ModelRequestId::new("req_guardian_adapter").unwrap(), + ClientApiKeyId::new("key_openai_contract").unwrap(), + ) + .with_upstream_adapters(Some(plan)), + NonZeroU32::MIN, + SystemTime::now() + Duration::from_secs(5), + account_policy().with_openai_guardian_reserved_concurrency(reserved), + AccountAttemptContext::new(BTreeSet::new(), None, None) + .with_account_scope(contract_account_scope()), + None, + CancellationToken::new(), + ); + let mut body = Map::from_iter([ + ("model".into(), json!("gpt-5.4")), + ("input".into(), json!("guardian scheduling contract")), + ]); + if let Some(kind) = subagent { + body.insert("client_metadata".into(), json!({"x-openai-subagent": kind})); + } + let mut protocol_context = Map::new(); + if let Some(kind) = turn_kind { + protocol_context.insert( + "turn_metadata".into(), + json!(json!({"subagent_kind": kind}).to_string()), + ); + } + let operation = Operation::Generate(GenerateRequest::from_protocol_payload( + ProtocolPayload::json_object("openai", body) + .unwrap() + .with_context(protocol_context), + )); + let result = provider + .execute(planned_request("openai", operation), context) + .await; + assert_eq!( + result.is_ok(), + allowed, + "{subagent:?}, {turn_kind:?}, reserve={reserved}" + ); + if allowed { + let mut stream = result.unwrap(); + assert_eq!( + stream.next().await.unwrap().unwrap_err().kind(), + ProviderErrorKind::Cancelled + ); + drop(receiver.await.unwrap()); + let requests = leases.requests.lock().unwrap(); + assert_eq!(requests.len(), 1); + assert_eq!(requests[0].max_concurrent().get(), 2); + } else { + assert!(leases.requests.lock().unwrap().is_empty()); + } + assert!(server.received_requests().await.unwrap().is_empty()); + } +} + #[tokio::test] async fn upstream_adapter_reloads_rotated_credentials_and_proxy_without_accepting_stale_failure() { use gateway_core::account::OutboundProxy; diff --git a/backend/crates/providers/xai/src/credential/selector.rs b/backend/crates/providers/xai/src/credential/selector.rs index 735c50af9..af6179118 100644 --- a/backend/crates/providers/xai/src/credential/selector.rs +++ b/backend/crates/providers/xai/src/credential/selector.rs @@ -230,6 +230,7 @@ impl GrokAccountSessionSelector { round_robin_cursor: scheduling.round_robin_cursor(), eligibility: request.eligibility(), account_scope: (!diagnostic).then(|| Arc::clone(request.account_scope())), + reserved_concurrency: 0, }; let wait_candidates = AccountSelector.wait_candidates(&candidates, &context); let capacity_context = context.clone(); diff --git a/backend/migrations/.frozen-sha256 b/backend/migrations/.frozen-sha256 index 824c8c7dc..8a91c0b7b 100644 --- a/backend/migrations/.frozen-sha256 +++ b/backend/migrations/.frozen-sha256 @@ -19,3 +19,4 @@ fcfae912ddefd54551ebf8e7e733442519d2862d2977134d548ebc1e75c1dc34 0017_unlimited 40a4d4b16e6ec2c2c659980d416b089fc170acd2255aaaf751c4a04c3790a941 0019_plugin_resources.sql e8c1339166ac02fb2d4a077424ee5f64f6dc20eb4d3f8ec4fcb09c0f2a0e61f6 0020_plugin_trust.sql 338037806358b23dc8b62a39e73b04ce4173c4556383e5df19b7e8c7123924ed 0021_account_warmup_slots.sql +3b6924605e7e70c074d409611d13f17625f98fae4e9dbe8012a6b3dc3aa1679f 0022_guardian_reserved_concurrency.sql diff --git a/backend/migrations/0022_guardian_reserved_concurrency.sql b/backend/migrations/0022_guardian_reserved_concurrency.sql new file mode 100644 index 000000000..c0e8df9d9 --- /dev/null +++ b/backend/migrations/0022_guardian_reserved_concurrency.sql @@ -0,0 +1,3 @@ +alter table runtime_settings + add column openai_guardian_reserved_concurrency bigint not null default 0 + check (openai_guardian_reserved_concurrency between 0 and 4294967295); diff --git a/docs/api.md b/docs/api.md index 0bda486c8..dc9f0c2d2 100644 --- a/docs/api.md +++ b/docs/api.md @@ -1218,6 +1218,7 @@ refreshConcurrency maxConcurrentPerAccount maxWaitingPerKey maxWaitingPerAccount +openaiGuardianReservedConcurrency concurrencyWaitTimeoutSeconds responsesMaxDecompressedBodyBytes requestIntervalMs @@ -1265,6 +1266,13 @@ accountWarmupModel 切换账号或内部重试不重新计时，等待同时计入请求总超时。该时限不用于中断已开始的上游生成。 设置更新请求须包含这三个字段，新请求使用更新后的快照 +`openaiGuardianReservedConcurrency`（默认 0，取值 0～4,294,967,295）为 Codex Guardian 自动审批保留账号并发，保存后对新请求生效。 +Guardian 以 `subagent_kind` 或 `client_metadata.x-openai-subagent` 值 `guardian` 识别。取值 R 大于 0 时， +有限上限为 L 的账号对其他 OpenAI 请求只开放 `max(L − R, 1)` 个名额，Guardian 可用满 L； +开启账号排队后，Guardian 排在同账号已有 Guardian 之后、全部普通等待者之前，不受单账号排队上限约束， +仍受总等待容量与等待时限约束。不限并发的账号和关闭排队时的其余行为不变。 +设置更新请求须包含该字段 + `responsesMaxDecompressedBodyBytes` 是压缩 Responses HTTP 请求的解压输出上限，单位字节，默认 67108864（64 MiB）。必须为正整数，且可表示为进程平台的 `isize`；管理端以整数 MiB 编辑。 保存并发布成功后，新请求使用新值；已鉴权请求沿用原快照，无需重启。调高上限会增加大请求的内存占用， diff --git a/docs/architecture.md b/docs/architecture.md index 6d9b808d6..ae6b5361e 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -544,12 +544,12 @@ OpenAI 启用账号排队时，可等待的已有亲和绑定优先保留；子 ### 并发等待 -Core 的 `concurrency` 拥有中立的有界等待位置与 FIFO 唤醒，不依赖账号、路由或执行会话； +Core 的 `concurrency` 拥有中立的有界等待位置、优先级与同级 FIFO 唤醒，不依赖账号、路由或执行会话； 账号策略仅引用其中的等待配置值。执行准入与各 Provider 选择器分别持有等待队列，按 Key/账号隔离。 运行并发仍由 Redis 原子准入与账号租约裁决，等待队列只保存当前进程中的等待位置，不复制运行计数， 也不持久化正文或在重启后重放请求。每个等待 owner 最多容纳 1,024 个等待者作为资源兜底 -新请求不能抢占已有等待队列；队首短周期重读容量，取消/完成等待通过 Drop 回收位置并唤醒后继。 +新请求不能越过同级或更高优先级的等待者；队首短周期重读容量，取消/完成等待通过 Drop 回收位置并唤醒后继。 Provider 等待重试仍在同一次未发送准备阶段重新读取账号资格、容量、目录和冷却，不增加 attempt； Key 的 RPM 在成功准入时才计数，金额限制在入队前及成功准入后检查。 账号并发与本地请求间隔可有限等待，权限、额度和上游冷却不能作为可等待容量。 @@ -557,6 +557,22 @@ Key 的 RPM 在成功准入时才计数，金额限制在入队前及成功准 切换等待层、账号或 Provider 不重置等待时限；没有发生排队时不启动该计时，也不据此中断已开始的上游生成。 队列满/超时属于本地容量拒绝，不作为上游限流反馈或 Provider 故障熔断证据 +#### Guardian 审批优先 + +Guardian 预留值由运行设置持久化并随调度快照冻结，每个请求沿用开始时的设置。 +Core 传递该值并拥有通用的并发上限与等待顺序，OpenAI Provider 负责识别请求类型并决定预留与优先级。 + +- OpenAI Provider 在选号前通过统一协议解析识别 `guardian`；原生发送与插件上游适配器使用相同分类。 + 请求头形成的协议上下文与正文 metadata 沿用编码器的既有优先顺序；其他协议不得仅凭同名字段获得该优先级。 + 适配器不因此继承原生会话亲和、续接或 WebSocket 约束，选号后的 attempt 中间件也不追溯改变已取得的租约。 +- `openaiGuardianReservedConcurrency = R > 0` 时，有限账号上限 L 对普通请求为 `max(L - R, 1)`， + Guardian 可用满 L，资格判断、策略投影、容量观测和原子租约共用同一请求级上限，数值边界见 [运行设置 API](api.md#8-运行设置) +- 开启账号排队时，Guardian 在普通等待者之前、已有 Guardian 之后；同级 FIFO，不抢占已运行请求。 + Guardian 可越过单账号等待数量上限，但仍受总等待容量、共享等待预算与截止时刻约束；取消及超时必须释放等待位置。 +- 配置为 0 时不启用预留或优先级；账号排队关闭时不引入隐式队列。Key 层准入仍按原有限额与排队规则执行。 + +等待者重新选号及取得租约前复核当前队首，异步读取期间被插队的普通请求继续等待，不凭已有 ticket 跳过优先顺序。 + ### Client Key 限额与结算 日金额、七天金额、并发和 RPM 按 Client Key 跨账号、跨 Provider 合计，零表示不限；修改限额不重置已用金额。 diff --git a/frontend/src/api/modules/settings.ts b/frontend/src/api/modules/settings.ts index f2f342d60..73de9c23d 100644 --- a/frontend/src/api/modules/settings.ts +++ b/frontend/src/api/modules/settings.ts @@ -29,6 +29,7 @@ export interface RuntimeSettings { refreshMarginSeconds: number refreshConcurrency: number maxConcurrentPerAccount: number + openaiGuardianReservedConcurrency: number requestIntervalMs: number maxWaitingPerKey: number maxWaitingPerAccount: number diff --git a/frontend/src/views/settings/components/ConcurrencyPolicyCard.vue b/frontend/src/views/settings/components/ConcurrencyPolicyCard.vue new file mode 100644 index 000000000..7724eabe4 --- /dev/null +++ b/frontend/src/views/settings/components/ConcurrencyPolicyCard.vue @@ -0,0 +1,79 @@ + + + diff --git a/frontend/src/views/settings/components/RequestQueueCard.vue b/frontend/src/views/settings/components/RequestQueueCard.vue deleted file mode 100644 index 25f6f82c2..000000000 --- a/frontend/src/views/settings/components/RequestQueueCard.vue +++ /dev/null @@ -1,23 +0,0 @@ - - - diff --git a/frontend/src/views/settings/components/RuntimeSettingsCard.vue b/frontend/src/views/settings/components/RuntimeSettingsCard.vue deleted file mode 100644 index 720f93456..000000000 --- a/frontend/src/views/settings/components/RuntimeSettingsCard.vue +++ /dev/null @@ -1,50 +0,0 @@ - - - diff --git a/frontend/src/views/settings/composables/useSettingsForm.ts b/frontend/src/views/settings/composables/useSettingsForm.ts index 824ae204c..27df7c7f0 100644 --- a/frontend/src/views/settings/composables/useSettingsForm.ts +++ b/frontend/src/views/settings/composables/useSettingsForm.ts @@ -32,6 +32,7 @@ export function useSettingsForm() { refreshMarginSeconds: null as number | null, refreshConcurrency: null as number | null, maxConcurrentPerAccount: null as number | null, + openaiGuardianReservedConcurrency: null as number | null, requestIntervalMs: null as number | null, maxWaitingPerKey: null as number | null, maxWaitingPerAccount: null as number | null, @@ -84,7 +85,7 @@ export function useSettingsForm() { mappings.value = saved.value.mappings.map(row => ({ ...row })) } - function numericModel(key: 'refreshMarginSeconds' | 'refreshConcurrency' | 'maxConcurrentPerAccount' | 'requestIntervalMs' | 'maxWaitingPerKey' | 'maxWaitingPerAccount' | 'concurrencyWaitTimeoutSeconds' | 'responsesMaxDecompressedBodyMiB' | 'accountAutoFreezeThreshold' | 'accountAutoFreezeWindowSeconds' | 'accountAutoFreezeDurationSeconds') { + function numericModel(key: 'refreshMarginSeconds' | 'refreshConcurrency' | 'maxConcurrentPerAccount' | 'openaiGuardianReservedConcurrency' | 'requestIntervalMs' | 'maxWaitingPerKey' | 'maxWaitingPerAccount' | 'concurrencyWaitTimeoutSeconds' | 'responsesMaxDecompressedBodyMiB' | 'accountAutoFreezeThreshold' | 'accountAutoFreezeWindowSeconds' | 'accountAutoFreezeDurationSeconds') { return computed({ get: () => (form[key] === null ? '' : String(form[key])), set: (value: string) => { @@ -101,6 +102,7 @@ export function useSettingsForm() { const refreshMarginSecondsValue = numericModel('refreshMarginSeconds') const refreshConcurrencyValue = numericModel('refreshConcurrency') const maxConcurrentPerAccountValue = numericModel('maxConcurrentPerAccount') + const openaiGuardianReservedConcurrencyValue = numericModel('openaiGuardianReservedConcurrency') const requestIntervalMsValue = numericModel('requestIntervalMs') const maxWaitingPerKeyValue = numericModel('maxWaitingPerKey') const maxWaitingPerAccountValue = numericModel('maxWaitingPerAccount') @@ -126,6 +128,7 @@ export function useSettingsForm() { form.refreshMarginSeconds = data.refreshMarginSeconds form.refreshConcurrency = data.refreshConcurrency form.maxConcurrentPerAccount = data.maxConcurrentPerAccount + form.openaiGuardianReservedConcurrency = data.openaiGuardianReservedConcurrency form.requestIntervalMs = data.requestIntervalMs form.maxWaitingPerKey = data.maxWaitingPerKey form.maxWaitingPerAccount = data.maxWaitingPerAccount @@ -211,8 +214,8 @@ export function useSettingsForm() { const savedSettings = saved.value if (saving.value || loading.value || !savedRequestLocation.value || !savedSettings) return - const { refreshMarginSeconds, refreshConcurrency, maxConcurrentPerAccount, requestIntervalMs, rotationStrategy, maxWaitingPerKey, maxWaitingPerAccount, concurrencyWaitTimeoutSeconds, responsesMaxDecompressedBodyMiB, accountAutoFreezeThreshold, accountAutoFreezeWindowSeconds, accountAutoFreezeDurationSeconds } = form - if (refreshMarginSeconds === null || refreshConcurrency === null || maxConcurrentPerAccount === null || requestIntervalMs === null || !rotationStrategy || maxWaitingPerKey === null || maxWaitingPerAccount === null || concurrencyWaitTimeoutSeconds === null) { + 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) { toast.warning('请完整填写并发、队列、凭据刷新参数和调度策略') return } @@ -220,6 +223,10 @@ export function useSettingsForm() { toast.warning('默认账号并发上限应为 0～4294967295 的整数，0 表示不限制') return } + if (!Number.isInteger(openaiGuardianReservedConcurrency) || openaiGuardianReservedConcurrency < 0 || openaiGuardianReservedConcurrency > 4294967295) { + toast.warning('自动审批预留并发应为 0～4294967295 的整数，0 表示关闭') + return + } if (responsesMaxDecompressedBodyMiB === null || !Number.isInteger(responsesMaxDecompressedBodyMiB) || responsesMaxDecompressedBodyMiB < 1 || !Number.isSafeInteger(responsesMaxDecompressedBodyMiB * MIB)) { toast.warning('Responses 解压上限应为有效的正整数（MiB）') @@ -286,6 +293,7 @@ export function useSettingsForm() { refreshMarginSeconds, refreshConcurrency, maxConcurrentPerAccount, + openaiGuardianReservedConcurrency, requestIntervalMs, maxWaitingPerKey, maxWaitingPerAccount, @@ -334,6 +342,7 @@ export function useSettingsForm() { refreshMarginSecondsValue, refreshConcurrencyValue, maxConcurrentPerAccountValue, + openaiGuardianReservedConcurrencyValue, requestIntervalMsValue, maxWaitingPerKeyValue, maxWaitingPerAccountValue, diff --git a/frontend/src/views/settings/index.vue b/frontend/src/views/settings/index.vue index 64ea43f57..b6b69c5a9 100644 --- a/frontend/src/views/settings/index.vue +++ b/frontend/src/views/settings/index.vue @@ -9,11 +9,10 @@ import AccountAutoFreezeCard from './components/AccountAutoFreezeCard.vue' import AccountWarmupCard from './components/AccountWarmupCard.vue' import SettingsBackupSection from './components/backup/SettingsBackupSection.vue' import ClientProfileCard from './components/ClientProfileCard.vue' +import ConcurrencyPolicyCard from './components/ConcurrencyPolicyCard.vue' import ModelAliasesCard from './components/ModelAliasesCard.vue' import RequestLocationCard from './components/RequestLocationCard.vue' -import RequestQueueCard from './components/RequestQueueCard.vue' import RotationStrategyCard from './components/RotationStrategyCard.vue' -import RuntimeSettingsCard from './components/RuntimeSettingsCard.vue' import SettingsAccessSection from './components/SettingsAccessSection.vue' import TokenRefreshCard from './components/TokenRefreshCard.vue' import { useSettingsForm } from './composables/useSettingsForm' @@ -61,6 +60,7 @@ const { refreshMarginSecondsValue, refreshConcurrencyValue, maxConcurrentPerAccountValue, + openaiGuardianReservedConcurrencyValue, requestIntervalMsValue, maxWaitingPerKeyValue, maxWaitingPerAccountValue, @@ -144,10 +144,6 @@ watch(section, (value) => {