feat(pool): 账号停用检测、OAuth refresh 失效标记与号池管理性能优化

- 新增 account_deactivated 关键词检测，覆盖 error_handler / health_policy / account_state / 前端
- 401 健康策略分级冷却：账号永久停用 1h，临时认证失败 60s
- OAuth refresh token 失败时标记 oauth_invalid（不停用 key），成功时清除非账号级标记
- 号池管理 API 使用 load_only + SQL 聚合替代全量拉取，减少查询开销
- Redis 冷却统计改用 batch_count_provider_cooldowns (SCAN) 替代逐 key 检查
- 前端时间线移除 executableTimeline 过滤层，Provider 选择改为非阻塞加载

frontend/src/features/usage/components/HorizontalRequestTimeline.vue

@@ -642,10 +642,6 @@ const normalizeTimelineStatus = (value: unknown): CandidateRecord['status'] => {
   return 'failed'
 }
 
-const isPlannedOnlyStatus = (status: CandidateRecord['status']): boolean => {
-  return status === 'available'
-}
-
 const extractPoolGroupId = (candidate: CandidateRecord): string | null => {
   const extra = candidate.extra_data
   if (!extra || typeof extra !== 'object' || Array.isArray(extra)) return null
@@ -671,22 +667,6 @@ const rawTimeline = computed<CandidateRecord[]>(() => {
     })
 })
 
