feat(peer-link)!: share upstream key concurrency and RPM limits

Use a fresh upstream key concurrency limit as the account default for scheduling, leases, and capacity reporting. Show disclosed limits alongside the upstream pool summary. BREAKING CHANGE: Both peers must support CPR Link schema v3. Refs: #587 Co-authored-by: Codex <noreply@openai.com>
2026-11-06 08:06:19 +00:00 · 2026-11-04 18:32:42 +08:00
parent 242ffb2324
commit 19ef484969
59 changed files with 511 additions and 158 deletions
@@ -197,6 +197,7 @@ async fn initialize_inner(
        auth,
        accounts: accounts.clone(),
        account_groups: Arc::new(DefaultAccountGroupService::new(
+            upstream_info.clone(),
            store.account_groups(),
            store.account_runtime(),
            snapshot.clone(),
@@ -73,6 +73,7 @@ pub struct AccountGroupUsage {
 /// 当前页账号组 membership 对应的持久账号事实；运行态在 Admin query service 合并
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub struct AccountGroupMemberFact {
+    pub inherited_version: Option<gateway_core::account::QuotaObservationVersion>,
    pub group_id: AccountGroupId,
    pub account_id: String,
    pub status: AccountStatusFacts,
@@ -219,7 +219,7 @@ pub struct AccountPageItem {
 pub struct AccountCapacity {
    /// 实时租约读取失败时为 `None`，不能当作空闲
    pub used_slots: Option<u64>,
-    /// 应用账号覆盖或全局默认值后的上限；`None` 表示不限
+    /// 应用账号覆盖或继承默认值后的普通并发上限；`None` 表示不限
    pub total_slots: Option<u64>,
 }

@@ -684,8 +684,10 @@ pub struct DashboardAccountUsage {
 }

 /// Dashboard 当前账号池的可重建运行时槽位事实
-#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+#[derive(Debug, Clone, PartialEq, Eq)]
 pub struct DashboardRuntimeSlots {
+    /// 仅未设置本地覆盖的账号可能继承有效互联观测
+    pub inherited_versions: Vec<gateway_core::account::QuotaObservationVersion>,
    /// 继承运行参数默认并发上限的可调度账号数
    pub inherited_accounts: u64,
    /// 已配置账号级并发上限的可调度账号槽位总数
@@ -17,6 +17,7 @@ pub struct PeerLinkLimits {
 pub struct PeerLinkExport {
    pub observed_at: DateTime<Utc>,
    pub expires_at: DateTime<Utc>,
+    pub key_limits: gateway_core::policy::RateLimits,
    pub budget: ClientBudgetStatus,
    pub budget_access: ClientBudgetAccess,
    pub budget_observed_at: DateTime<Utc>,
@@ -10,6 +10,8 @@ pub struct ProviderUpstreamInfo {
    pub stale: bool,
    pub connection_revision: String,
    pub last_error: Option<String>,
+    pub max_concurrency: Option<u64>,
+    pub requests_per_minute: Option<u64>,
    pub pool: Option<UpstreamPool>,
 }

@@ -38,6 +38,13 @@ pub trait PeerTransport: Send + Sync {
 /// 账号用例读取公共互联观测，显式刷新等待额度查询完成
 #[async_trait]
 pub trait PeerUpstreamService: gateway_core::account::UpstreamInfoSource {
+    /// 仅在指定持久观测与当前调度缓存一致时估算容量
+    fn observed_concurrency_limit(
+        &self,
+        version: &gateway_core::account::QuotaObservationVersion,
+        now: std::time::SystemTime,
+    ) -> Option<gateway_core::account::AccountConcurrencyLimit>;
+
    async fn account_changed(
        &self,
        id: &gateway_core::account::ProviderAccountId,
@@ -60,6 +60,7 @@ pub trait AccountGroupService: Send + Sync {
 }

 pub(crate) struct DefaultAccountGroupService {
+    upstream_info: Option<Arc<dyn crate::ports::peer::PeerUpstreamService>>,
    store: Arc<dyn AccountGroupStore>,
    runtime: Arc<dyn AccountRuntimeStore>,
    snapshot: Arc<dyn SnapshotControl>,
@@ -68,11 +69,13 @@ pub(crate) struct DefaultAccountGroupService {
 impl DefaultAccountGroupService {
    #[must_use]
    pub(crate) fn new(
+        upstream_info: Option<Arc<dyn crate::ports::peer::PeerUpstreamService>>,
        store: Arc<dyn AccountGroupStore>,
        runtime: Arc<dyn AccountRuntimeStore>,
        snapshot: Arc<dyn SnapshotControl>,
    ) -> Self {
        Self {
+            upstream_info,
            store,
            runtime,
            snapshot,
@@ -105,7 +108,7 @@ impl DefaultAccountGroupService {
                Default::default()
            }
        };
-        project_group_runtime(records, &members, &runtime);
+        project_group_runtime(records, &members, &runtime, self.upstream_info.as_deref());
        Ok(())
    }

@@ -215,6 +218,7 @@ fn project_group_runtime(
    records: &mut [AccountGroupRecord],
    members: &[AccountGroupMemberFact],
    runtime: &crate::model::accounts::AccountRuntimeSnapshot,
+    upstream: Option<&dyn crate::ports::peer::PeerUpstreamService>,
 ) {
    let now = SystemTime::now();
    let mut status_by_account = BTreeMap::new();
@@ -228,7 +232,14 @@ fn project_group_runtime(
            .or_insert_with(|| resolve_account_status(&status, now).status);
        slots_by_account
            .entry(member.account_id.as_str())
-            .or_insert(member.total_slots);
+            .or_insert_with(|| {
+                member
+                    .inherited_version
+                    .as_ref()
+                    .and_then(|version| upstream?.observed_concurrency_limit(version, now))
+                    .map(|limit| u64::from(limit.get()))
+                    .or(member.total_slots)
+            });
        accounts_by_group
            .entry(member.group_id.as_str())
            .or_default()
@@ -504,7 +504,11 @@ impl DefaultAccountsService {
            });
        Ok(AccountDirectoryItem {
            protection_rules: stored.protection_rules,
-            capacity: stored.capacity,
+            capacity: super::peer_link::upstream::account_capacity(
+                &stored.account,
+                stored.capacity,
+                &quota,
+            ),
            capabilities: provider
                .account_capabilities(account_id, &stored.account.authentication_kind),
            plan_type_display: self.providers.resolve_account_plan(
@@ -618,7 +622,11 @@ impl AccountsService for DefaultAccountsService {
                        used_slots: in_flight
                            .as_ref()
                            .map(|counts| counts.get(&item.account.id).copied().unwrap_or(0)),
-                        ..item.capacity
+                        ..super::peer_link::upstream::account_capacity(
+                            &item.account,
+                            item.capacity,
+                            &quota,
+                        )
                    },
                    capabilities,
                    plan_type_display: providers.resolve_account_plan(
@@ -133,7 +133,7 @@ impl DefaultObservabilityService {
            self.settings.load_runtime_settings(),
        )
        .map_err(|error| map_store_error(error, "dashboard"))?;
-        let runtime_slots = observation.runtime_slots;
+        let runtime_slots = observation.runtime_slots.as_ref();
        self.enrich_list_records(&mut observation.recent_requests)
            .await;
        self.enrich_dashboard_quotas(&mut observation.account_usage)
@@ -180,11 +180,19 @@ impl DefaultObservabilityService {
            .providers
            .dashboard_wire_profiles(&settings.request_profiles);
        let max_concurrent_per_account = u64::from(settings.values.max_concurrent_per_account);
-        let (inherited_accounts, overridden_slots) = runtime_slots
-            .as_ref()
+        let (mut inherited_accounts, mut overridden_slots) = runtime_slots
            .map_or((observation.provider_accounts.normal, 0), |slots| {
                (slots.inherited_accounts, slots.overridden_slots)
            });
+        if let (Some(peer), Some(slots)) = (&self.upstream_info, runtime_slots) {
+            let now = std::time::SystemTime::now();
+            for version in &slots.inherited_versions {
+                if let Some(limit) = peer.observed_concurrency_limit(version, now) {
+                    inherited_accounts -= 1;
+                    overridden_slots = overridden_slots.saturating_add(u64::from(limit.get()));
+                }
+            }
+        }
        let total_slots = (max_concurrent_per_account > 0 || inherited_accounts == 0).then(|| {
            inherited_accounts
                .saturating_mul(max_concurrent_per_account)
@@ -117,6 +117,7 @@ impl PeerLinkService for DefaultPeerLinkService {
                .budget
                .limits
                .access(key.budget.daily_used_usd, key.budget.weekly_used_usd),
+            key_limits: key.limits,
            budget: key.budget,
            budget_observed_at,
            pool: None,
@@ -25,6 +25,7 @@ use gateway_core::{
    error::StoreError,
    task::{WorkerContribution, WorkerDefinitionError},
 };
+pub(crate) use projection::account_capacity;
 use service::PeerInfoService;

 /// 组合根独立注册公共互联 Worker，不属于任何具体 Provider 的任务集合
@@ -62,6 +63,24 @@ pub fn initialize(

 #[async_trait::async_trait]
 impl PeerUpstreamService for PeerInfoService {
+    fn observed_concurrency_limit(
+        &self,
+        version: &gateway_core::account::QuotaObservationVersion,
+        now: std::time::SystemTime,
+    ) -> Option<gateway_core::account::AccountConcurrencyLimit> {
+        let observations = self
+            .observations
+            .read()
+            .unwrap_or_else(std::sync::PoisonError::into_inner);
+        let cached = observations.get(&version.account_id)?;
+        if cached.observed_at != Some(version.observed_at)
+            || cached.connection.credential_revision != version.credential_revision
+        {
+            return None;
+        }
+        cached.scheduling.as_ref()?.concurrency_limit(now)
+    }
+
    async fn account_changed(&self, id: &ProviderAccountId) -> Result<(), StoreError> {
        PeerInfoService::account_changed(self, id).await
    }
@@ -16,6 +16,26 @@ use gateway_protocol::peer::{PeerComponentStatus, PeerLink};

 use super::{PeerInfoService, quota, snapshot};

+pub(crate) fn account_capacity(
+    account: &crate::model::accounts::AccountRecord,
+    mut capacity: crate::model::accounts::AccountCapacity,
+    quota: &ProviderQuota,
+) -> crate::model::accounts::AccountCapacity {
+    if account.concurrency_limit.is_none()
+        && let Some(info) = &quota.upstream
+        && !info.stale
+        && info
+            .observed_at
+            .is_some_and(|at| snapshot::fresh_at(at.into(), SystemTime::now()))
+        && let Some(limit) = info
+            .max_concurrency
+            .and_then(gateway_core::account::AccountConcurrencyLimit::from_key_limit)
+    {
+        capacity.total_slots = Some(u64::from(limit.get()));
+    }
+    capacity
+}
+
 impl PeerInfoService {
    pub(super) fn project_peer(
        connection: &UpstreamConnection,
@@ -30,6 +50,7 @@ impl PeerInfoService {
        let stale = peer.as_ref().is_none_or(|peer| {
            !snapshot::component_fresh(Some(&peer.observed_at), Some(&peer.expires_at), now)
        });
+        let limits = peer.and_then(|peer| peer.key_limits);
        let info = ProviderUpstreamInfo {
            discovered: peer.is_some() || unavailable.is_some(),
            observed_at,
@@ -38,6 +59,8 @@ impl PeerInfoService {
            last_error: last_error
                .or(unavailable)
                .map(|error| error.as_str().to_owned()),
+            max_concurrency: limits.and_then(|limits| limits.max_concurrency),
+            requests_per_minute: limits.and_then(|limits| limits.requests_per_minute),
            pool: peer
                .as_ref()
                .and_then(|peer| peer.pool.as_ref())
@@ -63,6 +63,7 @@ struct ModelObservation {

 pub(super) struct SchedulingObservation {
    validity: Validity,
+    concurrency_limit: Option<gateway_core::account::AccountConcurrencyLimit>,
    exhausted_budgets: Vec<Validity>,
    models: HashMap<String, Option<ModelObservation>>,
 }
@@ -166,43 +167,52 @@ impl SchedulingObservation {
        }
        Some(Self {
            validity,
+            concurrency_limit: peer
+                .key_limits
+                .and_then(|limits| limits.max_concurrency)
+                .and_then(gateway_core::account::AccountConcurrencyLimit::from_key_limit),
            exhausted_budgets,
            models,
        })
    }

    pub(super) fn fresh(&self, now: SystemTime) -> bool {
-        self.validity.fresh(now)
-            && now
-                .duration_since(self.validity.observed_at)
-                .unwrap_or_default()
-                <= snapshot::FRESH_FOR
+        self.validity.fresh(now) && snapshot::fresh_at(self.validity.observed_at, now)
    }

-    pub(super) fn signal(&self, model: &str, now: SystemTime) -> Option<UpstreamSchedulingSignal> {
+    pub(super) fn concurrency_limit(
+        &self,
+        now: SystemTime,
+    ) -> Option<gateway_core::account::AccountConcurrencyLimit> {
+        self.concurrency_limit.filter(|_| self.fresh(now))
+    }
+
+    pub(super) fn signal(
+        &self,
+        model: Option<&str>,
+        now: SystemTime,
+    ) -> Option<UpstreamSchedulingSignal> {
        if !self.fresh(now) {
            return None;
        }
-        if self
+        let budget_exhausted = self
            .exhausted_budgets
            .iter()
-            .any(|validity| validity.fresh(now))
-        {
-            return Some(UpstreamSchedulingSignal {
-                exhausted: true,
-                remaining_percent: None,
-            });
-        }
-        let model = self.models.get(model)?.as_ref()?;
-        if !model.validity.fresh(now) {
+            .any(|validity| validity.fresh(now));
+        let model = model
+            .and_then(|id| self.models.get(id)?.as_ref())
+            .filter(|model| model.validity.fresh(now));
+        if !budget_exhausted && model.is_none() && self.concurrency_limit.is_none() {
            return None;
        }
        Some(UpstreamSchedulingSignal {
-            exhausted: model.exhausted,
-            remaining_percent: model
-                .account
-                .as_ref()
-                .and_then(|account| account.remaining_percent(now)),
+            concurrency_limit: self.concurrency_limit,
+            exhausted: budget_exhausted || model.is_some_and(|model| model.exhausted),
+            remaining_percent: if budget_exhausted {
+                None
+            } else {
+                model.and_then(|model| model.account.as_ref()?.remaining_percent(now))
+            },
        })
    }
 }
@@ -290,7 +290,7 @@ impl UpstreamInfoSource for PeerInfoService {
    fn signal(
        &self,
        account: &ProviderAccount,
-        model: &str,
+        model: Option<&str>,
        now: SystemTime,
    ) -> Option<UpstreamSchedulingSignal> {
        if !account.enabled() {
@@ -89,10 +89,12 @@ pub(super) fn datetime(value: &str) -> Option<DateTime<Utc>> {
 }

 pub(super) fn fresh(observed: &str, now: SystemTime) -> bool {
-    timestamp(observed).is_some_and(|observed| {
-        observed <= now + Duration::from_secs(60)
-            && now.duration_since(observed).unwrap_or_default() <= FRESH_FOR
-    })
+    timestamp(observed).is_some_and(|observed| fresh_at(observed, now))
+}
+
+pub(super) fn fresh_at(observed: SystemTime, now: SystemTime) -> bool {
+    observed <= now + Duration::from_secs(60)
+        && now.duration_since(observed).unwrap_or_default() <= FRESH_FOR
 }

 pub(super) fn component_fresh(
@@ -162,6 +164,9 @@ pub(super) fn parse(
    }
    // 未授权能力不能借同名数据块越过逐层公开边界
    let disclosed = |capability| peer.capabilities.iter().any(|value| value == capability);
+    if !disclosed("key_limits") {
+        peer.key_limits = None;
+    }
    if !disclosed("key_usage") {
        peer.usage = None;
    }
@@ -373,6 +373,7 @@ fn group_record() -> AccountGroupRecord {

 fn member(account_id: &str, total_slots: u64) -> AccountGroupMemberFact {
    AccountGroupMemberFact {
+        inherited_version: None,
        group_id: group_id(),
        account_id: account_id.to_owned(),
        status: AccountStatusFacts {
@@ -2686,12 +2686,15 @@ async fn peer_account_usage_should_follow_budget_windows_without_borrowing_pool_
    for longest in [86_400, 604_800] {
        let provider = FakeProviderAdmin::new("openai", events());
        let mut account = account_record("openai");
+        account.concurrency_limit = None;
        account.authentication_kind = "api_key".to_owned();
        account.created_at = Utc::now() - TimeDelta::days(60);
        let store = FakeAccountStore::with_account(account, events());
        let reset_at = Utc::now() + TimeDelta::hours(1);
        let mut quota = empty_quota();
        quota.upstream = Some(ProviderUpstreamInfo {
+            max_concurrency: Some(7),
+            requests_per_minute: Some(60),
            discovered: true,
            observed_at: Some(Utc::now()),
            stale: false,
@@ -2762,6 +2765,8 @@ async fn peer_account_usage_should_follow_budget_windows_without_borrowing_pool_
            page.items[0].usage.as_ref().unwrap().total_tokens,
            Some(1_234)
        );
+        assert_eq!(page.items[0].capacity.total_slots, Some(7));
+        assert_eq!(page.items[0].account.concurrency_limit, None);
        let queries = store.quota_window_queries();
        assert_eq!(queries.len(), window_usage.len());
        for (query, expected) in queries.iter().zip(&window_usage) {
@@ -143,7 +143,7 @@ impl UpstreamInfoSource for PendingPeer {
    fn signal(
        &self,
        _: &ProviderAccount,
-        _: &str,
+        _: Option<&str>,
        _: std::time::SystemTime,
    ) -> Option<UpstreamSchedulingSignal> {
        None
@@ -152,6 +152,14 @@ impl UpstreamInfoSource for PendingPeer {

 #[async_trait]
 impl PeerUpstreamService for PendingPeer {
+    fn observed_concurrency_limit(
+        &self,
+        _: &gateway_core::account::QuotaObservationVersion,
+        _: std::time::SystemTime,
+    ) -> Option<gateway_core::account::AccountConcurrencyLimit> {
+        None
+    }
+
    async fn account_changed(&self, id: &ProviderAccountId) -> Result<(), StoreError> {
        self.changed.lock().unwrap().push(id.as_str().into());
        Ok(())
@@ -448,6 +448,7 @@ async fn dashboard_summary_should_project_rebuildable_runtime_slots() {
    let now = Utc::now();
    let store = Arc::new(FixtureObservabilityStore::new(observation_range(now)));
    store.replace_runtime_slots(Some(DashboardRuntimeSlots {
+        inherited_versions: Vec::new(),
        inherited_accounts: 2,
        overridden_slots: 5,
        used_slots: Some(2),
@@ -476,6 +477,7 @@ async fn dashboard_capacity_distinguishes_unlimited_inheritance_finite_overrides
        let now = Utc::now();
        let store = Arc::new(FixtureObservabilityStore::new(observation_range(now)));
        store.replace_runtime_slots(Some(DashboardRuntimeSlots {
+            inherited_versions: Vec::new(),
            inherited_accounts,
            overridden_slots,
            used_slots: Some(2),
@@ -1212,7 +1214,7 @@ impl ObservabilityStore for FixtureObservabilityStore {
            range: query.range,
            totals: Default::default(),
            provider_accounts: AccountPoolMetrics::default(),
-            runtime_slots: *self.runtime_slots.lock().expect("runtime slots"),
+            runtime_slots: self.runtime_slots.lock().expect("runtime slots").clone(),
            trend: self.trend.lock().expect("trend").clone(),
            account_usage: self.account_usage.lock().expect("account usage").clone(),
            recent_requests: Vec::new(),
@@ -241,7 +241,7 @@ async fn discovery_uses_provider_capabilities_and_publishes_neutral_signals() {
        let reader = UpstreamInfoReader::default();
        assert!(
            reader
-                .signal(&initial_accounts[0], "model", SystemTime::now())
+                .signal(&initial_accounts[0], Some("model"), SystemTime::now())
                .is_none()
        );
        let source = bundle.scheduling_source();
@@ -273,7 +273,11 @@ async fn discovery_uses_provider_capabilities_and_publishes_neutral_signals() {
        );
        for account in &initial_accounts[..2] {
            if Some(account.provider().as_str()) == failed_provider {
-                assert!(reader.signal(account, "model", SystemTime::now()).is_none());
+                assert!(
+                    reader
+                        .signal(account, Some("model"), SystemTime::now())
+                        .is_none()
+                );
                continue;
            }
            let observed = bundle
@@ -288,21 +292,21 @@ async fn discovery_uses_provider_capabilities_and_publishes_neutral_signals() {
            );
            assert!(
                reader
-                    .signal(account, "model", SystemTime::now())
+                    .signal(account, Some("model"), SystemTime::now())
                    .unwrap()
                    .exhausted
            );
        }
        assert!(
            reader
-                .signal(&initial_accounts[2], "model", SystemTime::now())
+                .signal(&initial_accounts[2], Some("model"), SystemTime::now())
                .is_none()
        );
        drop(worker);
        drop(bundle);
        assert!(
            reader
-                .signal(&initial_accounts[0], "model", SystemTime::now())
+                .signal(&initial_accounts[0], Some("model"), SystemTime::now())
                .is_none()
        );
    }
@@ -332,7 +336,7 @@ async fn account_edits_preserve_observation_and_refresh_budget_until_connection_
    assert!(
        bundle
            .scheduling_source()
-            .signal(&edited, "model", SystemTime::now())
+            .signal(&edited, Some("model"), SystemTime::now())
            .is_some()
    );
    let after = service.refresh_quota(edited.id()).await.unwrap();
@@ -347,7 +351,7 @@ async fn account_edits_preserve_observation_and_refresh_budget_until_connection_
    assert!(
        bundle
            .scheduling_source()
-            .signal(&reconnected, "model", SystemTime::now())
+            .signal(&reconnected, Some("model"), SystemTime::now())
            .is_none()
    );
    service.refresh_quota(reconnected.id()).await.unwrap();
@@ -392,7 +396,7 @@ impl PeerTransport for TestTransport {
            .push(account.id().as_str().to_owned());
        let now = Utc::now();
        Ok(serde_json::to_vec(&serde_json::json!({
-            "object":"cpr.peer", "schema_version":2,
+            "object":"cpr.peer", "schema_version":3,
            "observed_at":now.to_rfc3339(),
            "expires_at":(now + chrono::Duration::minutes(5)).to_rfc3339(),
            "capabilities":["key_usage"],
@@ -216,7 +216,7 @@ async fn known_peer_failures_keep_identity_and_only_revocation_discards_shared_d
        assert_eq!(
            bundle
                .scheduling_source()
-                .signal(&account, "model", SystemTime::now())
+                .signal(&account, Some("model"), SystemTime::now())
                .is_none(),
            clears_data
        );
@@ -14,6 +14,8 @@ pub struct AccountUpstreamInfoView {
    pub stale: bool,
    pub connection_revision: String,
    pub last_error: Option<String>,
+    pub max_concurrency: Option<u64>,
+    pub requests_per_minute: Option<u64>,
    pub pool: Option<UpstreamPoolView>,
 }

@@ -78,6 +80,8 @@ pub(crate) fn view(
        stale: info.stale,
        connection_revision: info.connection_revision,
        last_error: info.last_error,
+        max_concurrency: info.max_concurrency,
+        requests_per_minute: info.requests_per_minute,
        pool: info.pool.map(|pool| UpstreamPoolView {
            account_count: pool.account_count,
            states: UpstreamPoolStatesView {
@@ -4,14 +4,14 @@ use chrono::{DateTime, Utc};
 use gateway_admin::model::peer_link::{PeerLinkExport, PeerModelState as ModelState};
 use gateway_core::metering::Decimal;
 use gateway_protocol::peer::{
-    OBJECT, PeerBudgetWindow, PeerComponentStatus, PeerKeyUsage, PeerLink, PeerModelAvailability,
-    PeerModelCoverage, PeerModelState, PeerPlanCount, PeerPool, PeerPoolStates, PeerQuota,
-    PeerQuotaAccount, PeerQuotaWindow, PeerUsage, SCHEMA_VERSION,
+    OBJECT, PeerBudgetWindow, PeerComponentStatus, PeerKeyLimits, PeerKeyUsage, PeerLink,
+    PeerModelAvailability, PeerModelCoverage, PeerModelState, PeerPlanCount, PeerPool,
+    PeerPoolStates, PeerQuota, PeerQuotaAccount, PeerQuotaWindow, PeerUsage, SCHEMA_VERSION,
 };
 use std::time::SystemTime;

 pub(super) fn response(export: PeerLinkExport) -> PeerLink {
-    let mut capabilities = vec!["key_usage".to_owned()];
+    let mut capabilities = vec!["key_usage".to_owned(), "key_limits".to_owned()];
    let pool = export.pool.map(|pool| {
        capabilities.push("pool_summary".to_owned());
        PeerPool {
@@ -74,6 +74,10 @@ pub(super) fn response(export: PeerLinkExport) -> PeerLink {
        capabilities,
        observed_at: export.observed_at.to_rfc3339(),
        expires_at: export.expires_at.to_rfc3339(),
+        key_limits: Some(PeerKeyLimits {
+            max_concurrency: Some(export.key_limits.max_concurrency),
+            requests_per_minute: Some(export.key_limits.requests_per_minute),
+        }),
        usage: Some(PeerUsage {
            status: PeerComponentStatus::Available,
            observed_at: Some(export.budget_observed_at.to_rfc3339()),
@@ -884,6 +884,7 @@ impl AccountGroupStore for MemoryAccountGroupStore {
        }
        Ok(vec![
            AccountGroupMemberFact {
+                inherited_version: None,
                group_id: group_id(PRIMARY_GROUP_ID),
                account_id: "acct_group_ready".to_owned(),
                status: AccountStatusFacts {
@@ -898,6 +899,7 @@ impl AccountGroupStore for MemoryAccountGroupStore {
                total_slots: Some(1),
            },
            AccountGroupMemberFact {
+                inherited_version: None,
                group_id: group_id(PRIMARY_GROUP_ID),
                account_id: "acct_group_disabled".to_owned(),
                status: AccountStatusFacts {
@@ -702,6 +702,8 @@ fn peer_summary(
 ) -> gateway_admin::model::provider_credentials::ProviderQuota {
    gateway_admin::model::provider_credentials::ProviderQuota {
        upstream: Some(gateway_admin::model::upstream_info::ProviderUpstreamInfo {
+            max_concurrency: None,
+            requests_per_minute: None,
            discovered,
            observed_at: observed.then(chrono::Utc::now),
            stale: true,
@@ -13,7 +13,7 @@ impl UpstreamInfoSource for TestPeerInfo {
    fn signal(
        &self,
        _: &ProviderAccount,
-        _: &str,
+        _: Option<&str>,
        _: SystemTime,
    ) -> Option<UpstreamSchedulingSignal> {
        None
@@ -22,6 +22,14 @@ impl UpstreamInfoSource for TestPeerInfo {

 #[async_trait]
 impl PeerUpstreamService for TestPeerInfo {
+    fn observed_concurrency_limit(
+        &self,
+        _: &gateway_core::account::QuotaObservationVersion,
+        _: std::time::SystemTime,
+    ) -> Option<gateway_core::account::AccountConcurrencyLimit> {
+        None
+    }
+
    async fn account_changed(&self, _: &ProviderAccountId) -> Result<(), StoreError> {
        Ok(())
    }
@@ -69,6 +69,10 @@ async fn fixture(sharing: ClientPeerSharing, plaintext: &str) -> (AdminTestFixtu
        .await
        .unwrap()
        .record;
+    key.limits = RateLimits {
+        max_concurrency: 5,
+        requests_per_minute: 60,
+    };
    key.peer_sharing = sharing;
    key.groups = vec![group.clone()];
    key.name = "private-key-sentinel".into();
@@ -243,7 +247,11 @@ async fn exports_scoped_observations_even_when_the_key_budget_is_exhausted_witho
    assert_eq!(response.status(), StatusCode::OK);
    let body = response_json(response).await;
    assert_eq!(body["object"], "cpr.peer");
-    assert_eq!(body["schema_version"], 2);
+    assert_eq!(body["schema_version"], 3);
+    assert_eq!(
+        body["key_limits"],
+        json!({"max_concurrency":5,"requests_per_minute":60})
+    );
    assert_eq!(body["usage"]["data"]["windows"][0]["remaining"], "0");
    assert_eq!(body["usage"]["data"]["allowed"], false);
    assert_eq!(body["usage"]["data"]["windows"][0]["limit_reached"], true);
@@ -358,7 +366,10 @@ async fn plan_only_summary_limits_exported_data_and_preserves_unknown_on_quota_r
        body["pool"]["plans"],
        json!([{ "name": "Free", "count": 1 }])
    );
-    assert_eq!(body["capabilities"], json!(["key_usage", "pool_summary"]));
+    assert_eq!(
+        body["capabilities"],
+        json!(["key_usage", "key_limits", "pool_summary"])
+    );
    assert!(body.get("quota").is_none());

    *fixture.provider_quota.lock().unwrap() = None;
@@ -39,6 +39,12 @@ impl ProviderAccountId {
 pub struct AccountConcurrencyLimit(NonZeroU32);

 impl AccountConcurrencyLimit {
+    /// Key 的正数上限映射到账号可表达的范围，零沿用本机默认值
+    #[must_use]
+    pub fn from_key_limit(limit: u64) -> Option<Self> {
+        Self::new(u32::try_from(limit).unwrap_or(u32::MAX))
+    }
+
    #[must_use]
    pub const fn new(value: u32) -> Option<Self> {
        match NonZeroU32::new(value) {
@@ -470,6 +470,8 @@ impl AccountQuotaSignals {
 /// 账号持久事实与可重建运行信号的请求级组合
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub struct AccountCandidate {
+    /// 本次选号已确认的上游默认上限，与本地可编辑覆盖独立
+    pub inherited_concurrency: Option<super::AccountConcurrencyLimit>,
    pub account: ProviderAccount,
    pub signals: AccountRuntimeSignals,
 }
@@ -525,11 +527,16 @@ pub struct AccountSelectionContext {
 impl AccountSelectionContext {
    /// 本请求在该账号上可使用的并发上限；资格判断、租约与策略投影必须共用这一口径
    #[must_use]
-    pub fn concurrency_limit(&self, account: &ProviderAccount) -> AccountConcurrency {
+    pub fn concurrency_limit(&self, candidate: &AccountCandidate) -> AccountConcurrency {
        if self.reserved_concurrency > 0 {
            AccountConcurrency::new(self.reserved_concurrency)
        } else {
-            account.effective_concurrency(self.policy.max_concurrent_per_account())
+            let default = candidate
+                .inherited_concurrency
+                .map_or(self.policy.max_concurrent_per_account(), |limit| {
+                    limit.into_non_zero().into()
+                });
+            candidate.account.effective_concurrency(default)
        }
    }
 }
@@ -671,8 +678,7 @@ impl AccountSelector {
                )
            })
            .try_fold((0_u64, 0_u64), |(used, total), candidate| {
-                let capacity =
-                    u64::from(context.concurrency_limit(&candidate.account).limit()?.get());
+                let capacity = u64::from(context.concurrency_limit(candidate).limit()?.get());
                Some((
                    used.saturating_add(u64::from(candidate.signals.in_flight)),
                    total.saturating_add(capacity),
@@ -868,7 +874,7 @@ impl AccountSelector {
            return Some(AccountSchedulingBlocker::ConcurrencyLimit);
        }
        if context
-            .concurrency_limit(&candidate.account)
+            .concurrency_limit(candidate)
            .limit()
            .is_some_and(|limit| candidate.signals.in_flight >= limit.get())
        {
@@ -885,7 +891,7 @@ const SMART_RESET_HALF_SCORE_SECONDS: f64 = 3_600.0;

 fn capacity_utilization(candidate: &AccountCandidate, context: &AccountSelectionContext) -> f64 {
    context
-        .concurrency_limit(&candidate.account)
+        .concurrency_limit(candidate)
        .limit()
        .map_or(0.0, |limit| {
            f64::from(candidate.signals.in_flight) / f64::from(limit.get())
@@ -76,6 +76,7 @@ impl UpstreamInfoFailure {
 /// 调度侧只消费已发布的中立观测，不读取协议、凭据或持久存储
 #[derive(Debug, Clone, Copy, PartialEq)]
 pub struct UpstreamSchedulingSignal {
+    pub concurrency_limit: Option<super::AccountConcurrencyLimit>,
    pub exhausted: bool,
    pub remaining_percent: Option<f64>,
 }
@@ -84,7 +85,7 @@ pub trait UpstreamInfoSource: Send + Sync {
    fn signal(
        &self,
        account: &ProviderAccount,
-        model: &str,
+        model: Option<&str>,
        now: SystemTime,
    ) -> Option<UpstreamSchedulingSignal>;
 }
@@ -110,7 +111,7 @@ impl UpstreamInfoSource for UpstreamInfoReader {
    fn signal(
        &self,
        account: &ProviderAccount,
-        model: &str,
+        model: Option<&str>,
        now: SystemTime,
    ) -> Option<UpstreamSchedulingSignal> {
        self.source.get()?.upgrade()?.signal(account, model, now)
@@ -41,7 +41,7 @@ impl TraceContext {
                    "blocker": AccountSelector.scheduling_blocker(candidate, context)
                        .map(|blocker| format!("{blocker:?}")),
                    "inFlight": signals.in_flight,
-                    "concurrencyLimit": context.concurrency_limit(&candidate.account).get(),
+                    "concurrencyLimit": context.concurrency_limit(candidate).get(),
                    "quotaRemainingBasisPoints": signals.quota_remaining_rank,
                    "quotaResetAtUnixMs": signals.quota_reset_at.and_then(|reset|
                        reset.duration_since(std::time::UNIX_EPOCH).ok()
@@ -483,7 +483,7 @@ impl RequestPolicyContext {
                    .is_none(),
                weight: candidate.account.weight().get(),
                in_flight: candidate.signals.in_flight,
-                maximum_concurrency: context.concurrency_limit(&candidate.account).get(),
+                maximum_concurrency: context.concurrency_limit(candidate).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,
@@ -50,6 +50,7 @@ fn account(id: &str) -> ProviderAccount {

 fn candidate(id: &str, in_flight: u32, remaining: Option<u64>) -> AccountCandidate {
    AccountCandidate {
+        inherited_concurrency: None,
        account: account(id),
        signals: AccountRuntimeSignals {
            in_flight,
@@ -215,6 +216,7 @@ fn inconclusive_quota_observation_cannot_erase_confirmed_access() {
 fn diagnostic_selection_bypasses_all_local_account_eligibility() {
    let observed_at = SystemTime::now();
    let exhausted = AccountCandidate {
+        inherited_concurrency: None,
        account: account("acct_exhausted").with_account_facts(
            true,
            CredentialState::Ready,
@@ -233,6 +235,7 @@ fn diagnostic_selection_bypasses_all_local_account_eligibility() {
        },
    };
    let disabled = AccountCandidate {
+        inherited_concurrency: None,
        account: account("acct_disabled").with_account_facts(
            false,
            CredentialState::Ready,
@@ -1113,11 +1116,8 @@ fn reserved_concurrency_uses_its_own_limit_without_reducing_normal_capacity() {

    // 独立额度可以大于账号普通上限，极值不能回绕成无限并发
    reserved.reserved_concurrency = u32::MAX;
-    assert_eq!(
-        reserved.concurrency_limit(&candidates[0].account).get(),
-        u32::MAX
-    );
-    assert_eq!(normal.concurrency_limit(&candidates[0].account).get(), 5);
+    assert_eq!(reserved.concurrency_limit(&candidates[0]).get(), u32::MAX);
+    assert_eq!(normal.concurrency_limit(&candidates[0]).get(), 5);
 }

 #[test]
@@ -2114,6 +2114,7 @@ impl Provider for PolicySelectingProvider {
        )
        .with_scheduling(None, AccountWeight::new(100).unwrap());
        let candidates = [AccountCandidate {
+            inherited_concurrency: None,
            account,
            signals: AccountRuntimeSignals {
                in_flight: 0,
@@ -55,6 +55,7 @@ fn account_id(id: &str) -> ProviderAccountId {

 fn candidate(id: &str) -> AccountCandidate {
    AccountCandidate {
+        inherited_concurrency: None,
        account: ProviderAccount::new(
            account_id(id),
            ProviderKind::new("openai").unwrap(),
@@ -608,6 +608,7 @@ async fn model_router_can_delegate_after_a_managed_http_side_effect() {

 fn candidate(id: &str, weight: u16) -> AccountCandidate {
    AccountCandidate {
+        inherited_concurrency: None,
        account: ProviderAccount::new(
            ProviderAccountId::new(id).unwrap(),
            ProviderKind::new("example").unwrap(),
@@ -3,7 +3,7 @@
 use serde::{Deserialize, Serialize};

 pub const OBJECT: &str = "cpr.peer";
-pub const SCHEMA_VERSION: u32 = 2;
+pub const SCHEMA_VERSION: u32 = 3;
 pub const MAX_ACCOUNTS: usize = 3_000;
 pub const MAX_QUOTA_WINDOWS: usize = 6_000;
 pub const MAX_MODELS: usize = 2_000;
@@ -20,6 +20,8 @@ pub struct PeerLink {
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub usage: Option<PeerUsage>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub key_limits: Option<PeerKeyLimits>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub pool: Option<PeerPool>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub quota: Option<PeerQuota>,
@@ -27,6 +29,15 @@ pub struct PeerLink {
    pub models: Vec<PeerModelAvailability>,
 }

+/// 当前认证密钥公开的请求上限，缺失表示未知，零表示不限
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
+pub struct PeerKeyLimits {
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub max_concurrency: Option<u64>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub requests_per_minute: Option<u64>,
+}
+
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
 #[serde(rename_all = "snake_case")]
 pub enum PeerComponentStatus {
@@ -243,7 +243,7 @@ impl AccountGroupStore for PgAccountGroupRepository {
                    account.authentication_kind, account.credential_revision, account.outbound_proxy_id, account.outbound_proxy_url,
                    account.has_refresh_token, account.access_token_expires_at,
                    account.next_refresh_at, account.enabled, account.concurrency_limit,
-                    account.weight, account.model_access_json, account.credential_state, account.quota_access_state,
+                    account.quota_observed_at, account.weight, account.model_access_json, account.credential_state, account.quota_access_state,
                    account.quota_evidence, account.quota_access_observed_at,
                    account.quota_reset_at, account.last_error_reason,
                    account.last_error_message, account.credential_observed_at,
@@ -277,8 +277,10 @@ impl AccountGroupStore for PgAccountGroupRepository {
                .map_err(|source| {
                    invalid("invalid default account concurrency").with_source(source)
                })?;
+                let version = super::provider_accounts::quota_version_from_row(&row)?;
                let account = account_summary_from_row(row)?;
                Ok(AccountGroupMemberFact {
+                    inherited_version: version.filter(|_| account.concurrency_limit.is_none()),
                    group_id,
                    account_id: account.id,
                    status: AccountStatusFacts {
@@ -123,6 +123,10 @@ impl PgObservabilityRepository {
            None
        };
        Ok(admin_observability::DashboardRuntimeSlots {
+            inherited_versions: normal_accounts
+                .iter()
+                .filter_map(|account| account.quota_version.clone())
+                .collect(),
            inherited_accounts,
            overridden_slots,
            used_slots,
@@ -13,28 +13,10 @@ pub(super) async fn versions(
        .collect::<Vec<_>>();
    let rows = sqlx::query("select id, credential_revision, quota_observed_at from provider_accounts where id = any($1) and quota_observed_at is not null")
        .bind(ids).fetch_all(&repository.pool).await.map_err(database_error)?;
-    rows.into_iter()
-        .map(|row| {
-            Ok(QuotaObservationVersion {
-                account_id: CoreProviderAccountId::new(
-                    row.try_get::<String, _>("id").map_err(database_error)?,
-                )
-                .map_err(|_| CoreStoreError::new(CoreStoreErrorKind::InvalidData))?,
-                credential_revision: CoreCredentialRevision::new(
-                    u64::try_from(
-                        row.try_get::<i64, _>("credential_revision")
-                            .map_err(database_error)?,
-                    )
-                    .map_err(|_| CoreStoreError::new(CoreStoreErrorKind::InvalidData))?,
-                )
-                .map_err(|_| CoreStoreError::new(CoreStoreErrorKind::InvalidData))?,
-                observed_at: row
-                    .try_get::<DateTime<Utc>, _>("quota_observed_at")
-                    .map_err(database_error)?
-                    .into(),
-            })
-        })
-        .collect()
+    rows.iter()
+        .filter_map(|row| super::quota_version_from_row(row).transpose())
+        .collect::<crate::StoreResult<Vec<_>>>()
+        .map_err(crate::core_store_error)
 }

 pub(super) async fn write(
@@ -435,6 +435,21 @@ pub(crate) fn require_core_update(updated: bool) -> Result<(), CoreStoreError> {
    }
 }

+pub(crate) fn quota_version_from_row(
+    row: &sqlx::postgres::PgRow,
+) -> StoreResult<Option<gateway_core::account::QuotaObservationVersion>> {
+    let Some(observed_at) = get::<Option<DateTime<Utc>>>(row, "quota_observed_at")? else {
+        return Ok(None);
+    };
+    Ok(Some(gateway_core::account::QuotaObservationVersion {
+        account_id: CoreProviderAccountId::new(get::<String>(row, "id")?)
+            .map_err(|_| invalid("invalid account ID"))?,
+        credential_revision: CoreCredentialRevision::new(to_u64(get(row, "credential_revision")?)?)
+            .map_err(|_| invalid("invalid credential revision"))?,
+        observed_at: observed_at.into(),
+    }))
+}
+
 pub(crate) fn account_summary_from_row(
    row: sqlx::postgres::PgRow,
 ) -> StoreResult<ProviderAccountSummary> {
@@ -64,6 +64,7 @@ pub(crate) async fn load_cooldown(
 pub(crate) struct ProviderAccountStatus {
    pub id: String,
    pub concurrency_limit: Option<gateway_core::account::AccountConcurrencyLimit>,
+    pub quota_version: Option<gateway_core::account::QuotaObservationVersion>,
    pub facts: AccountStatusFacts,
 }

@@ -72,7 +73,7 @@ pub(crate) async fn load_account_statuses<'e, E: sqlx::Executor<'e, Database = s
 ) -> crate::StoreResult<Vec<ProviderAccountStatus>> {
    use super::rows::{account_status_facts_from_row, get};
    let rows = sqlx::query(
-        "select id, enabled, concurrency_limit, credential_state, access_token_expires_at,
+        "select id, enabled, concurrency_limit, credential_revision, quota_observed_at, credential_state, access_token_expires_at,
                quota_access_state, quota_evidence, quota_access_observed_at, quota_reset_at,
                last_error_reason, last_error_message
         from provider_accounts",
@@ -91,6 +92,11 @@ pub(crate) async fn load_account_statuses<'e, E: sqlx::Executor<'e, Database = s
                })
                .transpose()?;
            Ok(ProviderAccountStatus {
+                quota_version: if concurrency_limit.is_none() {
+                    super::quota_version_from_row(row)?
+                } else {
+                    None
+                },
                id: get(row, "id")?,
                concurrency_limit,
                facts: account_status_facts_from_row(row)?,
@@ -332,31 +332,36 @@ impl CodexCredentialSelector {
        self
    }

-    fn quota_signals(
+    fn scheduling_signals(
        &self,
        account: &ProviderAccount,
        model: CodexSelectionModel<'_>,
        strategy: RotationStrategy,
-    ) -> Option<AccountQuotaSignals> {
+    ) -> (
+        Option<AccountQuotaSignals>,
+        Option<gateway_core::account::AccountConcurrencyLimit>,
+    ) {
        if account.authentication_kind() == CODEX_AUTHENTICATION_KIND_OAUTH {
-            return self.quota.scheduling_signals(account);
+            return (self.quota.scheduling_signals(account), None);
        }
-        if strategy != RotationStrategy::Smart {
-            return None;
-        }
-        let signal =
-            self.upstream_info
-                .as_ref()?
-                .signal(account, model.model()?, SystemTime::now())?;
-        let rank = if signal.exhausted {
-            Some(0)
-        } else {
-            signal
-                .remaining_percent
-                .map(|percent| (percent.clamp(0.0, 100.0) * 100.0).round() as u64)
+        let Some(signal) = self
+            .upstream_info
+            .as_ref()
+            .and_then(|source| source.signal(account, model.model(), SystemTime::now()))
+        else {
+            return (None, None);
        };
-        // 选号只读取 Smart 额度分，额度恢复由互联同步事务独立提交
-        Some(AccountQuotaSignals::new(None, rank))
+        let quota = (strategy == RotationStrategy::Smart).then(|| {
+            let rank = if signal.exhausted {
+                Some(0)
+            } else {
+                signal
+                    .remaining_percent
+                    .map(|percent| (percent.clamp(0.0, 100.0) * 100.0).round() as u64)
+            };
+            AccountQuotaSignals::new(None, rank)
+        });
+        (quota, signal.concurrency_limit)
    }

    pub(crate) async fn select_with_cyber_policy(
@@ -777,6 +782,11 @@ impl CodexCredentialSelector {
                    let health = self
                        .account_feedback
                        .scheduling_signals(&self.provider_kind, account.id());
+                    let (quota, inherited_concurrency) = self.scheduling_signals(
+                        &account,
+                        model,
+                        request.attempt.account_selection_policy().strategy(),
+                    );
                    let signals = scheduling
                        .signals()
                        .get(account.id())
@@ -790,14 +800,14 @@ impl CodexCredentialSelector {
                            failure_rate_basis_points: None,
                            first_output_latency_ms: None,
                        })
-                        .with_provider_quota(self.quota_signals(
-                            &account,
-                            model,
-                            request.attempt.account_selection_policy().strategy(),
-                        ))
+                        .with_provider_quota(quota)
                        .with_rate_limit(rate_limits.get(account.id()).copied().flatten())
                        .with_runtime_health(health.0, health.1);
-                    AccountCandidate { account, signals }
+                    AccountCandidate {
+                        account,
+                        signals,
+                        inherited_concurrency,
+                    }
                })
                .collect::<Vec<_>>();
            let mut affinity = binding
@@ -1060,7 +1070,7 @@ impl CodexCredentialSelector {
                            self.provider_kind.clone(),
                            account.id().clone(),
                            account.revision(),
-                            context.concurrency_limit(&account),
+                            context.concurrency_limit(selected),
                            request.attempt.deadline(),
                        )
                        .with_metadata(request.attempt.lease_metadata())
@@ -11,7 +11,7 @@ use provider_openai::credential::ResponsesTransport;
 fn peer_document() -> Value {
    let now = Utc::now();
    json!({
-        "object":"cpr.peer", "schema_version":2,
+        "object":"cpr.peer", "schema_version":3,
        "observed_at":now.to_rfc3339(), "expires_at":(now + chrono::Duration::minutes(5)).to_rfc3339(),
        "capabilities":["key_usage", "pool_summary"],
        "usage":{"status":"available","observed_at":now.to_rfc3339(),"data":{"unit":"USD","allowed":true,"windows":[
@@ -480,7 +480,7 @@ async fn model_only_snapshots_publish_signals_without_budget_or_account_disclosu
        let server = MockServer::start().await;
        let now = Utc::now();
        let document = json!({
-            "object":"cpr.peer", "schema_version":2,
+            "object":"cpr.peer", "schema_version":3,
            "observed_at":now.to_rfc3339(), "expires_at":(now + chrono::Duration::minutes(5)).to_rfc3339(),
            "capabilities":["model_availability"],
            "models":[{"model":"gpt-5.4", "coverage":coverage, "state":state,
@@ -502,7 +502,7 @@ async fn model_only_snapshots_publish_signals_without_budget_or_account_disclosu
            reader
                .signal(
                    &accounts.account(id.as_str()).unwrap(),
-                    "gpt-5.4",
+                    Some("gpt-5.4"),
                    SystemTime::now()
                )
                .map(|signal| signal.exhausted),
@@ -778,7 +778,7 @@ async fn invalid_or_redirected_responses_never_replace_the_last_valid_snapshot()
    cycle(&worker).await;
    let observed_at = quota(&bundle, &id).await.observed_at;
    let mut wrong_schema = peer_document();
-    wrong_schema["schema_version"] = json!(99);
+    wrong_schema["schema_version"] = json!(2);
    let mut wrong_object = peer_document();
    wrong_object["object"] = json!("other.peer");
    let mut negative = peer_document();
@@ -899,6 +899,127 @@ async fn peer_sync_excludes_budget_exhausted_accounts_without_pool_disclosure()
    server.verify().await;
 }

+#[tokio::test]
+async fn peer_key_limits_set_default_capacity_and_lease_without_model_or_pool_disclosure() {
+    use gateway_core::account::{AccountConcurrencyLimit, AccountWeight};
+    for (limits, observed, stale, local_limit, expected) in [
+        (
+            Some(json!({"max_concurrency":5})),
+            (Some(5), None),
+            false,
+            None,
+            5,
+        ),
+        (
+            Some(json!({"max_concurrency":5,"requests_per_minute":60})),
+            (Some(5), Some(60)),
+            false,
+            Some(1),
+            1,
+        ),
+        (
+            Some(json!({"max_concurrency":0})),
+            (Some(0), None),
+            false,
+            None,
+            2,
+        ),
+        (
+            Some(json!({"requests_per_minute":60})),
+            (None, Some(60)),
+            false,
+            None,
+            2,
+        ),
+        (None, (None, None), false, None, 2),
+        (
+            Some(json!({"max_concurrency":5})),
+            (Some(5), None),
+            true,
+            None,
+            2,
+        ),
+    ] {
+        let server = MockServer::start().await;
+        let now = Utc::now();
+        let observed_at = if stale {
+            now - chrono::Duration::minutes(6)
+        } else {
+            now
+        };
+        let mut document = json!({
+            "object":"cpr.peer", "schema_version":3,
+            "capabilities": ["key_limits"],
+            "observed_at":observed_at.to_rfc3339(), "expires_at":(now + chrono::Duration::minutes(2)).to_rfc3339()
+        });
+        if let Some(limits) = limits {
+            document["key_limits"] = limits;
+        }
+        Mock::given(method("GET"))
+            .and(path("/custom/v1/cpr/link"))
+            .respond_with(ResponseTemplate::new(200).set_body_json(document))
+            .expect(1)
+            .mount(&server)
+            .await;
+        Mock::given(method("POST"))
+            .and(path("/custom/v1/responses"))
+            .respond_with(
+                ResponseTemplate::new(200)
+                    .insert_header("content-type", "text/event-stream")
+                    .set_body_string(COMPLETED_SESSION_SSE),
+            )
+            .expect(1)
+            .mount(&server)
+            .await;
+        let (mut bundle, accounts, _, id, _config) = fixture(&server).await;
+        accounts.set_scheduling(
+            id.as_str(),
+            local_limit.and_then(AccountConcurrencyLimit::new),
+            AccountWeight::DEFAULT,
+        );
+        cycle(&worker(&mut bundle)).await;
+        let info = quota(&bundle, &id).await.upstream.unwrap();
+        assert_eq!((info.max_concurrency, info.requests_per_minute), observed);
+        let version = accounts
+            .get_quota_versions(std::slice::from_ref(&id))
+            .await
+            .unwrap()
+            .pop()
+            .unwrap();
+        let inherited = bundle
+            .admin_provider()
+            .observed_concurrency_limit(&version, SystemTime::now());
+        assert_eq!(
+            inherited.map(|limit| u64::from(limit.get())),
+            observed.0.filter(|limit| !stale && *limit > 0)
+        );
+        let mut stream = execute_peer_request(&bundle, "round_robin").await.unwrap();
+        let capacity = stream
+            .metadata()
+            .selection_observation()
+            .unwrap()
+            .capacity()
+            .unwrap();
+        assert_eq!(capacity.total_slots(), u64::from(expected));
+        assert_eq!(
+            bundle
+                .leases
+                .requests
+                .lock()
+                .unwrap()
+                .last()
+                .unwrap()
+                .max_concurrent()
+                .get(),
+            expected
+        );
+        while let Some(event) = stream.next().await {
+            event.unwrap();
+        }
+        server.verify().await;
+    }
+}
+
 #[tokio::test]
 async fn lists_reuse_versioned_summaries_and_details_keep_node_windows() {
    let server = MockServer::start().await;
@@ -1238,6 +1359,7 @@ async fn execute_peer_request(
 }

 struct PeerFixture {
+    leases: Arc<TestLeaseCoordinator>,
    provider: provider_openai::ProviderBundle,
    peer: gateway_admin::PeerUpstreamBundle,
    cooldowns: Arc<MemoryCooldownPort>,
@@ -1263,11 +1385,12 @@ async fn peer_fixture(
 ) -> PeerFixture {
    let reader = Arc::new(gateway_core::account::UpstreamInfoReader::default());
    let cooldowns = Arc::new(MemoryCooldownPort::new());
+    let leases = Arc::new(TestLeaseCoordinator::default());
    let provider = provider_openai::initialize(
        config,
        ProviderStorePorts::new(
            accounts.clone(),
-            Arc::new(TestLeaseCoordinator::default()),
+            leases.clone(),
            Arc::new(MemorySessionAffinity::default()),
            Arc::new(MemorySessionExclusions::default()),
            Arc::new(TestCatalogCache::default()),
@@ -1295,6 +1418,7 @@ async fn peer_fixture(
    .unwrap();
    reader.bind(&peer.scheduling_source()).unwrap();
    PeerFixture {
+        leases,
        provider,
        peer,
        cooldowns,
@@ -80,6 +80,7 @@ async fn old_weekly_quota_should_outweigh_one_in_flight_title_request() {
    let candidates = accounts
        .into_iter()
        .map(|account| AccountCandidate {
+            inherited_concurrency: None,
            signals: AccountRuntimeSignals {
                in_flight: u32::from(account.id().as_str() == "acct_74"),
                last_started_at: None,
@@ -241,7 +241,11 @@ impl GrokAccountSessionSelector {
                        .ok_or(GrokSessionSelectorError::Unavailable(None))?
                        .with_provider_quota(self.quota.scheduling_signals(&account))
                        .with_runtime_health(health.0, health.1);
-                    Ok(AccountCandidate { account, signals })
+                    Ok(AccountCandidate {
+                        account,
+                        signals,
+                        inherited_concurrency: None,
+                    })
                })
                .collect::<Result<Vec<_>, GrokSessionSelectorError>>()?;
            if let Some(required) = request.required_account() {
@@ -319,11 +323,7 @@ impl GrokAccountSessionSelector {
                            self.provider_kind.clone(),
                            selected_id.clone(),
                            selected_revision,
-                            selected.account.effective_concurrency(
-                                request
-                                    .account_selection_policy()
-                                    .max_concurrent_per_account(),
-                            ),
+                            context.concurrency_limit(selected),
                            request.deadline(),
                        )
                        .with_metadata(request.lease_metadata.clone())
@@ -399,17 +399,17 @@ PostgreSQL/Redis 启动密码。日常校验使用 `config --quiet`

 ### CPR 互联

-在上游的「API 密钥」中为专用 Key 选择账号分组并开启「CPR 互联」，按需公开账号池、额度和套餐信息。
+在上游的「API 密钥」中为专用密钥选择账号分组并开启「CPR 互联」，即可共享密钥预算、并发上限和 RPM，并按需公开账号池、账号额度和套餐信息。
 共享 Key 的实际推理只使用这些分组中的直连 OAuth 账号；分组内的 API Key 转供账号不会参与

 在下游导入或编辑 OpenAI API Key 账号，填写上游 API 地址和该 Key，后台自动发现 CPR 互联能力。
 地址包含自定义前缀时保留完整前缀，后台在该地址下请求 `cpr/link`。
-列表与详情显示已保存的观测，手动刷新等待上游查询完成；旧观测和查询失败分别提示，不要求上游实时可用
+列表与详情显示已保存的观测。展开账号可查看上游密钥限制，普通账号并发留空时跟随上游有效的正数上限。手动刷新等待上游查询完成，旧观测和查询失败分别提示

 第三方网关通过插件实现相同协议也可被识别，不按软件名称判定兼容。
 上游明确返回 Key 预算耗尽时，下游暂停该账号；后续同步确认允许使用后自动恢复，普通限流冷却独立保留。
 金额只供展示，缺少访问状态时不根据余额推断恢复。账号池额度观测作为 Smart 选择的参考，不能保证下一次请求成功
-互联两端需支持 CPR Link v2；协议、第三方接入与公开边界见 [CPR 互联 API](../docs/api.md#cpr-互联只读协议)
+互联两端均须使用 CPR Link v3；协议、第三方接入与公开边界见 [CPR 互联 API](../docs/api.md#cpr-互联只读协议)

 ### 批量维护代理

@@ -755,29 +755,31 @@ OpenAI 选号阶段确认本次可选账号全部额度耗尽时，HTTP 返回 `
 | 顶层必填字段 | 合同 |
 | --- | --- |
 | `object` | 固定为 `"cpr.peer"` |
-| `schema_version` | 当前为 `2`，不支持的版本按不兼容处理 |
-| `capabilities` | 当前 Key 获准公开的能力名称列表，最多 32 项；未知能力可以忽略 |
+| `schema_version` | 固定为 `3`，接入端只接受此版本 |
+| `capabilities` | 当前密钥 获准公开的能力名称列表，最多 32 项；未知能力可以忽略 |
 | `observed_at` | 本次快照组装时间，RFC3339 UTC 时间 |
 | `expires_at` | 快照到期时间，必须晚于 `observed_at`，且不超过其后 30 分钟 |

-`usage`、`pool`、`quota`、`models` 按能力独立提供，未提供的数据块直接省略。
+`usage`、`key_limits`、`pool`、`quota`、`models` 按能力独立提供，未提供的数据块直接省略。
 接入端只消费已声明能力的数据，缺失值保持未知，不解释为零余额、空池或额度恢复。
 未知字段可以忽略；没有独立节点 ID，连接由本地账号及其凭据、出口版本识别

 | 能力 | 对应数据 |
 | --- | --- |
 | `key_usage` | `usage`：当前认证 Key 的预算访问结论与可选金额窗口 |
+| `key_limits` | `key_limits`：当前认证密钥公开的并发上限和 RPM |
 | `pool_summary` | `pool`：账号数量、互斥状态计数及可选套餐分布 |
 | `quota_windows` | `quota.accounts`：匿名账号及各自额度窗口 |
 | `model_availability` | `models`：按当前 Key 公开的模型额度汇总，可独立于账号明细提供 |

-只公开预算访问结论的完整响应：
+提供请求限制和预算访问结论的响应示例：

 ```json
 {
  "object": "cpr.peer",
-  "schema_version": 2,
-  "capabilities": ["key_usage"],
+  "schema_version": 3,
+  "key_limits": { "max_concurrency": 5, "requests_per_minute": 60 },
+  "capabilities": ["key_usage", "key_limits"],
  "observed_at": "2026-11-01T12:00:00Z",
  "expires_at": "2026-11-01T12:05:00Z",
  "usage": {
@@ -788,6 +790,11 @@ OpenAI 选号阶段确认本次可选账号全部额度耗尽时，HTTP 返回 `
 }
 ```

+#### 密钥请求限制
+
+`key_limits` 形如 `{"max_concurrency":5,"requests_per_minute":60}`。上游可省略整个对象或其中任一项。已提供的值为非负整数，零表示该项不限；缺失表示未知。
+限制适用于整个密钥，有效期随顶层快照
+
 #### Key 预算

 `usage.status` 为 `available`、`unavailable` 或 `unknown`，未知状态按 `unknown` 处理。
@@ -883,7 +890,7 @@ CPR 默认路径为 `/v1/cpr/link`，使用同一数据面入口认证，不接

 | 共享配置 | 导出内容 |
 | --- | --- |
-| `enabled` | `key_usage`，包含明确访问结论及本地日、周美元账本；金额与周期口径见 [API Key 额度查询](#api-key-额度查询) |
+| `enabled` | `key_limits` 与 `key_usage`，包含密钥请求限制、明确访问结论及本地日、周美元账本；金额与周期口径见 [API Key 额度查询](#api-key-额度查询) |
 | `sharePool` | `pool_summary` |
 | `sharePlans` | 已公开账号池中的 `pool.plans` |
 | `shareQuota` | `quota_windows` 与 `model_availability` |
@@ -1085,8 +1092,7 @@ config 返回 `{ name, plaintextKey }`，仅读取服务端会话绑定的当前
 账号列表的 `search` 仅按字面量包含匹配名称和邮箱，不区分大小写；`%`、`_` 和 `\` 按普通字符处理

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

 并发明细返回 `accountId`、`entries`、`total`、`page`、`pageSize` 和 `observedAt`，只统计普通并发池。每条包含 `leaseId`、`requestId`、`clientKeyId`、`clientKeyName`、`model`（入口请求模型）、`clientIp`、`startedAt`、`startedAtDisplay` 和 `occupiedMs`。占用时长从取得账号租约开始，不包含排队；缺少入口事实时相应字段为 `null`，租约仍计入总数

@@ -1159,6 +1165,7 @@ OpenAI 主动额度刷新和正常响应携带的明确套餐会同步到账号
 | `stale` | 快照是否已过期；过期值仅供查看 |
 | `connectionRevision` | 当前连接身份的不可解释标识，用于隔离地址、密钥或出口变化前的展示结果 |
 | `lastError` | 当前实例最近同步失败分类：`unauthorized`、`unsupported`、`rate_limited`、`invalid_response`、`unavailable`；已保存的不支持结论及已确认互联的失权记录重启后仍返回对应分类，同步成功时为 `null` |
+| `maxConcurrency`、`requestsPerMinute` | 上游密钥的并发上限和 RPM，零表示不限，未观测到的项为 `null` |
 | `pool` | 上游授权公开的账号数量、状态数量及套餐数量，未公开时为 `null` |

 互联额度复用 `quota.windows`。上游 Key 账本按实际提供的周期展示，不补造缺失窗口；不限额也保留已用金额。
@@ -2042,7 +2049,8 @@ UA、平台参数及自动 Git 上下文版本由 [OpenAI 上游客户端身份]
 ### 并发与排队

 `maxConcurrentPerAccount` 是默认账号并发上限，初始值为 3，取值 0～4294967295；`0` 表示不限制。
-账号的 `concurrencyLimit: null` 继承该默认值，单独设置的正数上限仍优先生效。
+账号单独设置的正数 `concurrencyLimit` 优先。留空时，互联账号使用有效快照中密钥的正数并发上限，最大取 4294967295。
+无有效的上游正数上限时使用本机全局默认值。上游 RPM 供展示，由上游执行。
 无限并发仍统计在途请求，并遵守账号可用性与 Client Key 限制

 `maxWaitingPerKey` 与 `maxWaitingPerAccount` 是全局统一的普通排队容量，取值 0～1,000，默认 10，`0` 表示关闭；
@@ -1063,6 +1063,7 @@ OAuth 额度解析属于 Provider，CPR 发现、解析与投影属于 Admin；

 Admin 按连接身份及额度观察时间缓存管理摘要与调度投影，解析和投影在锁外完成。
 Core 定义中立调度信号与只读端口，组合根绑定观测来源；Provider 选号不持有控制面服务，弱引用避免互相保活。
+互联密钥的并发上限随有效观测参与普通账号选号和租约准入，账号手动配置优先。分组与概览按额度观测版本核对容量，账号配置与观测各自保存。
 账号和凭据提交后的共用通知更新原生 Provider 与互联派生状态，插件账号写入复用同一流程；
 互联按当前连接身份失效观测，普通信息编辑保留已有摘要与刷新退避，停用或删除时释放运行态。
 账号列表、概览与代理关联账号核对观测版本后复用预算与池摘要，详情读取逐节点窗口；手动刷新通过共用账号额度接口等待查询完成并返回结果。
@@ -189,6 +189,8 @@ export interface AccountUpstream {
  stale: boolean
  connectionRevision: string
  lastError: 'unauthorized' | 'unsupported' | 'rate_limited' | 'invalid_response' | 'unavailable' | null
+  maxConcurrency: number | null
+  requestsPerMinute: number | null
  pool: {
    accountCount: number
    states: { ready: number, cooldown: number, exhausted: number, disabled: number, error: number, unknown: number }
@@ -631,7 +631,9 @@
      "invalidResponse": "Invalid upstream data",
      "unavailable": "Upstream unavailable",
      "pool": "Shared pool",
-      "ready": "available",
+      "readyAccounts": "Ready / total accounts",
+      "keyConcurrency": "Upstream key concurrency limit",
+      "keyRpm": "Upstream key requests per minute",
      "otherPlans": "Other plans",
      "morePlans": "View {count} more plans",
      "nodeQuota": "Node quota",
@@ -79,7 +79,7 @@
      "invalidKey": "Use visible HTTP header characters, without spaces or line breaks",
      "requestsPerMinute": "Requests per minute",
      "peerSharing": "CPR Link",
-      "peerSharingHelp": "Allow other nodes to read this key’s budget; requests use only direct accounts in the selected groups",
+      "peerSharingHelp": "Share this key’s budget, concurrency limit and RPM with other nodes. Model requests use only direct accounts in the selected groups",
      "sharePool": "Share account pool summary",
      "shareQuota": "Share account quotas",
      "sharePlans": "Share plan breakdown",
@@ -631,7 +631,9 @@
      "invalidResponse": "上游数据异常",
      "unavailable": "上游暂不可用",
      "pool": "共享池",
-      "ready": "可用",
+      "readyAccounts": "可用账号 / 总账号",
+      "keyConcurrency": "上游密钥并发上限",
+      "keyRpm": "上游密钥每分钟请求数",
      "otherPlans": "其他套餐",
      "morePlans": "查看其余 {count} 种套餐",
      "nodeQuota": "节点额度",
@@ -79,7 +79,7 @@
      "invalidKey": "自定义 API密钥只能包含 HTTP 可传输的可见字符，不能包含空格或换行",
      "requestsPerMinute": "每分钟请求数",
      "peerSharing": "CPR 互联",
-      "peerSharingHelp": "允许其他节点查询此 Key 的额度，模型请求仅使用所选分组中的直连账号",
+      "peerSharingHelp": "允许其他节点查询此密钥的预算、并发上限和 RPM，模型请求仅使用所选分组中的直连账号",
      "sharePool": "共享账号池概览",
      "shareQuota": "共享账号额度",
      "sharePlans": "共享套餐构成",
@@ -631,7 +631,9 @@
      "invalidResponse": "上游資料異常",
      "unavailable": "上游暫不可用",
      "pool": "共享池",
-      "ready": "可用",
+      "readyAccounts": "可用帳號 / 總帳號",
+      "keyConcurrency": "上游密鑰並行上限",
+      "keyRpm": "上游密鑰每分鐘請求數",
      "otherPlans": "其他套餐",
      "morePlans": "查看其餘 {count} 種套餐",
      "nodeQuota": "節點額度",
@@ -79,7 +79,7 @@
      "invalidKey": "自定義 API密鑰只能包含 HTTP 可傳輸的可見字符，不能包含空格或換行",
      "requestsPerMinute": "每分鐘請求數",
      "peerSharing": "CPR 互聯",
-      "peerSharingHelp": "允許其他節點查詢此 Key 的額度，模型請求僅使用所選分組中的直連帳號",
+      "peerSharingHelp": "允許其他節點查詢此密鑰的預算、並行上限和 RPM，模型請求僅使用所選分組中的直連帳號",
      "sharePool": "共享帳號池概覽",
      "shareQuota": "共享帳號額度",
      "sharePlans": "共享方案組成",
@@ -1,17 +1,19 @@
 <script setup lang="ts">
 import type { AccountQuotaWindow, AccountUpstream } from '@/api'
 import { ZPopover, ZScrollbar } from '@codex-proxy/ui'
+import { Gauge, Network, Users } from '@lucide/vue'
 import { computed } from 'vue'
 import { useI18n } from 'vue-i18n'
 import AccountPlanBadge from '@/components/account/AccountPlanBadge.vue'
 import { poolNodes } from './pool'
 import PoolChart from './PoolChart.vue'

-const props = defineProps<{ pool?: AccountUpstream['pool'], windows: AccountQuotaWindow[] }>()
+const props = defineProps<{ upstream: AccountUpstream | null, windows: AccountQuotaWindow[] }>()
 const { t } = useI18n({ useScope: 'global' })
-const nodes = computed(() => poolNodes(props.windows, props.pool?.accountCount))
+const pool = computed(() => props.upstream?.pool)
+const nodes = computed(() => poolNodes(props.windows, pool.value?.accountCount))
 const plans = computed(() => {
-  const sorted = props.pool?.plans.slice().sort((left, right) => right.count - left.count) ?? []
+  const sorted = pool.value?.plans.slice().sort((left, right) => right.count - left.count) ?? []
  return {
    visible: sorted.slice(0, 3),
    overflow: sorted.slice(3),
@@ -22,12 +24,24 @@ const plans = computed(() => {
 <template>
  <section class="grid min-w-0 gap-1" :aria-label="t('accounts.upstream.pool')">
    <PoolChart v-if="nodes.length" :nodes="nodes" />
-    <div v-if="pool" class="flex min-h-5 min-w-0 items-end justify-between gap-x-3">
-      <div class="flex shrink-0 items-baseline gap-1 whitespace-nowrap" :aria-label="`${pool.states.ready} / ${pool.accountCount} ${t('accounts.upstream.ready')}`">
-        <strong class="font-mono text-z-xs leading-[normal] font-heavy tabular-nums" :class="pool.states.ready > 0 ? 'text-z-success-text' : 'text-z-text'">{{ pool.states.ready }}</strong>
-        <span class="font-mono text-z-xs leading-[normal] tabular-nums text-z-text-quaternary">/ {{ pool.accountCount }}</span>
+    <div v-if="pool || upstream?.maxConcurrency != null || upstream?.requestsPerMinute != null" class="flex min-h-5 min-w-0 flex-wrap items-end justify-between gap-x-3 gap-y-1.5">
+      <div class="flex flex-wrap items-center gap-x-3 gap-y-1 font-mono text-z-xs leading-none tabular-nums text-z-text-secondary">
+        <span v-if="pool" class="inline-flex items-center gap-1 whitespace-nowrap" :title="t('accounts.upstream.readyAccounts')">
+          <Users aria-hidden="true" :stroke-width="1.75" class="size-2.75 text-z-text-tertiary" />
+          <span class="sr-only">{{ t('accounts.upstream.readyAccounts') }}</span>
+          <strong class="font-heavy" :class="pool.states.ready > 0 ? 'text-z-success-text' : 'text-z-text'">{{ pool.states.ready }}</strong>
+          <span class="text-z-text-quaternary">/ {{ pool.accountCount }}</span>
+        </span>
+        <span v-if="upstream?.maxConcurrency != null" class="inline-flex items-center gap-1" :title="t('accounts.upstream.keyConcurrency')">
+          <Network aria-hidden="true" :stroke-width="1.75" class="size-2.25 text-z-text-tertiary" />
+          <span class="sr-only">{{ t('accounts.upstream.keyConcurrency') }}</span>{{ upstream.maxConcurrency || '∞' }}
+        </span>
+        <span v-if="upstream?.requestsPerMinute != null" class="inline-flex items-center gap-1" :title="t('accounts.upstream.keyRpm')">
+          <Gauge aria-hidden="true" :stroke-width="1.75" class="size-2.75 text-z-text-tertiary" />
+          <span class="sr-only">{{ t('accounts.upstream.keyRpm') }}</span>{{ upstream.requestsPerMinute || '∞' }}
+        </span>
      </div>
-      <div v-if="pool.plans.length" class="flex min-w-0 items-end justify-end gap-1.5">
+      <div v-if="pool?.plans.length" class="ml-auto flex min-w-0 flex-wrap items-end justify-end gap-1.5">
        <AccountPlanBadge
          v-for="plan in plans.visible" :key="plan.name"
          :plan-type="plan.name" :plan-type-display="plan.name" :count="plan.count"
@@ -32,7 +32,8 @@ const emit = defineEmits<{
 const { t } = useI18n({ useScope: 'global' })
 const poolWindows = usePoolQuotaWindows(() => props.account)
 const hasUpstreamPool = computed(() => props.account.upstream?.discovered && (
-  props.account.upstream.pool || poolWindows.value.some(window => window.sourceAccountId && !window.amount)
+  props.account.upstream.pool || props.account.upstream.maxConcurrency != null || props.account.upstream.requestsPerMinute != null
+  || poolWindows.value.some(window => window.sourceAccountId && !window.amount)
 ))

 const quotaEntries = computed(() => groupedAccountQuotaWindows(
@@ -138,7 +139,7 @@ watch(hasPersonalInfo, (available) => {
          />
        </div>
      </ZScrollbar>
-      <AccountUpstreamPool v-if="hasUpstreamPool" class="mt-auto" :pool="account.upstream?.pool" :windows="poolWindows" />
+      <AccountUpstreamPool v-if="hasUpstreamPool" class="mt-auto" :upstream="account.upstream" :windows="poolWindows" />
    </div>
  </section>