fix(openai): isolate guardian concurrency from ordinary requests

Give Guardian requests a separate per-account lease pool and wait queue while preserving ordinary concurrency and shared request intervals. Use the ordinary pool when the Guardian limit is zero, and align scheduling, diagnostics, settings copy, and documentation. Co-authored-by: Codex <noreply@openai.com>
2026-11-06 08:06:19 +00:00 · 2026-10-08 10:38:52 +08:00
parent 6c583e341a
commit 3315ddbff6
24 changed files with 624 additions and 299 deletions
@@ -90,18 +90,6 @@ 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<NonZeroU32> for AccountConcurrency {
@@ -206,6 +206,7 @@ impl AccountSelectionPolicy {
 /// Store 提供并发事实，Provider 叠加自己解释的额度事实；全部信号均可重建
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub struct AccountRuntimeSignals {
+    /// 本次调度容量池的在途数，不合并其他独立池
    pub in_flight: u32,
    pub last_started_at: Option<SystemTime>,
    pub quota_reset_at: Option<SystemTime>,
@@ -523,7 +524,7 @@ pub struct AccountSelectionContext {
    pub round_robin_cursor: u64,
    pub eligibility: AccountEligibilityPolicy,
    pub account_scope: Option<std::sync::Arc<crate::account::scope::FrozenAccountScope>>,
-    /// 本请求不可占用的每账号预留并发名额，由 Provider 按请求类别决定；0 表示不预留
+    /// 本请求使用的独立预留池上限，由 Provider 按请求类别决定；0 表示使用普通账号池
    pub reserved_concurrency: u32,
 }

@@ -531,9 +532,11 @@ 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)
+        if self.reserved_concurrency > 0 {
+            AccountConcurrency::new(self.reserved_concurrency)
+        } else {
+            account.effective_concurrency(self.policy.max_concurrent_per_account())
+        }
    }
 }

