feat(test,export,headers): 模型测试复用统一运行时、实时进度展示、导出增强与请求头大小写保留

- 模型测试 failover 从手动 FailoverEngine 改为 TaskService.execute_sync_candidates 统一运行时
- 前端新增实时 trace 轮询进度展示（候选状态、测试账号、进度条）
- 用户导出/导入支持明文 Key 优先（版本升至 1.2），新增 email_verified 字段
- SENSITIVE_CREDENTIAL_FIELDS 统一到 provider_ops/types.py，补充 refresh_token
- 请求头大小写保留机制（resolve_header_name_case + HeaderBuilder.add 语义修改）
- Codex envelope 移除合成头部，保留客户端原始请求头
- endpoint_checker 支持自定义超时透传
- 新增 x-forwarded-scheme 到上游丢弃头部列表

frontend/src/api/admin.ts

@@ -34,6 +34,12 @@ export interface OAuthProviderExport {
   is_enabled?: boolean
 }
 
+export interface SystemConfigExport {
+  key: string
+  value: unknown
+  description?: string | null
+}
+
 // 配置导出数据结构
 export interface ConfigExportData {
   version: string
@@ -42,6 +48,7 @@ export interface ConfigExportData {
   providers: ProviderExport[]
   ldap_config?: LDAPConfigExport | null
   oauth_providers?: OAuthProviderExport[]
+  system_configs?: SystemConfigExport[]
 }
 
 // 用户导出数据结构
@@ -54,6 +61,7 @@ export interface UsersExportData {
 
 export interface UserExport {
   email: string
+  email_verified?: boolean
   username: string
   password_hash: string
   role: string
@@ -69,6 +77,7 @@ export interface UserExport {
 }
 
 export interface UserApiKeyExport {
+  key?: string | null
   key_hash: string
   key_encrypted?: string | null
   name?: string | null
@@ -105,14 +114,18 @@ export interface ProviderExport {
   name: string
   description?: string | null
   website?: string | null
+  provider_type?: string
   billing_type?: string | null
   monthly_quota_usd?: number | null
   quota_reset_day?: number
-  rpm_limit?: number | null
   provider_priority?: number
+  keep_priority_on_conversion?: boolean
+  enable_format_conversion?: boolean
   is_active: boolean
   concurrent_limit?: number | null
   max_retries?: number | null
+  stream_first_byte_timeout?: number | null
+  request_timeout?: number | null
   proxy?: Record<string, unknown>
   config?: Record<string, unknown>
   endpoints: EndpointExport[]
@@ -123,19 +136,24 @@ export interface ProviderExport {
 export interface EndpointExport {
   api_format: string
   base_url: string
-  headers?: Record<string, unknown>
+  header_rules?: Record<string, unknown>[] | null
+  body_rules?: Record<string, unknown>[] | null
   max_retries?: number
   is_active: boolean
   custom_path?: string | null
   config?: Record<string, unknown>
+  format_acceptance_config?: Record<string, unknown> | null
   proxy?: Record<string, unknown>
 }
 
 export interface ProviderKeyExport {
   api_key: string
+  auth_type?: string
+  auth_config?: string | Record<string, unknown> | null
   name?: string | null
   note?: string | null
   api_formats: string[]
+  supported_endpoints?: string[]
   rate_multipliers?: Record<string, number> | null
   internal_priority?: number
   global_priority_by_format?: Record<string, number> | null
@@ -144,7 +162,13 @@ export interface ProviderKeyExport {
   capabilities?: Record<string, boolean>
   cache_ttl_minutes?: number
   max_probe_interval_minutes?: number
+  auto_fetch_models?: boolean
+  locked_models?: string[] | null
+  model_include_patterns?: string[] | null
+  model_exclude_patterns?: string[] | null
   is_active: boolean
+  proxy?: Record<string, unknown> | null
+  fingerprint?: Record<string, unknown> | null
 }
 
 export interface ModelExport {

frontend/src/api/endpoints/providers.ts

@@ -131,7 +131,9 @@ export interface TestModelResponse {
 }
 
 export async function testModel(data: TestModelRequest): Promise<TestModelResponse> {
-  const response = await client.post('/api/admin/provider-query/test-model', data)
+  const response = await client.post('/api/admin/provider-query/test-model', data, {
+    timeout: 10 * 60 * 1000,
+  })
   return response.data
 }
 
@@ -145,10 +147,12 @@ export interface TestModelFailoverRequest {
   api_format?: string
   endpoint_id?: string
   message?: string
+  request_id?: string
 }
 
 export interface TestAttemptDetail {
   candidate_index: number
+  retry_index?: number
   endpoint_api_format: string
   endpoint_base_url: string
   key_name: string | null
@@ -174,7 +178,9 @@ export interface TestModelFailoverResponse {
 }
 
 export async function testModelFailover(data: TestModelFailoverRequest): Promise<TestModelFailoverResponse> {
-  const response = await client.post('/api/admin/provider-query/test-model-failover', data)
+  const response = await client.post('/api/admin/provider-query/test-model-failover', data, {
+    timeout: 10 * 60 * 1000,
+  })
   return response.data
 }
 

frontend/src/api/requestTrace.ts

@@ -12,6 +12,7 @@ export interface CandidateRecord {
   endpoint_name?: string  // 端点显示名称（api_format）
   key_id?: string
   key_name?: string  // 密钥名称
+  key_account_label?: string  // 更适合展示的测试账号标签（优先 OAuth 邮箱）
   key_preview?: string  // 密钥脱敏预览（如 sk-***abc），OAuth 类型不返回
   key_auth_type?: string  // 密钥认证类型（api_key, service_account, oauth 等）
   key_oauth_plan_type?: string  // OAuth 账号套餐类型（free/plus/team/enterprise）

frontend/src/features/providers/components/provider-tabs/ModelTestDialog.vue

@@ -41,22 +41,138 @@
 
     <div
       v-else-if="testing"
-      class="flex flex-col items-center justify-center gap-3 py-10 text-center"
+      class="space-y-4 py-6"
     >
-      <Loader2 class="w-8 h-8 animate-spin text-primary" />
-      <div class="space-y-1">
-        <p class="text-sm font-medium">
-          正在测试模型
-        </p>
-        <p class="text-xs text-muted-foreground">
-          {{ selectingModelName || '-' }}
-        </p>
-        <p
-          v-if="selectedEndpoint"
-          class="text-xs text-muted-foreground"
+      <div class="flex flex-col items-center justify-center gap-3 text-center">
+        <Loader2 class="w-8 h-8 animate-spin text-primary" />
+        <div class="space-y-1">
+          <p class="text-sm font-medium">
+            正在测试模型
+          </p>
+          <p class="text-xs text-muted-foreground">
+            {{ selectingModelName || '-' }}
+          </p>
+          <p
+            v-if="selectedEndpoint"
+            class="text-xs text-muted-foreground"
+          >
+            端点：{{ formatApiFormat(selectedEndpoint.api_format) }} · {{ selectedEndpoint.base_url }}
+          </p>
+        </div>
+      </div>
+
+      <div class="rounded-lg border border-border/60 bg-muted/20 p-4 space-y-4">
+        <div class="space-y-2">
+          <div class="flex items-center justify-between gap-3 text-xs text-muted-foreground">
+            <span>实时进度</span>
+            <span>{{ liveTraceSummary.completed }}/{{ liveTraceSummary.total || 0 }}</span>
+          </div>
+          <div class="h-2 overflow-hidden rounded-full bg-muted">
+            <div
+              class="h-full bg-primary transition-all duration-300"
+              :style="{ width: `${liveProgressPercent}%` }"
+            />
+          </div>
+          <div class="flex flex-wrap gap-1.5">
+            <Badge
+              variant="secondary"
+              class="text-[10px] px-1.5 py-0"
+            >
+              待执行 {{ liveTraceSummary.available }}
+            </Badge>
+            <Badge
+              variant="outline"
+              class="text-[10px] px-1.5 py-0"
+            >
+              进行中 {{ liveTraceSummary.pending }}
+            </Badge>
+            <Badge
+              variant="success"
+              class="text-[10px] px-1.5 py-0"
+            >
+              成功 {{ liveTraceSummary.success }}
+            </Badge>
+            <Badge
+              variant="destructive"
+              class="text-[10px] px-1.5 py-0"
+            >
+              失败 {{ liveTraceSummary.failed }}
+            </Badge>
+            <Badge
+              variant="secondary"
+              class="text-[10px] px-1.5 py-0"
+            >
+              跳过 {{ liveTraceSummary.skipped }}
+            </Badge>
+          </div>
+        </div>
+
+        <div class="grid gap-3 sm:grid-cols-2">
+          <div class="rounded-md border border-border/60 bg-background/80 p-3 space-y-1">
+            <div class="text-xs text-muted-foreground">
+              测试账号
+            </div>
+            <div class="text-sm font-medium break-all">
+              {{ liveAccountTitle }}
+            </div>
+            <div class="text-xs text-muted-foreground break-all">
+              {{ liveAccountMeta }}
+            </div>
+          </div>
+          <div class="rounded-md border border-border/60 bg-background/80 p-3 space-y-1">
+            <div class="text-xs text-muted-foreground">
+              实时状态
+            </div>
+            <div class="text-sm font-medium">
+              {{ liveStatusTitle }}
+            </div>
+            <div class="text-xs text-muted-foreground break-all">
+              {{ liveStatusDetail }}
+            </div>
+          </div>
+        </div>
+
+        <div
+          v-if="requestId"
+          class="text-[11px] text-muted-foreground break-all"
         >
-          端点：{{ formatApiFormat(selectedEndpoint.api_format) }} · {{ selectedEndpoint.base_url }}
-        </p>
+          请求 ID：<code class="bg-muted px-1 py-0.5 rounded">{{ requestId }}</code>
+        </div>
+
+        <div
+          v-if="liveRecentCandidates.length > 0"
+          class="space-y-2"
+        >
+          <div class="text-xs font-medium text-muted-foreground">
+            最近状态
+          </div>
+          <div class="space-y-2">
+            <div
+              v-for="candidate in liveRecentCandidates"
+              :key="`${candidate.id}-${candidate.status}`"
+              class="flex items-start justify-between gap-3 rounded-md border border-border/50 bg-background/70 px-3 py-2 text-xs"
+            >
+              <div class="min-w-0 space-y-1">
+                <div class="flex items-center gap-2 min-w-0">
+                  <span class="text-muted-foreground shrink-0">{{ formatTraceCandidateIndex(candidate) }}</span>
+                  <Badge
+                    :variant="statusVariant(candidate.status)"
+                    class="text-[10px] px-1.5 py-0 shrink-0"
+                  >
+                    {{ candidate.status_code || statusLabel(candidate.status) }}
+                  </Badge>
+                  <span class="truncate font-medium">{{ formatTraceCandidateAccount(candidate) }}</span>
+                </div>
+                <div class="text-muted-foreground break-all">
+                  {{ traceCandidateDetail(candidate) }}
+                </div>
+              </div>
+              <div class="shrink-0 text-muted-foreground tabular-nums">
+                {{ candidate.latency_ms != null ? `${candidate.latency_ms}ms` : '' }}
+              </div>
+            </div>
+          </div>
+        </div>
       </div>
     </div>
 
@@ -118,7 +234,7 @@
         >
           <div class="flex items-center justify-between gap-2">
             <div class="flex items-center gap-1.5 min-w-0">
-              <span class="text-muted-foreground shrink-0">#{{ attempt.candidate_index }}</span>
+              <span class="text-muted-foreground shrink-0">{{ formatAttemptIndex(attempt) }}</span>
               <Badge
                 :variant="statusVariant(attempt.status)"
                 class="text-[10px] px-1.5 py-0 shrink-0"
@@ -209,7 +325,7 @@
               :class="attemptRowClass(attempt.status)"
             >
               <td class="pl-3 pr-1 py-2 text-muted-foreground">
-                {{ attempt.candidate_index }}
+                {{ formatAttemptIndex(attempt) }}
               </td>
               <td class="px-3 py-2">
                 <div
@@ -292,6 +408,7 @@ import { Dialog, Badge } from '@/components/ui'
 import Button from '@/components/ui/button.vue'
 import { formatApiFormat } from '@/api/endpoints/types/api-format'
 import type { TestModelFailoverResponse, TestAttemptDetail } from '@/api/endpoints/providers'
+import type { CandidateRecord, RequestTrace } from '@/api/requestTrace'
 
 type TestEndpointOption = {
   id: string
@@ -308,6 +425,8 @@ const props = defineProps<{
   endpoints?: TestEndpointOption[]
   selectedEndpoint?: TestEndpointOption | null
   testing?: boolean
+  trace?: RequestTrace | null
+  requestId?: string | null
   showEndpointSelector?: boolean
 }>()
 
@@ -318,6 +437,7 @@ const emit = defineEmits<{
 }>()
 
 const endpoints = computed(() => props.endpoints ?? [])
+const traceCandidates = computed(() => props.trace?.candidates ?? [])
 const showSelection = computed(() => props.open && !!props.showEndpointSelector && !props.testing && !props.result)
 const showResult = computed(() => !!props.result)
 const canReselect = computed(() => !!props.showEndpointSelector && endpoints.value.length > 1)
@@ -354,6 +474,82 @@ const hasEffectiveModel = computed(() => {
   return props.result.attempts.some(a => a.effective_model && a.effective_model !== props.result?.model)
 })
 
+const liveTraceSummary = computed(() => {
+  const summary = {
+    total: traceCandidates.value.length,
+    available: 0,
+    pending: 0,
+    success: 0,
+    failed: 0,
+    skipped: 0,
+    completed: 0,
+  }
+
+  for (const candidate of traceCandidates.value) {
+    if (candidate.status === 'available' || candidate.status === 'unused') summary.available += 1
+    if (candidate.status === 'pending' || candidate.status === 'streaming') summary.pending += 1
+    if (candidate.status === 'success') summary.success += 1
+    if (candidate.status === 'failed' || candidate.status === 'cancelled' || candidate.status === 'stream_interrupted') summary.failed += 1
+    if (candidate.status === 'skipped') summary.skipped += 1
+  }
+
+  summary.completed = summary.success + summary.failed + summary.skipped
+  return summary
+})
+
+const liveProgressPercent = computed(() => {
+  if (liveTraceSummary.value.total <= 0) return 6
+  const raw = Math.round((liveTraceSummary.value.completed / liveTraceSummary.value.total) * 100)
+  return Math.min(100, Math.max(raw, liveTraceSummary.value.pending > 0 ? 12 : 6))
+})
+
+const activeTraceCandidate = computed(() => {
+  const preferredStatuses = ['pending', 'streaming', 'failed', 'success', 'skipped', 'cancelled']
+  for (let index = traceCandidates.value.length - 1; index >= 0; index -= 1) {
+    const candidate = traceCandidates.value[index]
+    if (preferredStatuses.includes(candidate.status)) return candidate
+  }
+  return traceCandidates.value[0] ?? null
+})
+
+const liveAccountTitle = computed(() => {
+  const candidate = activeTraceCandidate.value
+  if (!candidate) return '等待分配测试账号'
+  return candidate.key_account_label || candidate.key_name || candidate.key_preview || '等待分配测试账号'
+})
+
+const liveAccountMeta = computed(() => {
+  const candidate = activeTraceCandidate.value
+  if (!candidate) return '候选创建后会显示测试账号和认证方式'
+  const parts: string[] = []
+  if (candidate.key_auth_type) parts.push(formatAuthType(candidate.key_auth_type))
+  if (candidate.key_oauth_plan_type) parts.push(candidate.key_oauth_plan_type)
+  if (candidate.key_preview && candidate.key_preview !== candidate.key_account_label) parts.push(candidate.key_preview)
+  return parts.join(' · ') || '正在等待候选进入执行阶段'
+})
+
+const liveStatusTitle = computed(() => {
+  const candidate = activeTraceCandidate.value
+  if (!candidate) return '正在创建测试请求'
+  if (candidate.status === 'pending' || candidate.status === 'streaming') {
+    return `正在测试 ${formatTraceCandidateIndex(candidate)}`
+  }
+  return statusLabel(candidate.status)
+})
+
+const liveStatusDetail = computed(() => {
+  const candidate = activeTraceCandidate.value
+  if (!candidate) return '等待后端写入候选状态'
+  return traceCandidateDetail(candidate)
+})
+
+const liveRecentCandidates = computed(() => {
+  return traceCandidates.value
+    .filter(candidate => !['available', 'unused'].includes(candidate.status))
+    .slice(-4)
+    .reverse()
+})
+
 function statusVariant(status: string) {
   if (status === 'success') return 'success' as const
   if (status === 'failed') return 'destructive' as const
@@ -364,6 +560,11 @@ function statusLabel(status: string) {
   if (status === 'success') return '成功'
   if (status === 'failed') return '失败'
   if (status === 'skipped') return '跳过'
+  if (status === 'pending') return '等待中'
+  if (status === 'streaming') return '测试中'
+  if (status === 'cancelled') return '已取消'
+  if (status === 'stream_interrupted') return '流中断'
+  if (status === 'available') return '待执行'
   return status
 }
 
@@ -379,6 +580,37 @@ function maskKey(key: string): string {
   return `${key.slice(0, 4)}...${key.slice(-4)}`
 }
 
+function formatAuthType(authType: string): string {
+  const lowered = authType.toLowerCase()
+  if (lowered === 'api_key') return 'API Key'
+  if (lowered === 'service_account') return 'Service Account'
+  if (lowered === 'oauth') return 'OAuth'
+  if (lowered === 'codex') return 'Codex OAuth'
+  if (lowered === 'antigravity') return 'Antigravity OAuth'
+  if (lowered === 'kiro') return 'Kiro OAuth'
+  return authType
+}
+
+function formatAttemptIndex(attempt: TestAttemptDetail): string {
+  const retryIndex = attempt.retry_index ?? 0
+  return retryIndex > 0 ? `#${attempt.candidate_index}.${retryIndex}` : `#${attempt.candidate_index}`
+}
+
+function formatTraceCandidateIndex(candidate: CandidateRecord): string {
+  return candidate.retry_index > 0 ? `#${candidate.candidate_index}.${candidate.retry_index}` : `#${candidate.candidate_index}`
+}
+
+function formatTraceCandidateAccount(candidate: CandidateRecord): string {
+  return candidate.key_account_label || candidate.key_name || candidate.key_preview || '待分配账号'
+}
+
+function traceCandidateDetail(candidate: CandidateRecord): string {
+  if (candidate.skip_reason) return candidate.skip_reason
+  if (candidate.error_message) return candidate.error_message
+  if (candidate.endpoint_name) return `端点：${formatApiFormat(candidate.endpoint_name)}`
+  return '等待响应中…'
+}
+
 function attemptDetail(attempt: TestAttemptDetail): string {
   if (attempt.skip_reason) return attempt.skip_reason
   if (attempt.error_message) return attempt.error_message

frontend/src/features/providers/components/provider-tabs/ModelsTab.vue

@@ -219,6 +219,8 @@
     :endpoints="activeEndpoints"
     :selected-endpoint="selectedTestEndpoint"
     :testing="!!pendingTestModel && testingModelId === pendingTestModel.id"
+    :trace="testTrace"
+    :request-id="currentTestRequestId"
     :show-endpoint-selector="activeEndpoints.length > 1"
     @close="handleTestDialogClose"
     @back="handleTestDialogBack"
@@ -227,7 +229,8 @@
 </template>
 
 <script setup lang="ts">
-import { ref, computed } from 'vue'
+import { ref, computed, onBeforeUnmount } from 'vue'
+import { isAxiosError } from 'axios'
 import { useSmartPagination } from '@/composables/useSmartPagination'
 import { Box, Edit, Layers, Power, Copy, Loader2, Play } from 'lucide-vue-next'
 import Card from '@/components/ui/card.vue'
@@ -242,6 +245,7 @@ import {
   type TestModelFailoverResponse,
 } from '@/api/endpoints'
 import { updateModel } from '@/api/endpoints/models'
+import { requestTraceApi, type RequestTrace } from '@/api/requestTrace'
 import { parseApiError } from '@/utils/errorParser'
 import { formatApiFormat } from '@/api/endpoints/types/api-format'
 import type { ProviderWithEndpointsSummary } from '@/api/endpoints'
@@ -272,6 +276,10 @@ const testResultMode = ref<'global' | 'direct'>('global')
 const testDialogOpen = ref(false)
 const pendingTestModel = ref<Model | null>(null)
 const selectedTestEndpoint = ref<ProviderEndpoint | null>(null)
+const currentTestRequestId = ref<string | null>(null)
+const testTrace = ref<RequestTrace | null>(null)
+let tracePollTimer: ReturnType<typeof setInterval> | null = null
+let tracePollToken = 0
 // 使用 props 传入的数据，或使用本地数据
 const activeEndpoints = computed(() => (props.endpoints ?? []).filter(endpoint => endpoint.is_active))
 // 使用 props 传入的数据，或使用本地数据
@@ -304,6 +312,47 @@ function refresh() {
   emit('refresh')
 }
 
+function buildTestRequestId(): string {
+  const randomUUID = globalThis.crypto?.randomUUID?.bind(globalThis.crypto)
+  if (randomUUID) {
+    return `provider-test-${randomUUID().replace(/-/g, '').slice(0, 20)}`
+  }
+  return `provider-test-${Date.now().toString(36)}${Math.random().toString(36).slice(2, 10)}`
+}
+
+async function pollTestTrace(requestId: string, token: number) {
+  try {
+    const trace = await requestTraceApi.getRequestTrace(requestId, { attemptedOnly: false })
+    if (tracePollToken !== token || currentTestRequestId.value !== requestId) return
+    testTrace.value = trace
+  } catch (err: unknown) {
+    if (isAxiosError(err) && err.response?.status === 404) return
+  }
+}
+
+function stopTestTracePolling(options: { clearState?: boolean } = {}) {
+  tracePollToken += 1
+  if (tracePollTimer) {
+    clearInterval(tracePollTimer)
+    tracePollTimer = null
+  }
+  if (options.clearState !== false) {
+    currentTestRequestId.value = null
+    testTrace.value = null
+  }
+}
+
+function startTestTracePolling(requestId: string) {
+  stopTestTracePolling()
+  currentTestRequestId.value = requestId
+  testTrace.value = null
+  const token = ++tracePollToken
+  void pollTestTrace(requestId, token)
+  tracePollTimer = setInterval(() => {
+    void pollTestTrace(requestId, token)
+  }, 800)
+}
+
 // 格式化价格显示
 function formatPrice(price: number | null | undefined): string {
   if (price === null || price === undefined) return '-'
@@ -437,6 +486,7 @@ async function toggleModelActive(model: Model) {
 }
 
 function resetTestDialogState() {
+  stopTestTracePolling()
   testDialogOpen.value = false
   pendingTestModel.value = null
   selectedTestEndpoint.value = null
@@ -466,6 +516,8 @@ async function runModelTest(model: Model, endpoint?: ProviderEndpoint) {
   testingModelId.value = model.id
   testDialogOpen.value = true
   selectedTestEndpoint.value = endpoint ?? null
+  const requestId = buildTestRequestId()
+  startTestTracePolling(requestId)
   try {
     const modelName = model.global_model_name || model.provider_model_name
 
@@ -476,6 +528,7 @@ async function runModelTest(model: Model, endpoint?: ProviderEndpoint) {
       api_format: endpoint?.api_format,
       endpoint_id: endpoint?.id,
       message: 'hello',
+      request_id: requestId,
     })
 
     if (result.success) {
@@ -489,9 +542,11 @@ async function runModelTest(model: Model, endpoint?: ProviderEndpoint) {
       resetTestDialogState()
       return
     }
+    stopTestTracePolling({ clearState: false })
     testResultMode.value = 'global'
     testResult.value = result
   } catch (err: unknown) {
+    stopTestTracePolling()
     showError(`模型测试失败: ${parseApiError(err, '测试请求失败')}`)
     if (activeEndpoints.value.length <= 1) {
       resetTestDialogState()
@@ -526,4 +581,8 @@ async function testModelConnection(model: Model) {
 defineExpose({
   reload: refresh
 })
+
+onBeforeUnmount(() => {
+  stopTestTracePolling()
+})
 </script>

frontend/src/features/providers/components/provider-tabs/TestResultDialog.vue

@@ -58,7 +58,7 @@
         >
           <div class="flex items-center justify-between gap-2">
             <div class="flex items-center gap-1.5 min-w-0">
-              <span class="text-muted-foreground shrink-0">#{{ attempt.candidate_index }}</span>
+              <span class="text-muted-foreground shrink-0">{{ formatAttemptIndex(attempt) }}</span>
               <Badge
                 :variant="statusVariant(attempt.status)"
                 class="text-[10px] px-1.5 py-0 shrink-0"
@@ -154,7 +154,7 @@
               :class="attemptRowClass(attempt.status)"
             >
               <td class="pl-3 pr-1 py-2 text-muted-foreground">
-                {{ attempt.candidate_index }}
+                {{ formatAttemptIndex(attempt) }}
               </td>
               <td class="px-3 py-2">
                 <div
@@ -267,6 +267,11 @@ function statusLabel(status: string) {
   if (status === 'success') return '成功'
   if (status === 'failed') return '失败'
   if (status === 'skipped') return '跳过'
+  if (status === 'pending') return '等待中'
+  if (status === 'streaming') return '测试中'
+  if (status === 'cancelled') return '已取消'
+  if (status === 'stream_interrupted') return '流中断'
+  if (status === 'available') return '待执行'
   return status
 }
 
@@ -282,6 +287,11 @@ function maskKey(key: string): string {
   return `${key.slice(0, 4)}...${key.slice(-4)}`
 }
 
+function formatAttemptIndex(attempt: TestAttemptDetail): string {
+  const retryIndex = attempt.retry_index ?? 0
+  return retryIndex > 0 ? `#${attempt.candidate_index}.${retryIndex}` : `#${attempt.candidate_index}`
+}
+
 function attemptDetail(attempt: TestAttemptDetail): string {
   if (attempt.skip_reason) return attempt.skip_reason
   if (attempt.error_message) return attempt.error_message

src/api/admin/monitoring/trace.py

@@ -39,6 +39,7 @@ class CandidateResponse(BaseModel):
     endpoint_name: str | None = None  # 端点显示名称（api_format）
     key_id: str | None = None
     key_name: str | None = None  # 密钥名称
+    key_account_label: str | None = None  # 更适合展示的测试账号标签（优先 OAuth 邮箱）
     key_preview: str | None = None  # 密钥脱敏预览（如 sk-***abc），OAuth 类型不返回
     key_auth_type: str | None = None  # 密钥认证类型（api_key, service_account, oauth）
     key_oauth_plan_type: str | None = None  # OAuth 账号套餐类型（free/plus/team/enterprise）
@@ -257,6 +258,7 @@ class AdminGetRequestTraceAdapter(AdminApiAdapter):
         key_ids = {c.key_id for c in candidates if c.key_id}
         key_map: dict[str, str] = {}
         key_preview_map: dict[str, str] = {}
+        key_account_label_map: dict[str, str | None] = {}
         key_capabilities_map: dict[str, dict | None] = {}
         key_auth_type_map: dict[str, str] = {}
         key_oauth_plan_map: dict[str, str | None] = {}
@@ -268,6 +270,7 @@ class AdminGetRequestTraceAdapter(AdminApiAdapter):
             keys = db.query(ProviderAPIKey).filter(ProviderAPIKey.id.in_(key_ids)).all()
             for k in keys:
                 key_map[k.id] = k.name
+                key_account_label_map[k.id] = k.name
                 key_capabilities_map[k.id] = k.capabilities
 
                 is_oauth = k.auth_type == "oauth"
@@ -286,6 +289,9 @@ class AdminGetRequestTraceAdapter(AdminApiAdapter):
                         try:
                             decrypted_config = crypto_service.decrypt(k.auth_config)
                             auth_config = json.loads(decrypted_config)
+                            email = auth_config.get("email")
+                            if isinstance(email, str) and email.strip():
+                                key_account_label_map[k.id] = email.strip()
                             oauth_plan_type = auth_config.get("plan_type")
                             if not oauth_plan_type:
                                 ag_tier = auth_config.get("tier")
@@ -348,6 +354,9 @@ class AdminGetRequestTraceAdapter(AdminApiAdapter):
                 endpoint_map.get(candidate.endpoint_id) if candidate.endpoint_id else None
             )
             key_name = key_map.get(candidate.key_id) if candidate.key_id else None
+            key_account_label = (
+                key_account_label_map.get(candidate.key_id) if candidate.key_id else None
+            )
             key_preview = key_preview_map.get(candidate.key_id) if candidate.key_id else None
             key_auth_type = key_auth_type_map.get(candidate.key_id) if candidate.key_id else None
             key_oauth_plan_type = (
@@ -370,6 +379,7 @@ class AdminGetRequestTraceAdapter(AdminApiAdapter):
                     endpoint_name=endpoint_name,
                     key_id=candidate.key_id,
                     key_name=key_name,
+                    key_account_label=key_account_label,
                     key_preview=key_preview,
                     key_auth_type=key_auth_type,
                     key_oauth_plan_type=key_oauth_plan_type,

src/api/admin/provider_query.py

@@ -7,9 +7,10 @@ from __future__ import annotations
 
 import asyncio
 import json
-import time
 from typing import TYPE_CHECKING, Any
+from uuid import uuid4
 
+import httpx
 from fastapi import APIRouter, Depends, HTTPException
 from pydantic import BaseModel
 from sqlalchemy.orm import Session, joinedload
@@ -207,12 +208,14 @@ class TestModelFailoverRequest(BaseModel):
     api_format: str | None = None  # 指定 API 格式（endpoint signature）
     endpoint_id: str | None = None  # 指定仅使用该端点测试
     message: str | None = "Hello"
+    request_id: str | None = None
 
 
 class TestAttemptDetail(BaseModel):
     """单次测试尝试的详情"""
 
     candidate_index: int
+    retry_index: int = 0
     endpoint_api_format: str
     endpoint_base_url: str
     key_name: str | None = None
@@ -1180,6 +1183,264 @@ def _filter_test_candidates_by_endpoint(
     ]
 
 
+def _parse_jsonish(value: Any) -> Any:
+    if isinstance(value, str):
+        try:
+            return json.loads(value)
+        except (json.JSONDecodeError, TypeError, ValueError):
+            return value
+    return value
+
+
+def _resolve_test_effective_model(
+    *,
+    provider: Provider,
+    candidate: Any,
+    request: TestModelFailoverRequest,
+    gm_obj: Any,
+    key: Any | None = None,
+) -> str:
+    effective_model = request.model_name
+    if request.mode != "global":
+        return effective_model
+
+    current_key = key or getattr(candidate, "key", None)
+    pool_mapping = (
+        getattr(current_key, "_pool_mapping_matched_model", None) if current_key else None
+    )
+    mapping_matched_model = pool_mapping or getattr(candidate, "mapping_matched_model", None)
+    if mapping_matched_model:
+        return str(mapping_matched_model)
+
+    if not gm_obj:
+        return effective_model
+
+    gm_id_str = str(gm_obj.id)
+    endpoint = getattr(candidate, "endpoint", None)
+    ep_format = str(getattr(endpoint, "api_format", "") or "")
+    for model in provider.models or []:
+        if not getattr(model, "is_active", False):
+            continue
+        if str(getattr(model, "global_model_id", "") or "") != gm_id_str:
+            continue
+        selected = model.select_provider_model_name(affinity_key=None, api_format=ep_format)
+        if selected:
+            return str(selected)
+    return effective_model
+
+
+def _build_test_candidate_meta(
+    *,
+    candidates: list[ProviderCandidate],
+    provider: Provider,
+    request: TestModelFailoverRequest,
+    gm_obj: Any,
+) -> tuple[dict[tuple[int, str], dict[str, Any]], dict[int, dict[str, Any]]]:
+    from src.services.scheduling.schemas import PoolCandidate
+
+    by_pair: dict[tuple[int, str], dict[str, Any]] = {}
+    by_candidate: dict[int, dict[str, Any]] = {}
+
+    for candidate_index, candidate in enumerate(candidates):
+        endpoint = candidate.endpoint
+        base_meta = {
+            "endpoint_api_format": str(getattr(endpoint, "api_format", "") or ""),
+            "endpoint_base_url": str(getattr(endpoint, "base_url", "") or "")[:80],
+            "effective_model": _resolve_test_effective_model(
+                provider=provider,
+                candidate=candidate,
+                request=request,
+                gm_obj=gm_obj,
+            ),
+        }
+        by_candidate[candidate_index] = base_meta
+
+        key = getattr(candidate, "key", None)
+        if key is not None and getattr(key, "id", None):
+            by_pair[(candidate_index, str(key.id))] = dict(base_meta)
+
+        if isinstance(candidate, PoolCandidate):
+            for pool_key in candidate.pool_keys or []:
+                if not getattr(pool_key, "id", None):
+                    continue
+                by_pair[(candidate_index, str(pool_key.id))] = {
+                    "endpoint_api_format": base_meta["endpoint_api_format"],
+                    "endpoint_base_url": base_meta["endpoint_base_url"],
+                    "effective_model": _resolve_test_effective_model(
+                        provider=provider,
+                        candidate=candidate,
+                        request=request,
+                        gm_obj=gm_obj,
+                        key=pool_key,
+                    ),
+                }
+
+    return by_pair, by_candidate
+
+
+def _maybe_mark_test_oauth_key_invalid(
+    *,
+    db: Session,
+    key: Any,
+    auth_type: str,
+    error_payload: Any,
+) -> None:
+    if auth_type != "oauth" or not isinstance(error_payload, dict):
+        return
+
+    error_obj = error_payload.get("error")
+    if not isinstance(error_obj, dict):
+        return
+
+    error_message = str(error_obj.get("message", "") or "")
+    if error_obj.get("code") != 403:
+        return
+    if (
+        "verify" not in error_message.lower()
+        and "permission" not in str(error_obj.get("status", "") or "").lower()
+    ):
+        return
+
+    from datetime import datetime, timezone
+
+    from src.services.provider.oauth_token import OAUTH_ACCOUNT_BLOCK_PREFIX
+
+    key.oauth_invalid_at = datetime.now(timezone.utc)
+    key.oauth_invalid_reason = f"{OAUTH_ACCOUNT_BLOCK_PREFIX}Google 要求验证账号"
+    key.is_active = False
+    db.commit()
+
+
+def _extract_test_response_or_raise(
+    *,
+    response: dict[str, Any],
+    endpoint: Any,
+    provider_name: str,
+    auth_type: str,
+    api_key: Any,
+    db: Session,
+) -> dict[str, Any]:
+    status_code = int(response.get("status_code", 0) or 0)
+    response_payload = response.get("response", {})
+    parsed_payload = _parse_jsonish(response_payload)
+    if isinstance(parsed_payload, dict) and "response_body" in parsed_payload:
+        parsed_payload = _parse_jsonish(parsed_payload.get("response_body"))
+
+    if isinstance(parsed_payload, dict) and "error" in parsed_payload:
+        _maybe_mark_test_oauth_key_invalid(
+            db=db,
+            key=api_key,
+            auth_type=auth_type,
+            error_payload=parsed_payload,
+        )
+        error_obj = parsed_payload["error"]
+        error_code = error_obj.get("code") if isinstance(error_obj, dict) else status_code or 500
+        error_message = (
+            error_obj.get("message") if isinstance(error_obj, dict) else str(error_obj or "")
+        )
+        error_status = error_obj.get("status") if isinstance(error_obj, dict) else None
+        from src.core.exceptions import EmbeddedErrorException
+
+        raise EmbeddedErrorException(
+            provider_name=provider_name,
+            error_code=int(error_code) if error_code else None,
+            error_message=str(error_message or ""),
+            error_status=str(error_status) if error_status else None,
+        )
+
+    if status_code == 200 and not response.get("error"):
+        return parsed_payload if isinstance(parsed_payload, dict) else response_payload
+
+    error_meta = response_payload if isinstance(response_payload, dict) else {}
+    error_type = str(error_meta.get("error_type", "") or "")
+    error_message = str(response.get("error", "") or "")
+    if not error_message and isinstance(parsed_payload, dict):
+        embedded_error = parsed_payload.get("error")
+        if isinstance(embedded_error, dict):
+            error_message = str(embedded_error.get("message", "") or "")
+        elif embedded_error:
+            error_message = str(embedded_error)
+    if not error_message and isinstance(parsed_payload, str):
+        error_message = parsed_payload
+    if not error_message and status_code:
+        error_message = f"HTTP {status_code}"
+
+    request_obj = httpx.Request("POST", str(getattr(endpoint, "base_url", "") or ""))
+    if status_code > 0:
+        body_text = error_message[:4000] if error_message else ""
+        synthetic_response = httpx.Response(
+            status_code=status_code,
+            request=request_obj,
+            text=body_text,
+            headers=response.get("headers", {}),
+        )
+        http_error = httpx.HTTPStatusError(
+            message=body_text or f"HTTP {status_code}",
+            request=request_obj,
+            response=synthetic_response,
+        )
+        http_error.upstream_response = body_text  # type: ignore[attr-defined]
+        raise http_error
+
+    if error_type == "timeout":
+        raise httpx.TimeoutException(error_message or "Request timeout")
+
+    if error_type in {"network_error", "connection_failed"}:
+        raise httpx.ConnectError(error_message or "Connection failed", request=request_obj)
+
+    from src.core.exceptions import ProviderNotAvailableException
+
+    raise ProviderNotAvailableException(
+        error_message or "服务暂时不可用，请稍后重试",
+        provider_name=provider_name,
+        upstream_response=error_message or None,
+    )
+
+
+def _build_test_attempts_from_candidate_keys(
+    *,
+    candidate_keys: list[Any],
+    candidate_meta_by_pair: dict[tuple[int, str], dict[str, Any]],
+    candidate_meta_by_index: dict[int, dict[str, Any]],
+) -> list[TestAttemptDetail]:
+    attempts: list[TestAttemptDetail] = []
+
+    for candidate_key in candidate_keys:
+        status = str(getattr(candidate_key, "status", "") or "").strip().lower()
+        if status in {"", "available", "unused"}:
+            continue
+
+        candidate_index = int(getattr(candidate_key, "candidate_index", 0) or 0)
+        retry_index = int(getattr(candidate_key, "retry_index", 0) or 0)
+        key_id = str(getattr(candidate_key, "key_id", "") or "")
+        meta = candidate_meta_by_pair.get((candidate_index, key_id)) or candidate_meta_by_index.get(
+            candidate_index, {}
+        )
+
+        attempts.append(
+            TestAttemptDetail(
+                candidate_index=candidate_index,
+                retry_index=retry_index,
+                endpoint_api_format=str(meta.get("endpoint_api_format", "") or ""),
+                endpoint_base_url=str(meta.get("endpoint_base_url", "") or ""),
+                key_name=getattr(candidate_key, "key_name", None),
+                key_id=key_id,
+                auth_type=str(getattr(candidate_key, "auth_type", "") or ""),
+                effective_model=(
+                    str(meta.get("effective_model")) if meta.get("effective_model") else None
+                ),
+                status=status,
+                skip_reason=getattr(candidate_key, "skip_reason", None),
+                error_message=getattr(candidate_key, "error_message", None),
+                status_code=getattr(candidate_key, "status_code", None),
+                latency_ms=getattr(candidate_key, "latency_ms", None),
+            )
+        )
+
+    attempts.sort(key=lambda attempt: (attempt.candidate_index, attempt.retry_index))
+    return attempts
+
+
 @router.post("/test-model-failover")
 async def test_model_failover(
     request: TestModelFailoverRequest,
@@ -1193,11 +1454,13 @@ async def test_model_failover(
     - global: 模拟外部请求，用全局模型名走候选解析（限定当前 Provider）
     - direct: 直接测试 provider_model_name，在当前 Provider 内多 Key 故障转移
     """
-    from src.services.candidate.failover import FailoverEngine
-    from src.services.candidate.policy import RetryMode, RetryPolicy, SkipPolicy
-    from src.services.task.protocol import AttemptKind, AttemptResult
+    from src.core.exceptions import ProviderNotAvailableException
+    from src.services.candidate.recorder import CandidateRecorder
+    from src.services.scheduling.candidate_builder import CandidateBuilder
+    from src.services.scheduling.candidate_sorter import CandidateSorter
+    from src.services.scheduling.scheduling_config import SchedulingConfig
+    from src.services.task import TaskService
 
-    # 1. 加载 Provider
     provider = (
         db.query(Provider)
         .options(
@@ -1214,9 +1477,8 @@ async def test_model_failover(
     if request.mode not in ("global", "direct"):
         raise HTTPException(status_code=400, detail="mode must be 'global' or 'direct'")
 
-    # 2. 构建候选列表
-    candidates = []
-    gm_obj = None  # GlobalModel 对象，global 模式下用于 fallback 映射
+    candidates: list[ProviderCandidate] = []
+    gm_obj = None
     endpoint_by_id = {
         str(getattr(ep, "id", "") or ""): ep
         for ep in (provider.endpoints or [])
@@ -1231,21 +1493,14 @@ async def test_model_failover(
         if request.api_format and ep_format != request.api_format:
             raise HTTPException(status_code=400, detail="endpoint_id does not match api_format")
 
+    client_format = request.api_format
     if request.mode == "global":
-        # 模拟外部请求：走 CandidateBuilder 候选解析
-        from src.services.scheduling.candidate_builder import CandidateBuilder
-        from src.services.scheduling.candidate_sorter import CandidateSorter
-        from src.services.scheduling.scheduling_config import SchedulingConfig
-
         sorter = CandidateSorter(SchedulingConfig())
         builder = CandidateBuilder(sorter)
 
-        # 确定 client_format
-        client_format = request.api_format
         if not client_format and requested_endpoint is not None:
             client_format = str(getattr(requested_endpoint, "api_format", "") or "")
         if not client_format:
-            # 取第一个活跃端点的格式
             for ep in provider.endpoints or []:
                 if getattr(ep, "is_active", False):
                     client_format = str(getattr(ep, "api_format", "") or "")
@@ -1256,11 +1511,9 @@ async def test_model_failover(
                 status_code=400, detail="No active endpoint found to determine API format"
             )
 
-        # 从 GlobalModel 提取 model_mappings（正则映射规则，用于 Key.allowed_models 匹配）
         from src.services.cache.model_cache import ModelCacheService
 
         model_mappings: list[str] = []
-        gm_obj = None
         try:
             gm_obj = await ModelCacheService.get_global_model_by_name(db, request.model_name)
             if gm_obj and isinstance(gm_obj.config, dict):
@@ -1285,7 +1538,10 @@ async def test_model_failover(
             candidates = []
         candidates = _filter_test_candidates_by_endpoint(candidates, request.endpoint_id)
     else:
-        # 直接测试：简单匹配 Endpoint + Key
+        if not client_format and requested_endpoint is not None:
+            client_format = str(getattr(requested_endpoint, "api_format", "") or "")
+        if not client_format and provider.endpoints:
+            client_format = str(getattr(provider.endpoints[0], "api_format", "") or "")
         candidates = _build_direct_test_candidates(
             provider=provider,
             api_format=request.api_format,
@@ -1303,266 +1559,165 @@ async def test_model_failover(
             error="No available candidates found for this model",
         ).model_dump()
 
-    # 3. 定义 attempt_func
-    attempts: list[TestAttemptDetail] = []
-    p_type = str(getattr(provider, "provider_type", "") or "").lower()
+    request_payload = {
+        "model": request.model_name,
+        "messages": [{"role": "user", "content": request.message or "Hello"}],
+        "max_tokens": 30,
+        "temperature": 0.7,
+        "stream": True,
+    }
+    request_id = str(request.request_id or f"provider-test-{uuid4().hex[:12]}")
+    request_timeout = float(getattr(provider, "request_timeout", 0) or TimeoutDefaults.HTTP_REQUEST)
+    provider_type = str(getattr(provider, "provider_type", "") or "").lower()
 
-    async def _attempt_func(candidate: Any) -> AttemptResult:
-        start_time = time.monotonic()
-        endpoint = candidate.endpoint
-        key = candidate.key
-        candidate_idx = getattr(candidate, "_utf_candidate_index", 0)
+    async def _request_func(provider_obj: Any, endpoint: Any, key: Any, candidate: Any) -> Any:
+        effective_proxy = resolve_effective_proxy(
+            getattr(provider_obj, "proxy", None), getattr(key, "proxy", None)
+        )
+        try:
+            api_key_value, auth_config = await _resolve_key_auth(
+                key,
+                provider_obj,
+                provider_proxy_config=effective_proxy,
+            )
+        except _KeyAuthError as e:
+            raise RuntimeError(e.message) from e
 
         auth_type = str(getattr(key, "auth_type", "api_key") or "api_key").lower()
-        extra_headers: dict[str, str] = {}
-        oauth_meta: dict = {}
-        effective_model = request.model_name
-        attempt_recorded = False
+        extra_headers = get_extra_headers_from_endpoint(endpoint) or {}
+        if auth_type == "oauth":
+            account_id = (auth_config or {}).get("account_id")
+            if account_id:
+                extra_headers["chatgpt-account-id"] = str(account_id)
 
-        try:
-            # 解析 Key（复用统一的认证解析逻辑）
-            effective_proxy = resolve_effective_proxy(
-                getattr(provider, "proxy", None), getattr(key, "proxy", None)
-            )
-            try:
-                api_key_value, auth_config = await _resolve_key_auth(
-                    key, provider, provider_proxy_config=effective_proxy
-                )
-            except _KeyAuthError as e:
-                raise Exception(e.message) from e
-            oauth_meta = auth_config or {}
+        effective_model = _resolve_test_effective_model(
+            provider=provider,
+            candidate=candidate,
+            request=request,
+            gm_obj=gm_obj,
+            key=key,
+        )
+        adapter_class = get_adapter_for_format(endpoint.api_format)
+        if not adapter_class:
+            raise ValueError(f"Unknown API format: {endpoint.api_format}")
 
-            # OAuth 额外头
-            if auth_type == "oauth":
-                account_id = oauth_meta.get("account_id")
-                if account_id:
-                    extra_headers["chatgpt-account-id"] = str(account_id)
-
-            ep_extra = get_extra_headers_from_endpoint(endpoint) or {}
-            extra_headers.update(ep_extra)
-
-            # 确定实际模型名
-            effective_model = request.model_name
-            if request.mode == "global":
-                if candidate.mapping_matched_model:
-                    effective_model = candidate.mapping_matched_model
-                elif gm_obj:
-                    # Fallback: 从 Provider.Model.provider_model_mappings 获取映射
-                    # 与正常请求流程中 _get_mapped_model() 的逻辑一致
-                    gm_id_str = str(gm_obj.id)
-                    for m in provider.models or []:
-                        if not getattr(m, "is_active", False):
-                            continue
-                        if str(getattr(m, "global_model_id", "")) != gm_id_str:
-                            continue
-                        ep_format = str(getattr(endpoint, "api_format", "") or "")
-                        effective_model = m.select_provider_model_name(
-                            affinity_key=None, api_format=ep_format
-                        )
-                        logger.info(
-                            "[test-failover] Fallback mapping: {} -> {} "
-                            "(provider_model_name={}, has_provider_model_mappings={})",
-                            request.model_name,
-                            effective_model,
-                            m.provider_model_name,
-                            bool(m.provider_model_mappings),
-                        )
-                        break
-                    else:
-                        logger.info(
-                            "[test-failover] No matching Model found for gm_id={} in provider={}",
-                            gm_id_str,
-                            provider.name,
-                        )
-
-            # 获取 adapter
-            adapter_class = get_adapter_for_format(endpoint.api_format)
-            if not adapter_class:
-                raise Exception(f"Unknown API format: {endpoint.api_format}")
-
-            # 构建测试请求
-            check_request = {
+        response = await adapter_class.check_endpoint(
+            None,
+            endpoint.base_url,
+            api_key_value,
+            {
+                **request_payload,
                 "model": effective_model,
-                "messages": [{"role": "user", "content": request.message or "Hello"}],
-                "max_tokens": 30,
-                "temperature": 0.7,
-                "stream": True,
-            }
-
-            body_rules = getattr(endpoint, "body_rules", None)
-            header_rules = getattr(endpoint, "header_rules", None)
-
-            # 执行检查
-            response = await adapter_class.check_endpoint(
-                None,
-                endpoint.base_url,
-                api_key_value,
-                check_request,
-                extra_headers if extra_headers else None,
-                body_rules=body_rules,
-                header_rules=header_rules,
-                db=db,
-                user=current_user,
-                provider_name=provider.name,
-                provider_id=str(provider.id),
-                api_key_id=str(key.id),
-                model_name=effective_model,
-                auth_type=auth_type,
-                provider_type=p_type if p_type else None,
-                decrypted_auth_config=oauth_meta if oauth_meta else None,
-                provider_endpoint=endpoint,
-                provider_api_key=key,
-                proxy_config=effective_proxy,
-            )
-
-            latency_ms = int((time.monotonic() - start_time) * 1000)
-            status_code = response.get("status_code", 0)
-
-            # 检查响应是否有错误
-            has_error = bool(response.get("error")) or status_code != 200
-            if not has_error:
-                resp_data = response.get("response", {})
-                resp_body = resp_data.get("response_body", {})
-                if isinstance(resp_body, str):
-                    try:
-                        parsed = json.loads(resp_body)
-                    except (json.JSONDecodeError, ValueError):
-                        parsed = resp_body
-                else:
-                    parsed = resp_body
-                if isinstance(parsed, dict) and "error" in parsed:
-                    has_error = True
-
-            if has_error:
-                error_msg = str(response.get("error", ""))[:300]
-                if not error_msg and status_code != 200:
-                    error_msg = f"HTTP {status_code}"
-                if not error_msg and isinstance(parsed, dict) and "error" in parsed:
-                    err_val = parsed["error"]
-                    error_msg = str(
-                        err_val.get("message", err_val) if isinstance(err_val, dict) else err_val
-                    )[:300]
-                attempts.append(
-                    TestAttemptDetail(
-                        candidate_index=candidate_idx,
-                        endpoint_api_format=str(endpoint.api_format),
-                        endpoint_base_url=str(endpoint.base_url)[:80],
-                        key_name=getattr(key, "name", None),
-                        key_id=str(key.id),
-                        auth_type=auth_type,
-                        effective_model=effective_model,
-                        status="failed",
-                        error_message=error_msg,
-                        status_code=status_code,
-                        latency_ms=latency_ms,
-                    )
-                )
-                attempt_recorded = True
-                raise Exception(f"Upstream error: status={status_code}, error={error_msg}")
-
-            # 成功
-            attempts.append(
-                TestAttemptDetail(
-                    candidate_index=candidate_idx,
-                    endpoint_api_format=str(endpoint.api_format),
-                    endpoint_base_url=str(endpoint.base_url)[:80],
-                    key_name=getattr(key, "name", None),
-                    key_id=str(key.id),
-                    auth_type=auth_type,
-                    effective_model=effective_model,
-                    status="success",
-                    status_code=status_code,
-                    latency_ms=latency_ms,
-                )
-            )
-
-            return AttemptResult(
-                kind=AttemptKind.SYNC_RESPONSE,
-                http_status=status_code,
-                http_headers={},
-                response_body=response.get("response", response),
-            )
-
-        except Exception as exc:
-            latency_ms = int((time.monotonic() - start_time) * 1000)
-            # has_error 路径已记录带 status_code 的详细 attempt，此处仅补录早期异常
-            if not attempt_recorded:
-                attempts.append(
-                    TestAttemptDetail(
-                        candidate_index=candidate_idx,
-                        endpoint_api_format=str(endpoint.api_format),
-                        endpoint_base_url=str(endpoint.base_url)[:80],
-                        key_name=getattr(key, "name", None),
-                        key_id=str(key.id),
-                        auth_type=auth_type,
-                        effective_model=effective_model,
-                        status="failed",
-                        error_message=str(exc)[:300],
-                        latency_ms=latency_ms,
-                    )
-                )
-            raise
-
-    # 4. 预设 candidate index（FailoverEngine 也会 setattr，此处兜底防止 setattr 失败）
-    for i, cand in enumerate(candidates):
-        cand._utf_candidate_index = i  # type: ignore[attr-defined]
-
-    # 5. 执行故障转移
-    try:
-        engine = FailoverEngine(db)
-        result = await engine.execute(
-            candidates=candidates,
-            attempt_func=_attempt_func,
-            retry_policy=RetryPolicy(mode=RetryMode.DISABLED),
-            skip_policy=SkipPolicy(),
-            request_id=None,
+            },
+            extra_headers if extra_headers else None,
+            body_rules=getattr(endpoint, "body_rules", None),
+            header_rules=getattr(endpoint, "header_rules", None),
+            db=db,
+            user=current_user,
+            provider_name=provider_obj.name,
+            provider_id=str(provider_obj.id),
+            api_key_id=str(key.id),
+            model_name=effective_model,
+            auth_type=auth_type,
+            provider_type=provider_type if provider_type else None,
+            decrypted_auth_config=auth_config if auth_config else None,
+            provider_endpoint=endpoint,
+            provider_api_key=key,
+            proxy_config=effective_proxy,
+            timeout_seconds=request_timeout,
+        )
+        return _extract_test_response_or_raise(
+            response=response,
+            endpoint=endpoint,
+            provider_name=str(provider_obj.name),
+            auth_type=auth_type,
+            api_key=key,
+            db=db,
         )
 
-        # 补充 skipped 候选到 attempts
-        for i, cand in enumerate(candidates):
-            if cand.is_skipped and not any(a.candidate_index == i for a in attempts):
-                attempts.append(
-                    TestAttemptDetail(
-                        candidate_index=i,
-                        endpoint_api_format=str(cand.endpoint.api_format),
-                        endpoint_base_url=str(cand.endpoint.base_url)[:80],
-                        key_name=getattr(cand.key, "name", None),
-                        key_id=str(cand.key.id),
-                        auth_type=str(getattr(cand.key, "auth_type", "") or ""),
-                        status="skipped",
-                        skip_reason=cand.skip_reason,
-                    )
-                )
+    candidate_recorder = CandidateRecorder(db)
+    task_service = TaskService(db)
+    exec_result = None
+    run_error: Exception | None = None
 
-        attempts.sort(key=lambda a: a.candidate_index)
+    try:
+        exec_result = await task_service.execute_sync_candidates(
+            api_format=client_format or "openai:chat",
+            model_name=request.model_name,
+            candidates=candidates,
+            request_func=_request_func,
+            request_id=request_id,
+            current_user=current_user,
+            user_api_key=None,
+            is_stream=False,
+            capability_requirements=None,
+            request_body_ref={"body": dict(request_payload)},
+            request_headers=None,
+            request_body=dict(request_payload),
+            affinity_key=f"provider-test:{provider.id}",
+            create_pending_usage=False,
+            enable_cache_affinity=False,
+        )
+    except Exception as exc:
+        run_error = exc
+        logger.error("[test-model-failover] Error: {}", exc)
 
-        # 提取成功时的数据
-        data = None
-        if result.success and result.attempt_result:
-            data = {
+    try:
+        candidate_keys = candidate_recorder.get_candidate_keys(request_id)
+    except Exception:
+        candidate_keys = list(exec_result.candidate_keys) if exec_result else []
+
+    candidate_meta_by_pair, candidate_meta_by_index = _build_test_candidate_meta(
+        candidates=candidates,
+        provider=provider,
+        request=request,
+        gm_obj=gm_obj,
+    )
+    attempts = _build_test_attempts_from_candidate_keys(
+        candidate_keys=candidate_keys,
+        candidate_meta_by_pair=candidate_meta_by_pair,
+        candidate_meta_by_index=candidate_meta_by_index,
+    )
+    total_attempts = sum(1 for attempt in attempts if attempt.status != "skipped")
+
+    if exec_result and exec_result.success:
+        return TestModelFailoverResponse(
+            success=True,
+            model=request.model_name,
+            provider={"id": str(provider.id), "name": provider.name},
+            attempts=attempts,
+            total_candidates=len(candidates),
+            total_attempts=exec_result.attempt_count,
+            data={
                 "stream": True,
-                "response": result.attempt_result.response_body,
-            }
-
-        return TestModelFailoverResponse(
-            success=result.success,
-            model=request.model_name,
-            provider={"id": str(provider.id), "name": provider.name},
-            attempts=attempts,
-            total_candidates=len(candidates),
-            total_attempts=result.attempt_count,
-            data=data,
-            error=result.error_message if not result.success else None,
+                "response": exec_result.response,
+            },
+            error=None,
         ).model_dump()
 
-    except Exception as e:
-        logger.error("[test-model-failover] Error: {}", e)
-        return TestModelFailoverResponse(
-            success=False,
-            model=request.model_name,
-            provider={"id": str(provider.id), "name": provider.name},
-            attempts=attempts,
-            total_candidates=len(candidates),
-            total_attempts=0,
-            error=str(e)[:500],
-        ).model_dump()
+    error_message = None
+    if run_error is not None:
+        if isinstance(run_error, ProviderNotAvailableException) and getattr(
+            run_error, "upstream_response", None
+        ):
+            error_message = str(run_error.upstream_response)[:500]
+        if not error_message:
+            error_message = str(run_error)
+    if not error_message:
+        failed_attempt = next(
+            (attempt for attempt in reversed(attempts) if attempt.error_message),
+            None,
+        )
+        error_message = (
+            failed_attempt.error_message if failed_attempt else "服务暂时不可用，请稍后重试"
+        )
+
+    return TestModelFailoverResponse(
+        success=False,
+        model=request.model_name,
+        provider={"id": str(provider.id), "name": provider.name},
+        attempts=attempts,
+        total_candidates=len(candidates),
+        total_attempts=total_attempts,
+        error=str(error_message)[:500],
+    ).model_dump()

src/api/admin/system.py

@@ -22,6 +22,7 @@ from src.database import get_db
 from src.models.api import SystemSettingsRequest, SystemSettingsResponse
 from src.models.database import ApiKey, Provider, Usage, User
 from src.services.email.email_template import EmailTemplate
+from src.services.provider_ops.types import SENSITIVE_CREDENTIAL_FIELDS
 from src.services.system.config import SystemConfigService
 from src.utils.cache_decorator import cache_result
 
@@ -899,16 +900,7 @@ class AdminExportConfigAdapter(AdminApiAdapter):
     """导出提供商和模型配置"""
 
     # Provider Ops 中需要解密的敏感字段
-    SENSITIVE_CREDENTIALS = {
-        "api_key",
-        "password",
-        "session_token",
-        "session_cookie",
-        "token_cookie",
-        "auth_cookie",
-        "cookie_string",
-        "cookie",
-    }
+    SENSITIVE_CREDENTIALS = SENSITIVE_CREDENTIAL_FIELDS
 
     @staticmethod
     def _normalize_api_formats(raw_formats: Any) -> list[str]:
@@ -1180,16 +1172,7 @@ class AdminImportConfigAdapter(AdminApiAdapter):
     """导入提供商和模型配置"""
 
     # Provider Ops 中需要加密的敏感字段
-    SENSITIVE_CREDENTIALS = {
-        "api_key",
-        "password",
-        "session_token",
-        "session_cookie",
-        "token_cookie",
-        "auth_cookie",
-        "cookie_string",
-        "cookie",
-    }
+    SENSITIVE_CREDENTIALS = SENSITIVE_CREDENTIAL_FIELDS
 
     @staticmethod
     def _extract_import_key_api_formats(
@@ -1962,8 +1945,45 @@ class AdminImportConfigAdapter(AdminApiAdapter):
 
 
 class AdminExportUsersAdapter(AdminApiAdapter):
+    @staticmethod
+    def _serialize_api_key(key: ApiKey, include_is_standalone: bool = False) -> dict[str, Any]:
+        """序列化用户 API Key 为导出格式。"""
+        from src.core.crypto import crypto_service
+
+        data = {
+            "key_hash": key.key_hash,
+            "name": key.name,
+            "balance_used_usd": key.balance_used_usd,
+            "current_balance_usd": key.current_balance_usd,
+            "allowed_providers": key.allowed_providers,
+            "allowed_api_formats": key.allowed_api_formats,
+            "allowed_models": key.allowed_models,
+            "rate_limit": key.rate_limit,
+            "concurrent_limit": key.concurrent_limit,
+            "force_capabilities": key.force_capabilities,
+            "is_active": key.is_active,
+            "expires_at": key.expires_at.isoformat() if key.expires_at else None,
+            "auto_delete_on_expiry": key.auto_delete_on_expiry,
+            "total_requests": key.total_requests,
+            "total_cost_usd": key.total_cost_usd,
+        }
+
+        if key.key_encrypted:
+            try:
+                data["key"] = crypto_service.decrypt(key.key_encrypted, silent=True)
+            except Exception:
+                logger.warning(
+                    "[USERS_EXPORT] API Key 解密失败，回退为 legacy 密文字段: key_id={}", key.id
+                )
+                data["key_encrypted"] = key.key_encrypted
+
+        if include_is_standalone:
+            data["is_standalone"] = key.is_standalone
+
+        return data
+
     async def handle(self, context: ApiRequestContext) -> Any:  # type: ignore[override]
-        """导出用户数据（保留加密数据，排除管理员）"""
+        """导出用户数据（优先导出解密后的完整 Key，排除管理员）"""
         from datetime import datetime, timezone
 
         from src.core.enums import UserRole
@@ -1971,30 +1991,6 @@ class AdminExportUsersAdapter(AdminApiAdapter):
 
         db = context.db
 
-        def _serialize_api_key(key: ApiKey, include_is_standalone: bool = False) -> dict:
-            """序列化 API Key 为导出格式"""
-            data = {
-                "key_hash": key.key_hash,
-                "key_encrypted": key.key_encrypted,
-                "name": key.name,
-                "balance_used_usd": key.balance_used_usd,
-                "current_balance_usd": key.current_balance_usd,
-                "allowed_providers": key.allowed_providers,
-                "allowed_api_formats": key.allowed_api_formats,
-                "allowed_models": key.allowed_models,
-                "rate_limit": key.rate_limit,
-                "concurrent_limit": key.concurrent_limit,
-                "force_capabilities": key.force_capabilities,
-                "is_active": key.is_active,
-                "expires_at": key.expires_at.isoformat() if key.expires_at else None,
-                "auto_delete_on_expiry": key.auto_delete_on_expiry,
-                "total_requests": key.total_requests,
-                "total_cost_usd": key.total_cost_usd,
-            }
-            if include_is_standalone:
-                data["is_standalone"] = key.is_standalone
-            return data
-
         # 导出 Users（排除管理员）
         users = db.query(User).filter(User.is_deleted.is_(False), User.role != UserRole.ADMIN).all()
         users_data = []
@@ -2006,12 +2002,13 @@ class AdminExportUsersAdapter(AdminApiAdapter):
                 .all()
             )
             api_keys_data = [
-                _serialize_api_key(key, include_is_standalone=True) for key in api_keys
+                self._serialize_api_key(key, include_is_standalone=True) for key in api_keys
             ]
 
             users_data.append(
                 {
                     "email": user.email,
+                    "email_verified": user.email_verified,
                     "username": user.username,
                     "password_hash": user.password_hash,
                     "role": user.role.value if user.role else "user",
@@ -2029,10 +2026,10 @@ class AdminExportUsersAdapter(AdminApiAdapter):
 
         # 导出独立余额 Keys（管理员创建的，不属于普通用户）
         standalone_keys = db.query(ApiKey).filter(ApiKey.is_standalone.is_(True)).all()
-        standalone_keys_data = [_serialize_api_key(key) for key in standalone_keys]
+        standalone_keys_data = [self._serialize_api_key(key) for key in standalone_keys]
 
         return {
-            "version": "1.1",
+            "version": "1.2",
             "exported_at": datetime.now(timezone.utc).isoformat(),
             "users": users_data,
             "standalone_keys": standalone_keys_data,
@@ -2040,6 +2037,22 @@ class AdminExportUsersAdapter(AdminApiAdapter):
 
 
 class AdminImportUsersAdapter(AdminApiAdapter):
+    @staticmethod
+    def _resolve_api_key_material(key_data: dict[str, Any]) -> tuple[str | None, str | None]:
+        """解析用户 API Key 导入材料，优先使用明文 key。"""
+        from src.core.crypto import crypto_service
+        from src.models.database import ApiKey
+
+        plaintext_key = key_data.get("key")
+        if isinstance(plaintext_key, str):
+            normalized = plaintext_key.strip()
+            if normalized:
+                return ApiKey.hash_key(normalized), crypto_service.encrypt(normalized)
+
+        key_hash = str(key_data.get("key_hash") or "").strip() or None
+        key_encrypted = key_data.get("key_encrypted")
+        return key_hash, key_encrypted
+
     async def handle(self, context: ApiRequestContext) -> Any:  # type: ignore[override]
         """导入用户数据"""
         import uuid
@@ -2079,7 +2092,7 @@ class AdminImportUsersAdapter(AdminApiAdapter):
                 (None, "skipped"): key 已存在，跳过
                 (None, "invalid"): 数据无效，跳过
             """
-            key_hash = key_data.get("key_hash", "").strip()
+            key_hash, key_encrypted = self._resolve_api_key_material(key_data)
             if not key_hash:
                 return None, "invalid"
 
@@ -2103,7 +2116,7 @@ class AdminImportUsersAdapter(AdminApiAdapter):
                     id=str(uuid.uuid4()),
                     user_id=owner_id,
                     key_hash=key_hash,
-                    key_encrypted=key_data.get("key_encrypted"),
+                    key_encrypted=key_encrypted,
                     name=key_data.get("name"),
                     is_standalone=is_standalone or key_data.get("is_standalone", False),
                     balance_used_usd=key_data.get("balance_used_usd", 0.0),

src/api/handlers/base/chat_sync_executor.py

@@ -74,6 +74,7 @@ class SyncRequestContext:
     mapped_model_result: str | None = None
     sync_proxy_info: dict[str, Any] | None = None
     provider_response_json: dict[str, Any] | None = None  # 格式转换前的提供商原始响应
+    pool_summary: dict[str, Any] | None = None
 
 
 class ChatSyncExecutor:

src/api/handlers/base/endpoint_checker.py

@@ -74,6 +74,7 @@ async def run_endpoint_check(
     user: Any | None = None,  # User对象
     proxy_config: dict[str, Any] | None = None,  # 原始代理配置（支持 tunnel 模式）
     is_stream: bool | None = None,  # 显式流式标记（优先于 body/url 推断）
+    timeout: float | None = None,
 ) -> dict[str, Any]:
     """
     执行端点检查（重构版本，使用新的架构）：
@@ -97,6 +98,7 @@ async def run_endpoint_check(
         request_id=str(uuid.uuid4())[:8],
         proxy_config=proxy_config,
         is_stream=is_stream,
+        timeout=float(timeout) if timeout is not None else 30.0,
     )
 
     # 使用协调器执行检查
@@ -595,6 +597,7 @@ class HttpRequestExecutor:
         """执行HTTP请求（支持流式和非流式响应）"""
         start_time = time.time()
         request_id = request.request_id or str(uuid.uuid4())[:8]
+        effective_timeout = float(request.timeout if request.timeout is not None else self.timeout)
 
         # 检查是否是流式请求（优先显式参数，其次 body，最后 URL 推断）
         if request.is_stream is not None:
@@ -618,7 +621,7 @@ class HttpRequestExecutor:
 
             # 统一通过 build_proxy_client_kwargs 构建（支持 tunnel 模式 + 普通代理 + 系统默认回退）
             client_kwargs = build_proxy_client_kwargs(
-                proxy_config=request.proxy_config, timeout=self.timeout
+                proxy_config=request.proxy_config, timeout=effective_timeout
             )
 
             async with httpx.AsyncClient(**client_kwargs) as client:

src/api/handlers/base/handler_adapter_base.py

@@ -30,6 +30,7 @@ from src.core.api_format import (
     get_adapter_protected_keys_for_endpoint,
     get_auth_handler,
     get_default_auth_method_for_endpoint,
+    resolve_header_name_case,
 )
 from src.core.exceptions import (
     ProviderAuthException,
@@ -343,6 +344,7 @@ class HandlerAdapterBase(ApiAdapter):
         provider_api_key: Any | None = None,
         # 代理配置
         proxy_config: dict[str, Any] | None = None,
+        timeout_seconds: float | None = None,
     ) -> dict[str, Any]:
         """
         测试模型连接性（非流式）
@@ -458,7 +460,8 @@ class HandlerAdapterBase(ApiAdapter):
                 default_auth_header, _ = get_auth_config_for_endpoint(cls.FORMAT_ID)
                 if default_auth_header.lower() != "authorization":
                     headers.pop(default_auth_header, None)
-                headers["Authorization"] = f"Bearer {api_key}"
+                auth_header_name = resolve_header_name_case(extra_headers, "Authorization")
+                headers[auth_header_name] = f"Bearer {api_key}"
 
         # ---- Body ----
         body = cls.build_request_body(request_data, base_url=base_url, provider_type=provider_type)
@@ -527,6 +530,7 @@ class HandlerAdapterBase(ApiAdapter):
             api_key_id=api_key_id,
             model_name=effective_model_name,
             proxy_config=proxy_config,
+            timeout=timeout_seconds,
         )
 
     # =========================================================================

src/api/handlers/base/request_builder.py

@@ -24,6 +24,7 @@ from src.core.api_format import (
     HeaderBuilder,
     get_auth_config_for_endpoint,
     make_signature_key,
+    resolve_header_name_case,
 )
 from src.core.crypto import crypto_service
 from src.models.endpoint_models import _CONDITION_OPS, _TYPE_IS_VALUES, parse_re_flags
@@ -1250,7 +1251,7 @@ class PassthroughRequestBuilder(RequestBuilder):
             builder.add_many(effective_extra_headers)
 
         # 5. 设置认证头（最高优先级，上游始终使用 header 认证）
-        builder.add(auth_header, auth_value)
+        builder.add(resolve_header_name_case(original_headers, auth_header), auth_value)
 
         # 6. 确保有 Content-Type
         headers = builder.build()

src/api/handlers/gemini/adapter.py

@@ -14,7 +14,7 @@ from fastapi.responses import JSONResponse
 
 from src.api.handlers.base.chat_adapter_base import ChatAdapterBase, register_adapter
 from src.api.handlers.base.chat_handler_base import ChatHandlerBase
-from src.core.api_format import ApiFamily, get_auth_handler
+from src.core.api_format import ApiFamily, get_auth_handler, resolve_header_name_case
 from src.core.api_format.enums import AuthMethod
 from src.core.api_format.headers import BROWSER_FINGERPRINT_HEADERS
 from src.core.logger import logger
@@ -302,6 +302,7 @@ class GeminiChatAdapter(ChatAdapterBase):
         provider_api_key: Any | None = None,
         # 代理配置
         proxy_config: dict[str, Any] | None = None,
+        timeout_seconds: float | None = None,
     ) -> dict[str, Any]:
         """测试 Gemini API 模型连接性（非流式）"""
         from src.api.handlers.base.endpoint_checker import run_endpoint_check
@@ -382,7 +383,8 @@ class GeminiChatAdapter(ChatAdapterBase):
                 default_auth_header, _ = get_auth_config_for_endpoint(cls.FORMAT_ID)
                 if default_auth_header.lower() != "authorization":
                     headers.pop(default_auth_header, None)
-                headers["Authorization"] = f"Bearer {api_key}"
+                auth_header_name = resolve_header_name_case(extra_headers, "Authorization")
+                headers[auth_header_name] = f"Bearer {api_key}"
 
         body = cls.build_request_body(request_data)
 
@@ -432,6 +434,7 @@ class GeminiChatAdapter(ChatAdapterBase):
             api_key_id=api_key_id,
             model_name=effective_model_name,
             proxy_config=proxy_config,
+            timeout=timeout_seconds,
         )
 
 

src/core/api_format/__init__.py

@@ -44,6 +44,7 @@ from src.core.api_format.headers import (
     merge_headers_with_protection,
     normalize_headers,
     redact_headers_for_log,
+    resolve_header_name_case,
 )
 from src.core.api_format.metadata import (
     ENDPOINT_DEFINITIONS,
@@ -122,6 +123,7 @@ __all__ = [
     "merge_headers_with_protection",
     "filter_response_headers",
     "redact_headers_for_log",
+    "resolve_header_name_case",
     "build_adapter_base_headers_for_endpoint",
     "build_adapter_headers_for_endpoint",
     "get_adapter_protected_keys_for_endpoint",

src/core/api_format/headers.py

@@ -82,6 +82,7 @@ UPSTREAM_DROP_HEADERS: frozenset[str] = frozenset(
         "x-real-proto",
         "x-forwarded-for",
         "x-forwarded-proto",
+        "x-forwarded-scheme",
         "x-forwarded-host",
         "x-forwarded-port",
     }
@@ -189,6 +190,19 @@ def extract_client_api_key_for_endpoint(
     return value
 
 
+def resolve_header_name_case(
+    headers: dict[str, str] | None,
+    preferred_key: str,
+) -> str:
+    """Preserve original header casing when replacing an existing header."""
+    if headers:
+        preferred_lower = preferred_key.lower()
+        for key in headers.keys():
+            if str(key).lower() == preferred_lower:
+                return str(key)
+    return preferred_key
+
+
 def extract_client_api_key_for_endpoint_with_query(
     headers: dict[str, str],
     query_params: dict[str, str] | None,
@@ -279,8 +293,11 @@ class HeaderBuilder:
         self._headers: dict[str, tuple[str, str]] = {}
 
     def add(self, key: str, value: str) -> HeaderBuilder:
-        """添加单个头部（会覆盖同名头部）"""
-        self._headers[key.lower()] = (key, value)
+        """添加单个头部（会覆盖同名头部，但保留已存在 key 的原始大小写）"""
+        key_lower = key.lower()
+        existing = self._headers.get(key_lower)
+        stored_key = existing[0] if existing else key
+        self._headers[key_lower] = (stored_key, value)
         return self
 
     def add_many(self, headers: dict[str, str]) -> HeaderBuilder:
@@ -454,7 +471,7 @@ def build_upstream_headers_for_endpoint(
     if extra_headers:
         builder.add_many(extra_headers)
 
-    builder.add(auth_header, auth_value)
+    builder.add(resolve_header_name_case(original_headers, auth_header), auth_value)
 
     result = builder.build()
     if not any(k.lower() == "content-type" for k in result):

src/services/orchestration/candidate_resolver.py

@@ -193,8 +193,8 @@ class CandidateResolver:
         self,
         all_candidates: list[ProviderCandidate],
         request_id: str | None,
-        user_id: str,
-        user_api_key: ApiKey,
+        user_id: str | None,
+        user_api_key: ApiKey | None,
         required_capabilities: dict[str, bool] | None = None,
         *,
         expand_retries: bool = True,

src/services/orchestration/request_dispatcher.py

@@ -50,7 +50,8 @@ class RequestDispatcher:
         candidate_index: int,
         retry_index: int,
         candidate_record_id: str,
-        user_api_key: ApiKey,
+        user_api_key: ApiKey | None,
+        user_id: str | None,
         request_func: Callable[..., Any],
         request_id: str | None,
         api_format: str,
@@ -112,6 +113,7 @@ class RequestDispatcher:
             candidate_id=candidate_record_id,
             candidate_index=candidate_index,
             user_api_key=user_api_key,
+            user_id=user_id,
             request_func=request_func,
             request_id=request_id,
             api_format=api_format,

src/services/provider/adapters/codex/envelope.py

@@ -14,10 +14,8 @@ to extra_headers().
 
 from __future__ import annotations
 
-import uuid
 from typing import Any
 
-from src.config.settings import config
 from src.services.provider.adapters.codex.context import (
     CodexRequestContext,
     get_codex_request_context,
@@ -30,36 +28,12 @@ class CodexOAuthEnvelope:
     """Provider envelope hooks for Codex OAuth upstream."""
 
     name = "codex:oauth"
-    _CODEX_VERSION = "0.101.0"
-    _CODEX_ORIGINATOR = "codex_cli_rs"
 
     def extra_headers(self) -> dict[str, str] | None:
-        # Keep these headers provider-scoped to avoid leaking to other upstreams.
-        headers: dict[str, str] = {
-            # Codex upstream is strict about Content-Type; variants like
-            # "application/json; charset=utf-8" are rejected.
-            "Content-Type": "application/json",
-            "Version": self._CODEX_VERSION,
-            "Session_id": str(uuid.uuid4()),
-            "Connection": "Keep-Alive",
-            "Originator": self._CODEX_ORIGINATOR,
-        }
-
-        # Compact endpoint is non-stream; normal responses endpoint expects SSE.
-        ctx = get_codex_request_context()
-        is_compact = bool(ctx.is_compact) if ctx else False
-        headers["Accept"] = "application/json" if is_compact else "text/event-stream"
-
-        ua = str(getattr(config, "internal_user_agent_openai_cli", "") or "").strip()
-        if ua:
-            headers["User-Agent"] = ua
-
-        # Add chatgpt-account-id from context (set by wrap_request).
-        # Context is NOT cleared here — build_codex_url reads is_compact from it later.
-        if ctx and ctx.account_id:
-            headers["Chatgpt-Account-Id"] = ctx.account_id
-
-        return headers
+        # Codex desktop clients already send the protocol-specific headers they need.
+        # Preserve the original request headers as much as possible and avoid injecting
+        # synthetic CLI identity headers here.
+        return None
 
     def wrap_request(
         self,

src/services/provider_ops/service.py

@@ -24,6 +24,7 @@ from src.models.database import Provider
 from src.services.provider_ops.architectures import ProviderConnector
 from src.services.provider_ops.registry import get_registry
 from src.services.provider_ops.types import (
+    SENSITIVE_CREDENTIAL_FIELDS,
     ActionResult,
     ActionStatus,
     BalanceInfo,
@@ -94,17 +95,7 @@ class ProviderOpsService:
     """
 
     # 凭据中需要加密的字段
-    SENSITIVE_FIELDS = {
-        "api_key",
-        "password",
-        "refresh_token",
-        "session_token",
-        "session_cookie",
-        "token_cookie",
-        "auth_cookie",
-        "cookie_string",
-        "cookie",
-    }
+    SENSITIVE_FIELDS = SENSITIVE_CREDENTIAL_FIELDS
 
     def __init__(self, db: Session):
         self.db = db

src/services/provider_ops/types.py

@@ -9,6 +9,20 @@ from datetime import datetime, timezone
 from enum import Enum
 from typing import Any
 
+SENSITIVE_CREDENTIAL_FIELDS = frozenset(
+    {
+        "api_key",
+        "password",
+        "refresh_token",
+        "session_token",
+        "session_cookie",
+        "token_cookie",
+        "auth_cookie",
+        "cookie_string",
+        "cookie",
+    }
+)
+
 
 class ConnectorAuthType(str, Enum):
     """连接器认证类型"""

src/services/request/executor.py

@@ -69,7 +69,8 @@ class RequestExecutor:
         candidate: Any,
         candidate_id: str,
         candidate_index: int,
-        user_api_key: Any,
+        user_api_key: Any | None,
+        user_id: str | None = None,
         request_func: Callable[..., Any],
         request_id: str | None,
         api_format: str,
@@ -93,8 +94,8 @@ class RequestExecutor:
             provider_id=provider.id,
             endpoint_id=endpoint.id,
             key_id=key.id,
-            user_id=user_api_key.user_id,
-            api_key_id=user_api_key.id,
+            user_id=user_id if user_id is not None else getattr(user_api_key, "user_id", None),
+            api_key_id=getattr(user_api_key, "id", None),
             is_cached_user=is_cached_user,
         )
 

src/services/task/service.py

@@ -187,6 +187,329 @@ class TaskService:
             request_body=request_body,
         )
 
+    async def execute_sync_candidates(
+        self,
+        *,
+        api_format: str,
+        model_name: str,
+        candidates: list[Any],
+        request_func: Callable[..., Any],
+        request_id: str | None = None,
+        current_user: User | None = None,
+        user_api_key: ApiKey | None = None,
+        is_stream: bool = False,
+        capability_requirements: dict[str, bool] | None = None,
+        request_body_ref: dict[str, Any] | None = None,
+        request_headers: dict[str, Any] | None = None,
+        request_body: dict[str, Any] | None = None,
+        affinity_key: str | None = None,
+        create_pending_usage: bool = False,
+        enable_cache_affinity: bool = False,
+    ) -> ExecutionResult:
+        """Execute a pre-built candidate set through the unified SYNC runtime."""
+        from src.services.rate_limit.adaptive_rpm import get_adaptive_rpm_manager
+        from src.services.rate_limit.concurrency_manager import get_concurrency_manager
+        from src.services.request.executor import RequestExecutor
+
+        if not request_id:
+            request_id = str(uuid4())
+
+        api_format_norm = normalize_endpoint_signature(api_format)
+
+        priority_mode = SystemConfigService.get_config(
+            self.db,
+            "provider_priority_mode",
+            CacheAwareScheduler.PRIORITY_MODE_PROVIDER,
+        )
+        scheduling_mode = SystemConfigService.get_config(
+            self.db,
+            "scheduling_mode",
+            CacheAwareScheduler.SCHEDULING_MODE_CACHE_AFFINITY,
+        )
+        cache_scheduler = await get_cache_aware_scheduler(
+            self.redis,
+            priority_mode=priority_mode,
+            scheduling_mode=scheduling_mode,
+        )
+        await cache_scheduler._ensure_initialized()
+
+        concurrency_manager = await get_concurrency_manager()
+        adaptive_manager = get_adaptive_rpm_manager()
+        request_executor = RequestExecutor(
+            db=self.db,
+            concurrency_manager=concurrency_manager,
+            adaptive_manager=adaptive_manager,
+        )
+        candidate_resolver = CandidateResolver(
+            db=self.db,
+            cache_scheduler=cache_scheduler,
+        )
+        error_classifier = ErrorClassifier(
+            db=self.db,
+            cache_scheduler=cache_scheduler,
+            adaptive_manager=adaptive_manager,
+        )
+        request_dispatcher = RequestDispatcher(
+            db=self.db,
+            request_executor=request_executor,
+            cache_scheduler=cache_scheduler if enable_cache_affinity else None,
+        )
+
+        resolved_user = current_user
+        if resolved_user is None and user_api_key is not None:
+            try:
+                resolved_user = user_api_key.user if hasattr(user_api_key, "user") else None
+            except Exception:
+                resolved_user = None
+            if resolved_user is None and getattr(user_api_key, "user_id", None):
+                resolved_user = self.db.query(User).filter(User.id == user_api_key.user_id).first()
+
+        user_id: str | None = None
+        if resolved_user is not None and getattr(resolved_user, "id", None):
+            user_id = str(resolved_user.id)
+        elif user_api_key is not None and getattr(user_api_key, "user_id", None):
+            user_id = str(user_api_key.user_id)
+
+        resolved_affinity_key = affinity_key
+        if not resolved_affinity_key:
+            api_key_id = getattr(user_api_key, "id", None) if user_api_key is not None else None
+            resolved_affinity_key = str(api_key_id) if api_key_id else f"internal-test:{request_id}"
+
+        if create_pending_usage:
+            try:
+                UsageService.create_pending_usage(
+                    db=self.db,
+                    request_id=request_id,
+                    user=resolved_user,
+                    api_key=user_api_key,
+                    model=model_name,
+                    is_stream=is_stream,
+                    api_format=api_format_norm,
+                    request_headers=request_headers,
+                    request_body=request_body,
+                )
+            except Exception as exc:
+                logger.warning("创建 pending 使用记录失败: {}", str(exc))
+
+        all_candidates = list(candidates)
+        all_candidates, pool_traces = await self._apply_pool_reorder(
+            all_candidates, request_body=request_body
+        )
+
+        candidate_record_map = candidate_resolver.create_candidate_records(
+            all_candidates=all_candidates,
+            request_id=request_id,
+            user_id=user_id,
+            user_api_key=user_api_key,
+            required_capabilities=capability_requirements,
+        )
+
+        max_attempts = candidate_resolver.count_total_attempts(all_candidates)
+        last_error: Exception | None = None
+        last_candidate: Any | None = all_candidates[-1] if all_candidates else None
+
+        async def _attempt(candidate: Any) -> AttemptResult:
+            nonlocal last_candidate
+            last_candidate = candidate
+
+            candidate_index = int(getattr(candidate, "_utf_candidate_index", -1))
+            retry_index = int(getattr(candidate, "_utf_retry_index", 0))
+            candidate_record_id = str(getattr(candidate, "_utf_candidate_record_id", "") or "")
+            attempt_counter = int(getattr(candidate, "_utf_attempt_count", 0))
+            max_attempts_local = int(getattr(candidate, "_utf_max_attempts", max_attempts))
+
+            if not candidate_record_id:
+                from src.services.scheduling.schemas import PoolCandidate
+
+                pool_extra = (
+                    getattr(candidate.key, "_pool_extra_data", None)
+                    if isinstance(getattr(candidate.key, "_pool_extra_data", None), dict)
+                    else {}
+                )
+                extra_data: dict[str, Any] = {
+                    "needs_conversion": bool(getattr(candidate, "needs_conversion", False)),
+                    "provider_api_format": getattr(candidate, "provider_api_format", None) or None,
+                    "mapping_matched_model": getattr(candidate, "mapping_matched_model", None)
+                    or None,
+                    **pool_extra,
+                }
+                if isinstance(candidate, PoolCandidate):
+                    extra_data["pool_group_id"] = str(candidate.provider.id)
+                    extra_data["pool_key_index"] = int(
+                        getattr(candidate, "_pool_key_index", 0) or 0
+                    )
+
+                candidate_record = RequestCandidateService.create_candidate(
+                    db=self.db,
+                    request_id=request_id,
+                    candidate_index=candidate_index,
+                    retry_index=retry_index,
+                    user_id=user_id,
+                    api_key_id=(getattr(user_api_key, "id", None) if user_api_key else None),
+                    provider_id=str(candidate.provider.id),
+                    endpoint_id=str(candidate.endpoint.id),
+                    key_id=str(candidate.key.id),
+                    status="available",
+                    is_cached=bool(getattr(candidate, "is_cached", False)),
+                    extra_data=extra_data,
+                )
+                self.db.flush()
+                candidate_record_id = str(candidate_record.id)
+                candidate_record_map[(candidate_index, retry_index)] = candidate_record_id
+                setattr(candidate, "_utf_candidate_record_id", candidate_record_id)
+
+            (
+                response,
+                _provider_name,
+                attempt_id,
+                _provider_id,
+                _endpoint_id,
+                _key_id,
+                _first_byte_time_ms,
+            ) = await request_dispatcher.dispatch(
+                candidate=candidate,
+                candidate_index=candidate_index,
+                retry_index=retry_index,
+                candidate_record_id=candidate_record_id,
+                user_api_key=user_api_key,
+                user_id=user_id,
+                request_func=request_func,
+                request_id=request_id,
+                api_format=api_format_norm,
+                model_name=model_name,
+                affinity_key=resolved_affinity_key,
+                global_model_id=model_name,
+                attempt_counter=attempt_counter,
+                max_attempts=max_attempts_local,
+                is_stream=is_stream,
+            )
+            _ = (attempt_id, _provider_name, _provider_id, _endpoint_id, _key_id)
+
+            await self._pool_on_success(
+                candidate,
+                request_body,
+                ttfb_ms=_first_byte_time_ms,
+            )
+
+            if is_stream:
+                return AttemptResult(
+                    kind=AttemptKind.STREAM,
+                    http_status=200,
+                    http_headers={},
+                    stream_iterator=response,
+                )
+            return AttemptResult(
+                kind=AttemptKind.SYNC_RESPONSE,
+                http_status=200,
+                http_headers={},
+                response_body=response,
+            )
+
+        async def _handle_exec_err(
+            *,
+            exec_err: Any,
+            candidate: Any,
+            candidate_index: int,
+            retry_index: int,
+            max_retries_for_candidate: int,
+            record_id: str | None,
+            attempt_count: int,
+            max_attempts: int | None,
+        ) -> tuple[Any, int | None]:
+            nonlocal last_error, last_candidate
+            last_candidate = candidate
+            last_error = getattr(exec_err, "cause", None)
+
+            candidate_record_id = str(record_id or "") or str(
+                candidate_record_map.get((candidate_index, 0), "")
+            )
+
+            action = await self._handle_candidate_error(
+                exec_err=exec_err,
+                candidate=candidate,
+                candidate_record_id=candidate_record_id,
+                retry_index=retry_index,
+                max_retries_for_candidate=max_retries_for_candidate,
+                affinity_key=resolved_affinity_key,
+                api_format=api_format_norm,
+                global_model_id=model_name,
+                request_id=request_id,
+                attempt=attempt_count,
+                max_attempts=int(max_attempts or 0),
+                request_body_ref=request_body_ref,
+                error_classifier=error_classifier,
+            )
+
+            if action == "continue":
+                new_max = None
+                if request_body_ref and request_body_ref.get("_rectified_this_turn", False):
+                    request_body_ref["_rectified_this_turn"] = False
+                    new_max = max(max_retries_for_candidate, retry_index + 2)
+                return ("retry", new_max)
+
+            if action == "break":
+                return ("continue", None)
+
+            if action == "raise":
+                if last_error is not None:
+                    self._attach_metadata_to_error(
+                        last_error, last_candidate, model_name, api_format_norm
+                    )
+                    raise last_error
+                raise
+
+            return ("continue", None)
+
+        engine = FailoverEngine(
+            self.db,
+            error_classifier=error_classifier,
+            recorder=self._recorder,
+        )
+        result = await engine.execute(
+            candidates=all_candidates,
+            attempt_func=_attempt,
+            retry_policy=RetryPolicy.for_sync_task(),
+            skip_policy=SkipPolicy(),
+            request_id=request_id,
+            user_id=user_id,
+            api_key_id=(
+                str(user_api_key.id) if user_api_key and getattr(user_api_key, "id", None) else None
+            ),
+            candidate_record_map=candidate_record_map,
+            max_attempts=max_attempts,
+            execution_error_handler=_handle_exec_err,
+        )
+
+        if result.success:
+            if pool_traces and result.key_id:
+                try:
+                    attempted_key_ids: set[str] = set()
+                    for ck in result.candidate_keys or []:
+                        status = str(getattr(ck, "status", "") or "").strip().lower()
+                        if status in {"", "available", "pending", "skipped", "unused"}:
+                            continue
+                        kid = getattr(ck, "key_id", None)
+                        if isinstance(kid, str) and kid:
+                            attempted_key_ids.add(kid)
+                    if not attempted_key_ids:
+                        attempted_key_ids.add(str(result.key_id))
+
+                    for pt in pool_traces:
+                        summary = pt.build_summary(
+                            result.key_id,
+                            attempted_key_ids=attempted_key_ids,
+                        )
+                        if summary:
+                            result.pool_summary = summary
+                            break
+                except Exception:
+                    pass
+            return result
+
+        self._raise_all_failed_exception(
+            request_id, max_attempts, last_candidate, model_name, api_format_norm, last_error
+        )
+
     @staticmethod
     def _extract_session_uuid(
         provider_type: str, request_body: dict[str, Any] | None
@@ -587,6 +910,7 @@ class TaskService:
                 retry_index=retry_index,
                 candidate_record_id=candidate_record_id,
                 user_api_key=user_api_key,
+                user_id=user_id,
                 request_func=request_func,
                 request_id=request_id,
                 api_format=api_format_norm,

tests/services/test_codex_request.py

@@ -1,8 +1,11 @@
 from __future__ import annotations
 
+from types import SimpleNamespace
+
 import jwt
 import pytest
 
+from src.api.handlers.base.request_builder import PassthroughRequestBuilder
 from src.services.provider.adapters.codex.request_patching import (
     maybe_patch_request_for_codex,
     patch_openai_cli_request_for_codex,
@@ -110,42 +113,49 @@ def test_openai_cli_normalizer_request_from_internal_codex_variant_defaults_stor
     assert out["store"] is False
 
 
-def test_codex_envelope_extra_headers_includes_sse_accept_and_session() -> None:
+def test_codex_envelope_extra_headers_does_not_inject_synthetic_headers() -> None:
     from src.services.provider.adapters.codex.envelope import codex_oauth_envelope
 
-    headers = codex_oauth_envelope.extra_headers() or {}
-    assert headers.get("Accept") == "text/event-stream"
-    assert headers.get("Originator") == "codex_cli_rs"
-    assert headers.get("Version") == "0.101.0"
-    assert headers.get("Connection") == "Keep-Alive"
-    assert isinstance(headers.get("Session_id"), str)
-    assert headers.get("Session_id")
+    assert codex_oauth_envelope.extra_headers() is None
 
 
-def test_codex_envelope_extra_headers_compact_uses_json_accept() -> None:
-    from src.services.provider.adapters.codex.context import (
-        CodexRequestContext,
-        set_codex_request_context,
+def test_codex_passthrough_builder_preserves_real_codex_headers() -> None:
+    from src.services.provider.adapters.codex.envelope import codex_oauth_envelope
+
+    builder = PassthroughRequestBuilder()
+    endpoint = SimpleNamespace(api_family="openai", endpoint_kind="cli", header_rules=None)
+    key = SimpleNamespace(api_key="unused")
+
+    headers = builder.build_headers(
+        original_headers={
+            "accept": "text/event-stream",
+            "content-type": "application/json",
+            "user-agent": "Codex Desktop/0.108.0-alpha.12",
+            "originator": "Codex Desktop",
+            "x-codex-turn-metadata": '{"turn_id":"abc"}',
+            "x-forwarded-scheme": "https",
+            "host": "aether.hetunai.cn",
+            "content-length": "123",
+        },
+        endpoint=endpoint,
+        key=key,
+        pre_computed_auth=("Authorization", "Bearer upstream-token"),
+        envelope=codex_oauth_envelope,
     )
-    from src.services.provider.adapters.codex.envelope import codex_oauth_envelope
 
-    set_codex_request_context(CodexRequestContext(is_compact=True))
-    headers = codex_oauth_envelope.extra_headers() or {}
-    assert headers.get("Accept") == "application/json"
-    set_codex_request_context(None)
-
-
-def test_codex_envelope_extra_headers_uses_account_id_header() -> None:
-    from src.services.provider.adapters.codex.context import (
-        CodexRequestContext,
-        set_codex_request_context,
-    )
-    from src.services.provider.adapters.codex.envelope import codex_oauth_envelope
-
-    set_codex_request_context(CodexRequestContext(account_id="acc_123"))
-    headers = codex_oauth_envelope.extra_headers() or {}
-    assert headers.get("Chatgpt-Account-Id") == "acc_123"
-    set_codex_request_context(None)
+    assert headers["accept"] == "text/event-stream"
+    assert headers["content-type"] == "application/json"
+    assert headers["user-agent"] == "Codex Desktop/0.108.0-alpha.12"
+    assert headers["originator"] == "Codex Desktop"
+    assert headers["x-codex-turn-metadata"] == '{"turn_id":"abc"}'
+    assert headers["Authorization"] == "Bearer upstream-token"
+    assert "Version" not in headers
+    assert "Session_id" not in headers
+    assert "Connection" not in headers
+    assert "Chatgpt-Account-Id" not in headers
+    assert "host" not in headers
+    assert "content-length" not in headers
+    assert "x-forwarded-scheme" not in headers
 
 
 def _encode_unsigned_jwt(payload: dict[str, object]) -> str:

tests/unit/test_admin_system_key_formats.py

@@ -1,4 +1,5 @@
 from src.api.admin.system import AdminExportConfigAdapter, AdminImportConfigAdapter
+from src.services.provider_ops.types import SENSITIVE_CREDENTIAL_FIELDS
 
 
 def test_export_key_api_formats_falls_back_to_provider_endpoints_when_none() -> None:
@@ -59,3 +60,57 @@ def test_import_key_api_formats_keeps_explicit_empty_list() -> None:
     )
 
     assert result == []
+
+
+class _FakeCrypto:
+    def encrypt(self, value: str) -> str:
+        return f"enc:{value}"
+
+    def decrypt(self, value: str) -> str:
+        return value.removeprefix("enc:")
+
+
+def test_provider_ops_sensitive_fields_include_refresh_token() -> None:
+    assert "refresh_token" in SENSITIVE_CREDENTIAL_FIELDS
+
+
+def test_export_provider_config_decrypts_refresh_token() -> None:
+    adapter = AdminExportConfigAdapter()
+
+    config = {
+        "provider_ops": {
+            "connector": {
+                "credentials": {
+                    "refresh_token": "enc:rt-1",
+                    "api_key": "enc:key-1",
+                }
+            }
+        }
+    }
+
+    result = adapter._decrypt_provider_config(config, _FakeCrypto())
+
+    assert result["provider_ops"]["connector"]["credentials"]["refresh_token"] == "rt-1"
+    assert result["provider_ops"]["connector"]["credentials"]["api_key"] == "key-1"
+    assert config["provider_ops"]["connector"]["credentials"]["refresh_token"] == "enc:rt-1"
+
+
+def test_import_provider_config_encrypts_refresh_token() -> None:
+    adapter = AdminImportConfigAdapter()
+
+    config = {
+        "provider_ops": {
+            "connector": {
+                "credentials": {
+                    "refresh_token": "rt-1",
+                    "api_key": "key-1",
+                }
+            }
+        }
+    }
+
+    result = adapter._encrypt_provider_config(config, _FakeCrypto())
+
+    assert result["provider_ops"]["connector"]["credentials"]["refresh_token"] == "enc:rt-1"
+    assert result["provider_ops"]["connector"]["credentials"]["api_key"] == "enc:key-1"
+    assert config["provider_ops"]["connector"]["credentials"]["refresh_token"] == "rt-1"

tests/unit/test_admin_system_users_export_import.py

@@ -0,0 +1,57 @@
+from src.api.admin.system import AdminExportUsersAdapter, AdminImportUsersAdapter
+from src.core.crypto import crypto_service
+from src.models.database import ApiKey
+
+
+def test_export_user_api_key_prefers_plaintext_key() -> None:
+    plaintext_key = "ak-user-plain-1"
+    key = ApiKey(
+        id="key-1",
+        user_id="user-1",
+        key_hash=ApiKey.hash_key(plaintext_key),
+        key_encrypted=crypto_service.encrypt(plaintext_key),
+        name="Demo Key",
+        is_standalone=False,
+        balance_used_usd=1.5,
+        current_balance_usd=8.5,
+        is_active=True,
+    )
+
+    data = AdminExportUsersAdapter._serialize_api_key(key, include_is_standalone=True)
+
+    assert data["key"] == plaintext_key
+    assert "key_encrypted" not in data
+    assert data["key_hash"] == ApiKey.hash_key(plaintext_key)
+    assert data["is_standalone"] is False
+
+
+def test_import_user_api_key_material_reencrypts_plaintext_key() -> None:
+    plaintext_key = "ak-user-plain-2"
+
+    key_hash, key_encrypted = AdminImportUsersAdapter._resolve_api_key_material(
+        {
+            "key": plaintext_key,
+            "key_hash": "stale-hash",
+            "key_encrypted": "stale-ciphertext",
+        }
+    )
+
+    assert key_hash == ApiKey.hash_key(plaintext_key)
+    assert key_encrypted is not None
+    assert crypto_service.decrypt(key_encrypted) == plaintext_key
+
+
+def test_import_user_api_key_material_keeps_legacy_encrypted_payload() -> None:
+    legacy_plaintext = "ak-user-legacy-1"
+    legacy_encrypted = crypto_service.encrypt(legacy_plaintext)
+    legacy_hash = ApiKey.hash_key(legacy_plaintext)
+
+    key_hash, key_encrypted = AdminImportUsersAdapter._resolve_api_key_material(
+        {
+            "key_hash": legacy_hash,
+            "key_encrypted": legacy_encrypted,
+        }
+    )
+
+    assert key_hash == legacy_hash
+    assert key_encrypted == legacy_encrypted

tests/unit/test_headers.py

@@ -65,7 +65,8 @@ class TestHeaderBuilder:
         builder.add("authorization", "b")
         built = builder.build()
         assert len(built) == 1
-        assert list(built.values()) == ["b"]
+        assert built["Authorization"] == "b"
+        assert "authorization" not in built
 
     def test_add_protected_does_not_override(self) -> None:
         builder = HeaderBuilder()
@@ -135,7 +136,8 @@ class TestBuildUpstreamHeaders:
             extra_headers={"User-Agent": "b"},
         )
         assert len([k for k in result if k.lower() == "user-agent"]) == 1
-        assert result["User-Agent"] == "b"
+        assert result["user-agent"] == "b"
+        assert "User-Agent" not in result
 
     def test_default_content_type(self) -> None:
         result = build_upstream_headers_for_endpoint({}, "openai:chat", "provider")
@@ -169,3 +171,35 @@ class TestCapabilityResolverHeaderParsing:
             request_headers={"x-require-capability": "context_1m"}
         )
         assert reqs == {"context_1m": True}
+
+
+class TestAuthHeaderCasePreservation:
+    def test_build_upstream_headers_preserves_lowercase_authorization_key(self) -> None:
+        result = build_upstream_headers_for_endpoint(
+            {"authorization": "Bearer client-token", "X-Test": "1"},
+            "openai:chat",
+            "provider",
+        )
+        assert "authorization" in result
+        assert "Authorization" not in result
+        assert result["authorization"] == "Bearer provider"
+
+    def test_passthrough_request_builder_preserves_lowercase_authorization_key(self) -> None:
+        from types import SimpleNamespace
+
+        from src.api.handlers.base.request_builder import PassthroughRequestBuilder
+
+        builder = PassthroughRequestBuilder()
+        endpoint = SimpleNamespace(api_family="openai", endpoint_kind="cli", header_rules=None)
+        key = SimpleNamespace(api_key="unused")
+
+        headers = builder.build_headers(
+            original_headers={"authorization": "Bearer client-token"},
+            endpoint=endpoint,
+            key=key,
+            pre_computed_auth=("Authorization", "Bearer provider-token"),
+        )
+
+        assert "authorization" in headers
+        assert "Authorization" not in headers
+        assert headers["authorization"] == "Bearer provider-token"

tests/unit/test_provider_query_failover_endpoint.py

@@ -2,7 +2,9 @@ from types import SimpleNamespace
 
 from src.api.admin.provider_query import (
     _build_direct_test_candidates,
+    _build_test_attempts_from_candidate_keys,
     _filter_test_candidates_by_endpoint,
+    _resolve_test_effective_model,
 )
 
 
@@ -36,3 +38,54 @@ def test_filter_test_candidates_by_endpoint_keeps_matching_candidates() -> None:
 
     assert {candidate.endpoint.id for candidate in filtered} == {endpoint_a.id}
     assert all(candidate.endpoint.id != endpoint_b.id for candidate in filtered)
+
+
+def test_resolve_test_effective_model_prefers_pool_key_mapping() -> None:
+    provider, endpoint_a, _endpoint_b = _build_provider()
+    candidate = _build_direct_test_candidates(provider, endpoint_id=endpoint_a.id)[0]  # type: ignore[arg-type]
+    pool_key = SimpleNamespace(id="pool-key", _pool_mapping_matched_model="mapped-model")
+    request = SimpleNamespace(mode="global", model_name="gpt-4")
+
+    effective = _resolve_test_effective_model(
+        provider=provider,  # type: ignore[arg-type]
+        candidate=candidate,
+        request=request,  # type: ignore[arg-type]
+        gm_obj=None,
+        key=pool_key,
+    )
+
+    assert effective == "mapped-model"
+
+
+def test_build_test_attempts_from_candidate_keys_includes_retry_index() -> None:
+    candidate_keys = [
+        SimpleNamespace(
+            candidate_index=2,
+            retry_index=1,
+            key_id="key-b",
+            key_name="Key B",
+            auth_type="api_key",
+            status="failed",
+            skip_reason=None,
+            error_message="timeout",
+            status_code=504,
+            latency_ms=1200,
+        )
+    ]
+
+    attempts = _build_test_attempts_from_candidate_keys(
+        candidate_keys=candidate_keys,
+        candidate_meta_by_pair={
+            (2, "key-b"): {
+                "endpoint_api_format": "openai:chat",
+                "endpoint_base_url": "https://example.test/v1",
+                "effective_model": "mapped-model",
+            }
+        },
+        candidate_meta_by_index={},
+    )
+
+    assert len(attempts) == 1
+    assert attempts[0].retry_index == 1
+    assert attempts[0].effective_model == "mapped-model"
+    assert attempts[0].endpoint_api_format == "openai:chat"