fix(accounts): start peer discovery after credential imports

Reuse quota routing for initial peer observation and preserve discovery state across ordinary account edits. Hide pending subscription badges and remove redundant provider quota labels without adding frontend polling. Co-authored-by: Codex <noreply@openai.com>
2026-11-06 08:06:19 +00:00 · 2026-11-01 03:10:18 +08:00
parent 1e57805598
commit cb79f4d5a5
16 changed files with 356 additions and 35 deletions
@@ -19,30 +19,60 @@ use gateway_core::{account::ProviderAccountId, routing::ProviderKind, runtime::S
 use super::{map_store_error, publish_committed};

 pub(super) async fn publish_credentials_and_observe_quota(
-    peer: Option<&dyn crate::ports::peer::PeerUpstreamService>,
+    peer: Option<&std::sync::Arc<dyn crate::ports::peer::PeerUpstreamService>>,
    provider: &std::sync::Arc<dyn crate::ports::provider::ProviderAdmin>,
+    accounts: &std::sync::Arc<dyn AccountStore>,
    snapshot: &dyn SnapshotControl,
    revision: crate::model::Revision,
    account_ids: &[ProviderAccountId],
    request_id: &str,
 ) -> Result<(), AdminError> {
-    notify_account_facts_changed(peer, provider.as_ref(), account_ids).await;
+    notify_account_facts_changed(peer.map(AsRef::as_ref), provider.as_ref(), account_ids).await;
    publish_committed(snapshot, revision).await?;

    // 凭据已经提交；额度只是可重建的观察，不能拖住管理请求或回滚提交结果
    let provider = provider.clone();
+    let peer = peer.cloned();
+    let accounts = accounts.clone();
    let account_ids = account_ids.to_vec();
    let request_id = request_id.to_owned();
    tokio::spawn(async move {
        for account_id in account_ids {
-            if let Err(error) = provider
-                .quota(ProviderQuotaRequest {
-                    account_id: account_id.clone(),
-                    refresh: true,
-                    rolling_usage: None,
-                })
+            let request = ProviderQuotaRequest {
+                account_id: account_id.clone(),
+                refresh: true,
+                rolling_usage: None,
+            };
+            let result = if let Some(peer) = peer.as_deref() {
+                let details = match required_credential(
+                    accounts.as_ref(),
+                    provider.provider_kind(),
+                    &account_id,
+                    "initial quota observation",
+                )
                .await
-            {
+                {
+                    Ok(details) => details,
+                    Err(error) => {
+                        tracing::warn!(request_id, account_id = %account_id, error = %error, "initial quota account lookup failed");
+                        continue;
+                    }
+                };
+                if !details.credential.enabled {
+                    continue;
+                }
+                // 首查与手动刷新共用认证能力分派，互联协议由 Admin 解释
+                super::peer_link::upstream::read_quota(
+                    Some(peer),
+                    provider.as_ref(),
+                    &details.credential,
+                    request,
+                )
+                .await
+            } else {
+                provider.quota(request).await
+            };
+            if let Err(error) = result {
                tracing::warn!(
                    request_id,
                    provider = %provider.provider_kind().as_str(),
@@ -92,8 +92,9 @@ impl ProviderCredentials {
            .map_err(|error| map_store_error(error, "provider credential import"))?;
        drop(proxy_reservation);
        publish_credentials_and_observe_quota(
-            self.upstream_info.as_deref(),
+            self.upstream_info.as_ref(),
            &self.provider,
+            &self.accounts,
            self.snapshot.as_ref(),
            result.config_revision,
            &result.credential_ids,
@@ -249,8 +250,9 @@ impl super::CredentialsService {
        let result = outcome.result;
        if outcome.newly_committed {
            publish_credentials_and_observe_quota(
-                self.upstream_info.as_deref(),
+                self.upstream_info.as_ref(),
                provider,
+                &self.accounts,
                self.snapshot.as_ref(),
                result.config_revision,
                std::slice::from_ref(&result.account_id),
@@ -61,7 +61,40 @@ impl PeerInfoService {
    }

    pub(crate) async fn account_changed(&self, id: &ProviderAccountId) -> Result<(), StoreError> {
-        self.forget(id);
+        let has_observation = self
+            .observations
+            .read()
+            .unwrap_or_else(std::sync::PoisonError::into_inner)
+            .contains_key(id);
+        let has_attempt = self
+            .attempts
+            .lock()
+            .unwrap_or_else(std::sync::PoisonError::into_inner)
+            .contains_key(id);
+        if !has_observation && !has_attempt {
+            return Ok(());
+        }
+        let Some(account) = self
+            .accounts
+            .get_account(id)
+            .await?
+            .filter(|account| self.supported(account))
+        else {
+            self.forget(id);
+            return Ok(());
+        };
+        let connection = UpstreamConnection::for_account(&account);
+        // 普通信息编辑保留观测与退避，连接变化由刷新状态中的身份比较处理
+        let mut observations = self
+            .observations
+            .write()
+            .unwrap_or_else(std::sync::PoisonError::into_inner);
+        if observations
+            .get(id)
+            .is_some_and(|cached| cached.connection != connection)
+        {
+            observations.remove(id);
+        }
        Ok(())
    }

@@ -173,8 +173,9 @@ impl PluginAccountAccess for DefaultPluginAccountAccess {
                .credential
                .credential_revision;
                publish_credentials_and_observe_quota(
-                    self.upstream_info.as_deref(),
+                    self.upstream_info.as_ref(),
                    &provider,
+                    &self.accounts,
                    self.snapshot.as_ref(),
                    result.config_revision,
                    std::slice::from_ref(&account_id),
@@ -217,8 +218,9 @@ impl PluginAccountAccess for DefaultPluginAccountAccess {
                    .credential_revision
                    .ok_or_else(|| AdminError::internal("账号保存未返回凭据版本"))?;
                publish_credentials_and_observe_quota(
-                    self.upstream_info.as_deref(),
+                    self.upstream_info.as_ref(),
                    &provider,
+                    &self.accounts,
                    self.snapshot.as_ref(),
                    result.config_revision,
                    std::slice::from_ref(&result.account_id),
@@ -279,6 +279,10 @@ impl FakeProviderAdmin {

 #[async_trait]
 impl ProviderAdmin for FakeProviderAdmin {
+    fn peer_authentication_kinds(&self) -> &[&str] {
+        &["api_key"]
+    }
+
    fn account_capabilities(
        &self,
        _account_id: &ProviderAccountId,
@@ -1,4 +1,5 @@
 //! 按 Provider 组织凭据导入、授权与提交边界测试

 mod openai;
+mod peer;
 mod xai;
@@ -0,0 +1,184 @@
+//! 验证凭据提交后的互联首查、原生额度分派与异步失败边界
+
+use std::sync::{Arc, Mutex};
+use std::time::Duration;
+
+use async_trait::async_trait;
+use gateway_admin::{
+    model::provider_credentials::{ImportCredentials, ProviderQuota, ProviderQuotaReadRequest},
+    ports::{
+        peer::PeerUpstreamService,
+        provider::{ProviderAdminError, ProviderAdminErrorKind},
+    },
+};
+use gateway_core::{
+    account::{ProviderAccount, ProviderAccountId, UpstreamInfoSource, UpstreamSchedulingSignal},
+    error::StoreError,
+    routing::ProviderKind,
+};
+
+use super::super::{
+    AdminHarness,
+    accounts::{FakeAccountStore, FakeProviderAdmin, account_record, context, document, events},
+};
+
+#[tokio::test]
+async fn imported_peer_accounts_should_start_observation_without_waiting_for_network() {
+    let events = events();
+    let provider = FakeProviderAdmin::new("openai", events.clone());
+    provider.set_import_account_ids(&["acct_first", "acct_second"]);
+    let mut first = account_record("openai");
+    first.id = "acct_first".into();
+    first.authentication_kind = "api_key".into();
+    let mut second = first.clone();
+    second.id = "acct_second".into();
+    let store = FakeAccountStore::with_account(first.clone(), events);
+    store.set_accounts(vec![first, second]);
+    let peer = Arc::new(PendingPeer::default());
+    let services = AdminHarness::new()
+        .provider(provider.clone())
+        .accounts(store)
+        .upstream_info(peer.clone())
+        .build()
+        .await;
+
+    let result = tokio::time::timeout(
+        Duration::from_secs(1),
+        services
+            .credentials()
+            .for_provider(&ProviderKind::new("openai").unwrap())
+            .unwrap()
+            .import_document(ImportCredentials {
+                outbound_proxy_id: None,
+                settings: None,
+                context: context("import-peer"),
+                document: document(),
+            }),
+    )
+    .await
+    .expect("import must not wait for peer response")
+    .expect("import committed");
+    peer.wait_for_requests(1).await;
+    assert_eq!(peer.changed.lock().unwrap().len(), 2);
+    assert_eq!(result.credential_ids.len(), 2);
+    assert!(provider.quota_requests().is_empty());
+    peer.gate.add_permits(2);
+    peer.wait_for_requests(2).await;
+    assert_eq!(
+        *peer.requests.lock().unwrap(),
+        ["acct_first", "acct_second"]
+    );
+    assert!(provider.quota_requests().is_empty());
+}
+
+#[tokio::test]
+async fn native_import_should_use_provider_quota_when_peer_service_is_available() {
+    let events = events();
+    let provider = FakeProviderAdmin::new("openai", events.clone());
+    provider.set_import_account_ids(&["acct_test"]);
+    let store = FakeAccountStore::new("openai", events);
+    let peer = Arc::new(PendingPeer::default());
+    let services = AdminHarness::new()
+        .provider(provider.clone())
+        .accounts(store)
+        .upstream_info(peer.clone())
+        .build()
+        .await;
+
+    services
+        .credentials()
+        .for_provider(&ProviderKind::new("openai").unwrap())
+        .unwrap()
+        .import_document(ImportCredentials {
+            outbound_proxy_id: None,
+            settings: None,
+            context: context("import-native"),
+            document: document(),
+        })
+        .await
+        .expect("native import");
+    provider.wait_for_quota_requests(1).await;
+    assert!(peer.requests.lock().unwrap().is_empty());
+    assert_eq!(
+        provider.quota_requests()[0].account_id.as_str(),
+        "acct_test"
+    );
+}
+
+struct PendingPeer {
+    changed: Mutex<Vec<String>>,
+    requests: Mutex<Vec<String>>,
+    started: tokio::sync::Notify,
+    gate: tokio::sync::Semaphore,
+}
+
+impl Default for PendingPeer {
+    fn default() -> Self {
+        Self {
+            changed: Mutex::default(),
+            requests: Mutex::default(),
+            started: tokio::sync::Notify::new(),
+            gate: tokio::sync::Semaphore::new(0),
+        }
+    }
+}
+
+impl PendingPeer {
+    async fn wait_for_requests(&self, count: usize) {
+        tokio::time::timeout(Duration::from_secs(1), async {
+            loop {
+                let notified = self.started.notified();
+                if self.requests.lock().unwrap().len() >= count {
+                    return;
+                }
+                notified.await;
+            }
+        })
+        .await
+        .expect("initial peer observation started");
+    }
+}
+
+impl UpstreamInfoSource for PendingPeer {
+    fn signal(
+        &self,
+        _: &ProviderAccount,
+        _: &str,
+        _: std::time::SystemTime,
+    ) -> Option<UpstreamSchedulingSignal> {
+        None
+    }
+}
+
+#[async_trait]
+impl PeerUpstreamService for PendingPeer {
+    async fn account_changed(&self, id: &ProviderAccountId) -> Result<(), StoreError> {
+        self.changed.lock().unwrap().push(id.as_str().into());
+        Ok(())
+    }
+    fn forget(&self, _: &ProviderAccountId) {}
+    async fn refresh_quota(
+        &self,
+        id: &ProviderAccountId,
+    ) -> Result<ProviderQuota, ProviderAdminError> {
+        self.requests.lock().unwrap().push(id.as_str().into());
+        self.started.notify_one();
+        self.gate.acquire().await.unwrap().forget();
+        Err(ProviderAdminError::new(ProviderAdminErrorKind::Unavailable))
+    }
+    async fn read_quota(&self, _: &ProviderAccountId) -> Result<Option<ProviderQuota>, StoreError> {
+        panic!("unexpected cached read")
+    }
+    async fn read_summary(
+        &self,
+        _: &ProviderAccountId,
+    ) -> Result<Option<ProviderQuota>, StoreError> {
+        panic!("unexpected summary read")
+    }
+    async fn read_quotas(
+        &self,
+        _: &[ProviderQuotaReadRequest<'_>],
+    ) -> Vec<Result<ProviderQuota, ProviderAdminError>> {
+        panic!("unexpected batch read")
+    }
+}
@@ -112,6 +112,7 @@ pub(super) struct AdminHarness {
    plugin_store: Arc<dyn gateway_admin::ports::plugins::PluginStore>,
    plugin_inspector: Arc<dyn gateway_admin::ports::plugins::PluginPackageInspector>,
    client_key_verifier: Arc<dyn ClientKeyVerifier>,
+    upstream_info: Option<Arc<dyn gateway_admin::ports::peer::PeerUpstreamService>>,
    service_middleware: gateway_admin::public_service::PlanSource,
 }

@@ -142,6 +143,7 @@ impl AdminHarness {
            plugin_store: Arc::new(plugins::TestPluginPorts),
            plugin_inspector: Arc::new(plugins::TestPluginPorts),
            client_key_verifier: Arc::new(UnavailableClientKeyVerifier),
+            upstream_info: None,
            service_middleware: Arc::new(|| None),
        }
    }
@@ -161,6 +163,14 @@ impl AdminHarness {
        self
    }

+    pub(super) fn upstream_info(
+        mut self,
+        peer: Arc<dyn gateway_admin::ports::peer::PeerUpstreamService>,
+    ) -> Self {
+        self.upstream_info = Some(peer);
+        self
+    }
+
    pub(super) fn account_runtime(mut self, store: Arc<dyn AccountRuntimeStore>) -> Self {
        self.account_runtime = store;
        self
@@ -287,7 +297,7 @@ impl AdminHarness {
                Arc::new(plugins::TestPluginPorts),
            ),
            gateway_admin::AdminRuntimePorts {
-                upstream_info: None,
+                upstream_info: self.upstream_info,
                timezone: Default::default(),
                service_middleware: self.service_middleware,
                plugin_preparation: Arc::new(plugins::TestPluginPorts),
@@ -33,21 +33,22 @@ async fn discovery_uses_provider_capabilities_and_publishes_neutral_signals() {
        });
        let registry = ProviderAdminRegistry::new(providers).unwrap();
        let accounts = Arc::new(TestAccounts {
-            accounts: vec![
+            accounts: Mutex::new(vec![
                account("acct_first", "openai", "peer_test"),
                account("acct_second", "xai", "peer_test"),
                account("acct_native", "xai", "oauth"),
-            ],
+            ]),
            records: Mutex::default(),
            failed_provider,
        });
        let transport = Arc::new(TestTransport(Mutex::new(Vec::new())));
        let mut bundle =
            initialize_peer_upstream(accounts.clone(), &registry, transport.clone()).unwrap();
+        let initial_accounts = accounts.accounts.lock().unwrap().clone();
        let reader = UpstreamInfoReader::default();
        assert!(
            reader
-                .signal(&accounts.accounts[0], "model", SystemTime::now())
+                .signal(&initial_accounts[0], "model", SystemTime::now())
                .is_none()
        );
        let source = bundle.scheduling_source();
@@ -77,7 +78,7 @@ async fn discovery_uses_provider_capabilities_and_publishes_neutral_signals() {
                vec!["acct_first", "acct_second"]
            }
        );
-        for account in &accounts.accounts[..2] {
+        for account in &initial_accounts[..2] {
            if Some(account.provider().as_str()) == failed_provider {
                assert!(reader.signal(account, "model", SystemTime::now()).is_none());
                continue;
@@ -101,19 +102,65 @@ async fn discovery_uses_provider_capabilities_and_publishes_neutral_signals() {
        }
        assert!(
            reader
-                .signal(&accounts.accounts[2], "model", SystemTime::now())
+                .signal(&initial_accounts[2], "model", SystemTime::now())
                .is_none()
        );
        drop(worker);
        drop(bundle);
        assert!(
            reader
-                .signal(&accounts.accounts[0], "model", SystemTime::now())
+                .signal(&initial_accounts[0], "model", SystemTime::now())
                .is_none()
        );
    }
 }

+#[tokio::test]
+async fn account_edits_preserve_observation_and_refresh_budget_until_connection_changes() {
+    let mut provider = UnavailableProvider::new("openai");
+    provider.peer_enabled = true;
+    let registry =
+        ProviderAdminRegistry::new([Arc::new(provider) as Arc<dyn ProviderAdmin>]).unwrap();
+    let initial = account("acct_peer", "openai", "peer_test");
+    let accounts = Arc::new(TestAccounts {
+        accounts: Mutex::new(vec![initial.clone()]),
+        records: Mutex::default(),
+        failed_provider: None,
+    });
+    let transport = Arc::new(TestTransport(Mutex::new(Vec::new())));
+    let bundle = initialize_peer_upstream(accounts.clone(), &registry, transport.clone()).unwrap();
+    let service = bundle.service();
+    service.refresh_quota(initial.id()).await.unwrap();
+    let before = service.read_summary(initial.id()).await.unwrap().unwrap();
+
+    let edited = initial.clone().with_notes(Some("仅修改备注".into()));
+    *accounts.accounts.lock().unwrap() = vec![edited.clone()];
+    service.account_changed(edited.id()).await.unwrap();
+    assert!(
+        bundle
+            .scheduling_source()
+            .signal(&edited, "model", SystemTime::now())
+            .is_some()
+    );
+    let after = service.refresh_quota(edited.id()).await.unwrap();
+    assert_eq!(after.upstream, before.upstream);
+    assert_eq!(transport.0.lock().unwrap().len(), 1);
+
+    let reconnected = edited.with_outbound_proxy(Some(
+        OutboundProxy::parse("http://proxy.invalid:8080").unwrap(),
+    ));
+    *accounts.accounts.lock().unwrap() = vec![reconnected.clone()];
+    service.account_changed(reconnected.id()).await.unwrap();
+    assert!(
+        bundle
+            .scheduling_source()
+            .signal(&reconnected, "model", SystemTime::now())
+            .is_none()
+    );
+    service.refresh_quota(reconnected.id()).await.unwrap();
+    assert_eq!(transport.0.lock().unwrap().len(), 2);
+}
+
 fn account(id: &str, provider: &str, authentication: &str) -> ProviderAccount {
    ProviderAccount::new(
        ProviderAccountId::new(id).unwrap(),
@@ -165,7 +212,7 @@ impl PeerTransport for TestTransport {
 }

 struct TestAccounts {
-    accounts: Vec<ProviderAccount>,
+    accounts: Mutex<Vec<ProviderAccount>>,
    records: Mutex<BTreeMap<ProviderAccountId, QuotaObservation>>,
    failed_provider: Option<&'static str>,
 }
@@ -188,6 +235,8 @@ impl ProviderAccountStore for TestAccounts {
    ) -> Result<Option<ProviderAccount>, StoreError> {
        Ok(self
            .accounts
+            .lock()
+            .unwrap()
            .iter()
            .find(|account| account.id() == account_id)
            .cloned())
@@ -204,6 +253,8 @@ impl ProviderAccountStore for TestAccounts {
        }
        Ok(self
            .accounts
+            .lock()
+            .unwrap()
            .iter()
            .filter(|account| account.provider() == provider)
            .cloned()
@@ -227,6 +278,8 @@ impl ProviderAccountStore for TestAccounts {
        id: &ProviderAccountId,
    ) -> Result<LoadedCredential, StoreError> {
        self.accounts
+            .lock()
+            .unwrap()
            .iter()
            .find(|account| account.id() == id)
            .map(loaded)
@@ -1012,7 +1012,8 @@ Provider 声明可连接的认证种类，并从本轮凭据中提取连接材
 Host 执行有界 HTTP 请求，复用组合根注入的证书策略和账号出口；Store 通过通用账号额度端口保存不透明观测，不解释互联或 Provider 协议。
 互联与原生额度共用 `provider_quota_json`、`quota_observed_at`、凭据版本和额度访问事实，每个账号仅保存当前来源的快照。
 OAuth 额度解析属于 Provider，CPR 发现、解析与投影属于 Admin；账号用例按能力分派读写与刷新，不将两种协议合并为同一解析服务。
-互联刷新沿用运行期排程和 Host Worker 租约，分批轮转账号，同一账号的并发刷新通过运行期锁合并；重启后根据已有快照恢复查询，不持久化领取或请求代次。
+凭据新建、导入或授权提交后，后台首查复用同一额度分派，不等待周期同步或阻塞保存请求。
+互联周期刷新沿用运行期排程和 Host Worker 租约，分批轮转账号，同一账号的并发刷新通过运行期锁合并；重启后根据已有快照恢复查询，不持久化领取或请求代次。
 额度时间戳支持轻量缓存校验，普通账号和凭据查询不加载额度正文，完整观测分批读取

 共享端按 Core 的预算规则从本地账本生成明确的访问状态，与本地准入共用口径；金额字段只供接入端展示。
@@ -1023,7 +1024,8 @@ OAuth 额度解析属于 Provider，CPR 发现、解析与投影属于 Admin；

 Admin 按连接身份及额度观察时间缓存管理摘要与调度投影，解析和投影在锁外完成。
 Core 定义中立调度信号与只读端口，组合根绑定观测来源；Provider 选号不持有控制面服务，弱引用避免互相保活。
-账号和凭据提交后的共用通知同时失效原生 Provider 与互联派生状态，插件账号写入复用同一流程。
+账号和凭据提交后的共用通知更新原生 Provider 与互联派生状态，插件账号写入复用同一流程；
+互联按当前连接身份失效观测，普通信息编辑保留已有摘要与刷新退避，停用或删除时释放运行态。
 账号列表、概览与代理关联账号核对观测版本后复用预算与池摘要，详情读取逐节点窗口；手动刷新通过共用账号额度接口等待查询完成并返回结果。
 界面复用现有额度窗口，保留上游原始观测时间；Key 金额、原生账号百分比和本机 Token 用量分别解释。
 请求与字段合同见 [CPR 互联 API](api.md#cpr-互联只读协议)
@@ -2,6 +2,7 @@
 import type { AccountUpstream } from '@/api'
 import { computed } from 'vue'

+import { accountUpstreamPending } from '@/utils/account-upstream'
 import { stableVisualIndex } from '@/utils/color'
 import AccountUpstreamBadge from './AccountUpstreamBadge.vue'

@@ -62,7 +63,7 @@ const paletteClass = computed(() => {
 <template>
  <AccountUpstreamBadge v-if="upstream?.discovered && upstream.observedAt" :size="size" />
  <span
-    v-else
+    v-else-if="!accountUpstreamPending(upstream)"
    class="inline-flex min-w-0 max-w-full items-center justify-center whitespace-nowrap leading-none shadow-z-tertiary"
    :class="[sizeClass, paletteClass]"
    :title="displayText || undefined"
@@ -25,7 +25,6 @@
      "reset": "Reset:",
      "remainingQuota": "Remaining {label} quota",
      "accountLimit": "Account limit",
-      "quota": "{provider}",
      "lastRefreshed": "Last refreshed: {refreshedAtDisplay}",
      "viewPersonalInformation": "View personal information",
      "forecastWeeklyMonthlyQuota": "Forecast weekly/monthly quota",
@@ -25,7 +25,6 @@
      "reset": "重置：",
      "remainingQuota": "{label}剩余额度",
      "accountLimit": "账号额度",
-      "quota": "{provider} 额度",
      "lastRefreshed": "最近刷新: {refreshedAtDisplay}",
      "viewPersonalInformation": "查看个人信息",
      "forecastWeeklyMonthlyQuota": "预测周/月额度",
@@ -25,7 +25,6 @@
      "reset": "重置：",
      "remainingQuota": "{label}剩餘額度",
      "accountLimit": "賬號額度",
-      "quota": "{provider} 額度",
      "lastRefreshed": "最近刷新: {refreshedAtDisplay}",
      "viewPersonalInformation": "查看個人信息",
      "forecastWeeklyMonthlyQuota": "預測周/月額度",
@@ -1,6 +1,10 @@
 import type { AccountUpstream } from '@/api'
 import { t } from '@/i18n'

+export function accountUpstreamPending(upstream: AccountUpstream | null | undefined): boolean {
+  return Boolean(upstream && !upstream.discovered && !upstream.lastError)
+}
+
 export function accountUpstreamErrorMessage(reason: AccountUpstream['lastError']): string {
  if (!reason)
    return ''
@@ -9,7 +9,7 @@ import { computed, shallowRef, watch } from 'vue'
 import { useI18n } from 'vue-i18n'
 import AccountPlanBadge from '@/components/account/AccountPlanBadge.vue'
 import AccountUpstreamStatus from '@/components/account/AccountUpstreamStatus.vue'
-import { formatProviderLabel } from '@/utils/providers'
+import { accountUpstreamPending } from '@/utils/account-upstream'
 import { formatRelativeTime } from '@/utils/relative-time'
 import { usePoolQuotaWindows } from '../../composables/usePoolQuotaWindows'
 import { groupedAccountQuotaWindows, orderedPanelQuotaWindows } from '../../constants'
@@ -43,6 +43,8 @@ const quotaEntries = computed(() => groupedAccountQuotaWindows(
 ))
 const profileOpen = shallowRef(false)
 const forecastOpen = shallowRef(false)
+const hasPlanBadge = computed(() => !accountUpstreamPending(props.account.upstream)
+  && (Boolean(props.account.planType) || props.account.authenticationKind === 'api_key'))
 const hasForecast = computed(() => props.account.authenticationKind !== 'api_key')
 const hasPersonalInfo = computed(() => props.account.capabilities.profile || props.account.capabilities.subscription)
 const canRefresh = computed(() => props.account.capabilities.quotaRefresh || props.account.upstream?.discovered)
@@ -74,13 +76,9 @@ watch(hasPersonalInfo, (available) => {
        v-if="account.capabilities.quota || quotaEntries.length > 0 || account.upstream?.discovered"
        class="col-span-2 row-start-2 m-0 flex min-w-0 items-center gap-1.5 text-z-xs font-emphasis text-z-text-secondary"
      >
-        <span class="shrink-0 whitespace-nowrap">{{ t('accounts.quota.quota', { provider: formatProviderLabel(account.provider) }) }}</span>
-        <template v-if="account.planType || account.authenticationKind === 'api_key'">
-          <span class="shrink-0">·</span>
-          <AccountPlanBadge :upstream="account.upstream" :authentication-kind="account.authenticationKind" :plan-type="account.planType" :plan-type-display="account.planTypeDisplay" size="sm" class="shrink-0" />
-        </template>
+        <AccountPlanBadge v-if="hasPlanBadge" :upstream="account.upstream" :authentication-kind="account.authenticationKind" :plan-type="account.planType" :plan-type-display="account.planTypeDisplay" size="sm" class="shrink-0" />
        <template v-if="showUpstreamStatus || lastRefreshed">
-          <span class="shrink-0">·</span>
+          <span v-if="hasPlanBadge" class="shrink-0">·</span>
          <AccountUpstreamStatus v-if="showUpstreamStatus && account.upstream" :upstream="account.upstream" />
          <span v-else-if="lastRefreshed" class="min-w-0 truncate" :title="lastRefreshed">{{ lastRefreshed }}</span>
        </template>