-// 仅保留”已进入执行链路”的节点，过滤预创建但未执行的 available 占位记录。
-// unused 候选：代表因前序候选成功而未被执行的备选项。
-//   - 号池内 unused key：不显示（号池只展示实际参与调度的 key）
-//   - 非号池 unused：每个 candidate_index 保留 retry_index=0（代表该候选存在但未执行）
-const executableTimeline = computed<CandidateRecord[]>(() => {
-  return rawTimeline.value.filter(candidate => {
-    if (isPlannedOnlyStatus(candidate.status)) return false
-    if (candidate.status === 'unused') {
-      // 号池内 unused key 不需要展示
-      if (extractPoolGroupId(candidate) !== null) return false
-      // 非号池的 unused retry slot 只保留首个
-      return candidate.retry_index === 0
-    }
-    return true
-  })
-})
 
 const schedulingAudit = computed<Record<string, unknown> | null>(() => {
   const metadata = props.requestMetadata
@@ -698,7 +678,7 @@ const schedulingAudit = computed<Record<string, unknown> | null>(() => {
 
 const poolAttemptCandidates = computed<CandidateRecord[]>(() => {
   // 新链路：优先使用后端写入的 extra_data.pool_group_id。
-  const fromTrace = executableTimeline.value.filter((candidate) => extractPoolGroupId(candidate) !== null)
+  const fromTrace = rawTimeline.value.filter((candidate) => extractPoolGroupId(candidate) !== null)
   if (fromTrace.length > 0) {
     return fromTrace
   }
@@ -823,8 +803,8 @@ const poolAttemptKeySet = computed<Set<string>>(() => {
 })
 
 const timeline = computed<CandidateRecord[]>(() => {
-  if (poolAttemptCandidates.value.length === 0) return executableTimeline.value
+  if (poolAttemptCandidates.value.length === 0) return rawTimeline.value
-  return executableTimeline.value.filter(
+  return rawTimeline.value.filter(
     (candidate) => !poolAttemptKeySet.value.has(makeAttemptKey(candidate.candidate_index, candidate.retry_index)),
   )
 })

frontend/src/utils/accountBlock.ts

@@ -4,11 +4,15 @@ const ACCOUNT_BLOCK_REASON_KEYWORDS = [
   'account blocked',
   'account has been disabled',
   'account disabled',
+  'account has been deactivated',
+  'account_deactivated',
+  'account deactivated',
   'organization has been disabled',
   'organization_disabled',
   'validation_required',
   'verify your account',
   'suspended',
+  'deactivated',
   // Kiro quota refresher 写入的确切文本
   '账户已封禁',
   // Antigravity quota refresher 写入的确切文本

frontend/src/views/admin/PoolManagement.vue

@@ -1426,7 +1426,10 @@ async function selectProvider(id: string) {
     clearTimeout(keysSearchDebounceTimer)
     keysSearchDebounceTimer = null
   }
-  await Promise.all([loadKeys(), loadProviderData(id)])
+  const keysTask = loadKeys()
+  // Provider summary is non-blocking for key list rendering.
+  void loadProviderData(id)
+  await keysTask
   if (requestId !== selectProviderRequestId) return
 }
 

src/api/admin/pool/routes.py

@@ -16,8 +16,8 @@ from datetime import datetime, timezone
 from typing import Any
 
 from fastapi import APIRouter, Depends, Query, Request
-from sqlalchemy import func
+from sqlalchemy import case, func
-from sqlalchemy.orm import Session
+from sqlalchemy.orm import Session, load_only
 
 from src.api.base.admin_adapter import AdminApiAdapter
 from src.api.base.context import ApiRequestContext
@@ -644,7 +644,20 @@ class AdminListSchedulingPresetsAdapter(AdminApiAdapter):
 class AdminPoolOverviewAdapter(AdminApiAdapter):
     async def handle(self, context: ApiRequestContext) -> Any:  # type: ignore[override]
         db = context.db
-        providers = db.query(Provider).order_by(Provider.provider_priority.asc()).all()
+        providers = (
+            db.query(Provider)
+            .options(
+                load_only(
+                    Provider.id,
+                    Provider.name,
+                    Provider.provider_type,
+                    Provider.provider_priority,
+                    Provider.config,
+                )
+            )
+            .order_by(Provider.provider_priority.asc())
+            .all()
+        )
 
         # 仅保留号池调度已开启的 Provider。
         enabled_providers: list[Provider] = []
@@ -655,53 +668,39 @@ class AdminPoolOverviewAdapter(AdminApiAdapter):
             enabled_providers.append(p)
             pool_provider_ids.append(str(p.id))
 
-        key_ids_by_provider: dict[str, list[str]] = {pid: [] for pid in pool_provider_ids}
+        key_stats_by_provider: dict[str, dict[str, int]] = {}
-        key_stats_by_provider: dict[str, dict[str, int]] = {
-            pid: {"total": 0, "active": 0} for pid in pool_provider_ids
-        }
         if pool_provider_ids:
             key_rows = (
                 db.query(
                     ProviderAPIKey.provider_id,
-                    ProviderAPIKey.id,
+                    func.count(ProviderAPIKey.id).label("total"),
-                    ProviderAPIKey.is_active,
+                    func.coalesce(
+                        func.sum(case((ProviderAPIKey.is_active.is_(True), 1), else_=0)),
+                        0,
+                    ).label("active"),
                 )
                 .filter(ProviderAPIKey.provider_id.in_(pool_provider_ids))
+                .group_by(ProviderAPIKey.provider_id)
                 .all()
             )
-            for provider_id, key_id, is_active in key_rows:
+            for provider_id, total, active in key_rows:
                 pid = str(provider_id)
-                kid = str(key_id)
+                key_stats_by_provider[pid] = {
-                key_ids_by_provider.setdefault(pid, []).append(kid)
+                    "total": int(total or 0),
-                stats = key_stats_by_provider.setdefault(pid, {"total": 0, "active": 0})
+                    "active": int(active or 0),
-                stats["total"] += 1
+                }
-                if is_active:
-                    stats["active"] += 1
 
-        # Redis 冷却状态并发获取，避免逐 Provider 串行等待。
+        # Redis 冷却状态按 Provider 统计，避免先拉取全量 key_id 再逐个检查。
         cooldown_count_by_provider: dict[str, int] = {}
         cooldown_targets = [
-            (pid, key_ids) for pid, key_ids in key_ids_by_provider.items() if key_ids
+            pid
+            for pid in pool_provider_ids
+            if key_stats_by_provider.get(pid, {}).get("total", 0) > 0
         ]
         if cooldown_targets:
-            cooldown_results = await asyncio.gather(
+            cooldown_count_by_provider = await pool_redis.batch_count_provider_cooldowns(
-                *[
+                cooldown_targets
-                    pool_redis.batch_get_cooldowns(pid, key_ids)
-                    for pid, key_ids in cooldown_targets
-                ],
-                return_exceptions=True,
             )
-            for (pid, _key_ids), result in zip(cooldown_targets, cooldown_results, strict=False):
-                if isinstance(result, Exception):
-                    logger.warning(
-                        "池管理概览读取冷却状态失败",
-                        extra={"provider_id": pid, "error": str(result)},
-                    )
-                    cooldown_count_by_provider[pid] = 0
-                else:
-                    cooldown_count_by_provider[pid] = sum(
-                        1 for value in result.values() if value is not None
-                    )
 
         items: list[PoolOverviewItem] = []
         for p in enabled_providers:
@@ -742,7 +741,44 @@ class AdminListPoolKeysAdapter(AdminApiAdapter):
         provider_type = str(getattr(provider, "provider_type", "custom") or "custom")
 
         # Base query
-        q = db.query(ProviderAPIKey).filter(ProviderAPIKey.provider_id == pid)
+        q = (
+            db.query(ProviderAPIKey)
+            .options(
+                load_only(
+                    ProviderAPIKey.id,
+                    ProviderAPIKey.provider_id,
+                    ProviderAPIKey.name,
+                    ProviderAPIKey.auth_type,
+                    ProviderAPIKey.auth_config,
+                    ProviderAPIKey.is_active,
+                    ProviderAPIKey.expires_at,
+                    ProviderAPIKey.oauth_invalid_at,
+                    ProviderAPIKey.oauth_invalid_reason,
+                    ProviderAPIKey.api_formats,
+                    ProviderAPIKey.rate_multipliers,
+                    ProviderAPIKey.internal_priority,
+                    ProviderAPIKey.rpm_limit,
+                    ProviderAPIKey.cache_ttl_minutes,
+                    ProviderAPIKey.max_probe_interval_minutes,
+                    ProviderAPIKey.note,
+                    ProviderAPIKey.allowed_models,
+                    ProviderAPIKey.capabilities,
+                    ProviderAPIKey.auto_fetch_models,
+                    ProviderAPIKey.locked_models,
+                    ProviderAPIKey.model_include_patterns,
+                    ProviderAPIKey.model_exclude_patterns,
+                    ProviderAPIKey.proxy,
+                    ProviderAPIKey.fingerprint,
+                    ProviderAPIKey.health_by_format,
+                    ProviderAPIKey.circuit_breaker_by_format,
+                    ProviderAPIKey.request_count,
+                    ProviderAPIKey.last_used_at,
+                    ProviderAPIKey.created_at,
+                    ProviderAPIKey.upstream_metadata,
+                )
+            )
+            .filter(ProviderAPIKey.provider_id == pid)
+        )
 
         if self.search:
             escaped = self.search.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_")

src/services/orchestration/error_handler.py

@@ -184,6 +184,19 @@ class ErrorHandlerService:
                     reason="AWS 账号被暂停",
                     provider=provider,
                 )
+            # 401 account_deactivated -> 标记 OAuth key 为账号被永久停用
+            elif (
+                status_code == 401
+                and key
+                and str(getattr(key, "auth_type", "") or "").lower() == "oauth"
+                and self._is_account_deactivated(error_response_text)
+            ):
+                self._mark_oauth_key_blocked(
+                    key,
+                    request_id,
+                    reason="账号已被停用 (account_deactivated)",
+                    provider=provider,
+                )
             return
 
         # 限流错误
@@ -347,6 +360,27 @@ class ErrorHandlerService:
             return True
         return False
 
+    @staticmethod
+    def _is_account_deactivated(error_text: str | None) -> bool:
+        """
+        检测 401 错误是否为账号被永久停用 (deactivated)
+
+        匹配条件：
+        - 错误文本包含 "account_deactivated" (OpenAI error code)
+        - 错误文本包含 "account has been deactivated"
+        - 错误文本包含 "account deactivated"
+        """
+        if not error_text:
+            return False
+        search_text = error_text.lower()
+        if "account_deactivated" in search_text:
+            return True
+        if "account has been deactivated" in search_text:
+            return True
+        if "account deactivated" in search_text:
+            return True
+        return False
+
     def _mark_oauth_key_blocked(
         self,
         key: ProviderAPIKey,

src/services/provider/auth.py

@@ -63,6 +63,14 @@ def _persist_refreshed_token(
     key.api_key = crypto_service.encrypt(access_token)
     key.auth_config = crypto_service.encrypt(json.dumps(token_meta))
 
+    # 刷新成功 => 清除非账号级别的 oauth_invalid 标记（如 [REFRESH_FAILED]）
+    from src.services.provider.oauth_token import is_account_level_block
+
+    if not is_account_level_block(getattr(key, "oauth_invalid_reason", None)):
+        if getattr(key, "oauth_invalid_at", None) is not None:
+            key.oauth_invalid_at = None
+            key.oauth_invalid_reason = None
+
     sess = object_session(key)
     if sess is not None:
         sess.add(key)
@@ -75,6 +83,64 @@ def _persist_refreshed_token(
         )
 
 
+def _extract_refresh_error_detail(error_body: str) -> str:
+    """Best-effort extraction of error detail from refresh token error response."""
+    try:
+        data = json.loads(error_body)
+        if isinstance(data, dict):
+            err = data.get("error")
+            if isinstance(err, dict):
+                code = err.get("code") or ""
+                msg = err.get("message") or ""
+                return f"{code}: {msg}".strip(": ") if (code or msg) else ""
+            if isinstance(err, str):
+                return err
+            return str(data.get("error_description") or data.get("message") or "")
+    except Exception:
+        pass
+    return error_body[:200] if error_body else ""
+
+
+def _mark_refresh_token_invalid(
+    key: Any,
+    status_code: int,
+    error_body: str,
+) -> None:
+    """标记 refresh token 已失效（仅设置 oauth_invalid 标记，不停用 key）。
+
+    Access token 在过期前仍可正常使用。oauth_invalid_reason 使用 [REFRESH_FAILED]
+    前缀。注意：如果上游错误体中包含账号封禁关键词（如 "deactivated"），
+    该 reason 仍会被 account_state 的关键词匹配判定为 blocked，这是预期行为。
+    """
+    from datetime import datetime, timezone
+
+    detail = _extract_refresh_error_detail(error_body)
+    reason = f"[REFRESH_FAILED] Token 续期失败 ({status_code})"
+    if detail:
+        reason = f"{reason}: {detail}"
+
+    try:
+        key.oauth_invalid_at = datetime.now(timezone.utc)
+        key.oauth_invalid_reason = reason
+        # 不设置 is_active = False：access token 过期前 key 仍可调度
+
+        sess = object_session(key)
+        if sess is not None:
+            sess.add(key)
+            sess.commit()
+            logger.info(
+                "[OAUTH_REFRESH] key {} marked refresh_token invalid: {}",
+                str(getattr(key, "id", "?"))[:8],
+                reason[:120],
+            )
+    except Exception as exc:
+        logger.warning(
+            "[OAUTH_REFRESH] failed to mark key {} refresh invalid: {}",
+            str(getattr(key, "id", "?"))[:8],
+            str(exc),
+        )
+
+
 def _get_proxy_config(key: Any, endpoint: Any = None) -> Any:
     """获取有效代理配置（Key 级别优先于 Provider 级别）。"""
     try:
@@ -216,13 +282,25 @@ async def _refresh_generic_oauth_token(
 
             _persist_refreshed_token(key, access_token, token_meta)
     else:
+        error_body = ""
+        try:
+            error_body = resp.text or ""
+        except Exception:
+            pass
+
         logger.warning(
-            "OAuth token refresh failed: provider={}, key_id={}, status={}",
+            "OAuth token refresh failed: provider={}, key_id={}, status={}, body={}",
             provider_type,
             getattr(key, "id", "?"),
             resp.status_code,
+            error_body[:500],
         )
 
+        # 标记 refresh token 失效（不停用 key，access token 过期前仍可调度）。
+        # 注意：如果上游错误包含账号封禁关键词（如 "deactivated"），
+        # oauth_invalid_reason 会被 account_state 关键词匹配判定为 blocked，这是预期行为。
+        _mark_refresh_token_invalid(key, resp.status_code, error_body)
+
     return token_meta
 
 

src/services/provider/pool/account_state.py

@@ -16,11 +16,15 @@ ACCOUNT_BLOCK_REASON_KEYWORDS: tuple[str, ...] = (
     "account blocked",
     "account has been disabled",
     "account disabled",
+    "account has been deactivated",
+    "account_deactivated",
+    "account deactivated",
     "organization has been disabled",
     "organization_disabled",
     "validation_required",
     "verify your account",
     "suspended",
+    "deactivated",
     # Kiro quota refresher 写入的确切文本
     "账户已封禁",
     # Antigravity quota refresher 写入的确切文本

src/services/provider/pool/health_policy.py

@@ -4,7 +4,7 @@ Maps upstream HTTP status codes to pool-level actions:
 
 | Code         | Action                                                     |
 |--------------|------------------------------------------------------------|
-| 401          | Invalidate OAuth token cache + short cooldown              |
+| 401          | Invalidate OAuth token cache; permanent (deactivated) 1h, else 60s |
 | 402          | Long cooldown (payment issue)                              |
 | 403          | Graded cooldown: severe (suspended/banned) 1h, else 300s+  |
 | 400          | Check body for "organization has been disabled" -> cooldown |
@@ -31,15 +31,20 @@ _ACCOUNT_DISABLE_PATTERNS = (
     "organization_disabled",
     "account has been disabled",
     "account_disabled",
+    "account has been deactivated",
+    "account_deactivated",
+    "account deactivated",
 )
 
 # 需要更长冷却的账号异常语义（403 body 关键字）。
 _FORBIDDEN_ACCOUNT_PATTERNS = (
     "account suspended",
     "account banned",
+    "account deactivated",
     "subscription inactive",
     "suspended",
     "banned",
+    "deactivated",
 )
 
 _TRANSIENT_STATUS_COOLDOWN_REASON: dict[int, str] = {
@@ -149,13 +154,25 @@ async def _apply(
     # --- 401 Unauthorized ---------------------------------------------------
     if status_code == 401:
         await redis_ops.invalidate_oauth_token_cache(key_id)
-        # Set a short cooldown to avoid hammering while refresh happens.
+        # Check if the 401 body indicates a permanent account-level deactivation
-        await redis_ops.set_cooldown(provider_id, key_id, "auth_failed_401", ttl=60)
+        # (e.g. OpenAI "account_deactivated"). These deserve a long cooldown.
-        logger.info(
+        error_lower = error_msg.lower()
-            "Pool[{}]: key {} got 401, token cache invalidated + 60s cooldown",
+        is_permanent = any(p in error_lower for p in _ACCOUNT_DISABLE_PATTERNS)
-            provider_id[:8],
+        if is_permanent:
-            key_id[:8],
+            await redis_ops.set_cooldown(provider_id, key_id, "account_deactivated_401", ttl=3600)
-        )
+            logger.warning(
+                "Pool[{}]: key {} got 401 with account deactivation, cooldown 1h",
+                provider_id[:8],
+                key_id[:8],
+            )
+        else:
+            # Transient auth failure (e.g. expired token) — short cooldown.
+            await redis_ops.set_cooldown(provider_id, key_id, "auth_failed_401", ttl=60)
+            logger.info(
+                "Pool[{}]: key {} got 401, token cache invalidated + 60s cooldown",
+                provider_id[:8],
+                key_id[:8],
+            )
         return
 
     # --- 402 Payment Required ------------------------------------------------

src/services/provider/pool/redis_ops.py

@@ -16,9 +16,10 @@ provider_oauth_token_cache:{key_id}        STRING  -> access_token (TTL: expires
 
 from __future__ import annotations
 
+import asyncio
 import time
 import uuid
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Any
 
 from src.clients.redis_client import get_redis_client
 from src.core.logger import logger
@@ -603,6 +604,44 @@ async def batch_get_cooldown_ttls(provider_id: str, key_ids: list[str]) -> dict[
         return {k: None for k in key_ids}
 
 
+async def batch_count_provider_cooldowns(provider_ids: list[str]) -> dict[str, int]:
+    """Count cooldown entries per provider using per-provider SCAN.
+
+    Each provider's cooldown keys are scanned independently with a
+    targeted pattern ``ap:{pid}:cooldown:*``, avoiding full-keyspace traversal.
+    Multiple providers are scanned concurrently via ``asyncio.gather``.
+    """
+    if not provider_ids:
+        return {}
+
+    redis = await _get_redis()
+    if redis is None:
+        return {pid: 0 for pid in provider_ids}
+
+    async def _count_one(r: Any, pid: str) -> tuple[str, int]:
+        pattern = f"{PREFIX}:{pid}:cooldown:*"
+        count = 0
+        async for _key in r.scan_iter(match=pattern, count=200):
+            count += 1
+        return pid, count
+
+    try:
+        results = await asyncio.gather(
+            *[_count_one(redis, pid) for pid in provider_ids],
+            return_exceptions=True,
+        )
+        counts: dict[str, int] = {}
+        for r in results:
+            if isinstance(r, Exception):
+                continue
+            counts[r[0]] = r[1]
+        for pid in provider_ids:
+            counts.setdefault(pid, 0)
+        return counts
+    except Exception:
+        return {pid: 0 for pid in provider_ids}
+
+
 # ---------------------------------------------------------------------------
 # Stream timeout counter
 # ---------------------------------------------------------------------------