feat: 新增 Auth Group 多凭证调度与管理面板

- 新增 auth_group/auth_entry 配置模型、运行态管理与 SQLite 持久化
- 支持 provider 绑定 auth_group，并在代理链路中按 least_inflight 选择凭证
- 增加 429 冷却、401/403 禁用、请求数与 token 配额统计及重置能力
- 扩展 HookContext，暴露 auth_group_name 和 auth_entry_id
- 重构 Provider 管理页，新增 Auth Group 管理、运行态面板与 Auth Entries YAML 编辑
- 同步更新 README、架构文档、示例配置和相关测试

README.md

@@ -89,6 +89,32 @@
 
 ## Provider 配置
 
+### Auth Groups
+
+```yaml
+auth_groups:
+  - name: openai-shared
+    strategy: least_inflight
+    cooldown_seconds_on_429: 60
+    entries:
+      - id: key-a
+        headers:
+          Authorization: Bearer sk-your-openai-key-a
+        max_concurrency: 3
+      - id: key-b
+        headers:
+          Authorization: Bearer sk-your-openai-key-b
+        max_concurrency: 2
+```
+
+规则：
+
+- provider 可以引用 `auth_group`，也可以继续用 legacy `api_key`
+- `auth_group` 和 `api_key` 不能同时填写；也可以都不填，交给 hook 或无鉴权上游处理
+- `api_key` 仍然保留为 legacy 单 key 快捷写法
+- `auth_group` 支持并发控制、`429` 冷却、请求数配额、Token 配额
+- `401/403` 会禁用当前 entry，直到在管理页手动恢复
+
 ### 样例
 
 见 [config.sample.yaml](/d:/001Code/008llm/003LLM_Proxy/config.sample.yaml)。
@@ -102,7 +128,7 @@ providers:
     transport: http
     source_format: openai_chat
     target_format: openai_chat
-    api_key: sk-your-openai-key
+    auth_group: openai-shared
     verify_ssl: true
     model_list:
       - gpt-4.1
@@ -120,6 +146,10 @@ providers:
   - 上游真实协议
 - `target_format`
   - 下游看到的协议
+- `auth_group`
+  - 引用顶层 `auth_groups` 里的凭证池
+- `api_key`
+  - legacy 单 key 快捷写法；不能和 `auth_group` 同时填写，也可以两者都留空
 - `model_list`
   - 这个 provider 暴露给下游的模型名列表
 - `hook`
@@ -204,6 +234,8 @@ class Hook(BaseHook):
 - `provider_target_format`
 - `transport`
 - `stream`
+- `auth_group_name`
+- `auth_entry_id`
 - `last_status_code`
 - `last_error_type`
 
@@ -223,9 +255,15 @@ class Hook(BaseHook):
 
 ## Provider 页面
 
-Provider 管理页不再展示兼容矩阵，也不再让用户选择流格式。
+Provider 管理页现在拆成两块：
 
-页面上会提供 3 个帮助入口：
+- `Auth Groups`
+  - 维护 auth_group / auth_entry
+  - 查看运行态、冷却、禁用和配额
+- `Providers`
+  - 维护 provider 与它绑定的 `auth_group`、legacy `api_key`，或留空交给 hook
+
+页面上会继续提供 3 个帮助入口：
 
 - `transport`
 - `source_format`

config.sample.yaml

@@ -9,13 +9,34 @@ admin:
   username: admin
   password: admin
 
+auth_groups:
+  - name: openai-shared
+    strategy: least_inflight
+    cooldown_seconds_on_429: 60
+    entries:
+      - id: openai-key-a
+        enabled: true
+        headers:
+          Authorization: Bearer sk-your-openai-key-a
+        max_concurrency: 3
+        cooldown_seconds_on_429: 120
+        request_quota_per_minute: 60
+        request_quota_per_day: 3000
+        token_quota_per_minute: 120000
+        token_quota_per_day: 5000000
+      - id: openai-key-b
+        enabled: true
+        headers:
+          Authorization: Bearer sk-your-openai-key-b
+        max_concurrency: 2
+
 providers:
   - name: openai-chat
     api: https://api.openai.com/v1/chat/completions
     transport: http
     source_format: openai_chat
     target_format: openai_chat
-    api_key: sk-your-openai-key
+    auth_group: openai-shared
     verify_ssl: true
     timeout_seconds: 1200
     max_retries: 3
@@ -29,7 +50,7 @@ providers:
     transport: http
     source_format: openai_responses
     target_format: openai_chat
-    api_key: sk-your-openai-key
+    auth_group: openai-shared
     verify_ssl: true
     timeout_seconds: 1200
     max_retries: 3

config.yaml

@@ -13,3 +13,12 @@ providers:
   verify_ssl: false
   model_list:
   - kimi-k2.5
+auth_groups:
+- name: a
+  strategy: least_inflight
+  cooldown_seconds_on_429: 60
+  entries:
+  - id: a
+    enabled: true
+    headers:
+      a: a

docs/architecture-4plus1.md

@@ -52,6 +52,10 @@ downstream request
 - `ProviderManager`
   - 加载 provider 配置
   - 维护 `provider/model -> provider` 映射
+- `AuthGroupManager`
+  - 加载 `auth_groups`
+  - 选择 `auth_entry`
+  - 持久化冷却、禁用与配额运行态
 - `ProxyService`
   - 组装整条代理链路
 - `ExecutorRegistry`
@@ -68,6 +72,8 @@ downstream request
 Hook 组件除了 header / guard，还会收到最小重试上下文：
 
 - `retry`
+- `auth_group_name`
+- `auth_entry_id`
 - `last_status_code`
 - `last_error_type`
 
@@ -116,6 +122,7 @@ Provider 公共运行时字段只有：
 - `source_format`
 - `target_format`
 - `api_key`
+- `auth_group`
 - `proxy`
 - `timeout_seconds`
 - `max_retries`
@@ -137,6 +144,8 @@ Provider 公共运行时字段只有：
 Hook 运行时上下文还会暴露最小重试状态：
 
 - `retry`
+- `auth_group_name`
+- `auth_entry_id`
 - `last_status_code`
 - `last_error_type`
 

src/application/application.py

@@ -8,7 +8,12 @@ from pathlib import Path
 from flask import request
 
 from .app_context import AppContext
