From 90e241f74b5d25d8eb5a2c3025bf64dcfc17d78f Mon Sep 17 00:00:00 2001 From: Yongzhao <133014490+Yongzhaooo@users.noreply.github.com> Date: Thu, 1 Oct 2026 16:49:11 +0200 Subject: [PATCH] feat(openai): configure Guardian concurrency priority at runtime (#306) Configure Guardian reservations through persisted runtime settings and frozen request policies. Combine concurrency controls and queues in the admin UI, with the scheduling strategy shown first. Prioritize Guardian account waiters, recheck displaced queue heads before lease acquisition, and report request-specific available capacity. Preserve account, key and wait-budget limits. Co-authored-by: Yongzhao Chen Co-authored-by: Zyy Co-authored-by: Claude Opus 5.5 Co-authored-by: Codex --- .../gateway-admin/src/model/settings.rs | 3 + .../gateway-admin/tests/use_case/accounts.rs | 1 + .../tests/use_case/freeze_recovery.rs | 1 + .../tests/use_case/observability.rs | 1 + .../gateway-admin/tests/use_case/settings.rs | 1 + .../crates/gateway-api/src/admin/settings.rs | 4 + backend/crates/gateway-api/tests/admin/mod.rs | 2 + .../gateway-api/tests/admin/settings.rs | 51 +++++++ .../crates/gateway-core/src/account/model.rs | 12 ++ .../gateway-core/src/account/selection.rs | 40 +++-- .../crates/gateway-core/src/concurrency.rs | 115 ++++++++++++--- .../gateway-core/src/diagnostics/selection.rs | 1 + .../crates/gateway-core/src/engine/policy.rs | 6 +- .../gateway-core/src/settings/compiled.rs | 1 + .../gateway-core/src/settings/values.rs | 8 + .../crates/gateway-core/tests/account/mod.rs | 53 ++++++- .../crates/gateway-core/tests/concurrency.rs | 96 +++++++++++- .../gateway-core/tests/engine/execution.rs | 1 + .../gateway-core/tests/engine/policy.rs | 1 + .../gateway-core/tests/routing/snapshot.rs | 12 +- .../runtime/tests/generation/policy/mod.rs | 1 + .../runtime/tests/rpc/session.rs | 2 +- .../sdk/src/call/services/settings.rs | 3 + .../crates/gateway-store/src/admin_adapter.rs | 3 + .../src/postgres/runtime_settings.rs | 10 +- .../gateway-store/src/postgres/snapshot.rs | 8 +- .../tests/postgres/runtime_settings.rs | 17 ++- .../openai/src/credential/selector.rs | 30 +++- .../providers/openai/src/provider/mod.rs | 31 ++-- .../src/transport/protocol/responses.rs | 7 + .../tests/credential/quota/scheduling.rs | 1 + .../tests/provider/contract/capacity.rs | 137 ++++++++++++++++++ .../openai/tests/provider/contract/mod.rs | 1 + .../provider/contract/upstream_adapter.rs | 103 +++++++++++++ .../providers/xai/src/credential/selector.rs | 1 + backend/migrations/.frozen-sha256 | 1 + .../0022_guardian_reserved_concurrency.sql | 3 + docs/api.md | 8 + docs/architecture.md | 20 ++- frontend/src/api/modules/settings.ts | 1 + .../components/ConcurrencyPolicyCard.vue | 79 ++++++++++ .../settings/components/RequestQueueCard.vue | 23 --- .../components/RuntimeSettingsCard.vue | 50 ------- .../settings/composables/useSettingsForm.ts | 15 +- frontend/src/views/settings/index.vue | 13 +- 45 files changed, 820 insertions(+), 158 deletions(-) create mode 100644 backend/migrations/0022_guardian_reserved_concurrency.sql create mode 100644 frontend/src/views/settings/components/ConcurrencyPolicyCard.vue delete mode 100644 frontend/src/views/settings/components/RequestQueueCard.vue delete mode 100644 frontend/src/views/settings/components/RuntimeSettingsCard.vue 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) => {