feat: 引入 status_snapshot 统一 provider key 状态管理

- 新增 StatusSnapshot 模型，聚合 OAuth / 账号 / 配额三维状态
- 新增 StatusSnapshotStore 负责快照的持久化与查询
- 重构 response_builder / endpoint_models，基于 snapshot 输出状态字段
- 前端抽取 providerKeyStatus / oauthRefreshFeedback 工具函数，
  统一 PoolManagement、ProviderDetailDrawer、BatchDialog 的状态展示
- errorParser 增加已知 OAuth 错误的友好提示
- refresher 适配 snapshot 写入，account_state 扩展状态分类
- 新增 alembic 迁移及存量数据回填脚本
- 补充前后端单元测试

src/api/admin/pool/routes.py

@@ -9,11 +9,10 @@ Provides endpoints for managing account pools at scale:
 from __future__ import annotations
 
 import asyncio
-import json
 import re
 import time
 import uuid
-from dataclasses import dataclass, field
+from dataclasses import asdict, dataclass, field
 from datetime import datetime, timezone
 from typing import Any, cast
 
@@ -35,7 +34,9 @@ from src.models.database import Provider, ProviderAPIKey
 from src.services.billing.precision import to_money_decimal
 from src.services.provider.fingerprint import generate_fingerprint
 from src.services.provider.pool import redis_ops as pool_redis