-from ..config import ConfigManager, ProviderManager, build_provider_schemas
+from ..config import (
+    ConfigManager,
+    ProviderManager,
+    build_auth_group_schemas,
+    build_provider_schemas,
+)
 from ..presentation import (
     AuthenticationController,
     ProviderController,
@@ -17,8 +22,9 @@ from ..presentation import (
     WebController,
     create_flask_app,
 )
-from ..repositories import LogRepository, UserRepository
+from ..repositories import AuthGroupRepository, LogRepository, UserRepository
 from ..services import (
+    AuthGroupService,
     AuthenticationService,
     LogService,
     ModelDiscoveryService,
@@ -147,10 +153,14 @@ class Application:
         self._db_connection_factory = create_connection_factory(db_path)
         self._user_repository = UserRepository(self._db_connection_factory)
         self._log_repository = LogRepository(self._db_connection_factory)
+        self._auth_group_repository = AuthGroupRepository(self._db_connection_factory)
 
     def _setup_provider_manager(self) -> None:
         """加载 provider 配置并注册可用模型。"""
-        self._provider_manager = ProviderManager(self._ctx)
+        from ..config import AuthGroupManager
+
+        self._auth_group_manager = AuthGroupManager(self._ctx, self._auth_group_repository)
+        self._provider_manager = ProviderManager(self._ctx, self._auth_group_manager)
         self.reload_providers()
 
     def _setup_controllers(self) -> None:
@@ -158,9 +168,14 @@ class Application:
         auth_service = AuthenticationService(self._ctx)
         user_service = UserService(self._ctx, self._user_repository)
         self._user_service = user_service
-        proxy_service = ProxyService(self._ctx)
+        proxy_service = ProxyService(self._ctx, self._auth_group_manager)
         log_service = LogService(self._ctx, self._log_repository)
         provider_service = ProviderService(self._ctx, self.reload_providers)
+        auth_group_service = AuthGroupService(
+            self._ctx,
+            self.reload_providers,
+            self._auth_group_manager,
+        )
         model_discovery_service = ModelDiscoveryService(self._ctx)
         settings_service = SettingsService(self._ctx)
 
@@ -169,6 +184,7 @@ class Application:
         self._provider_controller = ProviderController(
             self._ctx,
             provider_service,
+            auth_group_service,
             model_discovery_service,
             settings_service,
             auth_service,
@@ -187,10 +203,18 @@ class Application:
     def reload_providers(self) -> None:
         self._config_manager.reload()
         config_dict = self._config_manager.get_raw_config()
+        auth_groups_config = config_dict.get("auth_groups", [])
+        if auth_groups_config is None:
+            auth_groups_config = []
         providers_config = config_dict.get("providers", [])
         if providers_config is None:
             providers_config = []
-        provider_schemas = build_provider_schemas(providers_config)
+        auth_group_schemas = build_auth_group_schemas(auth_groups_config)
+        provider_schemas = build_provider_schemas(
+            providers_config,
+            available_auth_group_names={schema.name for schema in auth_group_schemas},
+        )
+        self._auth_group_manager.load_auth_groups(auth_group_schemas)
         self._provider_manager.load_providers(provider_schemas)
 
     def run(self) -> None:

src/config/__init__.py

@@ -4,23 +4,38 @@
 
 from .config_manager import ConfigManager
 from .provider_config import (
+    AuthEntrySchema,
+    AuthGroupSchema,
     ProviderConfigSchema,
     ProviderRuntimeView,
     RuntimeProviderSpec,
+    build_auth_group_schemas,
     build_provider_schemas,
+    validate_auth_group_definitions,
+    validate_auth_group_provider_definitions,
 )
 
 __all__ = [
     "ConfigManager",
+    "AuthEntrySchema",
+    "AuthGroupSchema",
     "ProviderConfigSchema",
     "RuntimeProviderSpec",
     "ProviderRuntimeView",
+    "build_auth_group_schemas",
     "build_provider_schemas",
+    "validate_auth_group_definitions",
+    "validate_auth_group_provider_definitions",
+    "AuthGroupManager",
     "ProviderManager",
 ]
 
 
 def __getattr__(name: str):
+    if name == "AuthGroupManager":
+        from .auth_group_manager import AuthGroupManager
+
+        return AuthGroupManager
     if name == "ProviderManager":
         from .provider_manager import ProviderManager
 

src/config/auth_group_manager.py

@@ -0,0 +1,506 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+"""Auth group runtime manager."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from datetime import datetime, timedelta, timezone
+from email.utils import parsedate_to_datetime
+from typing import Any, Dict, Mapping, Optional, Sequence
+
+from ..application.app_context import AppContext, Logger
+from ..hooks import HookErrorType
+from ..repositories import AuthGroupRepository
+from ..utils.local_time import format_local_datetime, now_local_datetime, parse_local_datetime
+from .provider_config import (
+    AuthEntrySchema,
+    AuthGroupSchema,
+    DEFAULT_AUTH_GROUP_COOLDOWN_SECONDS_ON_429,
+    DEFAULT_AUTH_GROUP_STRATEGY,
+)
+
+
+@dataclass(frozen=True, slots=True)
+class SelectedAuthEntry:
+    """Resolved auth entry selection for a single upstream attempt."""
+
+    auth_group_name: str
+    entry_id: str
+    headers: tuple[tuple[str, str], ...]
+    legacy_api_key: bool = False
+
+    def headers_mapping(self) -> Dict[str, str]:
+        return dict(self.headers)
+
+
+class AuthGroupSelectionError(Exception):
+    """Raised when no usable auth entry can be selected."""
+
+    def __init__(
+        self,
+        message: str,
+        *,
+        status_code: int = 503,
+        error_type: str = "auth_group_unavailable",
+        error_code: str = "auth_group_unavailable",
+    ) -> None:
+        super().__init__(message)
+        self.message = message
+        self.status_code = status_code
+        self.error_type = error_type
+        self.error_code = error_code
+
+
+class AuthGroupManager:
+    """Select auth entries and persist their runtime state."""
+
+    def __init__(self, ctx: AppContext, repository: AuthGroupRepository):
+        self._logger: Logger = ctx.logger
+        self._repository = repository
+        self._groups_by_name: Dict[str, AuthGroupSchema] = {}
+        self._explicit_group_names: set[str] = set()
+        self._inflight_counts: Dict[tuple[str, str], int] = {}
+        self._rotation_cursor_by_group: Dict[str, int] = {}
+
+    def load_auth_groups(self, auth_groups: Sequence[AuthGroupSchema]) -> None:
+        self._groups_by_name = {group.name: group for group in auth_groups}
+        self._explicit_group_names = {group.name for group in auth_groups}
+        self._inflight_counts.clear()
+        self._rotation_cursor_by_group.clear()
+        self._logger.info("Loaded auth groups: %s", list(sorted(self._groups_by_name.keys())))
+
+    def register_legacy_provider_group(self, provider_name: str, api_key: str) -> str:
+        group_name = f"__legacy_provider__/{provider_name}"
+        if group_name not in self._groups_by_name:
+            entry = AuthEntrySchema(
+                id="legacy",
+                enabled=True,
+                headers=(("Authorization", f"Bearer {api_key}"),),
+            )
+            self._groups_by_name[group_name] = AuthGroupSchema(
+                name=group_name,
+                strategy=DEFAULT_AUTH_GROUP_STRATEGY,
+                cooldown_seconds_on_429=DEFAULT_AUTH_GROUP_COOLDOWN_SECONDS_ON_429,
+                entries=(entry,),
+            )
+        return group_name
+
+    def acquire(self, auth_group_name: Optional[str]) -> Optional[SelectedAuthEntry]:
+        normalized_group_name = str(auth_group_name or "").strip()
+        if not normalized_group_name:
+            return None
+
+        group = self._groups_by_name.get(normalized_group_name)
+        if group is None:
+            raise AuthGroupSelectionError(
+                f"Unknown auth_group: {normalized_group_name}",
+                error_code="unknown_auth_group",
+            )
+
+        now = now_local_datetime()
+        runtime_states = self._repository.list_group_runtime_states(group.name)
+        usage_by_entry = self._repository.list_current_usage(group.name, (entry.id for entry in group.entries), now)
+
+        candidates: list[tuple[int, int, int, AuthEntrySchema]] = []
+        for index, entry in enumerate(group.entries):
+            runtime_state = runtime_states.get(entry.id) or self._repository.get_entry_runtime_state(group.name, entry.id)
+            usage = usage_by_entry.get(entry.id) or {}
+            if not self._is_entry_available(group, entry, runtime_state, usage, now):
+                continue
+            inflight = self._inflight_counts.get((group.name, entry.id), 0)
+            candidates.append((inflight, self._rotation_distance(group.name, index, len(group.entries)), index, entry))
+
+        if not candidates:
+            raise AuthGroupSelectionError(
+                f"No available auth entries for auth_group: {group.name}",
+                error_code="auth_entry_unavailable",
+            )
+
+        _, _, selected_index, selected_entry = min(candidates, key=lambda item: (item[0], item[1], item[2]))
+        key = (group.name, selected_entry.id)
+        self._inflight_counts[key] = self._inflight_counts.get(key, 0) + 1
+        self._rotation_cursor_by_group[group.name] = (selected_index + 1) % max(len(group.entries), 1)
+
+        return SelectedAuthEntry(
+            auth_group_name=group.name,
+            entry_id=selected_entry.id,
+            headers=selected_entry.headers,
+            legacy_api_key=group.name not in self._explicit_group_names,
+        )
+
+    def mark_request_dispatched(self, selection: Optional[SelectedAuthEntry]) -> None:
+        if selection is None:
+            return
+        self._repository.increment_request_usage(
+            selection.auth_group_name,
+            selection.entry_id,
+            now_local_datetime(),
+        )
+
+    def finish(
+        self,
+        selection: Optional[SelectedAuthEntry],
+        *,
+        status_code: Optional[int] = None,
+        error_type: Optional[HookErrorType] = None,
+        error_message: Optional[str] = None,
+        response_headers: Optional[Mapping[str, Any]] = None,
+        usage: Optional[Mapping[str, Any]] = None,
+    ) -> None:
+        if selection is None:
+            return
+
+        key = (selection.auth_group_name, selection.entry_id)
+        current_inflight = self._inflight_counts.get(key, 0)
+        if current_inflight <= 1:
+            self._inflight_counts.pop(key, None)
+        else:
+            self._inflight_counts[key] = current_inflight - 1
+
+        group = self._groups_by_name.get(selection.auth_group_name)
+        if group is None:
+            return
+        entry = self._find_entry(group, selection.entry_id)
+        if entry is None:
+            return
+
+        runtime_state = self._repository.get_entry_runtime_state(group.name, entry.id)
+        disabled = bool(runtime_state.get("disabled"))
+        disabled_reason = runtime_state.get("disabled_reason")
+        cooldown_until = runtime_state.get("cooldown_until")
+        last_status_code: Optional[int] = runtime_state.get("last_status_code")
+        last_error_type: Optional[str] = runtime_state.get("last_error_type")
+        last_error_message: Optional[str] = runtime_state.get("last_error_message")
+
+        if status_code is not None and 200 <= int(status_code) < 400:
+            disabled = False
+            disabled_reason = None
+            cooldown_until = None
+            last_status_code = None
+            last_error_type = None
+            last_error_message = None
+            if usage:
+                prompt_tokens = int(usage.get("prompt_tokens") or 0)
+                completion_tokens = int(usage.get("completion_tokens") or 0)
+                total_tokens = int(usage.get("total_tokens") or 0)
+                if total_tokens > 0 or prompt_tokens > 0 or completion_tokens > 0:
+                    self._repository.increment_token_usage(
+                        group.name,
+                        entry.id,
+                        now_local_datetime(),
+                        prompt_tokens=prompt_tokens,
+                        completion_tokens=completion_tokens,
+                        total_tokens=total_tokens,
+                    )
+        elif status_code == 429:
+            disabled = False
+            disabled_reason = None
+            cooldown_until = format_local_datetime(
+                now_local_datetime()
+                + timedelta(seconds=self._resolve_cooldown_seconds(group, entry, response_headers))
+            )
+            last_status_code = int(status_code)
+            last_error_type = None
+            last_error_message = error_message
+        elif status_code in {401, 403}:
+            disabled = True
+            disabled_reason = f"http_{status_code}"
+            cooldown_until = None
+            last_status_code = int(status_code)
+            last_error_type = None
+            last_error_message = error_message
+        elif status_code is not None and int(status_code) >= 400:
+            disabled = bool(runtime_state.get("disabled"))
+            disabled_reason = runtime_state.get("disabled_reason")
+            last_status_code = int(status_code)
+            last_error_type = None
+            last_error_message = error_message
+        elif error_type is not None:
+            last_status_code = None
+            last_error_type = error_type.value
+            last_error_message = error_message
+
+        self._repository.save_entry_runtime_state(
+            group.name,
+            entry.id,
+            disabled=disabled,
+            disabled_reason=disabled_reason,
+            cooldown_until=cooldown_until,
+            last_status_code=last_status_code,
+            last_error_type=last_error_type,
+            last_error_message=last_error_message,
+        )
+
+    def list_auth_group_summaries(self) -> list[Dict[str, Any]]:
+        return [self._build_group_summary(group) for group in self.list_explicit_auth_groups()]
+
+    def get_auth_group_runtime(self, auth_group_name: str) -> Dict[str, Any]:
+        group = self._groups_by_name.get(str(auth_group_name).strip())
+        if group is None or group.name not in self._explicit_group_names:
+            raise ValueError(f"Auth group not found: {auth_group_name}")
+
+        now = now_local_datetime()
+        runtime_states = self._repository.list_group_runtime_states(group.name)
+        usage_by_entry = self._repository.list_current_usage(group.name, (entry.id for entry in group.entries), now)
+        entries = [
+            self._build_entry_runtime_view(
+                group,
+                entry,
+                runtime_states.get(entry.id) or self._repository.get_entry_runtime_state(group.name, entry.id),
+                usage_by_entry.get(entry.id) or {},
+                now,
+            )
+            for entry in group.entries
+        ]
+        return {
+            "name": group.name,
+            "strategy": group.strategy,
+            "cooldown_seconds_on_429": group.cooldown_seconds_on_429,
+            "summary": self._summarize_entries(entries),
+            "entries": entries,
+        }
+
+    def restore_entry(self, auth_group_name: str, entry_id: str) -> None:
+        group = self._groups_by_name.get(str(auth_group_name).strip())
+        if group is None or group.name not in self._explicit_group_names:
+            raise ValueError(f"Auth group not found: {auth_group_name}")
+        if self._find_entry(group, entry_id) is None:
+            raise ValueError(f"Auth entry not found: {entry_id}")
+        self._repository.restore_entry(group.name, entry_id)
+
+    def clear_entry_cooldown(self, auth_group_name: str, entry_id: str) -> None:
+        group, entry = self._resolve_explicit_entry(auth_group_name, entry_id)
+        runtime_state = self._repository.get_entry_runtime_state(group.name, entry.id)
+        self._repository.save_entry_runtime_state(
+            group.name,
+            entry.id,
+            disabled=bool(runtime_state.get("disabled")),
+            disabled_reason=runtime_state.get("disabled_reason"),
+            cooldown_until=None,
+            last_status_code=runtime_state.get("last_status_code"),
+            last_error_type=runtime_state.get("last_error_type"),
+            last_error_message=runtime_state.get("last_error_message"),
+        )
+
+    def set_entry_disabled(self, auth_group_name: str, entry_id: str, *, disabled: bool) -> None:
+        group, entry = self._resolve_explicit_entry(auth_group_name, entry_id)
+        runtime_state = self._repository.get_entry_runtime_state(group.name, entry.id)
+        self._repository.save_entry_runtime_state(
+            group.name,
+            entry.id,
+            disabled=bool(disabled),
+            disabled_reason="manual_disabled" if disabled else None,
+            cooldown_until=runtime_state.get("cooldown_until"),
+            last_status_code=runtime_state.get("last_status_code"),
+            last_error_type=runtime_state.get("last_error_type"),
+            last_error_message=runtime_state.get("last_error_message"),
+        )
+
+    def reset_entry_minute_usage(self, auth_group_name: str, entry_id: str) -> None:
+        group, entry = self._resolve_explicit_entry(auth_group_name, entry_id)
+        self._repository.reset_current_minute_usage(group.name, entry.id, now_local_datetime())
+
+    def reset_entry_runtime(self, auth_group_name: str, entry_id: str) -> None:
+        group, entry = self._resolve_explicit_entry(auth_group_name, entry_id)
+        now = now_local_datetime()
+        self._repository.restore_entry(group.name, entry.id)
+        self._repository.reset_current_minute_usage(group.name, entry.id, now)
+        self._repository.reset_current_day_usage(group.name, entry.id, now)
+
+    def list_explicit_auth_groups(self) -> tuple[AuthGroupSchema, ...]:
+        return tuple(
+            self._groups_by_name[name]
+            for name in sorted(self._explicit_group_names)
+            if name in self._groups_by_name
+        )
+
+    def _build_group_summary(self, group: AuthGroupSchema) -> Dict[str, Any]:
+        runtime = self.get_auth_group_runtime(group.name)
+        return {
+            "name": group.name,
+            "strategy": group.strategy,
+            "cooldown_seconds_on_429": group.cooldown_seconds_on_429,
+            "entries": [entry.to_mapping() for entry in group.entries],
+            "summary": runtime["summary"],
+        }
+
+    def _build_entry_runtime_view(
+        self,
+        group: AuthGroupSchema,
+        entry: AuthEntrySchema,
+        runtime_state: Mapping[str, Any],
+        usage: Mapping[str, Any],
+        now: datetime,
+    ) -> Dict[str, Any]:
+        inflight = self._inflight_counts.get((group.name, entry.id), 0)
+        cooldown_until_text = runtime_state.get("cooldown_until")
+        cooldown_until = parse_local_datetime(cooldown_until_text)
+        is_cooling_down = cooldown_until is not None and cooldown_until > now
+        is_disabled = (not entry.enabled) or bool(runtime_state.get("disabled"))
+        is_request_quota_exceeded = self._is_request_quota_exceeded(entry, usage)
+        is_token_quota_exceeded = self._is_token_quota_exceeded(entry, usage)
+        is_saturated = entry.max_concurrency is not None and inflight >= entry.max_concurrency
+
+        if is_disabled:
+            status = "disabled"
+        elif is_cooling_down:
+            status = "cooldown"
+        elif is_request_quota_exceeded or is_token_quota_exceeded:
+            status = "quota_exceeded"
+        elif is_saturated:
+            status = "saturated"
+        else:
+            status = "available"
+
+        return {
+            "id": entry.id,
+            "enabled": bool(entry.enabled),
+            "headers_count": len(entry.headers),
+            "max_concurrency": entry.max_concurrency,
+            "cooldown_seconds_on_429": entry.cooldown_seconds_on_429,
+            "request_quota_per_minute": entry.request_quota_per_minute,
+            "request_quota_per_day": entry.request_quota_per_day,
+            "token_quota_per_minute": entry.token_quota_per_minute,
+            "token_quota_per_day": entry.token_quota_per_day,
+            "inflight": inflight,
+            "cooldown_until": cooldown_until_text,
+            "disabled": is_disabled,
+            "disabled_reason": runtime_state.get("disabled_reason"),
+            "last_status_code": runtime_state.get("last_status_code"),
+            "last_error_type": runtime_state.get("last_error_type"),
+            "last_error_message": runtime_state.get("last_error_message"),
+            "updated_at": runtime_state.get("updated_at"),
+            "minute_request_count": int(usage.get("minute_request_count") or 0),
+            "day_request_count": int(usage.get("day_request_count") or 0),
+            "minute_prompt_tokens": int(usage.get("minute_prompt_tokens") or 0),
+            "day_prompt_tokens": int(usage.get("day_prompt_tokens") or 0),
+            "minute_completion_tokens": int(usage.get("minute_completion_tokens") or 0),
+            "day_completion_tokens": int(usage.get("day_completion_tokens") or 0),
+            "minute_total_tokens": int(usage.get("minute_total_tokens") or 0),
+            "day_total_tokens": int(usage.get("day_total_tokens") or 0),
+            "status": status,
+        }
+
+    @staticmethod
+    def _summarize_entries(entries: Sequence[Mapping[str, Any]]) -> Dict[str, int]:
+        summary = {
+            "entry_count": len(entries),
+            "available_count": 0,
+            "cooldown_count": 0,
+            "disabled_count": 0,
+        }
+        for entry in entries:
+            status = str(entry.get("status") or "")
+            if status == "available":
+                summary["available_count"] += 1
+            elif status == "cooldown":
+                summary["cooldown_count"] += 1
+            elif status == "disabled":
+                summary["disabled_count"] += 1
+        return summary
+
+    def _is_entry_available(
+        self,
+        group: AuthGroupSchema,
+        entry: AuthEntrySchema,
+        runtime_state: Mapping[str, Any],
+        usage: Mapping[str, Any],
+        now: datetime,
+    ) -> bool:
+        if not entry.enabled:
+            return False
+        if bool(runtime_state.get("disabled")):
+            return False
+        cooldown_until = parse_local_datetime(runtime_state.get("cooldown_until"))
+        if cooldown_until is not None and cooldown_until > now:
+            return False
+        inflight = self._inflight_counts.get((group.name, entry.id), 0)
+        if entry.max_concurrency is not None and inflight >= entry.max_concurrency:
+            return False
+        if self._is_request_quota_exceeded(entry, usage):
+            return False
+        if self._is_token_quota_exceeded(entry, usage):
+            return False
+        return True
+
+    @staticmethod
+    def _is_request_quota_exceeded(entry: AuthEntrySchema, usage: Mapping[str, Any]) -> bool:
+        minute_requests = int(usage.get("minute_request_count") or 0)
+        day_requests = int(usage.get("day_request_count") or 0)
+        if entry.request_quota_per_minute is not None and minute_requests >= entry.request_quota_per_minute:
+            return True
+        if entry.request_quota_per_day is not None and day_requests >= entry.request_quota_per_day:
+            return True
+        return False
+
+    @staticmethod
+    def _is_token_quota_exceeded(entry: AuthEntrySchema, usage: Mapping[str, Any]) -> bool:
+        minute_tokens = int(usage.get("minute_total_tokens") or 0)
+        day_tokens = int(usage.get("day_total_tokens") or 0)
+        if entry.token_quota_per_minute is not None and minute_tokens >= entry.token_quota_per_minute:
+            return True
+        if entry.token_quota_per_day is not None and day_tokens >= entry.token_quota_per_day:
+            return True
+        return False
+
+    def _resolve_cooldown_seconds(
+        self,
+        group: AuthGroupSchema,
+        entry: AuthEntrySchema,
+        response_headers: Optional[Mapping[str, Any]],
+    ) -> int:
+        retry_after_seconds = self._parse_retry_after_seconds(response_headers)
+        if retry_after_seconds is not None and retry_after_seconds > 0:
+            return retry_after_seconds
+        if entry.cooldown_seconds_on_429 is not None:
+            return entry.cooldown_seconds_on_429
+        return group.cooldown_seconds_on_429
+
+    @staticmethod
+    def _parse_retry_after_seconds(response_headers: Optional[Mapping[str, Any]]) -> Optional[int]:
+        if not response_headers:
+            return None
+        retry_after_value: Optional[Any] = None
+        for key, value in response_headers.items():
+            if str(key).lower() == "retry-after":
+                retry_after_value = value
+                break
+        if retry_after_value in (None, ""):
+            return None
+
+        try:
+            parsed_seconds = int(str(retry_after_value).strip())
+            return parsed_seconds if parsed_seconds > 0 else None
+        except (TypeError, ValueError):
+            pass
+
+        try:
+            retry_after_datetime = parsedate_to_datetime(str(retry_after_value).strip())
+        except (TypeError, ValueError, IndexError):
+            return None
+        if retry_after_datetime.tzinfo is None:
+            retry_after_datetime = retry_after_datetime.replace(tzinfo=timezone.utc)
+        delta_seconds = int((retry_after_datetime - datetime.now(timezone.utc)).total_seconds())
+        return delta_seconds if delta_seconds > 0 else None
+
+    def _rotation_distance(self, group_name: str, index: int, size: int) -> int:
+        cursor = self._rotation_cursor_by_group.get(group_name, 0)
+        return (index - cursor) % max(size, 1)
+
+    @staticmethod
+    def _find_entry(group: AuthGroupSchema, entry_id: str) -> Optional[AuthEntrySchema]:
+        normalized_entry_id = str(entry_id).strip()
+        for entry in group.entries:
+            if entry.id == normalized_entry_id:
+                return entry
+        return None
+
+    def _resolve_explicit_entry(self, auth_group_name: str, entry_id: str) -> tuple[AuthGroupSchema, AuthEntrySchema]:
+        group = self._groups_by_name.get(str(auth_group_name).strip())
+        if group is None or group.name not in self._explicit_group_names:
+            raise ValueError(f"Auth group not found: {auth_group_name}")
+        entry = self._find_entry(group, entry_id)
+        if entry is None:
+            raise ValueError(f"Auth entry not found: {entry_id}")
+        return group, entry

src/config/provider_config.py

@@ -1,11 +1,11 @@
 #!/usr/bin/env python3
 # -*- coding: utf-8 -*-
-"""Provider config schema, factory helpers, and validation."""
+"""Provider and auth group config schema, factory helpers, and validation."""
 
 from __future__ import annotations
 
 from dataclasses import dataclass
-from typing import Any, Dict, List, Mapping, Optional, Sequence
+from typing import Any, Dict, Iterable, List, Mapping, Optional, Sequence
 from urllib.parse import urlparse
 
 from ..utils.net import normalize_proxy_url
@@ -16,6 +16,8 @@ DEFAULT_PROVIDER_VERIFY_SSL = False
 DEFAULT_PROVIDER_TRANSPORT = "http"
 DEFAULT_PROVIDER_SOURCE_FORMAT = "openai_chat"
 DEFAULT_PROVIDER_TARGET_FORMAT = "openai_chat"
+DEFAULT_AUTH_GROUP_STRATEGY = "least_inflight"
+DEFAULT_AUTH_GROUP_COOLDOWN_SECONDS_ON_429 = 60
 SUPPORTED_PROVIDER_TRANSPORTS = {"http", "websocket"}
 SUPPORTED_PROVIDER_PROTOCOLS = {
     "openai_chat",
@@ -24,6 +26,7 @@ SUPPORTED_PROVIDER_PROTOCOLS = {
     "codex",
 }
 SUPPORTED_PROVIDER_API_SCHEMES = {"http", "https", "ws", "wss"}
+SUPPORTED_AUTH_GROUP_STRATEGIES = {DEFAULT_AUTH_GROUP_STRATEGY}
 SUPPORTED_PROVIDER_FIELDS = {
     "name",
     "api",
@@ -31,6 +34,7 @@ SUPPORTED_PROVIDER_FIELDS = {
     "source_format",
     "target_format",
     "api_key",
+    "auth_group",
     "proxy",
     "timeout_seconds",
     "max_retries",
@@ -38,6 +42,23 @@ SUPPORTED_PROVIDER_FIELDS = {
     "model_list",
     "hook",
 }
+SUPPORTED_AUTH_GROUP_FIELDS = {
+    "name",
+    "strategy",
+    "cooldown_seconds_on_429",
+    "entries",
+}
+SUPPORTED_AUTH_ENTRY_FIELDS = {
+    "id",
+    "enabled",
+    "headers",
+    "max_concurrency",
+    "cooldown_seconds_on_429",
+    "request_quota_per_minute",
+    "request_quota_per_day",
+    "token_quota_per_minute",
+    "token_quota_per_day",
+}
 
 
 def clean_optional_string(value: Any) -> Optional[str]:
@@ -117,6 +138,29 @@ def normalize_model_list(value: Any) -> List[str]:
     return models
 
 
+def normalize_headers(value: Any) -> Dict[str, str]:
+    if value is None:
+        return {}
+    if not isinstance(value, Mapping):
+        raise ValueError("headers must be an object")
+
+    headers: Dict[str, str] = {}
+    seen_names: set[str] = set()
+    for raw_key, raw_value in value.items():
+        header_name = clean_optional_string(raw_key)
+        header_value = clean_optional_string(raw_value)
+        if header_name is None:
+            raise ValueError("header name must not be empty")
+        if header_value is None:
+            raise ValueError(f"header value for {header_name} must not be empty")
+        normalized_name = header_name.lower()
+        if normalized_name in seen_names:
+            raise ValueError(f"duplicate header detected: {header_name}")
+        seen_names.add(normalized_name)
+        headers[header_name] = header_value
+    return headers
+
+
 def resolve_provider_transport(api: str, transport: Any = None) -> str:
     scheme = urlparse(api).scheme.lower()
     if scheme not in SUPPORTED_PROVIDER_API_SCHEMES:
@@ -151,12 +195,172 @@ def resolve_provider_protocol(
     return resolved
 
 
+def resolve_auth_group_strategy(value: Any) -> str:
+    resolved = (clean_optional_string(value) or DEFAULT_AUTH_GROUP_STRATEGY).lower()
+    if resolved not in SUPPORTED_AUTH_GROUP_STRATEGIES:
+        supported = ", ".join(sorted(SUPPORTED_AUTH_GROUP_STRATEGIES))
+        raise ValueError(f"Auth group strategy must be one of: {supported}")
+    return resolved
+
+
 def validate_provider_fields(config: Mapping[str, Any]) -> None:
     unknown_fields = sorted(str(key) for key in config.keys() if str(key) not in SUPPORTED_PROVIDER_FIELDS)
     if unknown_fields:
         raise ValueError(f"Unsupported provider field(s): {', '.join(unknown_fields)}")
 
 
+def validate_auth_group_fields(config: Mapping[str, Any]) -> None:
+    unknown_fields = sorted(str(key) for key in config.keys() if str(key) not in SUPPORTED_AUTH_GROUP_FIELDS)
+    if unknown_fields:
+        raise ValueError(f"Unsupported auth_group field(s): {', '.join(unknown_fields)}")
+
+
+def validate_auth_entry_fields(config: Mapping[str, Any]) -> None:
+    unknown_fields = sorted(str(key) for key in config.keys() if str(key) not in SUPPORTED_AUTH_ENTRY_FIELDS)
+    if unknown_fields:
+        raise ValueError(f"Unsupported auth_entry field(s): {', '.join(unknown_fields)}")
+
+
+@dataclass(frozen=True, slots=True)
+class AuthEntrySchema:
+    """Normalized auth entry configuration."""
+
+    id: str
+    enabled: bool = True
+    headers: tuple[tuple[str, str], ...] = ()
+    max_concurrency: Optional[int] = None
+    cooldown_seconds_on_429: Optional[int] = None
+    request_quota_per_minute: Optional[int] = None
+    request_quota_per_day: Optional[int] = None
+    token_quota_per_minute: Optional[int] = None
+    token_quota_per_day: Optional[int] = None
+
+    @classmethod
+    def from_mapping(cls, config: Mapping[str, Any]) -> "AuthEntrySchema":
+        if not isinstance(config, Mapping):
+            raise ValueError("Auth entry config must be an object")
+
+        validate_auth_entry_fields(config)
+        entry_id = clean_optional_string(config.get("id"))
+        if entry_id is None:
+            raise ValueError("Auth entry id is required")
+
+        headers = normalize_headers(config.get("headers"))
+        if not headers:
+            raise ValueError(f"Auth entry '{entry_id}' must define at least one header")
+
+        return cls(
+            id=entry_id,
+            enabled=parse_optional_bool(
+                config.get("enabled"),
+                default=True,
+                error_message="Auth entry enabled must be a boolean value",
+            )
+            is not False,
+            headers=tuple(headers.items()),
+            max_concurrency=parse_optional_positive_int(
+                config.get("max_concurrency"),
+                error_message="Auth entry max_concurrency must be a positive integer",
+            ),
+            cooldown_seconds_on_429=parse_optional_positive_int(
+                config.get("cooldown_seconds_on_429"),
+                error_message="Auth entry cooldown_seconds_on_429 must be a positive integer",
+            ),
+            request_quota_per_minute=parse_optional_positive_int(
+                config.get("request_quota_per_minute"),
+                error_message="Auth entry request_quota_per_minute must be a positive integer",
+            ),
+            request_quota_per_day=parse_optional_positive_int(
+                config.get("request_quota_per_day"),
+                error_message="Auth entry request_quota_per_day must be a positive integer",
+            ),
+            token_quota_per_minute=parse_optional_positive_int(
+                config.get("token_quota_per_minute"),
+                error_message="Auth entry token_quota_per_minute must be a positive integer",
+            ),
+            token_quota_per_day=parse_optional_positive_int(
+                config.get("token_quota_per_day"),
+                error_message="Auth entry token_quota_per_day must be a positive integer",
+            ),
+        )
+
+    def to_mapping(self) -> Dict[str, Any]:
+        config: Dict[str, Any] = {
+            "id": self.id,
+            "enabled": bool(self.enabled),
+            "headers": dict(self.headers),
+        }
+        if self.max_concurrency is not None:
+            config["max_concurrency"] = self.max_concurrency
+        if self.cooldown_seconds_on_429 is not None:
+            config["cooldown_seconds_on_429"] = self.cooldown_seconds_on_429
+        if self.request_quota_per_minute is not None:
+            config["request_quota_per_minute"] = self.request_quota_per_minute
+        if self.request_quota_per_day is not None:
+            config["request_quota_per_day"] = self.request_quota_per_day
+        if self.token_quota_per_minute is not None:
+            config["token_quota_per_minute"] = self.token_quota_per_minute
+        if self.token_quota_per_day is not None:
+            config["token_quota_per_day"] = self.token_quota_per_day
+        return config
+
+    def headers_mapping(self) -> Dict[str, str]:
+        return dict(self.headers)
+
+
+@dataclass(frozen=True, slots=True)
+class AuthGroupSchema:
+    """Normalized auth group configuration."""
+
+    name: str
+    strategy: str = DEFAULT_AUTH_GROUP_STRATEGY
+    cooldown_seconds_on_429: int = DEFAULT_AUTH_GROUP_COOLDOWN_SECONDS_ON_429
+    entries: tuple[AuthEntrySchema, ...] = ()
+
+    @classmethod
+    def from_mapping(cls, config: Mapping[str, Any]) -> "AuthGroupSchema":
+        if not isinstance(config, Mapping):
+            raise ValueError("Auth group config must be an object")
+
+        validate_auth_group_fields(config)
+        name = clean_optional_string(config.get("name"))
+        if name is None:
+            raise ValueError("Auth group name is required")
+
+        raw_entries = config.get("entries")
+        if not isinstance(raw_entries, list) or not raw_entries:
+            raise ValueError(f"Auth group '{name}' must define a non-empty entries list")
+
+        seen_entry_ids: set[str] = set()
+        entries: List[AuthEntrySchema] = []
+        for index, raw_entry in enumerate(raw_entries):
+            entry = AuthEntrySchema.from_mapping(raw_entry)
+            if entry.id in seen_entry_ids:
+                raise ValueError(f"Duplicate auth entry id detected in group '{name}': {entry.id}")
+            seen_entry_ids.add(entry.id)
+            entries.append(entry)
+
+        return cls(
+            name=name,
+            strategy=resolve_auth_group_strategy(config.get("strategy")),
+            cooldown_seconds_on_429=parse_optional_positive_int(
+                config.get("cooldown_seconds_on_429"),
+                default=DEFAULT_AUTH_GROUP_COOLDOWN_SECONDS_ON_429,
+                error_message="Auth group cooldown_seconds_on_429 must be a positive integer",
+            )
+            or DEFAULT_AUTH_GROUP_COOLDOWN_SECONDS_ON_429,
+            entries=tuple(entries),
+        )
+
+    def to_mapping(self) -> Dict[str, Any]:
+        return {
+            "name": self.name,
+            "strategy": self.strategy,
+            "cooldown_seconds_on_429": self.cooldown_seconds_on_429,
+            "entries": [entry.to_mapping() for entry in self.entries],
+        }
+
+
 @dataclass(frozen=True, slots=True)
 class ProviderConfigSchema:
     """Normalized provider configuration schema."""
@@ -167,6 +371,7 @@ class ProviderConfigSchema:
     source_format: str = DEFAULT_PROVIDER_SOURCE_FORMAT
     target_format: str = DEFAULT_PROVIDER_TARGET_FORMAT
     api_key: Optional[str] = None
+    auth_group: Optional[str] = None
     proxy: Optional[str] = None
     timeout_seconds: Optional[int] = None
     max_retries: Optional[int] = None
@@ -191,10 +396,14 @@ class ProviderConfigSchema:
 
         name = clean_optional_string(config.get("name"))
         api = clean_optional_string(config.get("api"))
+        api_key = clean_optional_string(config.get("api_key"))
+        auth_group = clean_optional_string(config.get("auth_group"))
         if name is None:
             raise ValueError("Provider name is required")
         if api is None:
             raise ValueError("Provider api is required")
+        if api_key and auth_group:
+            raise ValueError("Provider must define either auth_group or api_key, not both")
 
         return cls(
             name=name,
@@ -210,7 +419,8 @@ class ProviderConfigSchema:
                 default=DEFAULT_PROVIDER_TARGET_FORMAT,
                 field_name="target_format",
             ),
-            api_key=clean_optional_string(config.get("api_key")),
+            api_key=api_key,
+            auth_group=auth_group,
             proxy=normalize_proxy_url(config.get("proxy")),
             timeout_seconds=parse_optional_positive_int(config.get("timeout_seconds")),
             max_retries=parse_optional_positive_int(config.get("max_retries")),
@@ -230,6 +440,8 @@ class ProviderConfigSchema:
 
         if self.api_key is not None:
             config["api_key"] = self.api_key
+        if self.auth_group is not None:
+            config["auth_group"] = self.auth_group
         if self.proxy is not None:
             config["proxy"] = self.proxy
         if self.timeout_seconds is not None:
@@ -256,6 +468,7 @@ class RuntimeProviderSpec:
     source_format: str
     target_format: str
     api_key: str
+    auth_group: Optional[str]
     model_list: tuple[str, ...]
     proxy: Optional[str]
     timeout_seconds: int
@@ -276,6 +489,7 @@ class RuntimeProviderSpec:
             source_format=config.source_format,
             target_format=config.target_format,
             api_key=config.api_key or "",
+            auth_group=config.auth_group,
             model_list=config.model_list,
             proxy=config.proxy,
             timeout_seconds=config.timeout_seconds or DEFAULT_PROVIDER_TIMEOUT_SECONDS,
@@ -298,6 +512,8 @@ class ProviderRuntimeView:
     transport: str
     source_format: str
     target_format: str
+    auth_group: Optional[str]
+    legacy_api_key: bool
     model_list: tuple[str, ...]
     proxy: Optional[str]
     timeout_seconds: int
@@ -306,13 +522,15 @@ class ProviderRuntimeView:
     hook: Optional[str]
 
     @classmethod
-    def from_spec(cls, spec: RuntimeProviderSpec) -> "ProviderRuntimeView":
+    def from_spec(cls, spec: RuntimeProviderSpec, *, legacy_api_key: bool = False) -> "ProviderRuntimeView":
         return cls(
             name=spec.name,
             api=spec.api,
             transport=spec.transport,
             source_format=spec.source_format,
             target_format=spec.target_format,
+            auth_group=spec.auth_group,
+            legacy_api_key=legacy_api_key,
             model_list=spec.model_list,
             proxy=spec.proxy,
             timeout_seconds=spec.timeout_seconds,
@@ -322,9 +540,31 @@ class ProviderRuntimeView:
         )
 
 
+def build_auth_group_schemas(
+    auth_groups: Sequence[Mapping[str, Any]],
+) -> tuple[AuthGroupSchema, ...]:
+    seen_names: set[str] = set()
+    schemas: List[AuthGroupSchema] = []
+
+    for index, auth_group in enumerate(auth_groups):
+        if not isinstance(auth_group, Mapping):
+            raise ValueError(f"Auth group entry at index {index} must be an object")
+
+        schema = AuthGroupSchema.from_mapping(auth_group)
+        if schema.name in seen_names:
+            raise ValueError(f"Duplicate auth group name detected: {schema.name}")
+        seen_names.add(schema.name)
+        schemas.append(schema)
+
+    return tuple(schemas)
+
+
 def build_provider_schemas(
     providers: Sequence[Mapping[str, Any]],
+    *,
+    available_auth_group_names: Optional[Iterable[str]] = None,
 ) -> tuple[ProviderConfigSchema, ...]:
+    auth_group_names = set(available_auth_group_names or ())
     seen_names: set[str] = set()
     seen_model_keys: set[str] = set()
     schemas: List[ProviderConfigSchema] = []
@@ -336,6 +576,12 @@ def build_provider_schemas(
         schema = ProviderConfigSchema.from_mapping(provider)
         if schema.name in seen_names:
             raise ValueError(f"Duplicate provider name detected: {schema.name}")
+        if (
+            schema.auth_group
+            and available_auth_group_names is not None
+            and schema.auth_group not in auth_group_names
+        ):
+            raise ValueError(f"Provider references unknown auth_group: {schema.auth_group}")
         seen_names.add(schema.name)
 
         for model in schema.model_list:
@@ -349,5 +595,20 @@ def build_provider_schemas(
     return tuple(schemas)
 
 
+def validate_auth_group_definitions(auth_groups: Sequence[Mapping[str, Any]]) -> None:
+    build_auth_group_schemas(auth_groups)
+
+
 def validate_provider_definitions(providers: Sequence[Mapping[str, Any]]) -> None:
     build_provider_schemas(providers)
+
+
+def validate_auth_group_provider_definitions(
+    auth_groups: Sequence[Mapping[str, Any]],
+    providers: Sequence[Mapping[str, Any]],
+) -> None:
+    auth_group_schemas = build_auth_group_schemas(auth_groups)
+    build_provider_schemas(
+        providers,
+        available_auth_group_names={schema.name for schema in auth_group_schemas},
+    )

src/config/provider_manager.py

@@ -6,21 +6,24 @@ import hashlib
 import importlib.util
 import inspect
 import sys
+from dataclasses import replace
 from pathlib import Path
 from typing import Dict, Optional, Sequence
 
 from ..application.app_context import AppContext, Logger
 from ..external import LLMProvider
 from ..hooks import HookModule
+from .auth_group_manager import AuthGroupManager
 from .provider_config import ProviderConfigSchema, ProviderRuntimeView, RuntimeProviderSpec
 
 
 class ProviderManager:
     """管理模型到 provider 的运行时映射关系。"""
 
-    def __init__(self, ctx: AppContext):
+    def __init__(self, ctx: AppContext, auth_group_manager: AuthGroupManager):
         self._base_dir = ctx.root_path.resolve()
         self._logger: Logger = ctx.logger
+        self._auth_group_manager = auth_group_manager
         self._provider_by_model: Dict[str, LLMProvider] = {}
         self._provider_by_name: Dict[str, LLMProvider] = {}
         self._provider_views_by_name: Dict[str, ProviderRuntimeView] = {}
@@ -69,31 +72,50 @@ class ProviderManager:
             self._logger.warning("Provider '%s' skipped: model_list is empty", spec.name)
             return
 
+        legacy_auth_group_name: Optional[str] = None
+        resolved_auth_group = spec.auth_group
+        if resolved_auth_group is None and spec.api_key:
+            legacy_auth_group_name = self._auth_group_manager.register_legacy_provider_group(
+                spec.name,
+                spec.api_key,
+            )
+            resolved_auth_group = legacy_auth_group_name
+        runtime_spec = replace(spec, auth_group=resolved_auth_group)
+
         provider = LLMProvider(
-            name=spec.name,
-            api=spec.api,
-            transport=spec.transport,
-            source_format=spec.source_format,
-            target_format=spec.target_format,
+            name=runtime_spec.name,
+            api=runtime_spec.api,
+            transport=runtime_spec.transport,
+            source_format=runtime_spec.source_format,
+            target_format=runtime_spec.target_format,
             api_key=spec.api_key,
-            model_list=spec.model_list,
-            proxy=spec.proxy,
-            timeout_seconds=spec.timeout_seconds,
-            max_retries=spec.max_retries,
-            verify_ssl=spec.verify_ssl,
-            hook=self._load_hook(spec.hook),
+            auth_group=runtime_spec.auth_group,
+            model_list=runtime_spec.model_list,
+            proxy=runtime_spec.proxy,
+            timeout_seconds=runtime_spec.timeout_seconds,
+            max_retries=runtime_spec.max_retries,
+            verify_ssl=runtime_spec.verify_ssl,
+            hook=self._load_hook(runtime_spec.hook),
         )
 
-        self._provider_by_name[spec.name] = provider
-        self._provider_views_by_name[spec.name] = ProviderRuntimeView.from_spec(spec)
+        self._provider_by_name[runtime_spec.name] = provider
+        self._provider_views_by_name[runtime_spec.name] = ProviderRuntimeView.from_spec(
+            runtime_spec,
+            legacy_api_key=legacy_auth_group_name is not None,
+        )
 
-        for model in spec.model_list:
-            model_key = f"{spec.name}/{model}"
+        for model in runtime_spec.model_list:
+            model_key = f"{runtime_spec.name}/{model}"
             if model_key in self._provider_by_model:
                 raise ValueError(f"Duplicate provider model mapping detected: {model_key}")
             self._provider_by_model[model_key] = provider
 
-        self._logger.info("Loaded provider: %s, models: %s", spec.name, list(spec.model_list))
+        self._logger.info(
+            "Loaded provider: %s, models: %s, auth_group=%s",
+            runtime_spec.name,
+            list(runtime_spec.model_list),
+            runtime_spec.auth_group or "<none>",
+        )
 
     def _load_hook(self, hook_path: Optional[str]) -> Optional[HookModule]:
         """按路径加载 hook 模块，并做缓存复用。"""

src/external/llm_provider.py

@@ -18,6 +18,7 @@ class LLMProvider:
     source_format: str = "openai_chat"
     target_format: str = "openai_chat"
     api_key: Optional[str] = None
+    auth_group: Optional[str] = None
     model_list: tuple[str, ...] = ()
     proxy: Optional[str] = None
     timeout_seconds: int = 1200

src/hooks/contracts.py

@@ -43,6 +43,8 @@ class HookContext:
     provider_target_format: str = "openai_chat"
     transport: str = "http"
     stream: bool = False
+    auth_group_name: Optional[str] = None
+    auth_entry_id: Optional[str] = None
     last_status_code: Optional[int] = None
     last_error_type: Optional[HookErrorType] = None
 

src/presentation/provider_controller.py

@@ -6,6 +6,7 @@ from flask import Response, jsonify, request
 
 from ..application.app_context import AppContext
 from ..services import (
+    AuthGroupService,
     AuthenticationService,
     ModelDiscoveryService,
     ProviderService,
@@ -21,6 +22,7 @@ class ProviderController:
         self,
         ctx: AppContext,
         provider_service: ProviderService,
+        auth_group_service: AuthGroupService,
         model_discovery_service: ModelDiscoveryService,
         settings_service: SettingsService,
         auth_service: AuthenticationService,
@@ -28,6 +30,7 @@ class ProviderController:
         self._app = ctx.flask_app
         self._logger = ctx.logger
         self._provider_service = provider_service
+        self._auth_group_service = auth_group_service
         self._model_discovery_service = model_discovery_service
         self._settings_service = settings_service
         self._auth_service = auth_service
@@ -43,6 +46,31 @@ class ProviderController:
         self._app.route('/api/providers/<string:name>', methods=['GET'])(auth(self.get_provider))
         self._app.route('/api/providers/<string:name>', methods=['PUT'])(auth(self.update_provider))
         self._app.route('/api/providers/<string:name>', methods=['DELETE'])(auth(self.delete_provider))
+        self._app.route('/api/auth-groups', methods=['GET'])(auth(self.get_auth_groups))
+        self._app.route('/api/auth-groups', methods=['POST'])(auth(self.create_auth_group))
+        self._app.route('/api/auth-groups/import-entries', methods=['POST'])(auth(self.import_auth_group_entries))
+        self._app.route('/api/auth-groups/<string:name>', methods=['GET'])(auth(self.get_auth_group))
+        self._app.route('/api/auth-groups/<string:name>', methods=['PUT'])(auth(self.update_auth_group))
+        self._app.route('/api/auth-groups/<string:name>', methods=['DELETE'])(auth(self.delete_auth_group))
+        self._app.route('/api/auth-groups/<string:name>/runtime', methods=['GET'])(auth(self.get_auth_group_runtime))
+        self._app.route('/api/auth-groups/<string:name>/entries/<string:entry_id>/clear-cooldown', methods=['POST'])(
+            auth(self.clear_auth_group_entry_cooldown)
+        )
+        self._app.route('/api/auth-groups/<string:name>/entries/<string:entry_id>/disable', methods=['POST'])(
+            auth(self.disable_auth_group_entry)
+        )
+        self._app.route('/api/auth-groups/<string:name>/entries/<string:entry_id>/enable', methods=['POST'])(
+            auth(self.enable_auth_group_entry)
+        )
+        self._app.route('/api/auth-groups/<string:name>/entries/<string:entry_id>/reset-minute-usage', methods=['POST'])(
+            auth(self.reset_auth_group_entry_minute_usage)
+        )
+        self._app.route('/api/auth-groups/<string:name>/entries/<string:entry_id>/reset', methods=['POST'])(
+            auth(self.reset_auth_group_entry_runtime)
+        )
+        self._app.route('/api/auth-groups/<string:name>/entries/<string:entry_id>/restore', methods=['POST'])(
+            auth(self.restore_auth_group_entry)
+        )
 
     def get_providers(self) -> Response:
         try:
@@ -100,6 +128,160 @@ class ProviderController:
             self._logger.error('Error deleting provider: %s', exc)
             return jsonify({'error': str(exc)}), 500
 
+    def get_auth_groups(self) -> Response:
+        try:
+            return jsonify(self._auth_group_service.list_auth_groups())
+        except Exception as exc:
+            self._logger.error('Error getting auth groups: %s', exc)
+            return jsonify({'error': str(exc)}), 500
+
+    def get_auth_group(self, name: str) -> Response:
+        try:
+            auth_group = self._auth_group_service.get_auth_group(name)
+            if auth_group is None:
+                return jsonify({'error': 'Auth group not found'}), 404
+            return jsonify(auth_group)
+        except Exception as exc:
+            self._logger.error('Error getting auth group: %s', exc)
+            return jsonify({'error': str(exc)}), 500
+
+    def create_auth_group(self) -> Response:
+        try:
+            payload = request.get_json(silent=True) or {}
+            auth_group = self._auth_group_service.create_auth_group(payload)
+            self._logger.info('Auth group created: %s', auth_group.get('name'))
+            return jsonify(auth_group), 201
+        except ValueError as exc:
+            return jsonify({'error': str(exc)}), 400
+        except Exception as exc:
+            self._logger.error('Error creating auth group: %s', exc)
+            return jsonify({'error': str(exc)}), 500
+
+    def import_auth_group_entries(self) -> Response:
+        try:
+            payload = request.get_json(silent=True) or {}
+            entries = self._auth_group_service.import_auth_entries(payload.get('yaml', ''))
+            return jsonify({'entries': entries})
+        except ValueError as exc:
+            return jsonify({'error': str(exc)}), 400
+        except Exception as exc:
+            self._logger.error('Error importing auth group entries: %s', exc)
+            return jsonify({'error': str(exc)}), 500
+
+    def update_auth_group(self, name: str) -> Response:
+        try:
+            payload = request.get_json(silent=True) or {}
+            auth_group = self._auth_group_service.update_auth_group(name, payload)
+            self._logger.info('Auth group updated: %s -> %s', name, auth_group.get('name'))
+            return jsonify(auth_group)
+        except ValueError as exc:
+            message = str(exc)
+            status_code = 404 if 'not found' in message.lower() else 400
+            return jsonify({'error': message}), status_code
+        except Exception as exc:
+            self._logger.error('Error updating auth group: %s', exc)
+            return jsonify({'error': str(exc)}), 500
+
+    def delete_auth_group(self, name: str) -> Response:
+        try:
+            self._auth_group_service.delete_auth_group(name)
+            self._logger.info('Auth group deleted: %s', name)
+            return jsonify({'message': 'Auth group deleted successfully'})
+        except ValueError as exc:
+            message = str(exc)
+            status_code = 404 if 'not found' in message.lower() else 400
+            return jsonify({'error': message}), status_code
+        except Exception as exc:
+            self._logger.error('Error deleting auth group: %s', exc)
+            return jsonify({'error': str(exc)}), 500
+
+    def get_auth_group_runtime(self, name: str) -> Response:
+        try:
+            return jsonify(self._auth_group_service.get_auth_group_runtime(name))
+        except ValueError as exc:
+            return jsonify({'error': str(exc)}), 404
+        except Exception as exc:
+            self._logger.error('Error getting auth group runtime: %s', exc)
+            return jsonify({'error': str(exc)}), 500
+
+    def clear_auth_group_entry_cooldown(self, name: str, entry_id: str) -> Response:
+        try:
+            self._auth_group_service.clear_entry_cooldown(name, entry_id)
+            self._logger.info('Auth group entry cooldown cleared: %s/%s', name, entry_id)
+            return jsonify({'message': 'Auth entry cooldown cleared successfully'})
+        except ValueError as exc:
+            message = str(exc)
+            status_code = 404 if 'not found' in message.lower() else 400
+            return jsonify({'error': message}), status_code
+        except Exception as exc:
+            self._logger.error('Error clearing auth group entry cooldown: %s', exc)
+            return jsonify({'error': str(exc)}), 500
+
+    def disable_auth_group_entry(self, name: str, entry_id: str) -> Response:
+        try:
+            self._auth_group_service.set_entry_disabled(name, entry_id, disabled=True)
+            self._logger.info('Auth group entry disabled: %s/%s', name, entry_id)
+            return jsonify({'message': 'Auth entry disabled successfully'})
+        except ValueError as exc:
+            message = str(exc)
+            status_code = 404 if 'not found' in message.lower() else 400
+            return jsonify({'error': message}), status_code
+        except Exception as exc:
+            self._logger.error('Error disabling auth group entry: %s', exc)
+            return jsonify({'error': str(exc)}), 500
+
+    def enable_auth_group_entry(self, name: str, entry_id: str) -> Response:
+        try:
+            self._auth_group_service.set_entry_disabled(name, entry_id, disabled=False)
+            self._logger.info('Auth group entry enabled: %s/%s', name, entry_id)
+            return jsonify({'message': 'Auth entry enabled successfully'})
+        except ValueError as exc:
+            message = str(exc)
+            status_code = 404 if 'not found' in message.lower() else 400
+            return jsonify({'error': message}), status_code
+        except Exception as exc:
+            self._logger.error('Error enabling auth group entry: %s', exc)
+            return jsonify({'error': str(exc)}), 500
+
+    def reset_auth_group_entry_minute_usage(self, name: str, entry_id: str) -> Response:
+        try:
+            self._auth_group_service.reset_entry_minute_usage(name, entry_id)
+            self._logger.info('Auth group entry minute usage reset: %s/%s', name, entry_id)
+            return jsonify({'message': 'Auth entry minute usage reset successfully'})
+        except ValueError as exc:
+            message = str(exc)
+            status_code = 404 if 'not found' in message.lower() else 400
+            return jsonify({'error': message}), status_code
+        except Exception as exc:
+            self._logger.error('Error resetting auth group entry minute usage: %s', exc)
+            return jsonify({'error': str(exc)}), 500
+
+    def reset_auth_group_entry_runtime(self, name: str, entry_id: str) -> Response:
+        try:
+            self._auth_group_service.reset_entry_runtime(name, entry_id)
+            self._logger.info('Auth group entry runtime reset: %s/%s', name, entry_id)
+            return jsonify({'message': 'Auth entry runtime reset successfully'})
+        except ValueError as exc:
+            message = str(exc)
+            status_code = 404 if 'not found' in message.lower() else 400
+            return jsonify({'error': message}), status_code
+        except Exception as exc:
+            self._logger.error('Error resetting auth group entry runtime: %s', exc)
+            return jsonify({'error': str(exc)}), 500
+
+    def restore_auth_group_entry(self, name: str, entry_id: str) -> Response:
+        try:
+            self._auth_group_service.restore_entry(name, entry_id)
+            self._logger.info('Auth group entry restored: %s/%s', name, entry_id)
+            return jsonify({'message': 'Auth entry restored successfully'})
+        except ValueError as exc:
+            message = str(exc)
+            status_code = 404 if 'not found' in message.lower() else 400
+            return jsonify({'error': message}), status_code
+        except Exception as exc:
+            self._logger.error('Error restoring auth group entry: %s', exc)
+            return jsonify({'error': str(exc)}), 500
+
     def fetch_models(self) -> Response:
         try:
             result = self._model_discovery_service.fetch_models_preview(

src/presentation/static/css/providers.css

@@ -14,6 +14,7 @@
     align-items: center;
     justify-content: space-between;
     gap: 16px;
+    flex-wrap: wrap;
     padding: 16px 18px;
     border: 1px solid var(--line);
     border-radius: var(--radius-md);
@@ -21,6 +22,31 @@
     box-shadow: var(--shadow-card);
 }
 
+.providers-page .toolbar-meta {
+    display: flex;
+    align-items: center;
+    justify-content: space-between;
+    gap: 14px;
+    width: 100%;
+}
+
+.providers-page .btn-toolbar-secondary {
+    min-height: 40px;
+    padding: 0 16px;
+    border-color: rgba(58, 127, 237, 0.24);
+    background: linear-gradient(180deg, rgba(248, 251, 255, 0.98), rgba(238, 244, 252, 0.98));
+    color: var(--primary-active);
+    font-weight: 700;
+}
+
+.providers-page .btn-toolbar-secondary:hover,
+.providers-page .btn-toolbar-secondary:focus {
+    border-color: rgba(58, 127, 237, 0.46);
+    background: linear-gradient(180deg, #ffffff, rgba(232, 241, 255, 1));
+    color: var(--primary-active);
+    box-shadow: none;
+}
+
 .providers-page .toolbar-summary {
     display: flex;
     align-items: center;
@@ -28,6 +54,45 @@
     gap: 10px;
 }
 
+.providers-page .provider-tabs {
+    display: inline-flex;
+    align-items: center;
+    gap: 4px;
+    padding: 4px 5px;
+    border: 1px solid var(--line);
+    border-radius: var(--radius-sm);
+    background: #fff;
+    box-shadow: var(--shadow-card);
+}
+
+.providers-page .provider-tab-btn {
+    min-width: 104px;
+    min-height: 34px;
+    padding: 0 12px;
+    border: 1px solid transparent;
+    border-radius: 8px;
+    background: transparent;
+    color: var(--text-secondary);
+    font-size: 13px;
+    font-weight: 700;
+    cursor: pointer;
+    transition: background-color var(--transition), color var(--transition), border-color var(--transition), box-shadow var(--transition);
+}
+
+.providers-page .provider-tab-btn:hover,
+.providers-page .provider-tab-btn:focus-visible {
+    color: var(--nav-tab-hover-text);
+    background: var(--nav-tab-hover-bg);
+    border-color: var(--nav-tab-hover-border);
+    outline: none;
+}
+
+.providers-page .provider-tab-btn.is-active {
+    color: #fff;
+    background: var(--primary);
+    border-color: var(--primary);
+}
+
 .providers-page .toolbar-pill {
     display: inline-flex;
     align-items: center;
@@ -62,6 +127,56 @@
     background: transparent;
 }
 
+.providers-page .section-card {
+    padding: 18px;
+    border: 1px solid var(--line);
+    border-radius: var(--radius-md);
+    background: #fff;
+    box-shadow: var(--shadow-card);
+}
+
+.providers-page .section-card-spaced {
+    margin-bottom: 14px;
+}
+
+.providers-page .section-header {
+    display: flex;
+    align-items: flex-start;
+    justify-content: space-between;
+    gap: 12px;
+    margin-bottom: 12px;
+}
+
+.providers-page .section-header-actions {
+    display: flex;
+    align-items: center;
+    gap: 10px;
+    flex-wrap: wrap;
+}
+
+.providers-page .section-title-with-help {
+    display: inline-flex;
+    align-items: center;
+    gap: 8px;
+}
+
+.providers-page .section-title {
+    margin: 0;
+    font-size: 18px;
+    font-weight: 700;
+}
+
+.providers-page .section-subtitle,
+.providers-page .field-help-text {
+    margin: 6px 0 0;
+    color: var(--text-secondary);
+    font-size: 13px;
+}
+
+.providers-page .auth-group-summary {
+    margin-bottom: 14px;
+}
+
 .providers-page .providers-table-shell {
     overflow: auto;
     border: 1px solid var(--line);
@@ -70,6 +185,10 @@
     box-shadow: var(--shadow-card);
 }
 
+.providers-page .providers-tab-panel[hidden] {
+    display: none !important;
+}
+
 :root[data-theme="dark"] .providers-page .providers-table-shell {
     background: #161b22;
     border-color: #30363d;
@@ -114,6 +233,21 @@
     overflow-wrap: anywhere;
 }
 
+.providers-page .auth-group-runtime-shell .providers-table td,
+.providers-page .auth-group-runtime-shell .providers-table th {
+    vertical-align: middle;
+}
+
+.providers-page .auth-groups-table-shell .providers-table td,
+.providers-page .auth-groups-table-shell .providers-table th {
+    vertical-align: middle;
+}
+
+.providers-page .auth-groups-table-shell .provider-meta-line {
+    margin-top: 0;
+    align-items: center;
+}
+
 :root[data-theme="dark"] .providers-page .providers-table td {
     background: #161b22;
     border-bottom-color: rgba(139, 148, 158, 0.28);
@@ -144,6 +278,45 @@
     background: #161b22;
 }
 
+:root[data-theme="dark"] .providers-page .provider-tabs {
+    border-color: #313b47;
+    background: #0d1117;
+}
+
+:root[data-theme="dark"] .providers-page .provider-tab-btn {
+    color: #9da7b3;
+}
+
+:root[data-theme="dark"] .providers-page .provider-tab-btn:hover,
+:root[data-theme="dark"] .providers-page .provider-tab-btn:focus-visible {
+    color: var(--nav-tab-hover-text);
+    background: var(--nav-tab-hover-bg);
+}
+
+:root[data-theme="dark"] .providers-page .provider-tab-btn.is-active {
+    color: #fff;
+    background: var(--primary);
+    border-color: var(--primary);
+}
+
+:root[data-theme="dark"] .providers-page .btn-toolbar-secondary {
+    border-color: #315d8f;
+    background: linear-gradient(180deg, rgba(20, 29, 42, 0.98), rgba(16, 24, 35, 0.98));
+    color: #9bc7ff;
+}
+
+:root[data-theme="dark"] .providers-page .btn-toolbar-secondary:hover,
+:root[data-theme="dark"] .providers-page .btn-toolbar-secondary:focus {
+    border-color: #4a79b6;
+    background: linear-gradient(180deg, rgba(24, 35, 50, 1), rgba(19, 28, 41, 1));
+    color: #c7e0ff;
+}
+
+:root[data-theme="dark"] .providers-page .section-card {
+    border-color: #313b47;
+    background: #161b22;
+}
+
 .providers-page .providers-table tbody tr:nth-child(even) td {
     background: #fbfdff;
 }
@@ -177,6 +350,134 @@
     vertical-align: middle;
 }
 
+.providers-page .entry-editor-header {
+    display: flex;
+    align-items: center;
+    gap: 12px;
+    justify-content: space-between;
+    margin: 12px 0;
+}
+
+.providers-page .auth-entry-table {
+    border: 1px solid var(--line);
+    border-radius: var(--radius-md);
+    overflow: hidden;
+    background: rgba(255, 255, 255, 0.52);
+}
+
+.providers-page .auth-entry-table-header,
+.providers-page .auth-entry-summary {
+    display: grid;
+    grid-template-columns: 44px minmax(180px, 1.3fr) minmax(96px, 0.72fr) minmax(96px, 0.72fr) minmax(260px, 2fr) 88px;
+    gap: 12px;
+    align-items: center;
+}
+
+.providers-page .auth-entry-table-header {
+    padding: 12px 16px;
+    border-bottom: 1px solid var(--line);
+    background: rgba(17, 24, 39, 0.04);
+    color: var(--text-secondary);
+    font-size: 12px;
+    font-weight: 700;
+    letter-spacing: 0.02em;
+}
+
+.providers-page .auth-entry-table-toggle-column {
+    display: block;
+}
+
+.providers-page .auth-entry-list {
+    display: grid;
+    gap: 0;
+}
+
+.providers-page .auth-entry-card {
+    border-bottom: 1px solid var(--line);
+    background: rgba(255, 255, 255, 0.78);
+}
+
+.providers-page .auth-entry-list .auth-entry-card:last-child {
+    border-bottom: 0;
+}
+
+.providers-page .auth-entry-summary {
+    padding: 12px 16px;
+    cursor: pointer;
+    transition: background-color var(--transition);
+}
+
+.providers-page .auth-entry-summary:hover {
+    background: rgba(58, 127, 237, 0.04);
+}
+
+.providers-page .auth-entry-summary:focus-visible {
+    outline: 2px solid rgba(58, 127, 237, 0.35);
+    outline-offset: -2px;
+}
+
+.providers-page .auth-entry-summary-cell {
+    min-width: 0;
+    color: var(--text-primary);
+    font-size: 13px;
+    line-height: 1.45;
+}
+
+.providers-page .auth-entry-summary-toggle {
+    display: inline-flex;
+    align-items: center;
+    justify-content: center;
+}
+
+.providers-page .auth-entry-summary-id {
+    font-weight: 700;
+}
+
+.providers-page .auth-entry-summary-limits {
+    color: var(--text-secondary);
+}
+
+.providers-page .auth-entry-summary-actions {
+    display: inline-flex;
+    align-items: center;
+    justify-content: flex-start;
+    gap: 8px;
+    white-space: nowrap;
+}
+
+.providers-page .auth-entry-toggle-btn {
+    width: 32px;
+    min-width: 32px;
+    padding: 0;
+    font-size: 15px;
+    line-height: 1;
+}
+
+.providers-page .auth-entry-detail {
+    padding: 0 16px 16px;
+}
+
+.providers-page .auth-entry-card[data-expanded="true"] .auth-entry-summary {
+    border-bottom: 1px solid var(--line);
+    background: rgba(58, 127, 237, 0.05);
+}
+
+.providers-page .entry-header-editor {
+    margin-top: 8px;
+}
+
+.providers-page .auth-header-list {
+    display: grid;
+    gap: 8px;
+}
+
+.providers-page .auth-header-row {
+    display: grid;
+    grid-template-columns: 1fr 1fr auto;
+    gap: 8px;
+    align-items: center;
+}
+
 .providers-page .provider-api-text {
     display: -webkit-box;
     -webkit-line-clamp: 2;
@@ -285,6 +586,30 @@
     color: #f6deb1;
 }
 
+:root[data-theme="dark"] .providers-page .auth-entry-table {
+    border-color: #313b47;
+    background: rgba(15, 20, 27, 0.72);
+}
+
+:root[data-theme="dark"] .providers-page .auth-entry-table-header {
+    border-bottom-color: #313b47;
+    background: rgba(13, 17, 23, 0.78);
+}
+
+:root[data-theme="dark"] .providers-page .auth-entry-card {
+    border-bottom-color: #313b47;
+    background: rgba(15, 20, 27, 0.72);
+}
+
+:root[data-theme="dark"] .providers-page .auth-entry-summary:hover {
+    background: rgba(47, 129, 247, 0.08);
+}
+
+:root[data-theme="dark"] .providers-page .auth-entry-card[data-expanded="true"] .auth-entry-summary {
+    border-bottom-color: #313b47;
+    background: rgba(47, 129, 247, 0.08);
+}
+
 .providers-page .muted-text,
 .providers-page .loading,
 .providers-page .empty-state,
@@ -316,6 +641,18 @@
     overflow: visible;
 }
 
+.providers-page .runtime-action-buttons {
+    flex-wrap: nowrap;
+    white-space: nowrap;
+}
+
+.providers-page .auth-group-runtime-shell .runtime-actions-cell,
+.providers-page .auth-group-runtime-shell .providers-table th:last-child,
+.providers-page .auth-group-runtime-shell .providers-table td:last-child {
+    width: 1%;
+    white-space: nowrap;
+}
+
 .providers-page .switch {
     position: relative;
     display: inline-block;
@@ -682,10 +1019,20 @@
     gap: 0 16px;
 }
 
+.providers-page .auth-group-base-grid {
+    display: grid;
+    grid-template-columns: repeat(3, minmax(0, 1fr));
+    gap: 0 16px;
+}
+
 .providers-page .modal-dialog {
     max-width: min(1120px, calc(100vw - 32px));
 }
 
+.providers-page .auth-entry-yaml-modal-dialog {
+    max-width: min(1180px, calc(100vw - 32px));
+}
+
 .providers-page .modal-content {
     border: 1px solid var(--line) !important;
     border-radius: var(--radius-lg);
@@ -724,6 +1071,53 @@
     font-size: 14px;
 }
 
+.providers-page .entry-import-textarea {
+    min-height: 260px;
+    font-family: "JetBrains Mono", "Consolas", monospace;
+    font-size: 13px;
+    line-height: 1.5;
+}
+
+.providers-page .auth-entry-yaml-modal-dialog .entry-import-textarea {
+    height: min(58vh, 560px);
+    min-height: 420px;
+}
+
+.providers-page .field-header-description {
+    margin-top: 4px;
+    color: var(--text-secondary);
+    font-size: 13px;
+    line-height: 1.45;
+}
+
+.providers-page .runtime-error-detail-grid {
+    display: grid;
+    grid-template-columns: repeat(2, minmax(0, 1fr));
+    gap: 12px;
+}
+
+.providers-page .runtime-error-detail-item {
+    padding: 12px 14px;
+    border: 1px solid var(--line);
+    border-radius: var(--radius-md);
+    background: rgba(17, 24, 39, 0.02);
+}
+
+.providers-page .runtime-error-detail-label {
+    margin-bottom: 6px;
+    color: var(--text-secondary);
+    font-size: 12px;
+    font-weight: 700;
+}
+
+.providers-page .runtime-error-detail-value {
+    color: var(--text-primary);
+    font-size: 13px;
+    line-height: 1.5;
+    white-space: pre-wrap;
+    word-break: break-word;
+}
+
 .providers-page .btn-close {
     opacity: 0.72;
 }
@@ -1169,6 +1563,11 @@
     color: #bcc8d8;
 }
 
+:root[data-theme="dark"] .providers-page .runtime-error-detail-item {
+    border-color: #313b47;
+    background: rgba(13, 17, 23, 0.78);
+}
+
 :root[data-theme="dark"] .providers-page .field-meta-badge {
     border-color: #313b47;
     background: rgba(255, 255, 255, 0.04);
@@ -1314,12 +1713,27 @@
     }
 
     .providers-page .toolbar-card {
+        align-items: stretch;
+    }
+
+    .providers-page .toolbar-meta,
+    .providers-page .toolbar-summary,
+    .providers-page .provider-tabs {
+        width: 100%;
+    }
+
+    .providers-page .toolbar-meta,
+    .providers-page .section-header {
         flex-direction: column;
         align-items: stretch;
     }
 
-    .providers-page .toolbar-summary {
-        width: 100%;
+    .providers-page .provider-tabs {
+        justify-content: stretch;
+    }
+
+    .providers-page .provider-tab-btn {
+        flex: 1 1 0;
     }
 
     .providers-page .toolbar-pill {
@@ -1330,6 +1744,32 @@
         grid-template-columns: 1fr;
     }
 
+    .providers-page .auth-group-base-grid {
+        grid-template-columns: 1fr;
+    }
+
+    .providers-page .auth-entry-table-header {
+        display: none;
+    }
+
+    .providers-page .auth-entry-summary {
+        grid-template-columns: 44px 1fr;
+        gap: 10px;
+    }
+
+    .providers-page .auth-entry-summary-cell,
+    .providers-page .auth-entry-summary-actions {
+        grid-column: 2;
+    }
+
+    .providers-page .auth-entry-summary-actions {
+        justify-content: flex-start;
+    }
+
+    .providers-page .runtime-error-detail-grid {
+        grid-template-columns: 1fr;
+    }
+
     .providers-page .field-header {
         flex-direction: column;
         align-items: stretch;

src/presentation/templates/users.html

@@ -118,7 +118,7 @@
                 </div>
                 <div class="modal-footer border-top-0">
                     <button type="button" class="btn btn-secondary" data-bs-dismiss="modal">取消</button>
-                    <button type="button" class="btn btn-primary" onclick="saveUser()">保存修改</button>
+                    <button type="button" class="btn btn-primary" onclick="saveUser()">保存</button>
                 </div>
             </div>
         </div>

src/repositories/__init__.py

@@ -3,9 +3,11 @@
 """仓储层导出。"""
 
 from .user_repository import UserRepository
+from .auth_group_repository import AuthGroupRepository
 from .log_repository import LogRepository
 
 __all__ = [
     'UserRepository',
+    'AuthGroupRepository',
     'LogRepository',
 ]

src/repositories/auth_group_repository.py

@@ -0,0 +1,351 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+"""Auth group runtime state repository."""
+
+from __future__ import annotations
+
+from datetime import datetime
+from typing import Any, Dict, Iterable
+
+from ..utils.database import ConnectionFactory
+from ..utils.local_time import format_local_date, format_local_datetime, now_local_datetime_text
+
+
+class AuthGroupRepository:
+    """Persist auth entry runtime state and quota buckets."""
+
+    def __init__(self, get_connection: ConnectionFactory):
+        self._get_connection = get_connection
+        self._ensure_tables()
+
+    def _ensure_tables(self) -> None:
+        with self._get_connection() as conn:
+            cursor = conn.cursor()
+            cursor.execute(
+                """
+                CREATE TABLE IF NOT EXISTS auth_entry_runtime_state (
+                    auth_group_name TEXT NOT NULL,
+                    entry_id TEXT NOT NULL,
+                    disabled INTEGER NOT NULL DEFAULT 0,
+                    disabled_reason TEXT,
+                    cooldown_until TEXT,
+                    last_status_code INTEGER,
+                    last_error_type TEXT,
+                    last_error_message TEXT,
+                    updated_at TEXT NOT NULL,
+                    PRIMARY KEY (auth_group_name, entry_id)
+                )
+                """
+            )
+            cursor.execute(
+                """
+                CREATE TABLE IF NOT EXISTS auth_entry_usage_buckets (
+                    auth_group_name TEXT NOT NULL,
+                    entry_id TEXT NOT NULL,
+                    bucket_type TEXT NOT NULL,
+                    bucket_start TEXT NOT NULL,
+                    request_count INTEGER NOT NULL DEFAULT 0,
+                    prompt_tokens INTEGER NOT NULL DEFAULT 0,
+                    completion_tokens INTEGER NOT NULL DEFAULT 0,
+                    total_tokens INTEGER NOT NULL DEFAULT 0,
+                    updated_at TEXT NOT NULL,
+                    PRIMARY KEY (auth_group_name, entry_id, bucket_type, bucket_start)
+                )
+                """
+            )
+            cursor.execute(
+                "CREATE INDEX IF NOT EXISTS idx_auth_entry_runtime_group ON auth_entry_runtime_state(auth_group_name)"
+            )
+            cursor.execute(
+                "CREATE INDEX IF NOT EXISTS idx_auth_entry_usage_group ON auth_entry_usage_buckets(auth_group_name, bucket_type, bucket_start)"
+            )
+
+    @staticmethod
+    def _default_runtime_state(auth_group_name: str, entry_id: str) -> Dict[str, Any]:
+        return {
+            "auth_group_name": auth_group_name,
+            "entry_id": entry_id,
+            "disabled": False,
+            "disabled_reason": None,
+            "cooldown_until": None,
+            "last_status_code": None,
+            "last_error_type": None,
+            "last_error_message": None,
+            "updated_at": None,
+        }
+
+    def get_entry_runtime_state(self, auth_group_name: str, entry_id: str) -> Dict[str, Any]:
+        with self._get_connection() as conn:
+            cursor = conn.cursor()
+            cursor.execute(
+                """
+                SELECT auth_group_name, entry_id, disabled, disabled_reason, cooldown_until,
+                       last_status_code, last_error_type, last_error_message, updated_at
+                FROM auth_entry_runtime_state
+                WHERE auth_group_name = ? AND entry_id = ?
+                """,
+                (auth_group_name, entry_id),
+            )
+            row = cursor.fetchone()
+        if row is None:
+            return self._default_runtime_state(auth_group_name, entry_id)
+        return {
+            "auth_group_name": row["auth_group_name"],
+            "entry_id": row["entry_id"],
+            "disabled": bool(row["disabled"]),
+            "disabled_reason": row["disabled_reason"],
+            "cooldown_until": row["cooldown_until"],
+            "last_status_code": row["last_status_code"],
+            "last_error_type": row["last_error_type"],
+            "last_error_message": row["last_error_message"],
+            "updated_at": row["updated_at"],
+        }
+
+    def list_group_runtime_states(self, auth_group_name: str) -> Dict[str, Dict[str, Any]]:
+        with self._get_connection() as conn:
+            cursor = conn.cursor()
+            cursor.execute(
+                """
+                SELECT auth_group_name, entry_id, disabled, disabled_reason, cooldown_until,
+                       last_status_code, last_error_type, last_error_message, updated_at
+                FROM auth_entry_runtime_state
+                WHERE auth_group_name = ?
+                """,
+                (auth_group_name,),
+            )
+            rows = cursor.fetchall()
+
+        result: Dict[str, Dict[str, Any]] = {}
+        for row in rows:
+            result[row["entry_id"]] = {
+                "auth_group_name": row["auth_group_name"],
+                "entry_id": row["entry_id"],
+                "disabled": bool(row["disabled"]),
+                "disabled_reason": row["disabled_reason"],
+                "cooldown_until": row["cooldown_until"],
+                "last_status_code": row["last_status_code"],
+                "last_error_type": row["last_error_type"],
+                "last_error_message": row["last_error_message"],
+                "updated_at": row["updated_at"],
+            }
+        return result
+
+    def save_entry_runtime_state(
+        self,
+        auth_group_name: str,
+        entry_id: str,
+        *,
+        disabled: bool,
+        disabled_reason: str | None,
+        cooldown_until: str | None,
+        last_status_code: int | None,
+        last_error_type: str | None,
+        last_error_message: str | None,
+    ) -> None:
+        with self._get_connection() as conn:
+            cursor = conn.cursor()
+            cursor.execute(
+                """
+                INSERT INTO auth_entry_runtime_state (
+                    auth_group_name, entry_id, disabled, disabled_reason, cooldown_until,
+                    last_status_code, last_error_type, last_error_message, updated_at
+                )
+                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
+                ON CONFLICT(auth_group_name, entry_id)
+                DO UPDATE SET
+                    disabled = excluded.disabled,
+                    disabled_reason = excluded.disabled_reason,
+                    cooldown_until = excluded.cooldown_until,
+                    last_status_code = excluded.last_status_code,
+                    last_error_type = excluded.last_error_type,
+                    last_error_message = excluded.last_error_message,
+                    updated_at = excluded.updated_at
+                """,
+                (
+                    auth_group_name,
+                    entry_id,
+                    1 if disabled else 0,
+                    disabled_reason,
+                    cooldown_until,
+                    last_status_code,
+                    last_error_type,
+                    last_error_message,
+                    now_local_datetime_text(),
+                ),
+            )
+
+    def restore_entry(self, auth_group_name: str, entry_id: str) -> None:
+        self.save_entry_runtime_state(
+            auth_group_name,
+            entry_id,
+            disabled=False,
+            disabled_reason=None,
+            cooldown_until=None,
+            last_status_code=None,
+            last_error_type=None,
+            last_error_message=None,
+        )
+
+    def reset_current_minute_usage(self, auth_group_name: str, entry_id: str, when: datetime) -> None:
+        minute_bucket = self._minute_bucket_start(when)
+        with self._get_connection() as conn:
+            cursor = conn.cursor()
+            cursor.execute(
+                """
+                DELETE FROM auth_entry_usage_buckets
+                WHERE auth_group_name = ? AND entry_id = ?
+                  AND bucket_type = 'minute' AND bucket_start = ?
+                """,
+                (auth_group_name, entry_id, minute_bucket),
+            )
+
+    def reset_current_day_usage(self, auth_group_name: str, entry_id: str, when: datetime) -> None:
+        day_bucket = self._day_bucket_start(when)
+        with self._get_connection() as conn:
+            cursor = conn.cursor()
+            cursor.execute(
+                """
+                DELETE FROM auth_entry_usage_buckets
+                WHERE auth_group_name = ? AND entry_id = ?
+                  AND bucket_type = 'day' AND bucket_start = ?
+                """,
+                (auth_group_name, entry_id, day_bucket),
+            )
+
+    @staticmethod
+    def _minute_bucket_start(value: datetime) -> str:
+        return format_local_datetime(value.replace(second=0, microsecond=0))
+
+    @staticmethod
+    def _day_bucket_start(value: datetime) -> str:
+        return format_local_date(value)
+
+    def _increment_usage_bucket(
+        self,
+        auth_group_name: str,
+        entry_id: str,
+        *,
+        bucket_type: str,
+        bucket_start: str,
+        request_count: int = 0,
+        prompt_tokens: int = 0,
+        completion_tokens: int = 0,
+        total_tokens: int = 0,
+    ) -> None:
+        with self._get_connection() as conn:
+            cursor = conn.cursor()
+            cursor.execute(
+                """
+                INSERT INTO auth_entry_usage_buckets (
+                    auth_group_name, entry_id, bucket_type, bucket_start,
+                    request_count, prompt_tokens, completion_tokens, total_tokens, updated_at
+                )
+                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
+                ON CONFLICT(auth_group_name, entry_id, bucket_type, bucket_start)
+                DO UPDATE SET
+                    request_count = request_count + excluded.request_count,
+                    prompt_tokens = prompt_tokens + excluded.prompt_tokens,
+                    completion_tokens = completion_tokens + excluded.completion_tokens,
+                    total_tokens = total_tokens + excluded.total_tokens,
+                    updated_at = excluded.updated_at
+                """,
+                (
+                    auth_group_name,
+                    entry_id,
+                    bucket_type,
+                    bucket_start,
+                    int(request_count or 0),
+                    int(prompt_tokens or 0),
+                    int(completion_tokens or 0),
+                    int(total_tokens or 0),
+                    now_local_datetime_text(),
+                ),
+            )
+
+    def increment_request_usage(self, auth_group_name: str, entry_id: str, when: datetime) -> None:
+        minute_bucket = self._minute_bucket_start(when)
+        day_bucket = self._day_bucket_start(when)
+        self._increment_usage_bucket(
+            auth_group_name,
+            entry_id,
+            bucket_type="minute",
+            bucket_start=minute_bucket,
+            request_count=1,
+        )
+        self._increment_usage_bucket(
+            auth_group_name,
+            entry_id,
+            bucket_type="day",
+            bucket_start=day_bucket,
+            request_count=1,
+        )
+
+    def increment_token_usage(
+        self,
+        auth_group_name: str,
+        entry_id: str,
+        when: datetime,
+        *,
+        prompt_tokens: int,
+        completion_tokens: int,
+        total_tokens: int,
+    ) -> None:
+        minute_bucket = self._minute_bucket_start(when)
+        day_bucket = self._day_bucket_start(when)
+        for bucket_type, bucket_start in (("minute", minute_bucket), ("day", day_bucket)):
+            self._increment_usage_bucket(
+                auth_group_name,
+                entry_id,
+                bucket_type=bucket_type,
+                bucket_start=bucket_start,
+                prompt_tokens=prompt_tokens,
+                completion_tokens=completion_tokens,
+                total_tokens=total_tokens,
+            )
+
+    def get_current_usage(self, auth_group_name: str, entry_id: str, when: datetime) -> Dict[str, int]:
+        minute_bucket = self._minute_bucket_start(when)
+        day_bucket = self._day_bucket_start(when)
+
+        usage = {
+            "minute_request_count": 0,
+            "day_request_count": 0,
+            "minute_prompt_tokens": 0,
+            "day_prompt_tokens": 0,
+            "minute_completion_tokens": 0,
+            "day_completion_tokens": 0,
+            "minute_total_tokens": 0,
+            "day_total_tokens": 0,
+        }
+        with self._get_connection() as conn:
+            cursor = conn.cursor()
+            cursor.execute(
+                """
+                SELECT bucket_type, bucket_start, request_count, prompt_tokens, completion_tokens, total_tokens
+                FROM auth_entry_usage_buckets
+                WHERE auth_group_name = ? AND entry_id = ?
+                  AND ((bucket_type = 'minute' AND bucket_start = ?)
+                    OR (bucket_type = 'day' AND bucket_start = ?))
+                """,
+                (auth_group_name, entry_id, minute_bucket, day_bucket),
+            )
+            rows = cursor.fetchall()
+
+        for row in rows:
+            prefix = "minute" if row["bucket_type"] == "minute" else "day"
+            usage[f"{prefix}_request_count"] = int(row["request_count"] or 0)
+            usage[f"{prefix}_prompt_tokens"] = int(row["prompt_tokens"] or 0)
+            usage[f"{prefix}_completion_tokens"] = int(row["completion_tokens"] or 0)
+            usage[f"{prefix}_total_tokens"] = int(row["total_tokens"] or 0)
+        return usage
+
+    def list_current_usage(
+        self,
+        auth_group_name: str,
+        entry_ids: Iterable[str],
+        when: datetime,
+    ) -> Dict[str, Dict[str, int]]:
+        usage_by_entry: Dict[str, Dict[str, int]] = {}
+        for entry_id in entry_ids:
+            usage_by_entry[str(entry_id)] = self.get_current_usage(auth_group_name, str(entry_id), when)
+        return usage_by_entry

src/services/__init__.py

@@ -3,6 +3,7 @@
 """服务层导出。"""
 
 from .authentication_service import AuthenticationService
+from .auth_group_service import AuthGroupService
 from .log_service import LogService
 from .model_discovery_service import ModelDiscoveryService
 from .provider_service import ProviderService
@@ -12,6 +13,7 @@ from .user_service import UserService
 
 __all__ = [
     'AuthenticationService',
+    'AuthGroupService',
     'UserService',
     'LogService',
     'ModelDiscoveryService',

src/services/auth_group_service.py

@@ -0,0 +1,239 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+"""Auth group configuration and runtime management."""
+
+from __future__ import annotations
+
+from typing import Any, Callable, Dict, List, Mapping, Optional
+
+import yaml
+
+from ..application.app_context import AppContext
+from ..config import AuthGroupManager
+from ..config.provider_config import (
+    AuthEntrySchema,
+    AuthGroupSchema,
+    validate_auth_group_provider_definitions,
+)
+
+
+class AuthGroupService:
+    """Manage auth group config and expose runtime state."""
+
+    def __init__(
+        self,
+        ctx: AppContext,
+        reload_callback: Callable[[], None],
+        auth_group_manager: AuthGroupManager,
+    ):
+        self._config_manager = ctx.config_manager
+        self._logger = ctx.logger
+        self._reload_callback = reload_callback
+        self._auth_group_manager = auth_group_manager
+
+    def list_auth_groups(self) -> List[Dict[str, Any]]:
+        config = self._config_manager.get_raw_config()
+        auth_groups = self._extract_auth_groups(config)
+        providers = self._extract_providers(config)
+        summaries = {
+            item["name"]: item
+            for item in self._auth_group_manager.list_auth_group_summaries()
+        }
+        provider_counts = self._count_providers_by_auth_group(providers)
+        result: List[Dict[str, Any]] = []
+        for auth_group in auth_groups:
+            normalized = AuthGroupSchema.from_mapping(auth_group).to_mapping()
+            normalized["summary"] = summaries.get(normalized["name"], {}).get("summary", {})
+            normalized["provider_count"] = provider_counts.get(normalized["name"], 0)
+            result.append(normalized)
+        return result
+
+    def get_auth_group(self, name: str) -> Optional[Dict[str, Any]]:
+        config = self._config_manager.get_raw_config()
+        auth_groups = self._extract_auth_groups(config)
+        providers = self._extract_providers(config)
+        target = self._find_auth_group(auth_groups, name)
+        if target is None:
+            return None
+        normalized = AuthGroupSchema.from_mapping(target).to_mapping()
+        summaries = {
+            item["name"]: item
+            for item in self._auth_group_manager.list_auth_group_summaries()
+        }
+        normalized["summary"] = summaries.get(normalized["name"], {}).get("summary", {})
+        normalized["provider_count"] = self._count_providers_by_auth_group(providers).get(normalized["name"], 0)
+        return normalized
+
+    def create_auth_group(self, payload: Dict[str, Any]) -> Dict[str, Any]:
+        config = self._config_manager.get_raw_config()
+        auth_groups = self._extract_auth_groups(config)
+        providers = self._extract_providers(config)
+        auth_group = AuthGroupSchema.from_mapping(payload)
+        normalized = auth_group.to_mapping()
+
+        if self._find_auth_group(auth_groups, auth_group.name):
+            raise ValueError(f"Auth group already exists: {auth_group.name}")
+
+        auth_groups.append(normalized)
+        self._save_auth_groups(config, auth_groups, providers)
+        return normalized
+
+    def update_auth_group(self, current_name: str, payload: Dict[str, Any]) -> Dict[str, Any]:
+        config = self._config_manager.get_raw_config()
+        auth_groups = self._extract_auth_groups(config)
+        providers = self._extract_providers(config)
+        auth_group = AuthGroupSchema.from_mapping(payload)
+        normalized = auth_group.to_mapping()
+
+        target = self._find_auth_group(auth_groups, current_name)
+        if target is None:
+            raise ValueError(f"Auth group not found: {current_name}")
+
+        duplicate = self._find_auth_group(auth_groups, auth_group.name)
+        if duplicate is not None and duplicate is not target:
+            raise ValueError(f"Auth group already exists: {auth_group.name}")
+
+        if auth_group.name != str(current_name).strip():
+            for provider in providers:
+                if str(provider.get("auth_group") or "").strip() == str(current_name).strip():
+                    provider["auth_group"] = auth_group.name
+
+        auth_groups[auth_groups.index(target)] = normalized
+        self._save_auth_groups(config, auth_groups, providers)
+        return normalized
+
+    def delete_auth_group(self, name: str) -> None:
+        config = self._config_manager.get_raw_config()
+        auth_groups = self._extract_auth_groups(config)
+        providers = self._extract_providers(config)
+        target = self._find_auth_group(auth_groups, name)
+        if target is None:
+            raise ValueError(f"Auth group not found: {name}")
+
+        if any(str(provider.get("auth_group") or "").strip() == str(name).strip() for provider in providers):
+            raise ValueError(f"Auth group is still referenced by providers: {name}")
+
+        auth_groups.remove(target)
+        self._save_auth_groups(config, auth_groups, providers)
+
+    def get_auth_group_runtime(self, name: str) -> Dict[str, Any]:
+        runtime = self._auth_group_manager.get_auth_group_runtime(name)
+        config = self._config_manager.get_raw_config()
+        providers = self._extract_providers(config)
+        runtime["provider_count"] = self._count_providers_by_auth_group(providers).get(runtime["name"], 0)
+        return runtime
+
+    def restore_entry(self, auth_group_name: str, entry_id: str) -> None:
+        self._auth_group_manager.restore_entry(auth_group_name, entry_id)
+
+    def clear_entry_cooldown(self, auth_group_name: str, entry_id: str) -> None:
+        self._auth_group_manager.clear_entry_cooldown(auth_group_name, entry_id)
+
+    def set_entry_disabled(self, auth_group_name: str, entry_id: str, *, disabled: bool) -> None:
+        self._auth_group_manager.set_entry_disabled(auth_group_name, entry_id, disabled=disabled)
+
+    def reset_entry_minute_usage(self, auth_group_name: str, entry_id: str) -> None:
+        self._auth_group_manager.reset_entry_minute_usage(auth_group_name, entry_id)
+
+    def reset_entry_runtime(self, auth_group_name: str, entry_id: str) -> None:
+        self._auth_group_manager.reset_entry_runtime(auth_group_name, entry_id)
+
+    def import_auth_entries(self, yaml_text: str) -> List[Dict[str, Any]]:
+        raw_text = str(yaml_text or "").strip()
+        if not raw_text:
+            raise ValueError("YAML 内容不能为空")
+
+        try:
+            payload = yaml.safe_load(raw_text)
+        except yaml.YAMLError as exc:
+            raise ValueError(f"YAML 解析失败: {exc}") from exc
+
+        entries_payload = self._normalize_import_payload(payload)
+        if not entries_payload:
+            raise ValueError("请至少导入一个 Auth Entry")
+
+        seen_entry_ids: set[str] = set()
+        normalized_entries: List[Dict[str, Any]] = []
+        for item in entries_payload:
+            entry = AuthEntrySchema.from_mapping(item)
+            if entry.id in seen_entry_ids:
+                raise ValueError(f"检测到重复的 Auth Entry ID: {entry.id}")
+            seen_entry_ids.add(entry.id)
+            normalized_entries.append(entry.to_mapping())
+        return normalized_entries
+
+    def _save_auth_groups(
+        self,
+        config: Dict[str, Any],
+        auth_groups: List[Dict[str, Any]],
+        providers: List[Dict[str, Any]],
+    ) -> None:
+        validate_auth_group_provider_definitions(auth_groups, providers)
+        config["auth_groups"] = auth_groups
+        config["providers"] = providers
+        self._config_manager.write_raw_config(config)
+        self._reload_callback()
+
+    @staticmethod
+    def _extract_auth_groups(config: Dict[str, Any]) -> List[Dict[str, Any]]:
+        auth_groups = config.get("auth_groups", [])
+        if auth_groups is None:
+            auth_groups = []
+        if not isinstance(auth_groups, list):
+            raise ValueError("Config field 'auth_groups' must be a list")
+        for index, auth_group in enumerate(auth_groups):
+            if not isinstance(auth_group, dict):
+                raise ValueError(f"Auth group entry at index {index} must be an object")
+        return list(auth_groups)
+
+    @staticmethod
+    def _extract_providers(config: Dict[str, Any]) -> List[Dict[str, Any]]:
+        providers = config.get("providers", [])
+        if providers is None:
+            providers = []
+        if not isinstance(providers, list):
+            raise ValueError("Config field 'providers' must be a list")
+        for index, provider in enumerate(providers):
+            if not isinstance(provider, dict):
+                raise ValueError(f"Provider entry at index {index} must be an object")
+        return list(providers)
+
+    @staticmethod
+    def _find_auth_group(auth_groups: List[Dict[str, Any]], name: str) -> Optional[Dict[str, Any]]:
+        normalized_name = str(name).strip()
+        for auth_group in auth_groups:
+            if str(auth_group.get("name", "")).strip() == normalized_name:
+                return auth_group
+        return None
+
+    @staticmethod
+    def _count_providers_by_auth_group(providers: List[Dict[str, Any]]) -> Dict[str, int]:
+        counts: Dict[str, int] = {}
+        for provider in providers:
+            auth_group = str(provider.get("auth_group") or "").strip()
+            if not auth_group:
+                continue
+            counts[auth_group] = counts.get(auth_group, 0) + 1
+        return counts
+
+    @staticmethod
+    def _normalize_import_payload(payload: Any) -> List[Mapping[str, Any]]:
+        if isinstance(payload, list):
+            entries = payload
+        elif isinstance(payload, Mapping):
+            if "entries" in payload:
+                entries = payload.get("entries")
+            else:
+                entries = [payload]
+        else:
+            raise ValueError("YAML 必须是 Auth Entry 列表，或包含 entries 字段的对象")
+
+        if not isinstance(entries, list):
+            raise ValueError("YAML 中的 entries 必须是列表")
+
+        normalized_entries: List[Mapping[str, Any]] = []
+        for index, item in enumerate(entries):
+            if not isinstance(item, Mapping):
+                raise ValueError(f"第 {index + 1} 个 Auth Entry 必须是对象")
+            normalized_entries.append(item)
+        return normalized_entries

src/services/provider_service.py

@@ -9,7 +9,7 @@ from typing import Any, Callable, Dict, List, Optional
 from ..application.app_context import AppContext
 from ..config.provider_config import (
     ProviderConfigSchema,
-    validate_provider_definitions,
+    validate_auth_group_provider_definitions,
 )
 
 
@@ -78,7 +78,7 @@ class ProviderService:
         self._save_providers(config, providers)
 
     def _save_providers(self, config: Dict[str, Any], providers: List[Dict[str, Any]]) -> None:
-        self._validate_providers(providers)
+        self._validate_providers(config, providers)
         config['providers'] = providers
         self._config_manager.write_raw_config(config)
         self._reload_callback()
@@ -104,5 +104,20 @@ class ProviderService:
         return None
 
     @staticmethod
-    def _validate_providers(providers: List[Dict[str, Any]]) -> None:
-        validate_provider_definitions(providers)
+    def _extract_auth_groups(config: Dict[str, Any]) -> List[Dict[str, Any]]:
+        auth_groups = config.get('auth_groups', [])
+        if auth_groups is None:
+            auth_groups = []
+        if not isinstance(auth_groups, list):
+            raise ValueError("Config field 'auth_groups' must be a list")
+        for index, auth_group in enumerate(auth_groups):
+            if not isinstance(auth_group, dict):
+                raise ValueError(f'Auth group entry at index {index} must be an object')
+        return list(auth_groups)
+
+    @staticmethod
+    def _validate_providers(config: Dict[str, Any], providers: List[Dict[str, Any]]) -> None:
+        validate_auth_group_provider_definitions(
+            ProviderService._extract_auth_groups(config),
+            providers,
+        )

src/services/proxy_service.py

@@ -13,6 +13,7 @@ import websocket
 from flask import Response, stream_with_context
 
 from ..application.app_context import AppContext
+from ..config.auth_group_manager import AuthGroupManager, AuthGroupSelectionError, SelectedAuthEntry
 from ..executors import OpenedUpstreamResponse, build_default_executor_registry
 from ..external import LLMProvider
 from ..hooks import HookContext, HookErrorType
@@ -41,9 +42,10 @@ class ProxyErrorInfo:
 class ProxyService:
     """处理上游 LLM 代理请求。"""
 
-    def __init__(self, ctx: AppContext):
+    def __init__(self, ctx: AppContext, auth_group_manager: Optional[AuthGroupManager] = None):
         self._logger = ctx.logger
         self._root_path = ctx.root_path
+        self._auth_group_manager = auth_group_manager
         self._executor_registry = build_default_executor_registry(self._logger)
         self._translator_registry = build_default_translator_registry()
 
@@ -71,6 +73,7 @@ class ProxyService:
 
         def build_request(
             attempt: int,
+            selected_auth: Optional[SelectedAuthEntry],
         ) -> Tuple[Dict[str, str], Dict[str, Any], Dict[str, Any], HookContext]:
             initial_stream = bool(request_data.get("stream", False))
             request_ctx = HookContext(
@@ -84,13 +87,21 @@ class ProxyService:
                 provider_target_format=provider.target_format,
                 transport=provider.transport,
                 stream=initial_stream,
+                auth_group_name=(
+                    selected_auth.auth_group_name
+                    if selected_auth is not None
+                    else provider.auth_group
+                ),
+                auth_entry_id=(selected_auth.entry_id if selected_auth is not None else None),
                 last_status_code=previous_status_code,
                 last_error_type=previous_error_type,
             )
 
             headers = dict(request_headers)
             headers["content-type"] = "application/json"
-            if provider.api_key:
+            if selected_auth is not None:
+                headers.update(selected_auth.headers_mapping())
+            elif provider.api_key:
                 headers["authorization"] = f"Bearer {provider.api_key}"
             headers = provider.apply_header_hook(request_ctx, headers)
 
@@ -103,21 +114,54 @@ class ProxyService:
             return headers, guarded_body, translated_body, request_ctx
 
         for attempt in range(max_retries):
-            headers, guarded_body, translated_body, request_ctx = build_request(attempt)
-            requested_stream = request_ctx.stream
-            self._logger.info(
-                "Proxying upstream request: provider=%s transport=%s source=%s target=%s model=%s attempt=%s/%s stream=%s",
-                provider.name,
-                provider.transport,
-                provider.source_format,
-                provider.target_format,
-                translated_body.get("model"),
-                attempt + 1,
-                max_retries,
-                requested_stream,
-            )
+            selected_auth: Optional[SelectedAuthEntry] = None
+            attempt_finalized = False
+
+            def finalize_attempt(
+                *,
+                status_code: Optional[int] = None,
+                error_type: Optional[HookErrorType] = None,
+                error_message: Optional[str] = None,
+                response_headers: Optional[Dict[str, Any]] = None,
+                usage: Optional[Dict[str, Any]] = None,
+            ) -> None:
+                nonlocal attempt_finalized
+                if attempt_finalized:
+                    return
+                attempt_finalized = True
+                if self._auth_group_manager is None:
+                    return
+                self._auth_group_manager.finish(
+                    selected_auth,
+                    status_code=status_code,
+                    error_type=error_type,
+                    error_message=error_message,
+                    response_headers=response_headers,
+                    usage=usage,
+                )
 
             try:
+                if self._auth_group_manager is not None and provider.auth_group:
+                    selected_auth = self._auth_group_manager.acquire(provider.auth_group)
+
+                headers, guarded_body, translated_body, request_ctx = build_request(attempt, selected_auth)
+                requested_stream = request_ctx.stream
+                self._logger.info(
+                    "Proxying upstream request: provider=%s transport=%s source=%s target=%s model=%s attempt=%s/%s stream=%s auth_group=%s auth_entry=%s",
+                    provider.name,
+                    provider.transport,
+                    provider.source_format,
+                    provider.target_format,
+                    translated_body.get("model"),
+                    attempt + 1,
+                    max_retries,
+                    requested_stream,
+                    request_ctx.auth_group_name or "<none>",
+                    request_ctx.auth_entry_id or "<none>",
+                )
+                if self._auth_group_manager is not None:
+                    self._auth_group_manager.mark_request_dispatched(selected_auth)
+
                 opened = self._coerce_opened_response(
                     self._open_upstream_response(
                         provider,
@@ -125,15 +169,22 @@ class ProxyService:
                         translated_body,
                         requested_stream,
                         target_url,
-                    request_proxies,
-                    timeout_seconds,
-                    verify_ssl,
+                        request_proxies,
+                        timeout_seconds,
+                        verify_ssl,
                     )
                 )
 
                 if self._should_retry_status_code(opened.status_code) and attempt < max_retries - 1:
                     previous_status_code = opened.status_code
                     previous_error_type = None
+                    raw_response_headers = dict(getattr(opened.response, "headers", {}) or {})
+                    _, _, retry_summary = self._consume_upstream_error(provider, opened)
+                    finalize_attempt(
+                        status_code=opened.status_code,
+                        error_message=retry_summary,
+                        response_headers=raw_response_headers,
+                    )
                     self._logger.warning(
                         "Retryable upstream status (attempt %s/%s): provider=%s status=%s stream=%s, retrying",
                         attempt + 1,
@@ -142,7 +193,6 @@ class ProxyService:
                         opened.status_code,
                         opened.is_stream,
                     )
-                    opened.response.close()
                     continue
 
                 if requested_stream != opened.is_stream:
@@ -156,11 +206,14 @@ class ProxyService:
                     )
 
                 if opened.status_code >= 400:
-                    return (
-                        self._build_error_response(provider, opened),
-                        opened.status_code,
-                        None,
+                    raw_response_headers = dict(getattr(opened.response, "headers", {}) or {})
+                    response, error_summary = self._build_error_response(provider, opened)
+                    finalize_attempt(
+                        status_code=opened.status_code,
+                        error_message=error_summary,
+                        response_headers=raw_response_headers,
                     )
+                    return response, opened.status_code, None
 
                 if opened.is_stream:
                     response = self._build_stream_response(
@@ -172,6 +225,7 @@ class ProxyService:
                         opened=opened,
                         on_complete=on_complete,
                         forward_stream_usage=forward_stream_usage,
+                        finalize_attempt=finalize_attempt,
                     )
                 else:
                     response = self._build_nonstream_response(
@@ -182,6 +236,7 @@ class ProxyService:
                         translated_request=translated_body,
                         opened=opened,
                         on_complete=on_complete,
+                        finalize_attempt=finalize_attempt,
                     )
 
                 self._logger.info(
@@ -194,6 +249,21 @@ class ProxyService:
                     opened.is_stream,
                 )
                 return response, opened.status_code, None
+            except AuthGroupSelectionError as exc:
+                last_error = ProxyErrorInfo(
+                    message=exc.message,
+                    status_code=exc.status_code,
+                    error_type=exc.error_type,
+                    error_code=exc.error_code,
+                )
+                self._logger.warning(
+                    "Auth group unavailable: provider=%s auth_group=%s status=%s error=%s",
+                    provider.name,
+                    provider.auth_group or "<none>",
+                    exc.status_code,
+                    exc.message,
+                )
+                return None, exc.status_code, last_error
             except requests.exceptions.RequestException as exc:
                 previous_status_code = None
                 previous_error_type = self._classify_request_error(exc)
@@ -205,6 +275,10 @@ class ProxyService:
                     provider.name,
                     exc,
                 )
+                finalize_attempt(
+                    error_type=previous_error_type,
+                    error_message=str(exc),
+                )
                 if attempt < max_retries - 1:
                     continue
             except (websocket.WebSocketException, OSError) as exc:
@@ -218,8 +292,15 @@ class ProxyService:
                     provider.name,
                     exc,
                 )
+                finalize_attempt(
+                    error_type=previous_error_type,
+                    error_message=str(exc),
+                )
                 if attempt < max_retries - 1:
                     continue
+            except Exception:
+                finalize_attempt()
+                raise
 
         if last_error is None:
             last_error = ProxyErrorInfo(
@@ -241,18 +322,30 @@ class ProxyService:
         opened: OpenedUpstreamResponse,
         on_complete: Optional[Callable[[Dict[str, Any]], None]],
         forward_stream_usage: bool,
+        finalize_attempt: Optional[Callable[..., None]] = None,
     ) -> Response:
         response = opened.response
         meta = self._create_empty_meta()
         completed = False
         terminal_sent = False
 
-        def safe_on_complete() -> None:
+        def safe_on_complete(
+            *,
+            error_type: Optional[HookErrorType] = None,
+            error_message: Optional[str] = None,
+        ) -> None:
             nonlocal completed
             if completed:
                 return
             completed = True
-            if on_complete:
+            if finalize_attempt is not None:
+                finalize_attempt(
+                    status_code=(opened.status_code if error_type is None else None),
+                    error_type=error_type,
+                    error_message=error_message,
+                    usage=(meta if error_type is None else None),
+                )
+            if on_complete and error_type is None:
                 try:
                     on_complete(meta)
                 except Exception as exc:
@@ -261,6 +354,8 @@ class ProxyService:
         def generate() -> Iterator[bytes]:
             nonlocal terminal_sent
             state: Dict[str, Any] = {}
+            completion_error_type: Optional[HookErrorType] = None
+            completion_error_message: Optional[str] = None
             try:
                 upstream_chunks = response.iter_content(chunk_size=None)
                 for event in decode_stream_events(upstream_chunks, opened.stream_format):
@@ -302,11 +397,29 @@ class ProxyService:
                     )
                     if encoded_terminal:
                         yield encoded_terminal
+            except requests.exceptions.RequestException as exc:
+                completion_error_type = self._classify_request_error(exc)
+                completion_error_message = str(exc)
+                self._logger.error("Streamed HTTP upstream error: provider=%s error=%s", provider.name, exc)
+                raise
+            except (websocket.WebSocketException, OSError) as exc:
+                completion_error_type = self._classify_websocket_error(exc)
+                completion_error_message = str(exc)
+                self._logger.error("Streamed WebSocket upstream error: provider=%s error=%s", provider.name, exc)
+                raise
+            except Exception as exc:
+                completion_error_type = HookErrorType.TRANSPORT_ERROR
+                completion_error_message = str(exc)
+                self._logger.error("Streamed upstream processing error: provider=%s error=%s", provider.name, exc)
+                raise
             finally:
                 try:
                     response.close()
                 finally:
-                    safe_on_complete()
+                    safe_on_complete(
+                        error_type=completion_error_type,
+                        error_message=completion_error_message,
+                    )
 
         headers = self._filter_response_headers(getattr(response, "headers", {}))
         headers["Content-Type"] = "text/event-stream; charset=utf-8"
@@ -327,6 +440,7 @@ class ProxyService:
         translated_request: Dict[str, Any],
         opened: OpenedUpstreamResponse,
         on_complete: Optional[Callable[[Dict[str, Any]], None]],
+        finalize_attempt: Optional[Callable[..., None]] = None,
     ) -> Response:
         response = opened.response
         try:
@@ -349,6 +463,8 @@ class ProxyService:
                     on_complete(meta)
                 except Exception as exc:
                     self._logger.error("Error in on_complete callback: %s", exc)
+            if finalize_attempt is not None:
+                finalize_attempt(status_code=opened.status_code, usage=meta)
 
             response_body = encode_downstream_response_body(body_to_send, provider.target_format)
             headers = self._filter_response_headers(getattr(response, "headers", {}))
@@ -361,7 +477,11 @@ class ProxyService:
         finally:
             response.close()
 
-    def _build_error_response(self, provider: LLMProvider, opened: OpenedUpstreamResponse) -> Response:
+    def _consume_upstream_error(
+        self,
+        provider: LLMProvider,
+        opened: OpenedUpstreamResponse,
+    ) -> tuple[bytes, Dict[str, str], Optional[str]]:
         response = opened.response
         try:
             body = self._read_response_body(response)
@@ -379,10 +499,18 @@ class ProxyService:
             headers = self._filter_response_headers(getattr(response, "headers", {}))
             if opened.content_type:
                 headers["Content-Type"] = opened.content_type
-            return Response(body, status=opened.status_code, headers=headers)
+            return body, headers, summary
         finally:
             response.close()
 
+    def _build_error_response(
+        self,
+        provider: LLMProvider,
+        opened: OpenedUpstreamResponse,
+    ) -> tuple[Response, Optional[str]]:
+        body, headers, summary = self._consume_upstream_error(provider, opened)
+        return Response(body, status=opened.status_code, headers=headers), summary
+
     def _guard_stream_chunk(
         self,
         provider: LLMProvider,

tests/test_auth_groups.py

@@ -0,0 +1,452 @@
+import sys
+import tempfile
+import unittest
+from pathlib import Path
+
+from flask import Flask
+import yaml
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+
+from src.application.app_context import AppContext
+from src.config import build_auth_group_schemas, build_provider_schemas
+from src.config.auth_group_manager import AuthGroupManager, AuthGroupSelectionError
+from src.config.provider_config import AuthGroupSchema, ProviderConfigSchema
+from src.config.config_manager import ConfigManager
+from src.repositories import AuthGroupRepository
+from src.services.auth_group_service import AuthGroupService
+from src.utils.database import create_connection_factory
+from src.utils.local_time import parse_local_datetime
+
+
+class FakeLogger:
+    def info(self, msg: str, *args) -> None:
+        del msg, args
+
+    def warning(self, msg: str, *args) -> None:
+        del msg, args
+
+    def error(self, msg: str, *args) -> None:
+        del msg, args
+
+    def debug(self, msg: str, *args) -> None:
+        del msg, args
+
+
+class AuthGroupConfigTests(unittest.TestCase):
+    def test_provider_rejects_multiple_auth_bindings(self) -> None:
+        with self.assertRaisesRegex(ValueError, "either auth_group or api_key, not both"):
+            ProviderConfigSchema.from_mapping(
+                {
+                    "name": "demo",
+                    "api": "https://example.com/v1/chat/completions",
+                    "auth_group": "pool-a",
+                    "api_key": "sk-demo",
+                    "model_list": ["gpt-4.1"],
+                }
+            )
+
+    def test_provider_allows_empty_auth_binding_for_hook_or_public_upstream(self) -> None:
+        schema = ProviderConfigSchema.from_mapping(
+            {
+                "name": "demo",
+                "api": "https://example.com/v1/chat/completions",
+                "model_list": ["gpt-4.1"],
+            }
+        )
+
+        self.assertIsNone(schema.auth_group)
+        self.assertIsNone(schema.api_key)
+
+    def test_build_provider_schemas_rejects_unknown_auth_group_even_when_no_groups_exist(self) -> None:
+        with self.assertRaisesRegex(ValueError, "unknown auth_group: pool-a"):
+            build_provider_schemas(
+                [
+                    {
+                        "name": "demo",
+                        "api": "https://example.com/v1/chat/completions",
+                        "auth_group": "pool-a",
+                        "model_list": ["gpt-4.1"],
+                    }
+                ],
+                available_auth_group_names=set(),
+            )
+
+
+class AuthGroupManagerTests(unittest.TestCase):
+    def setUp(self) -> None:
+        self.temp_dir = tempfile.TemporaryDirectory()
+        db_path = Path(self.temp_dir.name) / "auth-groups.db"
+        self.ctx = AppContext(
+            logger=FakeLogger(),
+            config_manager=None,  # type: ignore[arg-type]
+            root_path=Path(self.temp_dir.name),
+            flask_app=Flask(__name__),
+        )
+        self.repository = AuthGroupRepository(create_connection_factory(db_path))
+        self.manager = AuthGroupManager(self.ctx, self.repository)
+        self.group = AuthGroupSchema.from_mapping(
+            {
+                "name": "pool-a",
+                "strategy": "least_inflight",
+                "cooldown_seconds_on_429": 30,
+                "entries": [
+                    {
+                        "id": "key-a",
+                        "headers": {"Authorization": "Bearer sk-a"},
+                        "max_concurrency": 2,
+                        "cooldown_seconds_on_429": 45,
+                    },
+                    {
+                        "id": "key-b",
+                        "headers": {"Authorization": "Bearer sk-b"},
+                        "max_concurrency": 2,
+                    },
+                ],
+            }
+        )
+        self.manager.load_auth_groups((self.group,))
+
+    def tearDown(self) -> None:
+        self.temp_dir.cleanup()
+
+    def test_least_inflight_selection_uses_rotation_to_spread_equal_load(self) -> None:
+        first = self.manager.acquire("pool-a")
+        second = self.manager.acquire("pool-a")
+
+        self.assertEqual("key-a", first.entry_id)
+        self.assertEqual("key-b", second.entry_id)
+
+    def test_finish_success_persists_request_and_token_usage(self) -> None:
+        selection = self.manager.acquire("pool-a")
+        self.manager.mark_request_dispatched(selection)
+        self.manager.finish(
+            selection,
+            status_code=200,
+            usage={
+                "prompt_tokens": 12,
+                "completion_tokens": 8,
+                "total_tokens": 20,
+            },
+        )
+
+        runtime = self.manager.get_auth_group_runtime("pool-a")
+        entry = next(item for item in runtime["entries"] if item["id"] == "key-a")
+
+        self.assertEqual("available", entry["status"])
+        self.assertEqual(1, entry["minute_request_count"])
+        self.assertEqual(20, entry["minute_total_tokens"])
+
+    def test_429_cools_down_only_current_entry_using_retry_after_priority(self) -> None:
+        selection = self.manager.acquire("pool-a")
+        self.manager.mark_request_dispatched(selection)
+        self.manager.finish(
+            selection,
+            status_code=429,
+            response_headers={"Retry-After": "120"},
+            error_message="rate limited",
+        )
+
+        runtime = self.manager.get_auth_group_runtime("pool-a")
+        cooled_entry = next(item for item in runtime["entries"] if item["id"] == "key-a")
+        other_entry = next(item for item in runtime["entries"] if item["id"] == "key-b")
+
+        self.assertEqual("cooldown", cooled_entry["status"])
+        cooldown_until = parse_local_datetime(cooled_entry["cooldown_until"])
+        self.assertIsNotNone(cooldown_until)
+        self.assertEqual("available", other_entry["status"])
+
+        next_selection = self.manager.acquire("pool-a")
+        self.assertEqual("key-b", next_selection.entry_id)
+
+    def test_clear_entry_cooldown_restores_entry_to_available(self) -> None:
+        selection = self.manager.acquire("pool-a")
+        self.manager.mark_request_dispatched(selection)
+        self.manager.finish(
+            selection,
+            status_code=429,
+            response_headers={"Retry-After": "120"},
+            error_message="rate limited",
+        )
+
+        self.manager.clear_entry_cooldown("pool-a", "key-a")
+        runtime = self.manager.get_auth_group_runtime("pool-a")
+        entry = next(item for item in runtime["entries"] if item["id"] == "key-a")
+
+        self.assertEqual("available", entry["status"])
+        self.assertIsNone(entry["cooldown_until"])
+
+    def test_401_disables_entry_until_restore(self) -> None:
+        selection = self.manager.acquire("pool-a")
+        self.manager.mark_request_dispatched(selection)
+        self.manager.finish(selection, status_code=401, error_message="bad key")
+
+        runtime = self.manager.get_auth_group_runtime("pool-a")
+        disabled_entry = next(item for item in runtime["entries"] if item["id"] == "key-a")
+        self.assertEqual("disabled", disabled_entry["status"])
+
+        next_selection = self.manager.acquire("pool-a")
+        self.assertEqual("key-b", next_selection.entry_id)
+
+        self.manager.restore_entry("pool-a", "key-a")
+        restored_runtime = self.manager.get_auth_group_runtime("pool-a")
+        restored_entry = next(item for item in restored_runtime["entries"] if item["id"] == "key-a")
+        self.assertEqual("available", restored_entry["status"])
+
+    def test_manual_disable_and_enable_entry_updates_runtime_status(self) -> None:
+        self.manager.set_entry_disabled("pool-a", "key-a", disabled=True)
+        disabled_runtime = self.manager.get_auth_group_runtime("pool-a")
+        disabled_entry = next(item for item in disabled_runtime["entries"] if item["id"] == "key-a")
+
+        self.assertEqual("disabled", disabled_entry["status"])
+        self.assertEqual("manual_disabled", disabled_entry["disabled_reason"])
+
+        self.manager.set_entry_disabled("pool-a", "key-a", disabled=False)
+        enabled_runtime = self.manager.get_auth_group_runtime("pool-a")
+        enabled_entry = next(item for item in enabled_runtime["entries"] if item["id"] == "key-a")
+
+        self.assertEqual("available", enabled_entry["status"])
+        self.assertIsNone(enabled_entry["disabled_reason"])
+
+    def test_reset_entry_minute_usage_clears_only_current_minute_bucket(self) -> None:
+        selection = self.manager.acquire("pool-a")
+        self.manager.mark_request_dispatched(selection)
+        self.manager.finish(
+            selection,
+            status_code=200,
+            usage={
+                "prompt_tokens": 10,
+                "completion_tokens": 5,
+                "total_tokens": 15,
+            },
+        )
+
+        before_reset = self.manager.get_auth_group_runtime("pool-a")
+        before_entry = next(item for item in before_reset["entries"] if item["id"] == "key-a")
+        self.assertEqual(1, before_entry["minute_request_count"])
+        self.assertEqual(15, before_entry["minute_total_tokens"])
+        self.assertEqual(1, before_entry["day_request_count"])
+        self.assertEqual(15, before_entry["day_total_tokens"])
+
+        self.manager.reset_entry_minute_usage("pool-a", "key-a")
+        after_reset = self.manager.get_auth_group_runtime("pool-a")
+        after_entry = next(item for item in after_reset["entries"] if item["id"] == "key-a")
+
+        self.assertEqual(0, after_entry["minute_request_count"])
+        self.assertEqual(0, after_entry["minute_total_tokens"])
+        self.assertEqual(1, after_entry["day_request_count"])
+        self.assertEqual(15, after_entry["day_total_tokens"])
+
+    def test_reset_entry_runtime_clears_runtime_flags_and_current_usage(self) -> None:
+        selection = self.manager.acquire("pool-a")
+        self.manager.mark_request_dispatched(selection)
+        self.manager.finish(
+            selection,
+            status_code=429,
+            response_headers={"Retry-After": "120"},
+            error_message="rate limited",
+        )
+        self.manager.set_entry_disabled("pool-a", "key-a", disabled=True)
+
+        before_reset = self.manager.get_auth_group_runtime("pool-a")
+        before_entry = next(item for item in before_reset["entries"] if item["id"] == "key-a")
+        self.assertEqual("disabled", before_entry["status"])
+        self.assertEqual("manual_disabled", before_entry["disabled_reason"])
+        self.assertIsNotNone(before_entry["cooldown_until"])
+        self.assertEqual(429, before_entry["last_status_code"])
+        self.assertEqual("rate limited", before_entry["last_error_message"])
+        self.assertEqual(1, before_entry["minute_request_count"])
+        self.assertEqual(1, before_entry["day_request_count"])
+
+        self.manager.reset_entry_runtime("pool-a", "key-a")
+
+        after_reset = self.manager.get_auth_group_runtime("pool-a")
+        after_entry = next(item for item in after_reset["entries"] if item["id"] == "key-a")
+        self.assertEqual("available", after_entry["status"])
+        self.assertFalse(after_entry["disabled"])
+        self.assertIsNone(after_entry["disabled_reason"])
+        self.assertIsNone(after_entry["cooldown_until"])
+        self.assertIsNone(after_entry["last_status_code"])
+        self.assertIsNone(after_entry["last_error_type"])
+        self.assertIsNone(after_entry["last_error_message"])
+        self.assertEqual(0, after_entry["minute_request_count"])
+        self.assertEqual(0, after_entry["day_request_count"])
+        self.assertEqual(0, after_entry["minute_total_tokens"])
+        self.assertEqual(0, after_entry["day_total_tokens"])
+
+
+class AuthGroupServiceTests(unittest.TestCase):
+    def setUp(self) -> None:
+        self.temp_dir = tempfile.TemporaryDirectory()
+        self.root_path = Path(self.temp_dir.name)
+        self.config_path = self.root_path / "config.yaml"
+        self.db_path = self.root_path / "requests.db"
+        self._write_config({"auth_groups": [], "providers": []})
+        self.config_manager = ConfigManager(self.config_path, self.root_path)
+        self.ctx = AppContext(
+            logger=FakeLogger(),
+            config_manager=self.config_manager,
+            root_path=self.root_path,
+            flask_app=Flask(__name__),
+        )
+        self.repository = AuthGroupRepository(create_connection_factory(self.db_path))
+        self.manager = AuthGroupManager(self.ctx, self.repository)
+        self.reload_count = 0
+
+        def reload_callback() -> None:
+            self.reload_count += 1
+            self.config_manager.reload()
+            raw = self.config_manager.get_raw_config()
+            self.manager.load_auth_groups(build_auth_group_schemas(raw.get("auth_groups", []) or []))
+
+        self.reload_callback = reload_callback
+        self.service = AuthGroupService(self.ctx, self.reload_callback, self.manager)
+        self.reload_callback()
+        self.reload_count = 0
+
+    def tearDown(self) -> None:
+        self.temp_dir.cleanup()
+
+    def _write_config(self, payload) -> None:
+        with open(self.config_path, "w", encoding="utf-8") as handle:
+            yaml.safe_dump(payload, handle, allow_unicode=True, sort_keys=False)
+
+    def test_create_auth_group_persists_and_lists_summary(self) -> None:
+        created = self.service.create_auth_group(
+            {
+                "name": "pool-a",
+                "strategy": "least_inflight",
+                "cooldown_seconds_on_429": 60,
+                "entries": [
+                    {
+                        "id": "key-a",
+                        "headers": {"Authorization": "Bearer sk-a"},
+                    }
+                ],
+            }
+        )
+
+        self.assertEqual("pool-a", created["name"])
+        self.assertEqual(1, self.reload_count)
+        listed = self.service.list_auth_groups()
+        self.assertEqual(["pool-a"], [item["name"] for item in listed])
+
+    def test_import_auth_entries_accepts_yaml_list(self) -> None:
+        entries = self.service.import_auth_entries(
+            """
+- id: key-a
+  headers:
+    Authorization: Bearer sk-a
+  max_concurrency: 3
+- id: key-b
+  headers:
+    x-api-key: abc
+"""
+        )
+
+        self.assertEqual(2, len(entries))
+        self.assertEqual("key-a", entries[0]["id"])
+        self.assertEqual("Bearer sk-a", entries[0]["headers"]["Authorization"])
+        self.assertEqual("key-b", entries[1]["id"])
+
+    def test_import_auth_entries_accepts_entries_wrapper(self) -> None:
+        entries = self.service.import_auth_entries(
+            """
+entries:
+  - id: key-a
+    headers:
+      Authorization: Bearer sk-a
+"""
+        )
+
+        self.assertEqual(1, len(entries))
+        self.assertEqual("key-a", entries[0]["id"])
+
+    def test_import_auth_entries_rejects_invalid_yaml_shape(self) -> None:
+        with self.assertRaisesRegex(ValueError, "entries 必须是列表"):
+            self.service.import_auth_entries(
+                """
+entries:
+  id: key-a
+  headers:
+    Authorization: Bearer sk-a
+"""
+            )
+
+    def test_import_auth_entries_rejects_duplicate_entry_ids(self) -> None:
+        with self.assertRaisesRegex(ValueError, "检测到重复的 Auth Entry ID: key-a"):
+            self.service.import_auth_entries(
+                """
+entries:
+  - id: key-a
+    headers:
+      Authorization: Bearer sk-a
+  - id: key-a
+    headers:
+      Authorization: Bearer sk-b
+"""
+            )
+
+    def test_update_auth_group_renames_provider_reference(self) -> None:
+        self._write_config(
+            {
+                "auth_groups": [
+                    {
+                        "name": "pool-a",
+                        "strategy": "least_inflight",
+                        "entries": [{"id": "key-a", "headers": {"Authorization": "Bearer sk-a"}}],
+                    }
+                ],
+                "providers": [
+                    {
+                        "name": "demo",
+                        "api": "https://example.com/v1/chat/completions",
+                        "auth_group": "pool-a",
+                        "model_list": ["gpt-4.1"],
+                    }
+                ],
+            }
+        )
+        self.reload_callback()
+
+        updated = self.service.update_auth_group(
+            "pool-a",
+            {
+                "name": "pool-b",
+                "strategy": "least_inflight",
+                "entries": [{"id": "key-a", "headers": {"Authorization": "Bearer sk-a"}}],
+            },
+        )
+
+        self.assertEqual("pool-b", updated["name"])
+        current_config = self.config_manager.get_raw_config()
+        self.assertEqual("pool-b", current_config["providers"][0]["auth_group"])
+
+    def test_delete_referenced_auth_group_is_rejected(self) -> None:
+        self._write_config(
+            {
+                "auth_groups": [
+                    {
+                        "name": "pool-a",
+                        "strategy": "least_inflight",
+                        "entries": [{"id": "key-a", "headers": {"Authorization": "Bearer sk-a"}}],
+                    }
+                ],
+                "providers": [
+                    {
+                        "name": "demo",
+                        "api": "https://example.com/v1/chat/completions",
+                        "auth_group": "pool-a",
+                        "model_list": ["gpt-4.1"],
+                    }
+                ],
+            }
+        )
+        self.reload_callback()
+
+        with self.assertRaisesRegex(ValueError, "still referenced"):
+            self.service.delete_auth_group("pool-a")
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_proxy_transport.py

@@ -43,6 +43,7 @@ class ProviderTransportTests(unittest.TestCase):
             {
                 "name": "codex",
                 "api": "wss://example.com/v1/chat/completions",
+                "api_key": "demo-key",
                 "model_list": ["gpt-4.1"],
             }
         )
@@ -54,6 +55,7 @@ class ProviderTransportTests(unittest.TestCase):
             {
                 "name": "codex",
                 "api": "https://example.com/v1/chat/completions",
+                "api_key": "demo-key",
                 "model_list": ["gpt-4.1"],
             }
         )
@@ -67,6 +69,7 @@ class ProviderTransportTests(unittest.TestCase):
             {
                 "name": "codex",
                 "api": "https://example.com/v1/chat/completions",
+                "api_key": "demo-key",
                 "transport": "websocket",
                 "model_list": ["gpt-4.1"],
             }
@@ -80,6 +83,7 @@ class ProviderTransportTests(unittest.TestCase):
                 {
                     "name": "bad-provider",
                     "api": "wss://example.com/v1/chat/completions",
+                    "api_key": "demo-key",
                     "transport": "http",
                     "model_list": ["demo"],
                 }
@@ -90,6 +94,7 @@ class ProviderTransportTests(unittest.TestCase):
             {
                 "name": "demo",
                 "api": "https://example.com/v1/chat/completions",
+                "api_key": "demo-key",
                 "model_list": ["gpt-4.1"],
             }
         )
@@ -106,6 +111,7 @@ class ProviderTransportTests(unittest.TestCase):
                 {
                     "name": "demo",
                     "api": "https://example.com/v1/chat/completions",
+                    "api_key": "demo-key",
                     "format": "openai_chat",
                     "model_list": ["gpt-4.1"],
                 }
@@ -116,6 +122,7 @@ class ProviderTransportTests(unittest.TestCase):
                 {
                     "name": "demo",
                     "api": "https://example.com/v1/chat/completions",
+                    "api_key": "demo-key",
                     "stream_format": "sse_json",
                     "model_list": ["gpt-4.1"],
                 }
@@ -256,6 +263,12 @@ class ProviderTemplateTransportTests(unittest.TestCase):
         self.assertIn('id="providerTransport"', html)
         self.assertIn('id="providerSourceFormat"', html)
         self.assertIn('id="providerTargetFormat"', html)
+        self.assertIn('id="providerAuthMode"', html)
+        self.assertIn('id="providerAuthGroup"', html)
+        self.assertIn('id="authGroupsContainer"', html)
+        self.assertIn('id="authGroupModal"', html)
+        self.assertIn('id="authEntryImportModal"', html)
+        self.assertIn('id="authGroupRuntimeModal"', html)
         self.assertNotIn('id="providerFormat"', html)
         self.assertNotIn('id="providerStreamFormat"', html)
 
@@ -267,10 +280,31 @@ class ProviderTemplateTransportTests(unittest.TestCase):
         self.assertIn('data-provider-help-topic="transport"', html)
         self.assertIn('data-provider-help-topic="source_format"', html)
         self.assertIn('data-provider-help-topic="target_format"', html)
+        self.assertIn('data-provider-help-topic="auth_group_field"', html)
+        self.assertIn('data-provider-help-topic="auth_groups_overview"', html)
+        self.assertIn('data-provider-help-topic="auth_group_strategy"', html)
+        self.assertIn('data-provider-help-topic="auth_entries_editor"', html)
         self.assertIn('data-provider-help-topic="fetch_models"', html)
         self.assertIn('id="providerHelpPopover"', html)
         self.assertIn("function toggleProviderHelp(", html)
         self.assertIn("function syncProviderHelpPopover()", html)
+        self.assertIn("function renderAuthGroups()", html)
+        self.assertIn("function saveAuthGroup()", html)
+        self.assertIn("function saveAuthEntriesFromYaml()", html)
+        self.assertIn("function disableAuthGroupEntry(", html)
+        self.assertIn("function enableAuthGroupEntry(", html)
+        self.assertIn("function resetAuthGroupEntryRuntime(", html)
+        self.assertIn("function openAuthEntryErrorModal(", html)
+        self.assertIn("function updateProviderAuthModeFields()", html)
+        self.assertIn("/api/auth-groups", html)
+        self.assertIn("/api/auth-groups/import-entries", html)
+        self.assertIn("'disable'", html)
+        self.assertIn("'enable'", html)
+        self.assertIn("'reset'", html)
+        self.assertIn("setupCustomSelect('providerAuthMode');", html)
+        self.assertIn("setupCustomSelect('providerAuthGroup');", html)
+        self.assertIn("renderCustomSelectOptions('providerAuthGroup');", html)
+        self.assertIn("setupCustomSelect('authGroupStrategy');", html)
         self.assertIn("setupCustomSelect('providerTransport');", html)
         self.assertIn("setupCustomSelect('providerSourceFormat');", html)
         self.assertIn("setupCustomSelect('providerTargetFormat');", html)
@@ -285,6 +319,33 @@ class ProviderTemplateTransportTests(unittest.TestCase):
         self.assertIn('toggleFilteredFetchedModels(this.checked)', html)
         self.assertIn('class="provider-model-cell"', html)
         self.assertIn('class="provider-meta-line"', html)
+        self.assertIn('class="providers-table-shell auth-groups-table-shell"', html)
+        self.assertIn('<th>Entry 数</th>', html)
+        self.assertIn('placeholder="例如 openai-shared，供 Provider 绑定引用"', html)
+        self.assertIn('placeholder="例如 60，表示该组默认遇到 429 冷却 60 秒"', html)
+        self.assertIn('YAML 编辑', html)
+        self.assertIn('id="authEntryImportYaml"', html)
+        self.assertIn('这里编辑的是当前 Auth Entries 的完整 YAML', html)
+        self.assertIn('插入 Entry 模板', html)
+        self.assertIn('必填：Entry 唯一 ID', html)
+        self.assertIn('可选：每分钟请求数上限', html)
+        self.assertIn('新增 Entry', html)
+        self.assertNotIn("新 Entry", html)
+        self.assertIn('<span>Header 数</span>', html)
+        self.assertIn('<span>限制概览</span>', html)
+        self.assertIn('class="auth-entry-table-toggle-column"', html)
+        self.assertIn("function toggleAuthEntryCard(", html)
+        self.assertIn('class="btn-action btn-edit auth-entry-toggle-btn"', html)
+        self.assertIn("function handleAuthEntrySummaryKeydown(", html)
+        self.assertIn('onclick="toggleAuthEntryCard(this)"', html)
+        self.assertNotIn('id="authEntryImportTemplate"', html)
+        self.assertNotIn('复制模板', html)
+        self.assertNotIn('填入模板', html)
+        self.assertIn("function buildAuthEntriesYamlText(", html)
+        self.assertIn("function insertAuthEntryYamlTemplate(", html)
+        self.assertIn("function getSingleAuthEntryYamlTemplate(", html)
+        self.assertIn("function saveAuthEntriesFromYaml(", html)
+        self.assertIn('id="authEntryErrorModal"', html)
         self.assertIn("showActionError('保存 Provider'", html)
         self.assertIn("showActionError('删除 Provider'", html)
         self.assertIn("showActionError('拉取模型'", html)
@@ -324,6 +385,8 @@ const sandbox = {{
       providerModelCount: {{ textContent: "" }},
       providerName: {{ value: " demo " }},
       providerApi: {{ value: " https://example.com/v1/chat/completions " }},
+      providerAuthMode: {{ value: "auth_group" }},
+      providerAuthGroup: {{ value: " shared-pool " }},
       providerTransport: {{ value: "http" }},
       providerSourceFormat: {{ value: "openai_chat" }},
       providerTargetFormat: {{ value: "openai_chat" }},
@@ -348,8 +411,12 @@ const collectedAfter = sandbox.collectFormData();
 
 process.stdout.write(JSON.stringify({{
   beforeModelList: collectedBefore.model_list,
+  beforeAuthGroup: collectedBefore.auth_group,
+  beforeApiKey: collectedBefore.api_key,
   afterTextarea: sandbox.document.elements.providerModelList.value,
   afterModelList: collectedAfter.model_list,
+  afterAuthGroup: collectedAfter.auth_group,
+  afterApiKey: collectedAfter.api_key,
   countText: sandbox.document.elements.providerModelCount.textContent,
   message: sandbox.messages[0]?.message || "",
 }}));
@@ -363,8 +430,12 @@ process.stdout.write(JSON.stringify({{
         payload = json.loads(completed.stdout.decode("utf-8"))
 
         self.assertEqual("beta\nAlpha\nalpha\nBeta", payload["beforeModelList"])
+        self.assertEqual("shared-pool", payload["beforeAuthGroup"])
+        self.assertEqual("", payload["beforeApiKey"])
         self.assertEqual("Alpha\nBeta\nalpha\nbeta", payload["afterTextarea"])
         self.assertEqual("Alpha\nBeta\nalpha\nbeta", payload["afterModelList"])
+        self.assertEqual("shared-pool", payload["afterAuthGroup"])
+        self.assertEqual("", payload["afterApiKey"])
         self.assertIn("4", payload["countText"])
         self.assertTrue(payload["message"])