@@ -723,7 +726,6 @@ impl AccountSelector {

        let candidate = match context.policy.strategy() {
            RotationStrategy::QuotaResetPriority => {
-                let default_concurrency = context.policy.max_concurrent_per_account();
                eligible.sort_by(|left, right| {
                    (
                        left.signals.quota_reset_at.is_none(),
@@ -734,8 +736,8 @@ impl AccountSelector {
                            right.signals.quota_reset_at,
                        ))
                        .then_with(|| {
-                            capacity_utilization(left, default_concurrency)
-                                .total_cmp(&capacity_utilization(right, default_concurrency))
+                            capacity_utilization(left, context)
+                                .total_cmp(&capacity_utilization(right, context))
                        })
                        .then_with(|| {
                            left.signals
@@ -751,13 +753,7 @@ impl AccountSelector {
                let index = context.round_robin_cursor as usize % eligible.len();
                eligible.get(index).copied()?
            }
-            RotationStrategy::Smart => select_smart_candidate(
-                &eligible,
-                context.policy.max_concurrent_per_account(),
-                context.round_robin_cursor,
-                context.policy.smart_scheduling(),
-                context.now,
-            )?,
+            RotationStrategy::Smart => select_smart_candidate(&eligible, context)?,
            RotationStrategy::Sticky => {
                eligible.sort_by_key(|candidate| {
                    (
@@ -843,17 +839,7 @@ impl AccountSelector {
        // 账号信号在队列锁外评分，锁内只读取实时队长并查表，避免扫描候选阻塞其他等待者
        let scores = candidates
            .iter()
-            .map(|candidate| {
-                (
-                    candidate.account.id(),
-                    smart_score(
-                        candidate,
-                        context.policy.max_concurrent_per_account(),
-                        config,
-                        context.now,
-                    ),
-                )
-            })
+            .map(|candidate| (candidate.account.id(), smart_score(candidate, context)))
            .collect::<HashMap<_, _>>();
        waiting
            .wait_with_priority(keys, |key, count| {
@@ -914,13 +900,9 @@ const SMART_LATENCY_HALF_SCORE_MS: f64 = 10_000.0;
 // 距重置一小时时得分减半；未知和已过期时间不提供重置奖励
 const SMART_RESET_HALF_SCORE_SECONDS: f64 = 3_600.0;

-fn capacity_utilization(
-    candidate: &AccountCandidate,
-    default_concurrency: AccountConcurrency,
-) -> f64 {
-    candidate
-        .account
-        .effective_concurrency(default_concurrency)
+fn capacity_utilization(candidate: &AccountCandidate, context: &AccountSelectionContext) -> f64 {
+    context
+        .concurrency_limit(&candidate.account)
        .limit()
        .map_or(0.0, |limit| {
            f64::from(candidate.signals.in_flight) / f64::from(limit.get())
@@ -929,38 +911,29 @@ fn capacity_utilization(

 fn select_smart_candidate<'a>(
    candidates: &[&'a AccountCandidate],
-    default_concurrency: AccountConcurrency,
-    cursor: u64,
-    config: SmartSchedulingConfig,
-    now: SystemTime,
+    context: &AccountSelectionContext,
 ) -> Option<&'a AccountCandidate> {
    let mut ranked = candidates
        .iter()
-        .map(|candidate| {
-            (
-                *candidate,
-                smart_score(candidate, default_concurrency, config, now),
-            )
-        })
+        .map(|candidate| (*candidate, smart_score(candidate, context)))
        .collect::<Vec<_>>();
    let best_score = ranked
        .iter()
        .map(|(_, score)| *score)
        .max_by(f64::total_cmp)?;
-    ranked.retain(|(_, score)| best_score - score <= config.score_tolerance());
+    ranked.retain(|(_, score)| {
+        best_score - score <= context.policy.smart_scheduling().score_tolerance()
+    });
    // 轮换顺序保持稳定，避免分数轻微交错与 cursor 同步后仍反复命中同一账号
    ranked.sort_unstable_by(|(left, _), (right, _)| left.account.id().cmp(right.account.id()));
-    let index = (cursor % ranked.len() as u64) as usize;
+    let index = (context.round_robin_cursor % ranked.len() as u64) as usize;
    Some(ranked[index].0)
 }

-pub(crate) fn smart_score(
-    candidate: &AccountCandidate,
-    default_concurrency: AccountConcurrency,
-    config: SmartSchedulingConfig,
-    now: SystemTime,
-) -> f64 {
-    let load = 1.0 - capacity_utilization(candidate, default_concurrency).clamp(0.0, 1.0);
+pub(crate) fn smart_score(candidate: &AccountCandidate, context: &AccountSelectionContext) -> f64 {
+    let config = context.policy.smart_scheduling();
+    let now = context.now;
+    let load = 1.0 - capacity_utilization(candidate, context).clamp(0.0, 1.0);
    let quota = candidate
        .signals
        .quota_remaining_rank
@@ -41,17 +41,14 @@ impl TraceContext {
                    "blocker": AccountSelector.scheduling_blocker(candidate, context)
                        .map(|blocker| format!("{blocker:?}")),
                    "inFlight": signals.in_flight,
-                    "concurrencyLimit": candidate.account
-                        .effective_concurrency(context.policy.max_concurrent_per_account()).get(),
+                    "concurrencyLimit": context.concurrency_limit(&candidate.account).get(),
                    "quotaRemainingBasisPoints": signals.quota_remaining_rank,
                    "quotaResetAtUnixMs": signals.quota_reset_at.and_then(|reset|
                        reset.duration_since(std::time::UNIX_EPOCH).ok()
                    ).map(|duration| duration.as_millis()),
                    "failureRateBasisPoints": signals.failure_rate_basis_points,
                    "firstOutputLatencyMs": signals.first_output_latency_ms,
-                    "smartScore": smart.then(|| smart_score(
-                        candidate, context.policy.max_concurrent_per_account(), context.policy.smart_scheduling(), context.now
-                    )),
+                    "smartScore": smart.then(|| smart_score(candidate, context)),
                })
            })
            .collect::<Vec<_>>();
@@ -9,6 +9,14 @@ use crate::{
 use futures::future::BoxFuture;
 use std::{collections::BTreeMap, fmt, num::NonZeroU32, time::Duration};

+/// 账号内独立计数的调度容量池；Provider 决定请求归属，Store 只隔离租约
+#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
+pub enum ProviderConcurrencyPool {
+    #[default]
+    Shared,
+    Reserved,
+}
+
 /// 一个 Provider 的完整可重建调度状态
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub struct ProviderSchedulingState {
@@ -46,6 +54,7 @@ pub struct ProviderSchedulingLeaseRequest {
    account_id: ProviderAccountId,
    credential_revision: CredentialRevision,
    max_concurrent: AccountConcurrency,
+    concurrency_pool: ProviderConcurrencyPool,
    request_interval: Duration,
    deadline: crate::lifecycle::Deadline,
    cancellation: crate::lifecycle::CancellationToken,
@@ -66,6 +75,7 @@ impl ProviderSchedulingLeaseRequest {
            account_id,
            credential_revision,
            max_concurrent: max_concurrent.into(),
+            concurrency_pool: ProviderConcurrencyPool::Shared,
            request_interval,
            deadline: deadline.into(),
            cancellation: crate::lifecycle::CancellationToken::new(),
@@ -78,6 +88,17 @@ impl ProviderSchedulingLeaseRequest {
        self
    }

+    #[must_use]
+    pub const fn with_concurrency_pool(mut self, pool: ProviderConcurrencyPool) -> Self {
+        self.concurrency_pool = pool;
+        self
+    }
+
+    #[must_use]
+    pub const fn concurrency_pool(&self) -> ProviderConcurrencyPool {
+        self.concurrency_pool
+    }
+
    #[must_use]
    pub fn cancellation(&self) -> crate::lifecycle::CancellationToken {
        self.cancellation.clone()
@@ -145,11 +166,13 @@ pub enum ProviderLeaseRequest {
 }

 pub trait ProviderLeasePort: Send + Sync {
+    /// 按目标容量池读取在途数，账号最小请求间隔仍跨池共享
    fn load_state<'a>(
        &'a self,
        client_api_key_id: &'a ClientApiKeyId,
        provider_kind: &'a ProviderKind,
        accounts: &'a [ProviderAccountId],
+        pool: ProviderConcurrencyPool,
    ) -> BoxFuture<'a, Result<ProviderSchedulingState, ProviderStoreError>>;

    fn try_acquire(
@@ -157,7 +180,7 @@ pub trait ProviderLeasePort: Send + Sync {
        request: ProviderLeaseRequest,
    ) -> BoxFuture<'_, Result<ProviderLeaseAcquisition, ProviderStoreError>>;

-    /// 读取指定账号当前的在途请求数；只用于容量熔断的峰值证据，
+    /// 读取指定账号普通池的在途请求数；只用于容量熔断的峰值证据，
    /// 支持租约信号的存储实现覆盖，否则视为不可观测（空映射）
    fn account_in_flight<'a>(
        &'a self,
@@ -4,9 +4,9 @@ mod error;
 pub use error::{ProviderStoreError, ProviderStoreErrorKind};
 mod lease;
 pub use lease::{
-    ProviderLeaseAcquisition, ProviderLeaseGuard, ProviderLeasePort, ProviderLeaseRequest,
-    ProviderRefreshCapacityRequest, ProviderRefreshLeaseRequest, ProviderSchedulingLeaseRequest,
-    ProviderSchedulingState,
+    ProviderConcurrencyPool, ProviderLeaseAcquisition, ProviderLeaseGuard, ProviderLeasePort,
+    ProviderLeaseRequest, ProviderRefreshCapacityRequest, ProviderRefreshLeaseRequest,
+    ProviderSchedulingLeaseRequest, ProviderSchedulingState,
 };
 mod affinity;
 pub use affinity::{
@@ -15,13 +15,12 @@ use std::time::{Duration, SystemTime};
 use serde_json::{Map, Value};

 use gateway_core::account::{
-    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,
+    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,
 };
 use gateway_core::routing::{
    ClientRoutingScope, FrozenAccountScope, ProviderKind, RuntimeAccount, RuntimeAccountDirectory,
@@ -1099,40 +1098,57 @@ fn due_probe_freeze_blocks_scheduling_and_remains_visible_as_rate_limited() {
 }

 #[test]
-fn reserved_concurrency_only_blocks_requests_that_cannot_use_the_reserved_slots() {
+fn reserved_concurrency_uses_its_own_limit_without_reducing_normal_capacity() {
+    let normal = context(RotationStrategy::Smart);
    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());
+    assert!(AccountSelector.select(&candidates, &normal).is_some());
    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);
+    let mut reserved = normal.clone();
+    reserved.reserved_concurrency = 3;
+    let candidates = [candidate_with_concurrency("acct_reserved", 2, 5)];
+    assert!(AccountSelector.select(&candidates, &reserved).is_some());
+    let capacity = AccountSelector
+        .capacity_snapshot(&candidates, &reserved)
+        .unwrap();
+    assert_eq!((capacity.used_slots(), capacity.total_slots()), (2, 3));
+    let full = [candidate_with_concurrency("acct_reserved", 3, 5)];
+    assert!(AccountSelector.select(&full, &reserved).is_none());
+    assert_eq!(AccountSelector.wait_candidates(&full, &reserved).len(), 1);
+    assert!(AccountSelector.select(&full, &normal).is_some());
+
+    // 独立额度可以大于账号普通上限，极值不能回绕成无限并发
+    reserved.reserved_concurrency = u32::MAX;
    assert_eq!(
-        AccountConcurrency::Unlimited.excluding_reserved(2),
-        AccountConcurrency::Unlimited
+        reserved.concurrency_limit(&candidates[0].account).get(),
+        u32::MAX
    );
+    assert_eq!(normal.concurrency_limit(&candidates[0].account).get(), 5);
+}
+
+#[test]
+fn reserved_capacity_stays_bounded_when_normal_concurrency_is_unlimited() {
+    let candidates = [candidate("acct_unlimited", 3, None)];
+    let mut normal = context(RotationStrategy::Smart);
+    normal.policy = AccountSelectionPolicy::new(
+        RotationStrategy::Smart,
+        gateway_core::account::AccountConcurrency::Unlimited,
+        Duration::ZERO,
+    );
+    assert!(AccountSelector.select(&candidates, &normal).is_some());
+    assert!(
+        AccountSelector
+            .capacity_snapshot(&candidates, &normal)
+            .is_none()
+    );
+    let mut reserved = normal;
+    reserved.reserved_concurrency = 3;
+    assert!(AccountSelector.select(&candidates, &reserved).is_none());
+    let capacity = AccountSelector
+        .capacity_snapshot(&candidates, &reserved)
+        .unwrap();
+    assert_eq!((capacity.used_slots(), capacity.total_slots()), (3, 3));
 }
@@ -13,6 +13,31 @@ use gateway_core::concurrency::{

 use super::{candidate_with_concurrency, context, weighted_candidate};

+#[test]
+fn reserved_pool_scoring_and_trace_use_its_capacity_instead_of_the_normal_limit() {
+    let mut context = context(RotationStrategy::Smart);
+    context.reserved_concurrency = 5;
+    context.policy = context.policy.with_smart_scheduling(
+        SmartSchedulingConfig::new([1.0, 0.0, 0.0, 0.0, 0.0, 0.0], false).unwrap(),
+    );
+    let candidates = [
+        candidate_with_concurrency("acct_large_normal", 2, 100),
+        candidate_with_concurrency("acct_small_normal", 1, 1),
+    ];
+    let selection = AccountSelector.select(&candidates, &context).unwrap();
+    assert_eq!(
+        selection.candidate().account.id().as_str(),
+        "acct_small_normal"
+    );
+    let trace = gateway_core::diagnostics::TraceContext::new("req_reserved_scores");
+    trace.account_selection(&candidates, &context, Some(&selection));
+    let snapshot = trace.snapshot().unwrap();
+    let selected = &snapshot["events"][0]["data"]["candidates"][0];
+    assert_eq!(selected["concurrencyLimit"], 5);
+    assert_eq!(selected["inFlight"], 1);
+    assert_eq!(selected["smartScore"], 0.8);
+}
+
 #[test]
 fn smart_config_rejects_invalid_weights_and_incomplete_wire_values() {
    for value in [-0.1, 10.1, 0.01, f64::NAN, f64::INFINITY] {
@@ -12,7 +12,7 @@ use gateway_core::account::{AccountRuntimeSignals, ProviderAccountId};
 use gateway_core::lifecycle::REQUEST_LEASE_TTL;
 use gateway_core::policy::ClientApiKeyId;
 use gateway_core::provider_ports::{
-    ProviderLeaseAcquisition, ProviderLeasePort, ProviderLeaseRequest,
+    ProviderConcurrencyPool, ProviderLeaseAcquisition, ProviderLeasePort, ProviderLeaseRequest,
    ProviderRefreshCapacityRequest, ProviderSchedulingLeaseRequest, ProviderSchedulingState,
    ProviderStoreError, ProviderStoreErrorKind,
 };
@@ -128,6 +128,7 @@ return tostring(cursor - 1)
 pub enum CredentialLeaseScope {
    Provider,
    ProviderAccount,
+    ProviderAccountReserved,
    OAuthRefreshCapacity,
    OAuthRefresh,
    ProviderTask,
@@ -138,11 +139,19 @@ impl CredentialLeaseScope {
        match self {
            Self::Provider => "provider",
            Self::ProviderAccount => "account",
+            Self::ProviderAccountReserved => "account-reserved",
            Self::OAuthRefreshCapacity => "refresh-capacity",
            Self::OAuthRefresh => "refresh",
            Self::ProviderTask => "task",
        }
    }
+
+    const fn scheduling(pool: ProviderConcurrencyPool) -> Self {
+        match pool {
+            ProviderConcurrencyPool::Shared => Self::ProviderAccount,
+            ProviderConcurrencyPool::Reserved => Self::ProviderAccountReserved,
+        }
+    }
 }

 #[derive(Debug, Clone, PartialEq, Eq)]
@@ -334,6 +343,7 @@ pub trait CredentialLeaseRepository: Send + Sync {
        request: &CredentialLeaseRequest,
        grant: &CredentialLeaseGrant,
    ) -> StoreResult<bool>;
+    /// 管理端容量与自适应并发观测只统计普通池，不把审批占用投影为普通并发压力
    async fn credential_runtime_signals(
        &self,
        resource_ids: &[String],
@@ -395,10 +405,16 @@ impl RedisCredentialLeaseRepository {
        let fingerprint = resource_fingerprint("credential lease", &request.resource_id)?;
        let tag = format!("{{{fingerprint}}}");
        let prefix = format!("{}:lease:{}:{tag}", self.namespace, request.scope.as_str());
+        // 两个容量池独立计数，但账号最小请求间隔仍由同一时间戳原子裁决
+        let interval_prefix = if request.scope == CredentialLeaseScope::ProviderAccountReserved {
+            format!("{}:lease:account:{tag}", self.namespace)
+        } else {
+            prefix.clone()
+        };
        Ok([
            format!("{prefix}:active"),
            format!("{prefix}:fence"),
-            format!("{prefix}:last-started"),
+            format!("{interval_prefix}:last-started"),
        ])
    }

@@ -472,9 +488,29 @@ impl RedisCredentialLeaseRepository {
        })
    }

-    async fn load_signal(&self, resource_id: String) -> StoreResult<CredentialRuntimeSignal> {
+    async fn runtime_signals(
+        &self,
+        resource_ids: &[String],
+        pool: ProviderConcurrencyPool,
+    ) -> StoreResult<Vec<CredentialRuntimeSignal>> {
+        for resource_id in resource_ids {
+            require_nonempty("credential runtime signal", "resource_id", resource_id)?;
+        }
+        let mut signals = Vec::with_capacity(resource_ids.len());
+        for ids in resource_ids.chunks(super::ACCOUNT_STATE_READ_CONCURRENCY) {
+            let batch = join_all(ids.iter().cloned().map(|id| self.load_signal(id, pool))).await;
+            signals.extend(batch.into_iter().collect::<StoreResult<Vec<_>>>()?);
+        }
+        Ok(signals)
+    }
+
+    async fn load_signal(
+        &self,
+        resource_id: String,
+        pool: ProviderConcurrencyPool,
+    ) -> StoreResult<CredentialRuntimeSignal> {
        let request = CredentialLeaseRequest {
-            scope: CredentialLeaseScope::ProviderAccount,
+            scope: CredentialLeaseScope::scheduling(pool),
            resource_id: resource_id.clone(),
            owner_id: "signal-reader".to_owned(),
            ttl: Duration::from_secs(1),
@@ -539,6 +575,7 @@ impl RedisProviderLeaseCoordinator {
    async fn load_signals(
        &self,
        accounts: &[ProviderAccountId],
+        pool: ProviderConcurrencyPool,
    ) -> Result<BTreeMap<ProviderAccountId, AccountRuntimeSignals>, ProviderStoreError> {
        let ids = accounts
            .iter()
@@ -546,7 +583,7 @@ impl RedisProviderLeaseCoordinator {
            .collect::<Vec<_>>();
        let signals = self
            .repository
-            .credential_runtime_signals(&ids)
+            .runtime_signals(&ids, pool)
            .await
            .map_err(|source| crate::provider_unavailable("load scheduling signals", source))?;
        signals
@@ -588,7 +625,7 @@ impl RedisProviderLeaseCoordinator {
        let acquisition = self
            .repository
            .try_acquire_bounded_lease(&CredentialBoundedLeaseRequest {
-                scope: CredentialLeaseScope::ProviderAccount,
+                scope: CredentialLeaseScope::scheduling(request.concurrency_pool()),
                resource_id: request.account_id().as_str().to_owned(),
                owner_id: self.owner_id("request"),
                max_concurrent: request.max_concurrent().get(),
@@ -646,9 +683,10 @@ impl ProviderLeasePort for RedisProviderLeaseCoordinator {
        client_api_key_id: &'a ClientApiKeyId,
        provider_kind: &'a ProviderKind,
        accounts: &'a [ProviderAccountId],
+        pool: ProviderConcurrencyPool,
    ) -> futures::future::BoxFuture<'a, Result<ProviderSchedulingState, ProviderStoreError>> {
        Box::pin(async move {
-            let signals = self.load_signals(accounts).await?;
+            let signals = self.load_signals(accounts, pool).await?;
            let round_robin_cursor = self
                .next_scheduling_cursor(client_api_key_id, provider_kind)
                .await?;
@@ -782,15 +820,8 @@ impl CredentialLeaseRepository for RedisCredentialLeaseRepository {
        &self,
        resource_ids: &[String],
    ) -> StoreResult<Vec<CredentialRuntimeSignal>> {
-        for resource_id in resource_ids {
-            require_nonempty("credential runtime signal", "resource_id", resource_id)?;
-        }
-        let mut signals = Vec::with_capacity(resource_ids.len());
-        for ids in resource_ids.chunks(super::ACCOUNT_STATE_READ_CONCURRENCY) {
-            let batch = join_all(ids.iter().cloned().map(|id| self.load_signal(id))).await;
-            signals.extend(batch.into_iter().collect::<StoreResult<Vec<_>>>()?);
-        }
-        Ok(signals)
+        self.runtime_signals(resource_ids, ProviderConcurrencyPool::Shared)
+            .await
    }

    async fn try_acquire_bounded_lease(
@@ -34,95 +34,223 @@ fn only_account_scheduling_leases_allow_unlimited_concurrency() {
 }

 #[tokio::test]
-async fn maintained_scheduling_lease_survives_initial_ttl_and_releases_on_drop() {
+async fn reserved_and_normal_leases_enforce_independent_atomic_limits_and_release() {
    let Some((repository, _, _)) = repository().await else {
        return;
    };
-    let mut request = scheduling_request("acct_long_request", "worker-long", 1, Duration::ZERO);
-    request.ttl = Duration::from_secs(1);
-    let lease = acquired(
+    let normal = scheduling_request("acct_independent", "normal", 2, Duration::ZERO);
+    let reserved = CredentialBoundedLeaseRequest {
+        scope: CredentialLeaseScope::ProviderAccountReserved,
+        max_concurrent: 3,
+        ..normal.clone()
+    };
+    let attempts = futures::future::join_all((0..20).map(|index| {
+        let request = if index % 2 == 0 {
+            normal.clone()
+        } else {
+            reserved.clone()
+        };
+        let repository = &repository;
+        async move {
+            (
+                request.scope,
+                repository
+                    .try_acquire_bounded_lease(&request)
+                    .await
+                    .unwrap(),
+            )
+        }
+    }))
+    .await;
+    let mut normal_leases = Vec::new();
+    let mut reserved_leases = Vec::new();
+    for (scope, acquisition) in attempts {
+        if let CredentialBoundedLeaseAcquisition::Acquired(lease) = acquisition {
+            if scope == CredentialLeaseScope::ProviderAccount {
+                normal_leases.push(lease);
+            } else {
+                reserved_leases.push(lease);
+            }
+        }
+    }
+    assert_eq!((normal_leases.len(), reserved_leases.len()), (2, 3));
+    assert_eq!(
        repository
-            .try_acquire_bounded_lease(&request)
+            .credential_runtime_signals(std::slice::from_ref(&normal.resource_id))
+            .await
+            .unwrap()[0]
+            .in_flight,
+        2
+    );
+
+    reserved_leases.pop().unwrap().release().await.unwrap();
+    assert!(matches!(
+        repository.try_acquire_bounded_lease(&normal).await.unwrap(),
+        CredentialBoundedLeaseAcquisition::Busy { .. }
+    ));
+    reserved_leases.push(acquired(
+        repository
+            .try_acquire_bounded_lease(&reserved)
            .await
            .unwrap(),
-    );
-    let cancellation = gateway_core::lifecycle::CancellationToken::new();
-    let lease = lease
-        .maintain(Default::default(), cancellation.clone())
-        .unwrap();
-    tokio::time::sleep(Duration::from_millis(1_100)).await;
-    assert!(!cancellation.is_cancelled());
+    ));
+    normal_leases.pop().unwrap().release().await.unwrap();
    assert!(matches!(
        repository
-            .try_acquire_bounded_lease(&request)
+            .try_acquire_bounded_lease(&reserved)
            .await
            .unwrap(),
        CredentialBoundedLeaseAcquisition::Busy { .. }
    ));
-    drop(lease);
-    tokio::time::timeout(Duration::from_secs(2), async {
-        loop {
-            if let CredentialBoundedLeaseAcquisition::Acquired(replacement) = repository
+    normal_leases.push(acquired(
+        repository.try_acquire_bounded_lease(&normal).await.unwrap(),
+    ));
+    for lease in normal_leases.into_iter().chain(reserved_leases) {
+        lease.release().await.unwrap();
+    }
+    assert_eq!(
+        repository
+            .credential_runtime_signals(std::slice::from_ref(&normal.resource_id))
+            .await
+            .unwrap()[0]
+            .in_flight,
+        0
+    );
+}
+
+#[tokio::test]
+async fn independent_capacity_pools_preserve_the_shared_account_request_interval() {
+    let Some((repository, _, _)) = repository().await else {
+        return;
+    };
+    for first_reserved in [false, true] {
+        let normal = scheduling_request(
+            &format!("acct_interval_{first_reserved}"),
+            "worker",
+            2,
+            Duration::from_secs(30),
+        );
+        let reserved = CredentialBoundedLeaseRequest {
+            scope: CredentialLeaseScope::ProviderAccountReserved,
+            ..normal.clone()
+        };
+        let (first, second) = if first_reserved {
+            (&reserved, &normal)
+        } else {
+            (&normal, &reserved)
+        };
+        acquired(repository.try_acquire_bounded_lease(first).await.unwrap())
+            .release()
+            .await
+            .unwrap();
+        assert!(
+            matches!(repository.try_acquire_bounded_lease(second).await.unwrap(), CredentialBoundedLeaseAcquisition::Busy { retry_after: Some(value) } if value > Duration::ZERO)
+        );
+    }
+}
+
+#[tokio::test]
+async fn maintained_scheduling_lease_survives_initial_ttl_and_releases_on_drop() {
+    for scope in [
+        CredentialLeaseScope::ProviderAccount,
+        CredentialLeaseScope::ProviderAccountReserved,
+    ] {
+        let Some((repository, _, _)) = repository().await else {
+            return;
+        };
+        let mut request = scheduling_request("acct_long_request", "worker-long", 1, Duration::ZERO);
+        request.scope = scope;
+        request.ttl = Duration::from_secs(1);
+        let lease = acquired(
+            repository
                .try_acquire_bounded_lease(&request)
                .await
-                .unwrap()
-            {
-                replacement.release().await.unwrap();
-                break;
+                .unwrap(),
+        );
+        let cancellation = gateway_core::lifecycle::CancellationToken::new();
+        let lease = lease
+            .maintain(Default::default(), cancellation.clone())
+            .unwrap();
+        tokio::time::sleep(Duration::from_millis(1_100)).await;
+        assert!(!cancellation.is_cancelled());
+        assert!(matches!(
+            repository
+                .try_acquire_bounded_lease(&request)
+                .await
+                .unwrap(),
+            CredentialBoundedLeaseAcquisition::Busy { .. }
+        ));
+        drop(lease);
+        tokio::time::timeout(Duration::from_secs(2), async {
+            loop {
+                if let CredentialBoundedLeaseAcquisition::Acquired(replacement) = repository
+                    .try_acquire_bounded_lease(&request)
+                    .await
+                    .unwrap()
+                {
+                    replacement.release().await.unwrap();
+                    break;
+                }
+                tokio::time::sleep(Duration::from_millis(10)).await;
            }
-            tokio::time::sleep(Duration::from_millis(10)).await;
-        }
-    })
-    .await
-    .expect("dropped request releases its slot");
+        })
+        .await
+        .expect("dropped request releases its slot");
+    }
 }

 #[tokio::test]
 async fn maintained_scheduling_lease_does_not_resurrect_a_lost_slot() {
-    let Some((repository, _, _)) = repository().await else {
-        return;
-    };
-    let request = scheduling_request("acct_lost_request", "worker-lost", 1, Duration::ZERO);
-    let lease = acquired(
+    for scope in [
+        CredentialLeaseScope::ProviderAccount,
+        CredentialLeaseScope::ProviderAccountReserved,
+    ] {
+        let Some((repository, _, _)) = repository().await else {
+            return;
+        };
+        let mut request = scheduling_request("acct_lost_request", "worker-lost", 1, Duration::ZERO);
+        request.scope = scope;
+        let lease = acquired(
+            repository
+                .try_acquire_bounded_lease(&request)
+                .await
+                .unwrap(),
+        );
+        let grant = lease.grant().unwrap().clone();
+        let raw = CredentialLeaseRequest {
+            scope: request.scope,
+            resource_id: request.resource_id.clone(),
+            owner_id: request.owner_id.clone(),
+            ttl: request.ttl,
+        };
        repository
-            .try_acquire_bounded_lease(&request)
+            .release_credential_lease(&raw, &grant)
            .await
-            .unwrap(),
-    );
-    let grant = lease.grant().unwrap().clone();
-    let raw = CredentialLeaseRequest {
-        scope: request.scope,
-        resource_id: request.resource_id.clone(),
-        owner_id: request.owner_id.clone(),
-        ttl: request.ttl,
-    };
-    repository
-        .release_credential_lease(&raw, &grant)
-        .await
-        .unwrap();
-    let cancellation = gateway_core::lifecycle::CancellationToken::new();
-    let lease = lease
-        .maintain(Default::default(), cancellation.clone())
-        .unwrap();
-    tokio::time::timeout(Duration::from_secs(2), cancellation.cancelled())
-        .await
-        .expect("lost lease cancels the request");
-    let replacement = acquired(
-        repository
-            .try_acquire_bounded_lease(&request)
+            .unwrap();
+        let cancellation = gateway_core::lifecycle::CancellationToken::new();
+        let lease = lease
+            .maintain(Default::default(), cancellation.clone())
+            .unwrap();
+        tokio::time::timeout(Duration::from_secs(2), cancellation.cancelled())
            .await
-            .unwrap(),
-    );
-    drop(lease);
-    tokio::time::sleep(Duration::from_millis(20)).await;
-    assert!(matches!(
-        repository
-            .try_acquire_bounded_lease(&request)
-            .await
-            .unwrap(),
-        CredentialBoundedLeaseAcquisition::Busy { .. }
-    ));
-    replacement.release().await.unwrap();
+            .expect("lost lease cancels the request");
+        let replacement = acquired(
+            repository
+                .try_acquire_bounded_lease(&request)
+                .await
+                .unwrap(),
+        );
+        drop(lease);
+        tokio::time::sleep(Duration::from_millis(20)).await;
+        assert!(matches!(
+            repository
+                .try_acquire_bounded_lease(&request)
+                .await
+                .unwrap(),
+            CredentialBoundedLeaseAcquisition::Busy { .. }
+        ));
+        replacement.release().await.unwrap();
+    }
 }

 #[tokio::test]
@@ -16,10 +16,10 @@ use gateway_core::account::{
 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,
-    ProviderSchedulingLeaseRequest, ProviderSessionAffinityKey, ProviderSessionAffinityPort,
-    ProviderSessionBinding, ProviderSessionExclusionPort, ProviderSessionExclusions,
-    ProviderStoreError,
+    ProviderConcurrencyPool, ProviderLeaseAcquisition, ProviderLeaseGuard, ProviderLeasePort,
+    ProviderLeaseRequest, ProviderSchedulingLeaseRequest, ProviderSessionAffinityKey,
+    ProviderSessionAffinityPort, ProviderSessionBinding, ProviderSessionExclusionPort,
+    ProviderSessionExclusions, ProviderStoreError,
 };
 use gateway_core::routing::ProviderKind;
 use secrecy::ExposeSecret;
@@ -114,7 +114,7 @@ struct CredentialSelectionInput<'a> {
    attempt: &'a AttemptContext,
    session_affinity_key: Option<&'a ProviderSessionAffinityKey>,
    session_affinity_observation: Option<&'a CodexSessionAffinity>,
-    /// Codex Guardian 自动审批请求；仅在配置了预留名额时获得预留与队列优先
+    /// Codex Guardian 自动审批请求；仅在配置独立额度时使用审批容量池
    guardian: bool,
 }

@@ -126,6 +126,7 @@ pub(crate) struct CodexCyberPolicyScope {

 pub struct CodexCredentialSelector {
    waiting: ConcurrencyWaitQueue<ProviderAccountId>,
+    reserved_waiting: ConcurrencyWaitQueue<ProviderAccountId>,
    provider_kind: ProviderKind,
    repository: CodexCredentialRepository,
    leases: Arc<dyn ProviderLeasePort>,
@@ -285,6 +286,7 @@ impl CodexCredentialSelector {
            cookie_policy,
            risk_recovery: Mutex::new(HashMap::new()),
            waiting: ConcurrencyWaitQueue::default(),
+            reserved_waiting: ConcurrencyWaitQueue::default(),
            account_feedback,
        }
    }
@@ -465,15 +467,24 @@ impl CodexCredentialSelector {
                queue_policy.timeout = Duration::from_secs(30);
            }
        }
-        // 预留名额只对普通请求生效；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 reserved_concurrency = if prioritized { reserve } else { 0 };
+        let concurrency_pool = if prioritized {
+            ProviderConcurrencyPool::Reserved
+        } else {
+            ProviderConcurrencyPool::Shared
+        };
        let mut waiting = CapacityWait::new(
-            &self.waiting,
+            if prioritized {
+                &self.reserved_waiting
+            } else {
+                &self.waiting
+            },
            queue_policy,
            request.attempt.deadline().at(),
            request.attempt.concurrency_wait_budget(),
@@ -619,6 +630,7 @@ impl CodexCredentialSelector {
                    request.attempt.client_api_key_ref(),
                    &self.provider_kind,
                    &account_ids,
+                    concurrency_pool,
                )
                .await?;
            let round_robin_cursor = scheduling.round_robin_cursor();
@@ -877,6 +889,7 @@ impl CodexCredentialSelector {
                            policy.request_interval(),
                            request.attempt.deadline(),
                        )
+                        .with_concurrency_pool(concurrency_pool)
                        .with_cancellation(request.attempt.cancellation().clone()),
                    ))
                    .await?
@@ -2545,6 +2545,7 @@ mod errors {
            _: &'a ClientApiKeyId,
            _: &'a ProviderKind,
            _: &'a [ProviderAccountId],
+            _pool: gateway_core::provider_ports::ProviderConcurrencyPool,
        ) -> BoxFuture<'a, Result<ProviderSchedulingState, ProviderStoreError>> {
            panic!("manual refresh does not use scheduling leases")
        }
@@ -124,6 +124,7 @@ impl ProviderLeasePort for RefreshLeases {
        _: &'a ClientApiKeyId,
        _: &'a ProviderKind,
        _: &'a [ProviderAccountId],
+        _pool: gateway_core::provider_ports::ProviderConcurrencyPool,
    ) -> BoxFuture<'a, Result<ProviderSchedulingState, ProviderStoreError>> {
        Box::pin(async { panic!("scheduled credential refresh does not load scheduling state") })
    }
@@ -1,4 +1,4 @@
-//! 验证 OpenAI 请求等待账号容量与 Guardian 预留槽位
+//! 验证 OpenAI 请求等待账号容量与 Guardian 独立容量池

 use futures::FutureExt;
 use gateway_core::account::AccountRuntimeSignals;
@@ -123,7 +123,11 @@ async fn queued_root_and_new_child_both_observe_the_same_account_capacity() {
    server.verify().await;
 }

-fn unqueued_context(request_id: &str, reserved: u32) -> AttemptContext {
+fn capacity_context(
+    request_id: &str,
+    reserved: u32,
+    queue: ConcurrencyQueuePolicy,
+) -> AttemptContext {
    AttemptContext::new(
        RequestAttemptContext::new(
            ModelRequestId::new(request_id).unwrap(),
@@ -131,7 +135,9 @@ fn unqueued_context(request_id: &str, reserved: u32) -> AttemptContext {
        ),
        NonZeroU32::new(1).unwrap(),
        SystemTime::now() + Duration::from_secs(5),
-        account_policy().with_openai_guardian_reserved_concurrency(reserved),
+        account_policy()
+            .with_openai_guardian_reserved_concurrency(reserved)
+            .with_queue(queue),
        AccountAttemptContext::new(BTreeSet::new(), None, None)
            .with_account_scope(contract_account_scope()),
        None,
@@ -158,7 +164,9 @@ fn subagent_operation(subagent: Option<&str>) -> Operation {
 }

 #[tokio::test]
-async fn guardian_requests_can_use_the_reserved_slot_that_normal_requests_cannot() {
+async fn guardian_and_normal_requests_use_independent_capacity_and_observations() {
+    use gateway_core::provider_ports::ProviderConcurrencyPool::{Reserved, Shared};
+
    let store = Arc::new(MemoryAccountStore::default());
    create_account(&store, "acct_guardian").await;
    let account = ProviderAccountId::new("acct_guardian").unwrap();
@@ -171,90 +179,188 @@ async fn guardian_requests_can_use_the_reserved_slot_that_normal_requests_cannot
                .insert_header("content-type", "text/event-stream")
                .set_body_string(CAPTURE_COMPLETED_SSE),
        )
-        .expect(3)
+        .expect(4)
        .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| {
+    let signals = |in_flight| AccountRuntimeSignals {
+        in_flight,
+        last_started_at: None,
+        quota_reset_at: None,
+        quota_remaining_rank: None,
+        cooldown: None,
+        failure_rate_basis_points: None,
+        first_output_latency_ms: None,
+    };
+    // 普通上限为 2，审批额度独立取配置值；关闭后新审批请求回到普通池
+    for (id, normal_count, reserved_count, reserve, subagent, expected) in [
+        (
+            "req_large_reserve_normal",
+            1,
+            10,
+            10,
+            None,
+            Some((Shared, 2)),
+        ),
+        ("req_normal_full", 2, 0, 1, None, None),
+        (
+            "req_guardian_independent",
+            2,
+            0,
+            1,
+            Some("guardian"),
+            Some((Reserved, 1)),
+        ),
+        ("req_guardian_full", 0, 1, 1, Some("guardian"), None),
+        (
+            "req_normal_independent",
+            1,
+            1,
+            1,
+            Some("collab_spawn"),
+            Some((Shared, 2)),
+        ),
+        (
+            "req_guardian_disabled",
+            1,
+            1,
+            0,
+            Some("guardian"),
+            Some((Shared, 2)),
+        ),
+    ] {
        leases
-            .requests
+            .signals
            .lock()
            .unwrap()
-            .iter()
-            .map(|request| request.max_concurrent().get())
-            .collect::<Vec<_>>()
-    };
-    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
+            .insert(account.clone(), signals(normal_count));
+        leases
+            .reserved_signals
+            .lock()
+            .unwrap()
+            .insert(account.clone(), signals(reserved_count));
+        leases.requests.lock().unwrap().clear();
+        let result = provider
            .clone()
            .execute(
-                planned_request("openai", subagent_operation(None)),
-                unqueued_context("req_guardian_normal_busy", 1),
+                planned_request("openai", subagent_operation(subagent)),
+                capacity_context(id, reserve, ConcurrencyQueuePolicy::default()),
            )
+            .await;
+        let Some((pool, limit)) = expected else {
+            assert!(result.is_err(), "{id} should wait for its own capacity");
+            assert!(leases.requests.lock().unwrap().is_empty());
+            continue;
+        };
+        let mut stream = result.unwrap_or_else(|error| panic!("{id}: {error:?}"));
+        let capacity = stream
+            .metadata()
+            .selection_observation()
+            .unwrap()
+            .capacity()
+            .unwrap();
+        let count = if pool == Shared {
+            normal_count
+        } else {
+            reserved_count
+        };
+        assert_eq!(
+            (capacity.used_slots(), capacity.total_slots()),
+            (u64::from(count) + 1, limit)
+        );
+        while let Some(event) = stream.next().await {
+            event.unwrap();
+        }
+        drop(stream);
+        let requests = leases.requests.lock().unwrap();
+        assert_eq!(requests.len(), 1);
+        assert_eq!(requests[0].concurrency_pool(), pool);
+        assert_eq!(u64::from(requests[0].max_concurrent().get()), limit);
+    }
+    server.verify().await;
+}
+
+#[tokio::test]
+async fn a_full_capacity_pool_does_not_block_the_other_pools_queue() {
+    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(4)
+        .mount(&server)
+        .await;
+    let provider = provider_with_affinity_and_base_url_and_leases(
+        &store,
+        Arc::new(MemorySessionAffinity::default()),
+        server.uri(),
+        leases.clone(),
+    );
+    let queue = ConcurrencyQueuePolicy {
+        max_waiting: 1,
+        timeout: Duration::from_secs(2),
+    };
+    for guardian_waits in [false, true] {
+        let blocked_signals = if guardian_waits {
+            &leases.reserved_signals
+        } else {
+            &leases.signals
+        };
+        blocked_signals.lock().unwrap().insert(
+            account.clone(),
+            AccountRuntimeSignals {
+                in_flight: if guardian_waits { 1 } else { 2 },
+                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 mut waiting = Box::pin(provider.clone().execute(
+            planned_request(
+                "openai",
+                subagent_operation(guardian_waits.then_some("guardian")),
+            ),
+            capacity_context("req_pool_waiter", 1, queue),
+        ));
+        assert!(waiting.as_mut().now_or_never().is_none());
+
+        let mut available = provider
+            .clone()
+            .execute(
+                planned_request(
+                    "openai",
+                    subagent_operation((!guardian_waits).then_some("guardian")),
+                ),
+                capacity_context("req_other_pool", 1, queue),
+            )
            .await
-            .is_err()
-    );
-    assert_eq!(limits(&leases), [1]);
+            .expect("the other pool has independent capacity and queue position");
+        while let Some(event) = available.next().await {
+            event.unwrap();
+        }
+        drop(available);

-    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();
+        blocked_signals.lock().unwrap().clear();
+        let mut resumed = waiting.await.expect("own pool capacity was released");
+        while let Some(event) = resumed.next().await {
+            event.unwrap();
+        }
+        drop(resumed);
    }
-    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;
 }
@@ -263,7 +263,7 @@ async fn selected_adapter_connection(

 #[tokio::test]
 async fn guardian_reservation_survives_upstream_adapters_and_metadata_precedence() {
-    // 上限 2、已有 1 个在途请求：预留启用时只有 Guardian 能继续取得租约
+    // 普通池上限 2 且已满，审批独立池仍有空位；关闭独立池后审批也受普通容量约束
    for (subagent, turn_kind, reserved, allowed) in [
        (Some("guardian"), None, 1, true),
        (None, Some("guardian"), 1, true),
@@ -271,7 +271,8 @@ async fn guardian_reservation_survives_upstream_adapters_and_metadata_precedence
        (Some("guardian"), Some("collab_spawn"), 1, false),
        (Some("collab_spawn"), None, 1, false),
        (None, None, 1, false),
-        (None, None, 0, true),
+        (None, None, 0, false),
+        (Some("guardian"), None, 0, false),
    ] {
        let store = Arc::new(MemoryAccountStore::default());
        create_account(&store, "acct_provider_contract").await;
@@ -279,7 +280,7 @@ async fn guardian_reservation_survives_upstream_adapters_and_metadata_precedence
        leases.signals.lock().unwrap().insert(
            ProviderAccountId::new("acct_provider_contract").unwrap(),
            AccountRuntimeSignals {
-                in_flight: 1,
+                in_flight: 2,
                last_started_at: None,
                quota_reset_at: None,
                quota_remaining_rank: None,
@@ -355,7 +356,11 @@ async fn guardian_reservation_survives_upstream_adapters_and_metadata_precedence
            drop(receiver.await.unwrap());
            let requests = leases.requests.lock().unwrap();
            assert_eq!(requests.len(), 1);
-            assert_eq!(requests[0].max_concurrent().get(), 2);
+            assert_eq!(requests[0].max_concurrent().get(), 1);
+            assert_eq!(
+                requests[0].concurrency_pool(),
+                gateway_core::provider_ports::ProviderConcurrencyPool::Reserved
+            );
        } else {
            assert!(leases.requests.lock().unwrap().is_empty());
        }
@@ -673,6 +673,7 @@ pub(crate) struct TestLeaseCoordinator {
    pub(crate) busy: Mutex<bool>,
    pub(crate) busy_accounts: Mutex<BTreeSet<ProviderAccountId>>,
    pub(crate) signals: Mutex<BTreeMap<ProviderAccountId, AccountRuntimeSignals>>,
+    pub(crate) reserved_signals: Mutex<BTreeMap<ProviderAccountId, AccountRuntimeSignals>>,
    round_robin_cursor: Mutex<u64>,
 }

@@ -682,9 +683,17 @@ impl ProviderLeasePort for TestLeaseCoordinator {
        _client_api_key_id: &'a ClientApiKeyId,
        _provider_kind: &'a ProviderKind,
        accounts: &'a [ProviderAccountId],
+        pool: gateway_core::provider_ports::ProviderConcurrencyPool,
    ) -> BoxFuture<'a, Result<ProviderSchedulingState, ProviderStoreError>> {
        Box::pin(async move {
-            let overrides = self.signals.lock().expect("scheduling signals lock");
+            let overrides = match pool {
+                gateway_core::provider_ports::ProviderConcurrencyPool::Shared => &self.signals,
+                gateway_core::provider_ports::ProviderConcurrencyPool::Reserved => {
+                    &self.reserved_signals
+                }
+            }
+            .lock()
+            .expect("scheduling signals lock");
            let signals = accounts
                .iter()
                .map(|account| {
@@ -191,6 +191,7 @@ impl GrokAccountSessionSelector {
                    request.client_api_key_id(),
                    &self.provider_kind,
                    &account_ids,
+                    gateway_core::provider_ports::ProviderConcurrencyPool::Shared,
                )
                .await
                .map_err(|source| {
@@ -715,6 +715,7 @@ impl ProviderLeasePort for TestLeases {
        _: &'a gateway_core::policy::ClientApiKeyId,
        _: &'a gateway_core::routing::ProviderKind,
        accounts: &'a [ProviderAccountId],
+        _pool: gateway_core::provider_ports::ProviderConcurrencyPool,
    ) -> BoxFuture<
        'a,
        Result<gateway_core::provider_ports::ProviderSchedulingState, ProviderStoreError>,
@@ -1255,6 +1256,7 @@ mod errors {
            _: &'a ClientApiKeyId,
            _: &'a ProviderKind,
            _: &'a [ProviderAccountId],
+            _pool: gateway_core::provider_ports::ProviderConcurrencyPool,
        ) -> BoxFuture<
            'a,
            Result<gateway_core::provider_ports::ProviderSchedulingState, ProviderStoreError>,
@@ -265,6 +265,7 @@ impl ProviderLeasePort for TestRefreshLeases {
        _: &'a gateway_core::policy::ClientApiKeyId,
        _: &'a gateway_core::routing::ProviderKind,
        _: &'a [gateway_core::account::ProviderAccountId],
+        _pool: gateway_core::provider_ports::ProviderConcurrencyPool,
    ) -> futures::future::BoxFuture<
        'a,
        Result<gateway_core::provider_ports::ProviderSchedulingState, ProviderStoreError>,
@@ -59,6 +59,7 @@ impl ProviderLeasePort for SchedulingCoordinator {
        _: &'a ClientApiKeyId,
        _: &'a gateway_core::routing::ProviderKind,
        _: &'a [ProviderAccountId],
+        _pool: gateway_core::provider_ports::ProviderConcurrencyPool,
    ) -> futures::future::BoxFuture<'a, Result<ProviderSchedulingState, ProviderStoreError>> {
        Box::pin(async move {
            Ok(ProviderSchedulingState::new(
@@ -1839,6 +1839,7 @@ impl ProviderLeasePort for DiagnosticLeasePort {
        _: &'a ClientApiKeyId,
        _: &'a ProviderKind,
        account_ids: &'a [gateway_core::account::ProviderAccountId],
+        _pool: gateway_core::provider_ports::ProviderConcurrencyPool,
    ) -> futures::future::BoxFuture<'a, Result<ProviderSchedulingState, ProviderStoreError>> {
        Box::pin(async move {
            Ok(ProviderSchedulingState::new(
@@ -644,7 +644,7 @@ config 返回 `{ name, plaintextKey }`，仅读取服务端会话绑定的当前
 - `sortBy`: `email`、`status`、`planType`、`usage`、`lastUsedAt`、`expiresAt`
 - `sortDirection`: `asc`、`desc`

-账号视图的 `capacity` 返回查询时的网关并发容量：`usedSlots` 是正在执行的请求占用数，不含排队请求，
+账号视图的 `capacity` 返回查询时的普通并发容量：`usedSlots` 不含独立审批池和排队请求，
 读取运行态失败时为 `null`；`totalSlots` 是应用账号独立配置或全局默认值后的上限，`null` 表示不限。
 该上限不代表上游实际允许的并发数

@@ -1177,7 +1177,7 @@ HTTP 请求头及新建 WS 的握手提示按当时的最终出站档位构造
 `providerCounts` 和 `clientKeyCount`。查询分组成员使用账号列表的 `groupId` 筛选，
 不提供独立的分组成员路由；账号的 Provider 不代表整个分组的 Provider。
 `capacity.totalSlots` 为 `number | null`：`null` 表示可用成员中存在继承无限并发的账号，`0` 表示没有可用槽位。
-`capacity.usedSlots` 继续返回实际在途数；Redis 不可用时为 `null`
+分组容量只统计普通池，`capacity.usedSlots` 返回其实际在途数；Redis 不可用时为 `null`

 分组费用按请求执行时实际服务账号的分组快照归属，不按 Client Key 绑定的分组分摊。
 账号属于多个组时，各组均包含该请求费用；之后调整账号分组不重写历史归属
@@ -1361,12 +1361,13 @@ OpenAI 严格亲和下后代线程的会话账号等待不随普通账号排队
 切换账号或内部重试不重新计时；若插件设置了请求总时限，等待也计入该时限。排队超时不用于中断已开始的上游生成。
 设置更新请求须包含这三个字段，新请求使用更新后的快照

-`openaiGuardianReservedConcurrency`（默认 0，取值 0～4,294,967,295）为 Codex Guardian 自动审批保留账号并发，保存后对新请求生效。
+`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 之后、全部普通等待者之前，不受单账号排队上限约束，
-仍受总等待容量与等待时限约束。不限并发的账号和关闭排队时的其余行为不变。
-设置更新请求须包含该字段
+普通请求保留账号原上限 L，Guardian 单独计数并最多运行 R 个，两类请求互不占用名额，也不借用对方空位。
+例如 L 为 10、R 为 3 时，可以同时运行 10 个普通请求和 3 个审批请求；普通并发不限时，审批仍受 R 限制。
+设为 0 时关闭独立额度，Guardian 使用普通并发。开启账号排队后，审批请求单独按 FIFO 等待，
+不受普通队列位置或单账号排队上限约束，仍受自身队列总等待容量与等待时限约束。
+两类请求仍遵守账号最小请求间隔、可用性和 Client Key 限额。设置更新请求须包含该字段

 `openaiAccountAffinity` 控制 OpenAI 账号亲和，默认 `strict`（严格），已有保存的模式保持不变：

@@ -1710,7 +1711,7 @@ errorCode, errorMessage, startedAt, completedAt, expiresAt, createdAt, updatedAt

 ### Dashboard 容量与账号用量

-Dashboard 的 `capacityInfo.maxConcurrentPerAccount` 为默认账号并发上限，`0` 表示不限制。
+Dashboard 的 `capacityInfo` 统计普通容量，不含独立审批池；`maxConcurrentPerAccount` 为默认账号并发上限，`0` 表示不限制。
 `capacityInfo.totalSlots` 为 `number | null`；可用账号池含无限并发账号时为 `null`，此时 `availableSlots` 也为 `null`。
 `usedSlots` 仍表示实际在途数，Redis 不可用时为 `null`；没有可用账号时 `totalSlots` 为 `0`

@@ -657,21 +657,23 @@ Key 的 RPM 在成功准入时才计数，金额限制在入队前及成功准
 切换等待层、账号或 Provider 不重置等待时限；没有发生排队时不启动该计时，也不据此中断已开始的上游生成。
 队列满/超时属于本地容量拒绝，不作为上游限流反馈或 Provider 故障熔断证据

-#### Guardian 审批优先
+#### Guardian 独立并发

-Guardian 预留值由运行设置持久化并随调度快照冻结，每个请求沿用开始时的设置。
-Core 传递该值并拥有通用的并发上限与等待顺序，OpenAI Provider 负责识别请求类型并决定预留与优先级。
+Guardian 独立额度由运行设置持久化并随调度快照冻结，每个请求沿用开始时的设置。
+Core 拥有容量池、上限与等待合同，OpenAI Provider 负责识别请求类型并选择普通池或审批池。

 - OpenAI Provider 在选号前通过统一协议解析识别 `guardian`；原生发送与插件上游适配器使用相同分类。
-  请求头形成的协议上下文与正文 metadata 沿用编码器的既有优先顺序；其他协议不得仅凭同名字段获得该优先级。
+  请求头形成的协议上下文与正文 metadata 沿用编码器的既有优先顺序；其他协议不得仅凭同名字段使用审批池。
  适配器不因此继承原生会话亲和、续接或 WebSocket 约束，选号后的 attempt 中间件也不追溯改变已取得的租约。
- `openaiGuardianReservedConcurrency = R > 0` 时，有限账号上限 L 对普通请求为 `max(L - R, 1)`，
-  Guardian 可用满 L，资格判断、策略投影、容量观测和原子租约共用同一请求级上限，数值边界见 [运行设置 API](api.md#8-运行设置)
- 开启账号排队时，Guardian 在普通等待者之前、已有 Guardian 之后；同级 FIFO，不抢占已运行请求。
-  Guardian 可越过单账号等待数量上限，但仍受总等待容量、共享等待预算与截止时刻约束；取消及超时必须释放等待位置。
- 配置为 0 时不启用预留或优先级；账号排队关闭时不引入隐式队列。Key 层准入仍按原有限额与排队规则执行。
+- `openaiGuardianReservedConcurrency = R > 0` 时，普通池使用账号原上限，审批池使用独立上限 R。
+  选号信号、负载评分、策略投影、请求容量观测和原子租约使用同一池的计数与上限，数值边界见 [运行设置 API](api.md#8-运行设置)
+- Store 按账号和容量池隔离租约，复用续期、取消与释放生命周期，不复制总计数。
+  最小请求间隔仍按账号共享；管理端普通容量和自适应并发下调的观测只计普通池
+- 两个池分别持有等待队列，任一池的队首不会阻塞另一池。审批等待者保持 FIFO，
+  可越过单账号等待数量上限，但仍受自身队列的总等待容量、共享等待预算与截止时刻约束；取消及超时必须释放等待位置
+- 配置为 0 时审批请求使用普通池；账号排队关闭时不引入隐式队列。Key 层准入仍按原有限额与排队规则执行

-等待者重新选号及取得租约前复核当前队首，异步读取期间被插队的普通请求继续等待，不凭已有 ticket 跳过优先顺序。
+等待者重新选号及取得租约前复核所属池的当前队首，不凭已有 ticket 跳过等待顺序

 ### Client Key 限额与结算

@@ -58,21 +58,21 @@ const openaiGuardianReservedConcurrency = defineModel<string>('openaiGuardianRes
      <BaseFormItem label="排队超时（秒）" description="密钥队列与账号队列共用的等待时限，从首次入队起计时，1～120 秒">
        <BaseInput v-model="concurrencyWaitTimeoutSeconds" aria-label="排队超时（秒）" type="number" min="1" max="120" step="1" />
      </BaseFormItem>
-      <BaseFormItem label="自动审批预留并发" description="为 Codex 自动审批预留每账号名额，0 表示关闭">
+      <BaseFormItem label="自动审批独立并发" description="每个账号额外的审批名额，0 表示共用普通并发">
        <template #label-extra>
          <BasePopover class="-my-1" trigger="hover-click" placement="top-start">
            <template #trigger="{ open }">
-              <button type="button" aria-label="自动审批预留并发说明" :aria-expanded="open" class="inline-flex size-6 shrink-0 cursor-pointer items-center justify-center rounded-cp-sm border-0 bg-transparent p-0 text-cp-text-tertiary outline-none transition-colors hover:text-cp-text focus-visible:ring-2 focus-visible:ring-cp-control-outline motion-reduce:transition-none">
+              <button type="button" aria-label="自动审批独立并发说明" :aria-expanded="open" class="inline-flex size-6 shrink-0 cursor-pointer items-center justify-center rounded-cp-sm border-0 bg-transparent p-0 text-cp-text-tertiary outline-none transition-colors hover:text-cp-text focus-visible:ring-2 focus-visible:ring-cp-control-outline motion-reduce:transition-none">
                <CircleAlert class="size-3.5" aria-hidden="true" />
              </button>
            </template>
            <div class="max-w-72 space-y-2 px-3 py-2 text-cp-sm leading-relaxed text-cp-text-secondary">
-              <p>仅对 OpenAI 账号生效，普通请求可用并发会相应减少，至少保留 1 个名额，不限并发的账号不扣减</p>
-              <p>大于 0 且开启账号排队时，审批请求优先等待，可超过账号队列容量，仍受总等待容量、排队超时与密钥限额约束</p>
+              <p>仅对 OpenAI 账号生效，普通并发为 10、审批名额为 3 时，两类请求分别最多运行 10 个和 3 个，互不占用名额</p>
+              <p>开启独立额度后，审批请求单独排队，可超过账号队列容量，仍受总等待容量、排队超时、请求间隔与密钥限额约束</p>
            </div>
          </BasePopover>
        </template>
-        <BaseInput v-model="openaiGuardianReservedConcurrency" aria-label="自动审批预留并发" type="number" min="0" max="4294967295" step="1" />
+        <BaseInput v-model="openaiGuardianReservedConcurrency" aria-label="自动审批独立并发" type="number" min="0" max="4294967295" step="1" />
      </BaseFormItem>
    </BaseForm>
  </BaseCard>
@@ -227,7 +227,7 @@ export function useSettingsForm() {
      return
    }
    if (!Number.isInteger(openaiGuardianReservedConcurrency) || openaiGuardianReservedConcurrency < 0 || openaiGuardianReservedConcurrency > 4294967295) {
-      toast.warning('自动审批预留并发应为 0～4294967295 的整数，0 表示关闭')
+      toast.warning('自动审批独立并发应为 0～4294967295 的整数，0 表示共用普通并发')
      return
    }
    if (responsesMaxDecompressedBodyMiB === null || !Number.isInteger(responsesMaxDecompressedBodyMiB) || responsesMaxDecompressedBodyMiB < 1