-from src.services.provider.pool.account_state import resolve_pool_account_state
+from src.services.provider.pool.account_state import (
+    resolve_pool_account_state,
+)
 from src.services.provider.pool.config import parse_pool_config
 from src.services.provider.pool.dimensions import get_preset_dimension_metas
 from src.services.provider.pool.scheduling_dimensions import (
@@ -45,6 +46,18 @@ from src.services.provider.pool.scheduling_dimensions import (
 )
 from src.services.provider_keys.key_side_effects import cleanup_key_references
 from src.services.provider_keys.quota_reader import get_quota_reader
+from src.services.provider_keys.status_snapshot_store import (
+    derive_oauth_expires_at as derive_persisted_oauth_expires_at,
+)
+from src.services.provider_keys.status_snapshot_store import (
+    extract_oauth_auth_config as extract_persisted_oauth_auth_config,
+)
+from src.services.provider_keys.status_snapshot_store import (
+    normalize_oauth_expires_at as normalize_persisted_oauth_expires_at,
+)
+from src.services.provider_keys.status_snapshot_store import (
+    resolve_provider_key_status_snapshot,
+)
 
 from .schemas import (
     BatchActionRequest,
@@ -252,10 +265,6 @@ def _build_account_quota(provider_type: str, upstream_metadata: Any) -> str | No
     return get_quota_reader(provider_type, upstream_metadata).display_summary()
 
 
-def _extract_quota_updated_at(provider_type: str, upstream_metadata: Any) -> int | None:
-    return get_quota_reader(provider_type, upstream_metadata).updated_at()
-
-
 def _normalize_oauth_plan_type(plan_type: Any, provider_type: str) -> str | None:
     if not isinstance(plan_type, str):
         return None
@@ -273,52 +282,17 @@ def _normalize_oauth_plan_type(plan_type: Any, provider_type: str) -> str | None
 
 
 def _extract_oauth_auth_config(key: ProviderAPIKey) -> dict[str, Any] | None:
-    if str(getattr(key, "auth_type", "") or "").strip().lower() != "oauth":
-        return None
-
-    auth_config_raw = getattr(key, "auth_config", None)
-    if not auth_config_raw:
-        return None
-
-    try:
-        decrypted = crypto_service.decrypt(auth_config_raw)
-        parsed = json.loads(decrypted)
-        if isinstance(parsed, dict):
-            return parsed
-    except Exception:
-        return None
-
-    return None
+    return extract_persisted_oauth_auth_config(key)
 
 
 def _normalize_oauth_expires_at(raw: Any) -> int | None:
-    value = _to_float(raw)
-    if value is None or value <= 0:
-        return None
-    # 兼容毫秒时间戳
-    if value > 1_000_000_000_000:
-        value /= 1000
-    return int(value)
+    return normalize_persisted_oauth_expires_at(raw)
 
 
 def _derive_oauth_expires_at(
     key: ProviderAPIKey, auth_config: dict[str, Any] | None = None
 ) -> int | None:
-    if str(getattr(key, "auth_type", "") or "").strip().lower() != "oauth":
-        return None
-
-    cfg = auth_config if isinstance(auth_config, dict) else _extract_oauth_auth_config(key)
-    if cfg:
-        for field in ("expires_at", "expiresAt", "expiry", "exp"):
-            expires_at = _normalize_oauth_expires_at(cfg.get(field))
-            if expires_at is not None:
-                return expires_at
-
-    # 兼容历史字段
-    expires_dt = getattr(key, "expires_at", None)
-    if isinstance(expires_dt, datetime):
-        return int(expires_dt.timestamp())
-    return None
+    return derive_persisted_oauth_expires_at(key, auth_config=auth_config)
 
 
 def _derive_oauth_plan_type(
@@ -864,7 +838,16 @@ def _has_no_weekly_limit(account_quota: Any) -> bool:
 def _detail_is_oauth_invalid(detail: PoolKeyDetail) -> bool:
     if _normalize_batch_text(detail.auth_type) != "oauth":
         return False
-    status_code = _normalize_batch_text(detail.account_status_code)
+    snapshot_oauth_code = _normalize_batch_text(getattr(detail.status_snapshot.oauth, "code", None))
+    if snapshot_oauth_code == "invalid":
+        return True
+    if snapshot_oauth_code == "expired":
+        return True
+    if snapshot_oauth_code == "check_failed":
+        return False
+    status_code = _normalize_batch_text(
+        getattr(detail.status_snapshot.account, "code", None) or detail.account_status_code
+    )
     if status_code in _TOKEN_ISSUE_CODES:
         return True
     if status_code in _ACCOUNT_BANNED_CODES or status_code == "oauth_request_failed":
@@ -883,9 +866,16 @@ def _detail_is_oauth_invalid(detail: PoolKeyDetail) -> bool:
 
 
 def _detail_is_banned(detail: PoolKeyDetail) -> bool:
+    snapshot_account_code = _normalize_batch_text(
+        getattr(detail.status_snapshot.account, "code", None)
+    )
+    if snapshot_account_code in _ACCOUNT_BANNED_CODES:
+        return True
     if _normalize_batch_text(detail.account_status_code) in _ACCOUNT_BANNED_CODES:
         return True
-    reason = _normalize_batch_text(detail.oauth_invalid_reason)
+    reason = _normalize_batch_text(
+        getattr(detail.status_snapshot.account, "reason", None) or detail.oauth_invalid_reason
+    )
     if reason and _BANNED_REASON_PATTERN.search(reason):
         return True
     for item in detail.scheduling_reasons or []:
@@ -916,12 +906,15 @@ def _matches_pool_key_search(
         detail.key_name,
         detail.auth_type,
         detail.oauth_plan_type,
-        detail.account_status_label,
-        detail.account_status_reason,
+        getattr(detail.status_snapshot.account, "label", None) or detail.account_status_label,
+        getattr(detail.status_snapshot.account, "reason", None) or detail.account_status_reason,
         detail.account_quota,
         "独立代理" if _detail_has_proxy(detail) else "未配置代理",
         "已启用" if detail.is_active else "已禁用",
-        detail.oauth_invalid_reason,
+        getattr(detail.status_snapshot.oauth, "reason", None) or detail.oauth_invalid_reason,
+        getattr(detail.status_snapshot.oauth, "label", None),
+        getattr(detail.status_snapshot.quota, "label", None),
+        getattr(detail.status_snapshot.quota, "reason", None),
     ]
     return any(keyword in _normalize_batch_text(part) for part in parts)
 
@@ -984,7 +977,9 @@ def _detail_is_schedulable(detail: PoolKeyDetail) -> bool:
 
     if not detail.is_active:
         return False
-    if detail.account_status_blocked:
+    if bool(
+        getattr(detail.status_snapshot.account, "blocked", False) or detail.account_status_blocked
+    ):
         return False
     if detail.cooldown_reason:
         return False
@@ -1081,11 +1076,15 @@ async def _serialize_pool_key_details(
         cost_limit = pcfg.cost_limit_per_key_tokens if pcfg else None
         latency_avg_raw = latency_avgs.get(kid)
         latency_avg_ms = float(latency_avg_raw) if latency_avg_raw is not None else None
-        account_state = resolve_pool_account_state(
+        oauth_auth_config = _extract_oauth_auth_config(k)
+        oauth_expires_at = _derive_oauth_expires_at(k, auth_config=oauth_auth_config)
+        status_snapshot = resolve_provider_key_status_snapshot(
+            k,
             provider_type=provider_type,
-            upstream_metadata=getattr(k, "upstream_metadata", None),
-            oauth_invalid_reason=getattr(k, "oauth_invalid_reason", None),
+            auth_config=oauth_auth_config,
+            oauth_expires_at=oauth_expires_at,
         )
+        account_state = status_snapshot.account
         (
             scheduling_status,
             scheduling_reason,
@@ -1153,7 +1152,7 @@ async def _serialize_pool_key_details(
         key_total_tokens = int(getattr(k, "total_tokens", 0) or 0)
         key_total_cost_usd = _serialize_money(getattr(k, "total_cost_usd", 0.0))
         key_last_used_at = getattr(k, "last_used_at", None)
-        oauth_auth_config = _extract_oauth_auth_config(k)
+        oauth_invalid_at = status_snapshot.oauth.invalid_at
 
         key_details.append(
             PoolKeyDetail(
@@ -1161,12 +1160,8 @@ async def _serialize_pool_key_details(
                 key_name=str(getattr(k, "name", "") or ""),
                 is_active=bool(k.is_active),
                 auth_type=str(getattr(k, "auth_type", "api_key") or "api_key"),
-                oauth_expires_at=_derive_oauth_expires_at(k, auth_config=oauth_auth_config),
-                oauth_invalid_at=(
-                    int(k.oauth_invalid_at.timestamp())
-                    if getattr(k, "oauth_invalid_at", None)
-                    else None
-                ),
+                oauth_expires_at=oauth_expires_at,
+                oauth_invalid_at=oauth_invalid_at,
                 oauth_invalid_reason=getattr(k, "oauth_invalid_reason", None),
                 oauth_plan_type=_derive_oauth_plan_type(
                     k, provider_type, auth_config=oauth_auth_config
@@ -1181,10 +1176,8 @@ async def _serialize_pool_key_details(
                 account_status_blocked=account_state.blocked,
                 account_status_recoverable=bool(getattr(account_state, "recoverable", False)),
                 account_status_source=getattr(account_state, "source", None),
-                quota_updated_at=_extract_quota_updated_at(
-                    provider_type,
-                    getattr(k, "upstream_metadata", None),
-                ),
+                status_snapshot=asdict(status_snapshot),
+                quota_updated_at=status_snapshot.quota.updated_at,
                 health_score=health_score,
                 circuit_breaker_open=any_circuit_open,
                 api_formats=api_formats,

src/api/admin/pool/schemas.py

@@ -6,6 +6,8 @@ from typing import Any
 
 from pydantic import BaseModel, ConfigDict, Field
 
+from src.models.status_snapshot import ProviderKeyStatusSnapshotResponse
+
 # ---------------------------------------------------------------------------
 # Overview
 # ---------------------------------------------------------------------------
@@ -80,20 +82,42 @@ class PoolKeyDetail(BaseModel):
     key_name: str
     is_active: bool
     auth_type: str = "api_key"
-    oauth_expires_at: int | None = None
-    oauth_invalid_at: int | None = None
-    oauth_invalid_reason: str | None = None
+    oauth_expires_at: int | None = Field(
+        default=None, description="兼容字段；优先使用 status_snapshot.oauth"
+    )
+    oauth_invalid_at: int | None = Field(
+        default=None, description="兼容字段；优先使用 status_snapshot.oauth"
+    )
+    oauth_invalid_reason: str | None = Field(
+        default=None, description="兼容字段；优先使用 status_snapshot.oauth"
+    )
     oauth_plan_type: str | None = None
     oauth_account_id: str | None = None
     oauth_account_name: str | None = None
     oauth_account_user_id: str | None = None
     oauth_organizations: list[OAuthOrganizationSummary] = Field(default_factory=list)
-    account_status_code: str | None = None
-    account_status_label: str | None = None
-    account_status_reason: str | None = None
-    account_status_blocked: bool = False
-    account_status_recoverable: bool = False
-    account_status_source: str | None = None
+    account_status_code: str | None = Field(
+        default=None, description="兼容字段；优先使用 status_snapshot.account"
+    )
+    account_status_label: str | None = Field(
+        default=None, description="兼容字段；优先使用 status_snapshot.account"
+    )
+    account_status_reason: str | None = Field(
+        default=None, description="兼容字段；优先使用 status_snapshot.account"
+    )
+    account_status_blocked: bool = Field(
+        default=False, description="兼容字段；优先使用 status_snapshot.account"
+    )
+    account_status_recoverable: bool = Field(
+        default=False, description="兼容字段；优先使用 status_snapshot.account"
+    )
+    account_status_source: str | None = Field(
+        default=None, description="兼容字段；优先使用 status_snapshot.account"
+    )
+    status_snapshot: ProviderKeyStatusSnapshotResponse = Field(
+        default_factory=ProviderKeyStatusSnapshotResponse,
+        description="统一的账号/OAuth/额度状态快照",
+    )
     quota_updated_at: int | None = None
     # 健康度聚合字段（与 Provider Key 列表口径一致）
     health_score: float = 1.0

src/api/admin/provider_oauth.py

@@ -51,6 +51,85 @@ from src.utils.auth_utils import require_admin
 router = APIRouter(prefix="/api/admin/provider-oauth", tags=["Provider OAuth"])
 
 
+def _normalize_oauth_refresh_error_message(
+    message: str | None,
+    *,
+    status_code: int | None = None,
+    error_code: str | None = None,
+    error_type: str | None = None,
+) -> str:
+    text = str(message or "").strip()
+    lowered = text.lower()
+    code = str(error_code or "").strip().lower()
+    err_type = str(error_type or "").strip().lower()
+
+    if code == "refresh_token_reused" or (
+        "already been used to generate a new access token" in lowered
+    ):
+        return "refresh_token 已被使用并轮换，请重新登录授权"
+
+    if code in {"invalid_grant", "invalid_refresh_token"} or (
+        "refresh token" in lowered
+        and any(keyword in lowered for keyword in ("expired", "revoked", "invalid"))
+    ):
+        return "refresh_token 无效、已过期或已撤销，请重新登录授权"
+
+    if err_type == "invalid_request_error" and text:
+        return text
+
+    if text:
+        return text
+    if status_code is not None:
+        return f"HTTP {status_code}"
+    return "未知错误"
+
+
+def _extract_oauth_refresh_error_reason(resp: httpx.Response) -> str:
+    status_code = int(resp.status_code)
+    message: str | None = None
+    error_code: str | None = None
+    error_type: str | None = None
+
+    try:
+        error_body = resp.json()
+        if isinstance(error_body, dict):
+            err = error_body.get("error")
+            if isinstance(err, dict):
+                raw_message = err.get("message") or err.get("error_description")
+                if raw_message is not None:
+                    message = str(raw_message).strip() or None
+                raw_code = err.get("code")
+                if raw_code is not None:
+                    error_code = str(raw_code).strip() or None
+                raw_type = err.get("type")
+                if raw_type is not None:
+                    error_type = str(raw_type).strip() or None
+            elif isinstance(err, str):
+                message = err.strip() or None
+            raw_message = error_body.get("message") or error_body.get("error_description")
+            if raw_message is not None and not message:
+                message = str(raw_message).strip() or None
+            raw_code = error_body.get("code")
+            if raw_code is not None and not error_code:
+                error_code = str(raw_code).strip() or None
+            raw_type = error_body.get("type")
+            if raw_type is not None and not error_type:
+                error_type = str(raw_type).strip() or None
+    except Exception:
+        pass
+
+    if not message:
+        text = str(getattr(resp, "text", "") or "").strip()
+        message = text[:300] if text else None
+
+    return _normalize_oauth_refresh_error_message(
+        message,
+        status_code=status_code,
+        error_code=error_code,
+        error_type=error_type,
+    )
+
+
 def _store_completed_oauth_sync(
     key_id: str,
     provider_type: str,
@@ -71,22 +150,31 @@ def _mark_refresh_failed_sync(key_id: str, reason: str) -> None:
         if not key:
             raise NotFoundException("Key 不存在", "key")
         current_reason = str(getattr(key, "oauth_invalid_reason", None) or "").strip()
-        if _should_preserve_refresh_failure_reason(current_reason):
+        merged_reason = _merge_refresh_failure_reason(current_reason, reason)
+        if merged_reason is None:
             return
         key.oauth_invalid_at = datetime.now(timezone.utc)
-        key.oauth_invalid_reason = reason
+        key.oauth_invalid_reason = merged_reason
 
 
-def _should_preserve_refresh_failure_reason(reason: str | None) -> bool:
+def _merge_refresh_failure_reason(current_reason: str | None, refresh_reason: str) -> str | None:
     from src.services.provider.oauth_token import is_account_level_block
     from src.services.provider.pool.account_state import OAUTH_EXPIRED_PREFIX
 
-    text = str(reason or "").strip()
-    if not text:
-        return False
-    if is_account_level_block(text):
-        return True
-    return text.startswith(OAUTH_EXPIRED_PREFIX)
+    current = str(current_reason or "").strip()
+    next_reason = str(refresh_reason or "").strip()
+    if not next_reason:
+        return current or None
+    if not current:
+        return next_reason
+    if current.startswith(OAUTH_EXPIRED_PREFIX):
+        return None
+    if is_account_level_block(current):
+        if "[REFRESH_FAILED]" in current:
+            head, _sep, _tail = current.partition("[REFRESH_FAILED]")
+            return f"{head.rstrip()}\n{next_reason}".strip()
+        return f"{current}\n{next_reason}"
+    return next_reason
 
 
 def _store_refreshed_oauth_sync(
@@ -1168,15 +1256,7 @@ async def refresh_oauth(
         )
 
         if resp.status_code < 200 or resp.status_code >= 300:
-            error_reason = f"HTTP {resp.status_code}"
-            try:
-                error_body = resp.json()
-                if "error" in error_body:
-                    error_reason = str(
-                        error_body.get("error_description") or error_body.get("error")
-                    )
-            except Exception:
-                error_reason = resp.text[:100] if resp.text else f"HTTP {resp.status_code}"
+            error_reason = _extract_oauth_refresh_error_reason(resp)
 
             if resp.status_code in (400, 401, 403):
                 await run_in_threadpool(
@@ -1194,7 +1274,7 @@ async def refresh_oauth(
                 refresh_error=error_reason,
             )
 
-            raise InvalidRequestException(f"token refresh 失败: {error_reason}")
+            raise InvalidRequestException(f"Token 刷新失败：{error_reason}")
 
         token = resp.json()
         access_token = str(token.get("access_token") or "")

src/models/database.py

@@ -30,6 +30,8 @@ from sqlalchemy import (
     String,
     Text,
     UniqueConstraint,
+    event,
+    inspect,
     text,
 )
 from sqlalchemy.dialects.postgresql import JSONB
@@ -2066,6 +2068,7 @@ class ProviderAPIKey(ExportMixin, Base):
     # OAuth 失效状态（账号被封、授权撤销、刷新失败等）
     oauth_invalid_at = Column(DateTime(timezone=True), nullable=True)  # 失效时间
     oauth_invalid_reason = Column(String(255), nullable=True)  # 失效原因
+    status_snapshot = Column(JSON, nullable=True, default=None)  # 结构化状态快照（兼容旧字段）
 
     # Key 级别的代理配置（覆盖 Provider 级别的代理设置）
     # 结构: {"node_id": "xxx", "enabled": true} 或 {"url": "socks5://...", "enabled": true}
@@ -2092,6 +2095,49 @@ class ProviderAPIKey(ExportMixin, Base):
     provider = relationship("Provider", back_populates="api_keys")
 
 
+_PROVIDER_API_KEY_STATUS_SNAPSHOT_FIELDS: tuple[str, ...] = (
+    "auth_type",
+    "auth_config",
+    "expires_at",
+    "oauth_invalid_at",
+    "oauth_invalid_reason",
+    "upstream_metadata",
+    "provider_id",
+)
+
+
+def _sync_provider_api_key_status_snapshot(
+    target: ProviderAPIKey,
+    *,
+    connection: Any | None,
+    force: bool,
+) -> None:
+    state = inspect(target)
+    if state is None:
+        return
+    if not force:
+        relevant_changed = any(
+            state.attrs[field].history.has_changes()
+            for field in _PROVIDER_API_KEY_STATUS_SNAPSHOT_FIELDS
+        )
+        if not relevant_changed and getattr(target, "status_snapshot", None) is not None:
+            return
+
+    from src.services.provider_keys.status_snapshot_store import sync_provider_key_status_snapshot
+
+    sync_provider_key_status_snapshot(target, connection=connection)
+
+
+@event.listens_for(ProviderAPIKey, "before_insert")
+def _provider_api_key_before_insert(mapper: Any, connection: Any, target: ProviderAPIKey) -> None:
+    _sync_provider_api_key_status_snapshot(target, connection=connection, force=True)
+
+
+@event.listens_for(ProviderAPIKey, "before_update")
+def _provider_api_key_before_update(mapper: Any, connection: Any, target: ProviderAPIKey) -> None:
+    _sync_provider_api_key_status_snapshot(target, connection=connection, force=False)
+
+
 def _generate_short_id(length: int = 12) -> str:
     """生成 Gemini 风格的短 ID（小写字母+数字）"""
     import secrets

src/models/endpoint_models.py

@@ -16,6 +16,7 @@ from src.models.admin_requests import (
     PoolAdvancedConfig,
     ProxyConfig,
 )
+from src.models.status_snapshot import ProviderKeyStatusSnapshotResponse
 
 # ========== Header Rule 类型定义 ==========
 # 请求头规则支持三种操作：
@@ -130,8 +131,7 @@ def _validate_condition(condition: Any, rule_label: str) -> None:
     op = condition.get("op")
     if not isinstance(op, str) or op not in _CONDITION_OPS:
         raise ValueError(
-            f"{rule_label}: condition.op 必须是 {sorted(_CONDITION_OPS)} 之一，"
-            f"当前值: {op!r}"
+            f"{rule_label}: condition.op 必须是 {sorted(_CONDITION_OPS)} 之一，" f"当前值: {op!r}"
         )
 
     path = condition.get("path")
@@ -152,15 +152,11 @@ def _validate_condition(condition: Any, rule_label: str) -> None:
     # matches 正则校验
     if op == "matches":
         if not isinstance(value, str) or not value:
-            raise ValueError(
-                f"{rule_label}: condition op=matches 的 value 必须为非空字符串"
-            )
+            raise ValueError(f"{rule_label}: condition op=matches 的 value 必须为非空字符串")
         try:
             re.compile(value)
         except re.error as e:
-            raise ValueError(
-                f"{rule_label}: condition op=matches 的 value 不是合法正则: {e}"
-            )
+            raise ValueError(f"{rule_label}: condition op=matches 的 value 不是合法正则: {e}")
 
     # in 校验
     if op == "in":
@@ -188,10 +184,7 @@ def _validate_header_rules(rules: list[HeaderRule]) -> list[HeaderRule]:
             raise ValueError(f"header_rules[{idx}]: 规则必须是 JSON 对象")
 
         action = rule.get("action")
-        if (
-            not isinstance(action, str)
-            or action.strip().lower() not in _HEADER_RULE_ACTIONS
-        ):
+        if not isinstance(action, str) or action.strip().lower() not in _HEADER_RULE_ACTIONS:
             raise ValueError(
                 f"header_rules[{idx}]: action 必须是 {sorted(_HEADER_RULE_ACTIONS)} 之一，"
                 f"当前值: {action!r}"
@@ -241,10 +234,7 @@ def _validate_body_rules(rules: list[BodyRule]) -> list[BodyRule]:
             raise ValueError(f"body_rules[{idx}]: 规则必须是 JSON 对象")
 
         action = rule.get("action")
-        if (
-            not isinstance(action, str)
-            or action.strip().lower() not in _BODY_RULE_ACTIONS
-        ):
+        if not isinstance(action, str) or action.strip().lower() not in _BODY_RULE_ACTIONS:
             raise ValueError(
                 f"body_rules[{idx}]: action 必须是 {sorted(_BODY_RULE_ACTIONS)} 之一，"
                 f"当前值: {action!r}"
@@ -255,9 +245,7 @@ def _validate_body_rules(rules: list[BodyRule]) -> list[BodyRule]:
         if action in {"set", "drop", "append", "insert", "regex_replace", "name_style"}:
             path = rule.get("path")
             if not isinstance(path, str) or not path.strip():
-                raise ValueError(
-                    f"body_rules[{idx}]: action={action!r} 必须提供非空 path"
-                )
+                raise ValueError(f"body_rules[{idx}]: action={action!r} 必须提供非空 path")
 
         # ---------- rename 校验 ----------
         if action == "rename":
@@ -278,15 +266,11 @@ def _validate_body_rules(rules: list[BodyRule]) -> list[BodyRule]:
         if action == "regex_replace":
             pattern = rule.get("pattern")
             if not isinstance(pattern, str) or not pattern:
-                raise ValueError(
-                    f"body_rules[{idx}]: regex_replace 必须提供非空 pattern 字符串"
-                )
+                raise ValueError(f"body_rules[{idx}]: regex_replace 必须提供非空 pattern 字符串")
 
             replacement = rule.get("replacement", "")
             if not isinstance(replacement, str):
-                raise ValueError(
-                    f"body_rules[{idx}]: regex_replace 的 replacement 必须为字符串"
-                )
+                raise ValueError(f"body_rules[{idx}]: regex_replace 的 replacement 必须为字符串")
 
             # 校验 flags
             flags_str = rule.get("flags", "")
@@ -312,9 +296,7 @@ def _validate_body_rules(rules: list[BodyRule]) -> list[BodyRule]:
             # 校验 count
             count = rule.get("count", 0)
             if not isinstance(count, int) or count < 0:
-                raise ValueError(
-                    f"body_rules[{idx}]: regex_replace 的 count 必须为非负整数"
-                )
+                raise ValueError(f"body_rules[{idx}]: regex_replace 的 count 必须为非负整数")
 
         # ---------- name_style 校验 ----------
         if action == "name_style":
@@ -347,9 +329,7 @@ class ProviderEndpointCreate(BaseModel):
         ),
     )
     base_url: str = Field(..., min_length=1, max_length=500, description="API 基础 URL")
-    custom_path: str | None = Field(
-        default=None, max_length=200, description="自定义请求路径"
-    )
+    custom_path: str | None = Field(default=None, max_length=200, description="自定义请求路径")
 
     # 请求头配置
     header_rules: list[HeaderRule] | None = Field(
@@ -411,9 +391,7 @@ class ProviderEndpointCreate(BaseModel):
 
     @field_validator("header_rules")
     @classmethod
-    def validate_header_rules(
-        cls, v: list[HeaderRule] | None
-    ) -> list[HeaderRule] | None:
+    def validate_header_rules(cls, v: list[HeaderRule] | None) -> list[HeaderRule] | None:
         """校验 header_rules 结构和 condition 合法性"""
         if v is None:
             return v
@@ -426,9 +404,7 @@ class ProviderEndpointUpdate(BaseModel):
     base_url: str | None = Field(
         default=None, min_length=1, max_length=500, description="API 基础 URL"
     )
-    custom_path: str | None = Field(
-        default=None, max_length=200, description="自定义请求路径"
-    )
+    custom_path: str | None = Field(default=None, max_length=200, description="自定义请求路径")
 
     # 请求头配置
     header_rules: list[HeaderRule] | None = Field(
@@ -442,9 +418,7 @@ class ProviderEndpointUpdate(BaseModel):
         description="请求体规则列表，支持 set/drop/rename/append/insert/regex_replace 操作",
     )
 
-    max_retries: int | None = Field(
-        default=None, ge=0, le=999, description="最大重试次数"
-    )
+    max_retries: int | None = Field(default=None, ge=0, le=999, description="最大重试次数")
     is_active: bool | None = Field(default=None, description="是否启用")
     config: dict[str, Any] | None = Field(default=None, description="额外配置")
     proxy: ProxyConfig | None = Field(default=None, description="代理配置")
@@ -477,9 +451,7 @@ class ProviderEndpointUpdate(BaseModel):
 
     @field_validator("header_rules")
     @classmethod
-    def validate_header_rules(
-        cls, v: list[HeaderRule] | None
-    ) -> list[HeaderRule] | None:
+    def validate_header_rules(cls, v: list[HeaderRule] | None) -> list[HeaderRule] | None:
         """校验 header_rules 结构和 condition 合法性"""
         if v is None:
             return v
@@ -499,14 +471,10 @@ class ProviderEndpointResponse(BaseModel):
     custom_path: str | None = None
 
     # 请求头配置
-    header_rules: list[HeaderRule] | None = Field(
-        default=None, description="请求头规则列表"
-    )
+    header_rules: list[HeaderRule] | None = Field(default=None, description="请求头规则列表")
 
     # 请求体配置
-    body_rules: list[BodyRule] | None = Field(
-        default=None, description="请求体规则列表"
-    )
+    body_rules: list[BodyRule] | None = Field(default=None, description="请求体规则列表")
 
     max_retries: int
 
@@ -517,9 +485,7 @@ class ProviderEndpointResponse(BaseModel):
     config: dict[str, Any] | None = None
 
     # 代理配置（响应中密码已脱敏）
-    proxy: dict[str, Any] | None = Field(
-        default=None, description="代理配置（密码已脱敏）"
-    )
+    proxy: dict[str, Any] | None = Field(default=None, description="代理配置（密码已脱敏）")
 
     # 格式转换配置
     format_acceptance_config: dict[str, Any] | None = Field(
@@ -544,9 +510,7 @@ class ProviderEndpointResponse(BaseModel):
 class EndpointAPIKeyCreate(BaseModel):
     """为 Provider 添加 API Key"""
 
-    provider_id: str | None = Field(
-        default=None, description="Provider ID（从 URL 获取）"
-    )
+    provider_id: str | None = Field(default=None, description="Provider ID（从 URL 获取）")
     api_formats: list[str] | None = Field(
         default=None,
         min_length=1,
@@ -569,9 +533,7 @@ class EndpointAPIKeyCreate(BaseModel):
             "oauth 时存储 token/refresh/expires_at 等（后端加密存储，不在响应中返回）"
         ),
     )
-    name: str = Field(
-        ..., min_length=1, max_length=100, description="密钥名称（必填，用于识别）"
-    )
+    name: str = Field(..., min_length=1, max_length=100, description="密钥名称（必填，用于识别）")
 
     # 成本计算
     rate_multipliers: dict[str, float] | None = Field(
@@ -580,9 +542,7 @@ class EndpointAPIKeyCreate(BaseModel):
     )
 
     # 优先级和限制（数字越小越优先）
-    internal_priority: int = Field(
-        default=50, description="Key 内部优先级（提供商优先模式）"
-    )
+    internal_priority: int = Field(default=50, description="Key 内部优先级（提供商优先模式）")
     # rpm_limit: NULL=自适应模式（系统自动学习），数字=固定限制模式
     rpm_limit: int | None = Field(
         default=None, ge=1, le=10000, description="RPM 限制（NULL=自适应模式）"
@@ -607,9 +567,7 @@ class EndpointAPIKeyCreate(BaseModel):
     )
 
     # 备注
-    note: str | None = Field(
-        default=None, max_length=500, description="备注说明（可选）"
-    )
+    note: str | None = Field(default=None, max_length=500, description="备注说明（可选）")
 
     # 自动获取模型
     auto_fetch_models: bool = Field(
@@ -649,9 +607,7 @@ class EndpointAPIKeyCreate(BaseModel):
         for fmt in v:
             normalized = normalize_signature_key(fmt)
             if resolve_endpoint_definition(normalized) is None:
-                raise ValueError(
-                    f"api_formats 必须是以下之一: {allowed}，当前值: {fmt}"
-                )
+                raise ValueError(f"api_formats 必须是以下之一: {allowed}，当前值: {fmt}")
             if normalized in seen:
                 continue  # 静默去重
             seen.add(normalized)
@@ -737,9 +693,7 @@ class EndpointAPIKeyUpdate(BaseModel):
             "oauth 时存储 token/refresh/expires_at 等（后端加密存储，不在响应中返回）"
         ),
     )
-    name: str | None = Field(
-        default=None, min_length=1, max_length=100, description="密钥名称"
-    )
+    name: str | None = Field(default=None, min_length=1, max_length=100, description="密钥名称")
     rate_multipliers: dict[str, float] | None = Field(
         default=None,
         description="按 endpoint signature 的成本倍率，如 {'claude:cli': 1.0, 'openai:cli': 0.8}",
@@ -774,9 +728,7 @@ class EndpointAPIKeyUpdate(BaseModel):
     )
     is_active: bool | None = Field(default=None, description="是否启用")
     note: str | None = Field(default=None, max_length=500, description="备注说明")
-    auto_fetch_models: bool | None = Field(
-        default=None, description="是否启用自动获取模型"
-    )
+    auto_fetch_models: bool | None = Field(default=None, description="是否启用自动获取模型")
     locked_models: list[str] | None = Field(
         default=None, description="被锁定的模型列表（刷新时不会被删除）"
     )
@@ -891,9 +843,7 @@ class EndpointAPIKeyResponse(BaseModel):
     )
     rpm_limit: int | None = None
     allowed_models: list[str] | None = None
-    capabilities: dict[str, bool] | None = Field(
-        default=None, description="Key 能力标签"
-    )
+    capabilities: dict[str, bool] | None = Field(default=None, description="Key 能力标签")
 
     # OAuth 相关
     oauth_expires_at: int | None = Field(
@@ -904,9 +854,7 @@ class EndpointAPIKeyResponse(BaseModel):
         default=None, description="OAuth 账号套餐类型（如 free/plus/team/enterprise）"
     )
     oauth_account_id: str | None = Field(default=None, description="OAuth 账号 ID")
-    oauth_account_name: str | None = Field(
-        default=None, description="OAuth 当前工作区/账号名称"
-    )
+    oauth_account_name: str | None = Field(default=None, description="OAuth 当前工作区/账号名称")
     oauth_account_user_id: str | None = Field(
         default=None,
         description="OAuth 账号-工作区联合 ID（如 Codex chatgpt_account_user_id）",
@@ -917,17 +865,19 @@ class EndpointAPIKeyResponse(BaseModel):
     )
     oauth_invalid_at: int | None = Field(
         default=None,
-        description="OAuth Token 失效时间（Unix 时间戳），如账号被封、授权撤销等",
+        description="OAuth Token 失效时间（Unix 时间戳，兼容字段；优先使用 status_snapshot.oauth）",
     )
     oauth_invalid_reason: str | None = Field(
-        default=None, description="OAuth Token 失效原因"
+        default=None, description="OAuth Token 失效原因（兼容字段；优先使用 status_snapshot.oauth）"
+    )
+    status_snapshot: ProviderKeyStatusSnapshotResponse = Field(
+        default_factory=ProviderKeyStatusSnapshotResponse,
+        description="统一的账号/OAuth/额度状态快照",
     )
 
     # 缓存与熔断配置
     cache_ttl_minutes: int = Field(default=5, description="缓存 TTL（分钟），0=禁用")
-    max_probe_interval_minutes: int = Field(
-        default=32, description="熔断探测间隔（分钟）"
-    )
+    max_probe_interval_minutes: int = Field(default=32, description="熔断探测间隔（分钟）")
 
     # 按 endpoint signature 的健康度数据
     health_by_format: dict[str, Any] | None = Field(
@@ -943,18 +893,10 @@ class EndpointAPIKeyResponse(BaseModel):
     last_failure_at: datetime | None = None
 
     # 聚合熔断器字段
-    circuit_breaker_open: bool = Field(
-        default=False, description="熔断器是否打开（任何格式）"
-    )
-    circuit_breaker_open_at: datetime | None = Field(
-        default=None, description="熔断器打开时间"
-    )
-    next_probe_at: datetime | None = Field(
-        default=None, description="下次进入半开状态时间"
-    )
-    half_open_until: datetime | None = Field(
-        default=None, description="半开状态结束时间"
-    )
+    circuit_breaker_open: bool = Field(default=False, description="熔断器是否打开（任何格式）")
+    circuit_breaker_open_at: datetime | None = Field(default=None, description="熔断器打开时间")
+    next_probe_at: datetime | None = Field(default=None, description="下次进入半开状态时间")
+    half_open_until: datetime | None = Field(default=None, description="半开状态结束时间")
     half_open_successes: int | None = Field(default=0, description="半开状态成功次数")
     half_open_failures: int | None = Field(default=0, description="半开状态失败次数")
     request_results_window: list[dict[str, Any]] | None = Field(
@@ -972,18 +914,12 @@ class EndpointAPIKeyResponse(BaseModel):
     is_active: bool
 
     # 自适应 RPM 信息
-    is_adaptive: bool = Field(
-        default=False, description="是否为自适应模式（rpm_limit=NULL）"
-    )
+    is_adaptive: bool = Field(default=False, description="是否为自适应模式（rpm_limit=NULL）")
     learned_rpm_limit: int | None = Field(None, description="学习到的 RPM 限制")
     effective_limit: int | None = Field(None, description="当前有效限制")
     # 滑动窗口利用率采样
-    utilization_samples: list[dict[str, Any]] | None = Field(
-        None, description="利用率采样窗口"
-    )
-    last_probe_increase_at: datetime | None = Field(
-        None, description="上次探测性扩容时间"
-    )
+    utilization_samples: list[dict[str, Any]] | None = Field(None, description="利用率采样窗口")
+    last_probe_increase_at: datetime | None = Field(None, description="上次探测性扩容时间")
     concurrent_429_count: int | None = None
     rpm_429_count: int | None = None
     last_429_at: datetime | None = None
@@ -995,9 +931,7 @@ class EndpointAPIKeyResponse(BaseModel):
     # 自动获取模型
     auto_fetch_models: bool = Field(default=False, description="是否启用自动获取模型")
     last_models_fetch_at: datetime | None = Field(None, description="最后获取模型时间")
-    last_models_fetch_error: str | None = Field(
-        None, description="最后获取模型错误信息"
-    )
+    last_models_fetch_error: str | None = Field(None, description="最后获取模型错误信息")
     locked_models: list[str] | None = Field(None, description="被锁定的模型列表")
     # 模型过滤规则
     model_include_patterns: list[str] | None = Field(None, description="模型包含规则")
@@ -1071,9 +1005,7 @@ class HealthStatusResponse(BaseModel):
 class HealthSummaryResponse(BaseModel):
     """健康状态摘要"""
 
-    endpoints: dict[str, int] = Field(
-        ..., description="Endpoint 统计 (total, active, unhealthy)"
-    )
+    endpoints: dict[str, int] = Field(..., description="Endpoint 统计 (total, active, unhealthy)")
     keys: dict[str, int] = Field(..., description="Key 统计 (total, active, unhealthy)")
 
 
@@ -1092,17 +1024,13 @@ class KeyPriorityItem(BaseModel):
     """单个 Key 优先级项"""
 
     key_id: str = Field(..., description="Key ID")
-    internal_priority: int = Field(
-        ..., ge=0, description="Key 内部优先级（数字越小越优先）"
-    )
+    internal_priority: int = Field(..., ge=0, description="Key 内部优先级（数字越小越优先）")
 
 
 class BatchUpdateKeyPriorityRequest(BaseModel):
     """批量更新 Key 优先级请求"""
 
-    priorities: list[KeyPriorityItem] = Field(
-        ..., min_length=1, description="Key 优先级列表"
-    )
+    priorities: list[KeyPriorityItem] = Field(..., min_length=1, description="Key 优先级列表")
 
 
 # ========== 提供商摘要（增强版） ==========
@@ -1114,9 +1042,7 @@ class ProviderUpdateRequest(BaseModel):
     name: str | None = Field(None, min_length=1, max_length=100)
     description: str | None = None
     website: str | None = Field(None, max_length=500, description="主站网站")
-    provider_priority: int | None = Field(
-        None, description="提供商优先级(数字越小越优先)"
-    )
+    provider_priority: int | None = Field(None, description="提供商优先级(数字越小越优先)")
     keep_priority_on_conversion: bool | None = Field(
         None,
         description="格式转换时是否保持优先级（True=保持原优先级，False=需要转换时降级）",
@@ -1130,9 +1056,7 @@ class ProviderUpdateRequest(BaseModel):
         None, description="计费类型：monthly_quota/pay_as_you_go/free_tier"
     )
     monthly_quota_usd: float | None = Field(None, ge=0, description="订阅配额（美元）")
-    quota_reset_day: int | None = Field(
-        None, ge=1, le=31, description="配额重置日（1-31）"
-    )
+    quota_reset_day: int | None = Field(None, ge=1, le=31, description="配额重置日（1-31）")
     quota_expires_at: datetime | None = Field(None, description="配额过期时间")
     # 请求配置（从 Endpoint 迁移）
     max_retries: int | None = Field(None, ge=0, le=10, description="最大重试次数")
@@ -1148,9 +1072,7 @@ class ProviderUpdateRequest(BaseModel):
         None, description="Claude Code 高级配置"
     )
     pool_advanced: PoolAdvancedConfig | None = Field(None, description="通用号池配置")
-    failover_rules: FailoverRulesConfig | None = Field(
-        None, description="故障转移规则配置"
-    )
+    failover_rules: FailoverRulesConfig | None = Field(None, description="故障转移规则配置")
 
 
 class ProviderWithEndpointsSummary(BaseModel):
@@ -1165,9 +1087,7 @@ class ProviderWithEndpointsSummary(BaseModel):
     )
     description: str | None = None
     website: str | None = None
-    provider_priority: int = Field(
-        default=100, description="提供商优先级(数字越小越优先)"
-    )
+    provider_priority: int = Field(default=100, description="提供商优先级(数字越小越优先)")
     keep_priority_on_conversion: bool = Field(
         default=False,
         description="格式转换时是否保持优先级（True=保持原优先级，False=需要转换时降级）",
@@ -1182,12 +1102,8 @@ class ProviderWithEndpointsSummary(BaseModel):
     billing_type: str | None = None
     monthly_quota_usd: float | None = None
     monthly_used_usd: float | None = None
-    quota_reset_day: int | None = Field(
-        default=None, description="配额重置周期（天数）"
-    )
-    quota_last_reset_at: datetime | None = Field(
-        default=None, description="当前周期开始时间"
-    )
+    quota_reset_day: int | None = Field(default=None, description="配额重置周期（天数）")
+    quota_last_reset_at: datetime | None = Field(default=None, description="当前周期开始时间")
     quota_expires_at: datetime | None = Field(default=None, description="配额过期时间")
 
     # 请求配置（从 Endpoint 迁移）
@@ -1197,18 +1113,12 @@ class ProviderWithEndpointsSummary(BaseModel):
     stream_first_byte_timeout: float | None = Field(
         default=None, description="流式请求首字节超时（秒）"
     )
-    request_timeout: float | None = Field(
-        default=None, description="非流式请求整体超时（秒）"
-    )
+    request_timeout: float | None = Field(default=None, description="非流式请求整体超时（秒）")
     claude_code_advanced: ClaudeCodeAdvancedConfig | None = Field(
         default=None, description="Claude Code 高级配置"
     )
-    pool_advanced: PoolAdvancedConfig | None = Field(
-        default=None, description="通用号池配置"
-    )
-    failover_rules: FailoverRulesConfig | None = Field(
-        default=None, description="故障转移规则配置"
-    )
+    pool_advanced: PoolAdvancedConfig | None = Field(default=None, description="通用号池配置")
+    failover_rules: FailoverRulesConfig | None = Field(default=None, description="故障转移规则配置")
 
     # Endpoint 统计
     total_endpoints: int = Field(default=0, description="总 Endpoint 数量")
@@ -1221,9 +1131,7 @@ class ProviderWithEndpointsSummary(BaseModel):
     # Model 统计
     total_models: int = Field(default=0, description="总模型数量")
     active_models: int = Field(default=0, description="活跃模型数量")
-    global_model_ids: list[str] = Field(
-        default=[], description="活跃模型关联的全局模型 ID 列表"
-    )
+    global_model_ids: list[str] = Field(default=[], description="活跃模型关联的全局模型 ID 列表")
 
     # API 格式列表
     api_formats: list[str] = Field(default=[], description="支持的 API 格式列表")
@@ -1241,9 +1149,7 @@ class ProviderWithEndpointsSummary(BaseModel):
     )
 
     # Provider Ops 配置状态
-    ops_configured: bool = Field(
-        default=False, description="是否配置了扩展操作（余额监控等）"
-    )
+    ops_configured: bool = Field(default=False, description="是否配置了扩展操作（余额监控等）")
     ops_architecture_id: str | None = Field(
         default=None, description="扩展操作使用的架构 ID（如 cubence, anyrouter）"
     )
@@ -1322,9 +1228,7 @@ class ApiFormatHealthMonitor(BaseModel):
     time_range_start: datetime | None = Field(
         default=None, description="时间线所覆盖区间的开始时间"
     )
-    time_range_end: datetime | None = Field(
-        default=None, description="时间线所覆盖区间的结束时间"
-    )
+    time_range_end: datetime | None = Field(default=None, description="时间线所覆盖区间的结束时间")
 
 
 class ApiFormatHealthMonitorResponse(BaseModel):
@@ -1358,19 +1262,13 @@ class PublicApiFormatHealthMonitor(BaseModel):
     skipped_count: int = Field(default=0, description="跳过次数")
     success_rate: float = Field(default=1.0, description="成功率")
     last_event_at: datetime | None = None
-    events: list[PublicHealthEvent] = Field(
-        default_factory=list, description="事件列表"
-    )
+    events: list[PublicHealthEvent] = Field(default_factory=list, description="事件列表")
     timeline: list[str] = Field(
         default_factory=list,
         description="Usage 表生成的健康时间线（healthy/warning/unhealthy/unknown）",
     )
-    time_range_start: datetime | None = Field(
-        default=None, description="时间线覆盖区间开始时间"
-    )
-    time_range_end: datetime | None = Field(
-        default=None, description="时间线覆盖区间结束时间"
-    )
+    time_range_start: datetime | None = Field(default=None, description="时间线覆盖区间开始时间")
+    time_range_end: datetime | None = Field(default=None, description="时间线覆盖区间结束时间")
 
 
 class PublicApiFormatHealthMonitorResponse(BaseModel):

src/models/status_snapshot.py

@@ -0,0 +1,40 @@
+from __future__ import annotations
+
+from pydantic import BaseModel, Field
+
+
+class OAuthStatusSnapshotResponse(BaseModel):
+    code: str = Field(default="none", description="OAuth 状态代码")
+    label: str | None = Field(default=None, description="OAuth 状态标签")
+    reason: str | None = Field(default=None, description="OAuth 状态原因")
+    expires_at: int | None = Field(default=None, description="OAuth 过期时间（Unix 时间戳）")
+    invalid_at: int | None = Field(default=None, description="OAuth 失效时间（Unix 时间戳）")
+    source: str | None = Field(default=None, description="OAuth 状态来源")
+    requires_reauth: bool = Field(default=False, description="是否需要重新授权")
+    expiring_soon: bool = Field(default=False, description="是否即将过期")
+
+
+class AccountStatusSnapshotResponse(BaseModel):
+    code: str = Field(default="ok", description="账号状态代码")
+    label: str | None = Field(default=None, description="账号状态标签")
+    reason: str | None = Field(default=None, description="账号状态原因")
+    blocked: bool = Field(default=False, description="是否为账号级阻塞")
+    source: str | None = Field(default=None, description="账号状态来源")
+    recoverable: bool = Field(default=False, description="是否为可恢复状态")
+
+
+class QuotaStatusSnapshotResponse(BaseModel):
+    code: str = Field(default="unknown", description="额度状态代码")
+    label: str | None = Field(default=None, description="额度状态标签")
+    reason: str | None = Field(default=None, description="额度状态原因")
+    exhausted: bool = Field(default=False, description="额度是否耗尽")
+    usage_ratio: float | None = Field(default=None, description="额度使用比例 [0, 1]")
+    updated_at: int | None = Field(default=None, description="额度刷新时间（Unix 时间戳）")
+    reset_seconds: float | None = Field(default=None, description="距离重置剩余秒数")
+    plan_type: str | None = Field(default=None, description="额度读取到的套餐类型")
+
+
+class ProviderKeyStatusSnapshotResponse(BaseModel):
+    oauth: OAuthStatusSnapshotResponse = Field(default_factory=OAuthStatusSnapshotResponse)
+    account: AccountStatusSnapshotResponse = Field(default_factory=AccountStatusSnapshotResponse)
+    quota: QuotaStatusSnapshotResponse = Field(default_factory=QuotaStatusSnapshotResponse)

src/services/provider/pool/account_state.py

@@ -6,6 +6,8 @@ from upstream metadata and OAuth invalid reasons.
 
 from __future__ import annotations
 
+import re
+import time
 from dataclasses import dataclass
 from typing import Any
 
@@ -106,6 +108,47 @@ class PoolAccountState:
     recoverable: bool = False
 
 
+@dataclass(frozen=True, slots=True)
+class OAuthStatusSnapshot:
+    code: str = "none"  # none / valid / expiring / expired / invalid / check_failed
+    label: str | None = None
+    reason: str | None = None
+    expires_at: int | None = None
+    invalid_at: int | None = None
+    source: str | None = None
+    requires_reauth: bool = False
+    expiring_soon: bool = False
+
+
+@dataclass(frozen=True, slots=True)
+class AccountStatusSnapshot:
+    code: str = "ok"
+    label: str | None = None
+    reason: str | None = None
+    blocked: bool = False
+    source: str | None = None
+    recoverable: bool = False
+
+
+@dataclass(frozen=True, slots=True)
+class QuotaStatusSnapshot:
+    code: str = "unknown"  # unknown / ok / exhausted
+    label: str | None = None
+    reason: str | None = None
+    exhausted: bool = False
+    usage_ratio: float | None = None
+    updated_at: int | None = None
+    reset_seconds: float | None = None
+    plan_type: str | None = None
+
+
+@dataclass(frozen=True, slots=True)
+class ProviderKeyStatusSnapshot:
+    oauth: OAuthStatusSnapshot
+    account: AccountStatusSnapshot
+    quota: QuotaStatusSnapshot
+
+
 def _is_truthy_flag(value: Any) -> bool:
     if isinstance(value, bool):
         return value
@@ -139,6 +182,26 @@ def _is_workspace_deactivated_reason(reason: str | None) -> bool:
     return bool(text and "deactivated_workspace" in text.lower())
 
 
+_TAGGED_REASON_PATTERN = re.compile(
+    r"(?:^|\n)\[(?P<tag>[A-Z_]+)\]\s*(?P<detail>.*?)(?=\n\[[A-Z_]+\]|\Z)",
+    re.S,
+)
+
+
+def _extract_tagged_reason_sections(reason: str | None) -> dict[str, str]:
+    text = _clean_text(reason)
+    if not text:
+        return {}
+    sections: dict[str, str] = {}
+    for match in _TAGGED_REASON_PATTERN.finditer(text):
+        tag = str(match.group("tag") or "").strip().upper()
+        if not tag or tag in sections:
+            continue
+        detail = str(match.group("detail") or "").strip()
+        sections[tag] = detail
+    return sections
+
+
 def _resolve_from_metadata(
     provider_type: str | None,
     upstream_metadata: Any,
@@ -266,6 +329,216 @@ def _resolve_from_oauth_invalid_reason(reason: str | None) -> PoolAccountState |
     return None
 
 
+def resolve_account_status_snapshot(
+    *,
+    provider_type: str | None,
+    upstream_metadata: Any,
+    oauth_invalid_reason: str | None,
+) -> AccountStatusSnapshot:
+    from_metadata = _resolve_from_metadata(provider_type, upstream_metadata)
+    if from_metadata is not None:
+        return AccountStatusSnapshot(
+            code=from_metadata.code or "ok",
+            label=from_metadata.label,
+            reason=from_metadata.reason,
+            blocked=from_metadata.blocked,
+            source=from_metadata.source,
+            recoverable=from_metadata.recoverable,
+        )
+
+    text = _clean_text(oauth_invalid_reason)
+    if not text:
+        return AccountStatusSnapshot()
+
+    tagged_sections = _extract_tagged_reason_sections(text)
+    if "ACCOUNT_BLOCK" in tagged_sections:
+        cleaned = tagged_sections["ACCOUNT_BLOCK"]
+        code, label = (
+            _classify_block_reason(cleaned) if cleaned else ("account_blocked", "账号异常")
+        )
+        return AccountStatusSnapshot(
+            code=code,
+            label=label,
+            reason=cleaned or "账号异常",
+            blocked=True,
+            source="oauth_invalid",
+        )
+
+    if text.startswith("["):
+        return AccountStatusSnapshot()
+
+    lowered = text.lower()
+    if any(keyword in lowered for keyword in ACCOUNT_BLOCK_REASON_KEYWORDS):
+        code, label = _classify_block_reason(text)
+        return AccountStatusSnapshot(
+            code=code,
+            label=label,
+            reason=text,
+            blocked=True,
+            source="oauth_invalid",
+        )
+
+    return AccountStatusSnapshot()
+
+
+def resolve_oauth_status_snapshot(
+    *,
+    auth_type: str | None,
+    oauth_expires_at: int | None,
+    oauth_invalid_at: int | None,
+    oauth_invalid_reason: str | None,
+    now_ts: int | None = None,
+) -> OAuthStatusSnapshot:
+    if str(auth_type or "").strip().lower() != "oauth":
+        return OAuthStatusSnapshot()
+
+    now = int(now_ts if now_ts is not None else time.time())
+    invalid_at = int(oauth_invalid_at) if isinstance(oauth_invalid_at, int) else None
+    tagged_sections = _extract_tagged_reason_sections(oauth_invalid_reason)
+    raw_reason = _clean_text(oauth_invalid_reason)
+
+    expired_reason = tagged_sections.get("OAUTH_EXPIRED")
+    if expired_reason:
+        return OAuthStatusSnapshot(
+            code="invalid",
+            label="已失效",
+            reason=expired_reason,
+            invalid_at=invalid_at,
+            expires_at=oauth_expires_at,
+            source="oauth_invalid",
+            requires_reauth=True,
+        )
+
+    refresh_failed_reason = tagged_sections.get("REFRESH_FAILED")
+    if refresh_failed_reason:
+        return OAuthStatusSnapshot(
+            code="invalid",
+            label="已失效",
+            reason=refresh_failed_reason,
+            invalid_at=invalid_at,
+            expires_at=oauth_expires_at,
+            source="oauth_refresh",
+            requires_reauth=True,
+        )
+
+    request_failed_reason = tagged_sections.get("REQUEST_FAILED")
+    if request_failed_reason:
+        return OAuthStatusSnapshot(
+            code="check_failed",
+            label="检查失败",
+            reason=request_failed_reason,
+            expires_at=oauth_expires_at,
+            source="oauth_request",
+        )
+
+    account_snapshot = resolve_account_status_snapshot(
+        provider_type=None,
+        upstream_metadata=None,
+        oauth_invalid_reason=raw_reason,
+    )
+    if account_snapshot.blocked:
+        if oauth_expires_at is None:
+            return OAuthStatusSnapshot()
+    elif raw_reason or invalid_at is not None:
+        return OAuthStatusSnapshot(
+            code="invalid",
+            label="已失效",
+            reason=raw_reason,
+            invalid_at=invalid_at,
+            expires_at=oauth_expires_at,
+            source="oauth_invalid",
+            requires_reauth=True,
+        )
+
+    expires_at = int(oauth_expires_at) if isinstance(oauth_expires_at, int) else None
+    if expires_at is None:
+        return OAuthStatusSnapshot()
+    if expires_at <= now:
+        return OAuthStatusSnapshot(
+            code="expired",
+            label="已过期",
+            reason="Token 已过期，请重新授权",
+            expires_at=expires_at,
+            source="expires_at",
+            requires_reauth=True,
+        )
+    expiring_soon = (expires_at - now) < 24 * 3600
+    return OAuthStatusSnapshot(
+        code="expiring" if expiring_soon else "valid",
+        label="即将过期" if expiring_soon else "有效",
+        expires_at=expires_at,
+        source="expires_at",
+        expiring_soon=expiring_soon,
+    )
+
+
+def resolve_quota_status_snapshot(
+    *,
+    provider_type: str | None,
+    upstream_metadata: Any,
+) -> QuotaStatusSnapshot:
+    normalized_provider = str(provider_type or "").strip().lower()
+    reader = get_quota_reader(normalized_provider, upstream_metadata)
+    quota_state = reader.is_exhausted()
+    usage_ratio = reader.usage_ratio()
+    updated_at = reader.updated_at()
+    reset_seconds = reader.reset_seconds()
+    plan_type = reader.plan_type()
+
+    if quota_state.exhausted:
+        return QuotaStatusSnapshot(
+            code="exhausted",
+            label="额度耗尽",
+            reason=quota_state.reason,
+            exhausted=True,
+            usage_ratio=usage_ratio,
+            updated_at=updated_at,
+            reset_seconds=reset_seconds,
+            plan_type=plan_type,
+        )
+
+    if any(value is not None for value in (usage_ratio, updated_at, reset_seconds, plan_type)):
+        return QuotaStatusSnapshot(
+            code="ok",
+            exhausted=False,
+            usage_ratio=usage_ratio,
+            updated_at=updated_at,
+            reset_seconds=reset_seconds,
+            plan_type=plan_type,
+        )
+
+    return QuotaStatusSnapshot()
+
+
+def build_provider_key_status_snapshot(
+    *,
+    auth_type: str | None,
+    oauth_expires_at: int | None,
+    oauth_invalid_at: int | None,
+    oauth_invalid_reason: str | None,
+    provider_type: str | None,
+    upstream_metadata: Any,
+    now_ts: int | None = None,
+) -> ProviderKeyStatusSnapshot:
+    account = resolve_account_status_snapshot(
+        provider_type=provider_type,
+        upstream_metadata=upstream_metadata,
+        oauth_invalid_reason=oauth_invalid_reason,
+    )
+    oauth = resolve_oauth_status_snapshot(
+        auth_type=auth_type,
+        oauth_expires_at=oauth_expires_at,
+        oauth_invalid_at=oauth_invalid_at,
+        oauth_invalid_reason=oauth_invalid_reason,
+        now_ts=now_ts,
+    )
+    quota = resolve_quota_status_snapshot(
+        provider_type=provider_type,
+        upstream_metadata=upstream_metadata,
+    )
+    return ProviderKeyStatusSnapshot(oauth=oauth, account=account, quota=quota)
+
+
 def resolve_pool_account_state(
     *,
     provider_type: str | None,
@@ -303,11 +576,19 @@ def should_auto_remove_account_state(state: PoolAccountState) -> bool:
 __all__ = [
     "ACCOUNT_BLOCK_REASON_KEYWORDS",
     "AUTO_REMOVABLE_ACCOUNT_STATE_CODES",
+    "AccountStatusSnapshot",
     "OAUTH_ACCOUNT_BLOCK_PREFIX",
     "OAUTH_EXPIRED_PREFIX",
     "OAUTH_REFRESH_FAILED_PREFIX",
     "OAUTH_REQUEST_FAILED_PREFIX",
+    "OAuthStatusSnapshot",
     "PoolAccountState",
+    "ProviderKeyStatusSnapshot",
+    "QuotaStatusSnapshot",
+    "build_provider_key_status_snapshot",
+    "resolve_account_status_snapshot",
+    "resolve_oauth_status_snapshot",
     "resolve_pool_account_state",
+    "resolve_quota_status_snapshot",
     "should_auto_remove_account_state",
 ]

src/services/provider_keys/key_query_service.py

@@ -204,6 +204,12 @@ def list_provider_keys_responses(
     provider = db.query(Provider).filter(Provider.id == provider_id).first()
     if not provider:
         raise NotFoundException(f"Provider {provider_id} 不存在")
+    provider_type = (
+        str(
+            getattr(provider, "provider_type", None) or getattr(provider, "type", None) or ""
+        ).strip()
+        or None
+    )
 
     keys = (
         db.query(ProviderAPIKey)
@@ -213,7 +219,7 @@ def list_provider_keys_responses(
         .limit(limit)
         .all()
     )
-    return [build_key_response(key) for key in keys]
+    return [build_key_response(key, provider_type=provider_type) for key in keys]
 
 
 def reveal_endpoint_key_payload(

src/services/provider_keys/quota_refresh/_helpers.py

@@ -0,0 +1,25 @@
+"""Shared helpers for quota refresh strategies."""
+
+from __future__ import annotations
+
+from typing import Any
+
+from src.models.database import ProviderAPIKey
+from src.services.provider.pool.account_state import OAUTH_REFRESH_FAILED_PREFIX
+
+
+def build_success_state_update(key: ProviderAPIKey) -> dict[str, Any]:
+    """配额刷新成功时的 state_updates 构建。
+
+    如果当前 key 携带 [REFRESH_FAILED] 标记，保留该标记（配额刷新不等于 token 刷新成功）。
+    """
+    current_reason = str(getattr(key, "oauth_invalid_reason", None) or "").strip()
+    if current_reason.startswith(OAUTH_REFRESH_FAILED_PREFIX):
+        return {
+            "oauth_invalid_at": getattr(key, "oauth_invalid_at", None),
+            "oauth_invalid_reason": current_reason,
+        }
+    return {
+        "oauth_invalid_at": None,
+        "oauth_invalid_reason": None,
+    }

src/services/provider_keys/quota_refresh/antigravity_refresher.py

@@ -12,6 +12,7 @@ from sqlalchemy.orm import Session
 from src.core.logger import logger
 from src.models.database import Provider, ProviderAPIKey, ProviderEndpoint
 from src.services.provider.auth import get_provider_auth
+from src.services.provider_keys.quota_refresh._helpers import build_success_state_update
 
 
 async def refresh_antigravity_key_quota(
@@ -110,10 +111,7 @@ async def refresh_antigravity_key_quota(
             upstream_meta["antigravity"]["forbidden_reason"] = None
             upstream_meta["antigravity"]["forbidden_at"] = None
         metadata_updates[key.id] = upstream_meta
-        state_updates[key.id] = {
-            "oauth_invalid_at": None,
-            "oauth_invalid_reason": None,
-        }
+        state_updates[key.id] = build_success_state_update(key)
         return {
             "key_id": key.id,
             "key_name": key.name,

src/services/provider_keys/quota_refresh/codex_refresher.py

@@ -26,6 +26,7 @@ from src.services.provider_keys.codex_usage_parser import (
     parse_codex_usage_headers,
     parse_codex_wham_usage_response,
 )
+from src.services.provider_keys.quota_refresh._helpers import build_success_state_update
 
 
 def _normalize_plan_type(value: Any) -> str | None:
@@ -277,10 +278,7 @@ async def refresh_codex_key_quota(
                 metadata_updates[key.id] = {
                     "codex": _build_quota_exhausted_fallback_metadata(oauth_plan_type)
                 }
-            state_updates[key.id] = {
-                "oauth_invalid_at": None,
-                "oauth_invalid_reason": None,
-            }
+            state_updates[key.id] = build_success_state_update(key)
             return {
                 "key_id": key.id,
                 "key_name": key.name,
@@ -348,10 +346,7 @@ async def refresh_codex_key_quota(
     if metadata:
         # 收集元数据，稍后统一更新数据库（存储到 codex 子对象）
         metadata_updates[key.id] = {"codex": metadata}
-        state_updates[key.id] = {
-            "oauth_invalid_at": None,
-            "oauth_invalid_reason": None,
-        }
+        state_updates[key.id] = build_success_state_update(key)
         return {
             "key_id": key.id,
             "key_name": key.name,

src/services/provider_keys/quota_refresh/kiro_refresher.py

@@ -13,6 +13,7 @@ from sqlalchemy.orm import Session
 from src.core.crypto import crypto_service
 from src.core.logger import logger
 from src.models.database import Provider, ProviderAPIKey, ProviderEndpoint
+from src.services.provider_keys.quota_refresh._helpers import build_success_state_update
 
 
 async def refresh_kiro_key_quota(
@@ -132,10 +133,7 @@ async def refresh_kiro_key_quota(
         metadata["banned_at"] = None
         # 收集元数据，稍后统一更新数据库（存储到 kiro 子对象）
         metadata_updates[key.id] = {"kiro": metadata}
-        state_updates[key.id] = {
-            "oauth_invalid_at": None,
-            "oauth_invalid_reason": None,
-        }
+        state_updates[key.id] = build_success_state_update(key)
 
         # 如果 auth_config 有更新（例如 token 刷新），也需要更新
         if updated_auth_config:

src/services/provider_keys/response_builder.py

@@ -5,7 +5,7 @@ Provider Key 响应对象构建器。
 from __future__ import annotations
 
 import json
-from datetime import datetime
+from dataclasses import asdict
 from typing import Any
 
 from src.core.crypto import crypto_service
@@ -14,10 +14,17 @@ from src.core.provider_oauth_utils import normalize_oauth_organizations
 from src.models.database import ProviderAPIKey
 from src.models.endpoint_models import EndpointAPIKeyResponse
 from src.services.provider_keys.auth_type import normalize_auth_type
+from src.services.provider_keys.status_snapshot_store import (
+    normalize_oauth_expires_at,
+    resolve_provider_key_status_snapshot,
+)
 
 
 def build_key_response(
-    key: ProviderAPIKey, api_key_plain: str | None = None
+    key: ProviderAPIKey,
+    api_key_plain: str | None = None,
+    *,
+    provider_type: str | None = None,
 ) -> EndpointAPIKeyResponse:
     """构建 Key 响应对象。"""
     auth_type = normalize_auth_type(getattr(key, "auth_type", "api_key"))
@@ -40,9 +47,7 @@ def build_key_response(
             masked_key = "***ERROR***"
 
     success_rate = success_count / request_count if request_count > 0 else 0.0
-    avg_response_time_ms = (
-        total_response_time_ms / success_count if success_count > 0 else 0.0
-    )
+    avg_response_time_ms = total_response_time_ms / success_count if success_count > 0 else 0.0
 
     is_adaptive = rpm_limit is None
     key_dict: dict[str, Any] = dict(getattr(key, "__dict__", {}))
@@ -57,66 +62,61 @@ def build_key_response(
     oauth_account_id = None
     oauth_account_name = None
     oauth_account_user_id = None
+    auth_config: dict[str, Any] | None = None
     oauth_organizations: list[dict[str, object]] = []
     encrypted_auth_config = key_dict.pop("auth_config", None)  # 移除敏感字段，避免泄露
-    if (
-        auth_type == "oauth"
-        and isinstance(encrypted_auth_config, str)
-        and encrypted_auth_config
-    ):
+    if auth_type == "oauth" and isinstance(encrypted_auth_config, str) and encrypted_auth_config:
         try:
             decrypted_config = crypto_service.decrypt(encrypted_auth_config)
             auth_config = json.loads(decrypted_config)
-            oauth_expires_at = auth_config.get("expires_at")
+            oauth_expires_at = normalize_oauth_expires_at(auth_config.get("expires_at"))
             oauth_email = auth_config.get("email")
-            oauth_plan_type = auth_config.get(
-                "plan_type"
-            )  # Codex: plus/free/team/enterprise
+            oauth_plan_type = auth_config.get("plan_type")  # Codex: plus/free/team/enterprise
             # Antigravity 使用 "tier" 字段（如 "PAID"/"FREE"），做小写化 fallback
             if not oauth_plan_type:
                 ag_tier = auth_config.get("tier")
                 if ag_tier and isinstance(ag_tier, str):
                     oauth_plan_type = ag_tier.lower()
-            oauth_account_id = auth_config.get(
-                "account_id"
-            )  # Codex: chatgpt_account_id
+            oauth_account_id = auth_config.get("account_id")  # Codex: chatgpt_account_id
             oauth_account_name = auth_config.get("account_name")
             oauth_account_user_id = auth_config.get("account_user_id")
-            oauth_organizations = normalize_oauth_organizations(
-                auth_config.get("organizations")
-            )
+            oauth_organizations = normalize_oauth_organizations(auth_config.get("organizations"))
         except Exception as e:
             logger.error("Failed to decrypt auth_config for key {}: {}", key.id, e)
 
+    if not provider_type:
+        provider_rel = getattr(key, "provider", None)
+        provider_type = (
+            str(getattr(provider_rel, "provider_type", None) or "").strip()
+            or str(getattr(provider_rel, "type", None) or "").strip()
+            or None
+        )
+
+    status_snapshot = resolve_provider_key_status_snapshot(
+        key,
+        provider_type=provider_type,
+        auth_config=auth_config,
+        oauth_expires_at=oauth_expires_at,
+    )
+
     # 从 health_by_format 计算汇总字段（便于列表展示）
     raw_health_by_format = getattr(key, "health_by_format", None)
-    health_by_format = (
-        raw_health_by_format if isinstance(raw_health_by_format, dict) else {}
-    )
+    health_by_format = raw_health_by_format if isinstance(raw_health_by_format, dict) else {}
     raw_circuit_by_format = getattr(key, "circuit_breaker_by_format", None)
-    circuit_by_format = (
-        raw_circuit_by_format if isinstance(raw_circuit_by_format, dict) else {}
-    )
+    circuit_by_format = raw_circuit_by_format if isinstance(raw_circuit_by_format, dict) else {}
 
     # 计算整体健康度（取所有格式中的最低值）
     if health_by_format:
-        health_scores = [
-            float(h.get("health_score") or 1.0) for h in health_by_format.values()
-        ]
+        health_scores = [float(h.get("health_score") or 1.0) for h in health_by_format.values()]
         min_health_score = min(health_scores) if health_scores else 1.0
         # 取最大的连续失败次数
         max_consecutive = max(
-            (
-                int(h.get("consecutive_failures") or 0)
-                for h in health_by_format.values()
-            ),
+            (int(h.get("consecutive_failures") or 0) for h in health_by_format.values()),
             default=0,
         )
         # 取最近的失败时间
         failure_times = [
-            h.get("last_failure_at")
-            for h in health_by_format.values()
-            if h.get("last_failure_at")
+            h.get("last_failure_at") for h in health_by_format.values() if h.get("last_failure_at")
         ]
         last_failure = max(failure_times) if failure_times else None
     else:
@@ -154,15 +154,9 @@ def build_key_response(
             "oauth_account_name": oauth_account_name,
             "oauth_account_user_id": oauth_account_user_id,
             "oauth_organizations": oauth_organizations,
-            "oauth_invalid_at": (
-                int(oauth_invalid_at.timestamp())
-                if isinstance(
-                    (oauth_invalid_at := getattr(key, "oauth_invalid_at", None)),
-                    datetime,
-                )
-                else None
-            ),
+            "oauth_invalid_at": status_snapshot.oauth.invalid_at,
             "oauth_invalid_reason": getattr(key, "oauth_invalid_reason", None),
+            "status_snapshot": asdict(status_snapshot),
         }
     )
 

src/services/provider_keys/status_snapshot_store.py

@@ -0,0 +1,293 @@
+from __future__ import annotations
+
+import json
+from dataclasses import asdict
+from datetime import datetime
+from typing import Any
+
+from sqlalchemy import select
+from sqlalchemy.engine import Connection
+
+from src.core.crypto import crypto_service
+from src.models.database import Provider, ProviderAPIKey
+from src.services.provider.pool.account_state import (
+    AccountStatusSnapshot,
+    OAuthStatusSnapshot,
+    ProviderKeyStatusSnapshot,
+    QuotaStatusSnapshot,
+    build_provider_key_status_snapshot,
+    resolve_oauth_status_snapshot,
+)
+
+
+def _clean_text(value: Any) -> str | None:
+    if not isinstance(value, str):
+        return None
+    text = value.strip()
+    return text or None
+
+
+def _coerce_bool(value: Any) -> bool:
+    if isinstance(value, bool):
+        return value
+    if isinstance(value, (int, float)):
+        return value != 0
+    if isinstance(value, str):
+        return value.strip().lower() in {"1", "true", "yes", "y"}
+    return False
+
+
+def _coerce_int(value: Any) -> int | None:
+    if isinstance(value, bool):
+        return None
+    if isinstance(value, int):
+        return value
+    if isinstance(value, float):
+        return int(value)
+    if isinstance(value, str):
+        text = value.strip()
+        if not text:
+            return None
+        try:
+            return int(float(text))
+        except ValueError:
+            return None
+    return None
+
+
+def _coerce_float(value: Any) -> float | None:
+    if isinstance(value, bool):
+        return None
+    if isinstance(value, (int, float)):
+        return float(value)
+    if isinstance(value, str):
+        text = value.strip()
+        if not text:
+            return None
+        try:
+            return float(text)
+        except ValueError:
+            return None
+    return None
+
+
+def extract_oauth_auth_config(key: ProviderAPIKey) -> dict[str, Any] | None:
+    if str(getattr(key, "auth_type", "") or "").strip().lower() != "oauth":
+        return None
+
+    auth_config_raw = getattr(key, "auth_config", None)
+    if not auth_config_raw:
+        return None
+
+    try:
+        decrypted = crypto_service.decrypt(auth_config_raw)
+        if isinstance(decrypted, str) and decrypted.strip():
+            parsed = json.loads(decrypted)
+            if isinstance(parsed, dict):
+                return parsed
+    except Exception:
+        pass
+
+    return None
+
+
+def normalize_oauth_expires_at(raw: Any) -> int | None:
+    value = _coerce_float(raw)
+    if value is None or value <= 0:
+        return None
+    if value > 1_000_000_000_000:
+        value /= 1000
+    return int(value)
+
+
+def hydrate_provider_key_status_snapshot(raw: Any) -> ProviderKeyStatusSnapshot | None:
+    if not isinstance(raw, dict):
+        return None
+
+    oauth_raw = raw.get("oauth") if isinstance(raw.get("oauth"), dict) else {}
+    account_raw = raw.get("account") if isinstance(raw.get("account"), dict) else {}
+    quota_raw = raw.get("quota") if isinstance(raw.get("quota"), dict) else {}
+
+    return ProviderKeyStatusSnapshot(
+        oauth=OAuthStatusSnapshot(
+            code=_clean_text(oauth_raw.get("code")) or "none",
+            label=_clean_text(oauth_raw.get("label")),
+            reason=_clean_text(oauth_raw.get("reason")),
+            expires_at=_coerce_int(oauth_raw.get("expires_at")),
+            invalid_at=_coerce_int(oauth_raw.get("invalid_at")),
+            source=_clean_text(oauth_raw.get("source")),
+            requires_reauth=_coerce_bool(oauth_raw.get("requires_reauth")),
+            expiring_soon=_coerce_bool(oauth_raw.get("expiring_soon")),
+        ),
+        account=AccountStatusSnapshot(
+            code=_clean_text(account_raw.get("code")) or "ok",
+            label=_clean_text(account_raw.get("label")),
+            reason=_clean_text(account_raw.get("reason")),
+            blocked=_coerce_bool(account_raw.get("blocked")),
+            source=_clean_text(account_raw.get("source")),
+            recoverable=_coerce_bool(account_raw.get("recoverable")),
+        ),
+        quota=QuotaStatusSnapshot(
+            code=_clean_text(quota_raw.get("code")) or "unknown",
+            label=_clean_text(quota_raw.get("label")),
+            reason=_clean_text(quota_raw.get("reason")),
+            exhausted=_coerce_bool(quota_raw.get("exhausted")),
+            usage_ratio=_coerce_float(quota_raw.get("usage_ratio")),
+            updated_at=_coerce_int(quota_raw.get("updated_at")),
+            reset_seconds=_coerce_float(quota_raw.get("reset_seconds")),
+            plan_type=_clean_text(quota_raw.get("plan_type")),
+        ),
+    )
+
+
+def resolve_provider_type_for_key(
+    key: ProviderAPIKey,
+    *,
+    provider_type: str | None = None,
+    connection: Connection | None = None,
+) -> str | None:
+    normalized = _clean_text(provider_type)
+    if normalized:
+        return normalized
+
+    provider_rel = getattr(key, "__dict__", {}).get("provider")
+    rel_type = _clean_text(getattr(provider_rel, "provider_type", None)) or _clean_text(
+        getattr(provider_rel, "type", None)
+    )
+    if rel_type:
+        return rel_type
+
+    provider_id = _clean_text(getattr(key, "provider_id", None))
+    if provider_id and connection is not None:
+        result = connection.execute(
+            select(Provider.provider_type).where(Provider.id == provider_id)
+        ).scalar_one_or_none()
+        return _clean_text(result)
+
+    return None
+
+
+def derive_oauth_expires_at(
+    key: ProviderAPIKey,
+    *,
+    auth_config: dict[str, Any] | None = None,
+) -> int | None:
+    if str(getattr(key, "auth_type", "") or "").strip().lower() != "oauth":
+        return None
+
+    cfg = auth_config if isinstance(auth_config, dict) else extract_oauth_auth_config(key)
+    if cfg:
+        for field in ("expires_at", "expiresAt", "expiry", "exp"):
+            expires_at = normalize_oauth_expires_at(cfg.get(field))
+            if expires_at is not None:
+                return expires_at
+
+    expires_dt = getattr(key, "expires_at", None)
+    if isinstance(expires_dt, datetime):
+        return int(expires_dt.timestamp())
+    return None
+
+
+def resolve_provider_key_status_snapshot(
+    key: ProviderAPIKey,
+    *,
+    provider_type: str | None = None,
+    connection: Connection | None = None,
+    auth_config: dict[str, Any] | None = None,
+    oauth_expires_at: int | None = None,
+    now_ts: int | None = None,
+) -> ProviderKeyStatusSnapshot:
+    persisted_snapshot = hydrate_provider_key_status_snapshot(getattr(key, "status_snapshot", None))
+    current_snapshot = _build_snapshot_from_current_fields(
+        key,
+        provider_type=provider_type,
+        connection=connection,
+        auth_config=auth_config,
+        oauth_expires_at=oauth_expires_at,
+        now_ts=now_ts,
+    )
+    if persisted_snapshot is None:
+        return current_snapshot
+
+    resolved_oauth_expires_at = current_snapshot.oauth.expires_at
+    if resolved_oauth_expires_at is None and persisted_snapshot.oauth.expires_at is not None:
+        resolved_oauth_expires_at = int(persisted_snapshot.oauth.expires_at)
+    resolved_oauth_invalid_at = current_snapshot.oauth.invalid_at
+    if resolved_oauth_invalid_at is None and persisted_snapshot.oauth.invalid_at is not None:
+        resolved_oauth_invalid_at = int(persisted_snapshot.oauth.invalid_at)
+    oauth_invalid_reason = _clean_text(getattr(key, "oauth_invalid_reason", None)) or (
+        persisted_snapshot.oauth.reason if persisted_snapshot is not None else None
+    )
+
+    return ProviderKeyStatusSnapshot(
+        oauth=resolve_oauth_status_snapshot(
+            auth_type=str(getattr(key, "auth_type", "api_key") or "api_key"),
+            oauth_expires_at=resolved_oauth_expires_at,
+            oauth_invalid_at=resolved_oauth_invalid_at,
+            oauth_invalid_reason=oauth_invalid_reason,
+            now_ts=now_ts,
+        ),
+        account=persisted_snapshot.account,
+        quota=persisted_snapshot.quota,
+    )
+
+
+def _build_snapshot_from_current_fields(
+    key: ProviderAPIKey,
+    *,
+    provider_type: str | None = None,
+    connection: Connection | None = None,
+    auth_config: dict[str, Any] | None = None,
+    oauth_expires_at: int | None = None,
+    now_ts: int | None = None,
+) -> ProviderKeyStatusSnapshot:
+    resolved_provider_type = resolve_provider_type_for_key(
+        key, provider_type=provider_type, connection=connection
+    )
+    oauth_auth_config = (
+        auth_config if isinstance(auth_config, dict) else extract_oauth_auth_config(key)
+    )
+    normalized_oauth_expires_at = normalize_oauth_expires_at(oauth_expires_at)
+    resolved_oauth_expires_at = (
+        normalized_oauth_expires_at
+        if normalized_oauth_expires_at is not None
+        else derive_oauth_expires_at(
+            key,
+            auth_config=oauth_auth_config,
+        )
+    )
+    raw_invalid_at = getattr(key, "oauth_invalid_at", None)
+    oauth_invalid_at = (
+        int(raw_invalid_at.timestamp()) if isinstance(raw_invalid_at, datetime) else None
+    )
+    oauth_invalid_reason = _clean_text(getattr(key, "oauth_invalid_reason", None))
+
+    return build_provider_key_status_snapshot(
+        auth_type=str(getattr(key, "auth_type", "api_key") or "api_key"),
+        oauth_expires_at=resolved_oauth_expires_at,
+        oauth_invalid_at=oauth_invalid_at,
+        oauth_invalid_reason=oauth_invalid_reason,
+        provider_type=resolved_provider_type,
+        upstream_metadata=getattr(key, "upstream_metadata", None),
+        now_ts=now_ts,
+    )
+
+
+def sync_provider_key_status_snapshot(
+    key: ProviderAPIKey,
+    *,
+    provider_type: str | None = None,
+    connection: Connection | None = None,
+    auth_config: dict[str, Any] | None = None,
+    oauth_expires_at: int | None = None,
+) -> dict[str, Any]:
+    snapshot = _build_snapshot_from_current_fields(
+        key,
+        provider_type=provider_type,
+        connection=connection,
+        auth_config=auth_config,
+        oauth_expires_at=oauth_expires_at,
+    )
+    snapshot_dict = asdict(snapshot)
+    key.status_snapshot = snapshot_dict
+    return snapshot_dict