feat(provider): 重构模型测试对话框，加固 Vertex AI 传输层

模型测试:
- 将消息输入替换为完整 JSON 请求体编辑器，支持格式化和校验
- 新增端点选择面板，测试前可选择目标端点
- 新增调试检查器，可查看每次尝试的请求/响应头和体
- 结果视图改用 HorizontalRequestTimeline 组件展示请求追踪
- endpoint_checker 返回完整调试数据，通过 candidate extra_data 持久化

Vertex AI:
- 改进上下文检测逻辑，不再仅依赖 provider_type，支持从 base_url 推断
- Service Account 密钥现支持自动拉取模型（使用 auth_config 而非 api_key）
- 移除 Gemini Developer API 回退，API Key 仅走 Express 模式
- 端点表单为 Vertex AI 显示格式特定的默认路径模板
- 密钥格式校验仅在 auth_type/api_formats 变更时执行

其他:
- 禁用 ClaudeCode 提供商类型创建入口
- Dialog 组件新增 closeOnBackdrop 属性

frontend/src/api/endpoints/providers.ts

@@ -194,6 +194,7 @@ export interface TestModelFailoverRequest {
   api_format?: string
   endpoint_id?: string
   message?: string
+  request_body?: Record<string, unknown>
   request_id?: string
   concurrency?: number
 }
@@ -212,6 +213,11 @@ export interface TestAttemptDetail {
   error_message?: string | null
   status_code?: number | null
   latency_ms?: number | null
+  request_url?: string | null
+  request_headers?: Record<string, unknown> | null
+  request_body?: unknown
+  response_headers?: Record<string, unknown> | null
+  response_body?: unknown
 }
 
 export interface TestModelFailoverResponse {

frontend/src/components/ui/dialog/Dialog.vue

@@ -109,6 +109,7 @@ const props = defineProps<{
   zIndex?: number // Custom z-index for nested dialogs (default: 60)
   noPadding?: boolean // Disable default content padding
   persistent?: boolean // Prevent closing on backdrop click
+  closeOnBackdrop?: boolean // Allow closing on backdrop click (default: true)
 }>()
 
 // Emits 定义
@@ -145,7 +146,7 @@ function handleClose() {
 
 // 处理背景点击
 function handleBackdropClick() {
-  if (!props.persistent) {
+  if (!props.persistent && props.closeOnBackdrop !== false) {
     handleClose()
   }
 }

frontend/src/composables/useModelTest.ts

@@ -15,6 +15,7 @@ export interface StartTestParams {
   apiFormat?: string
   endpointId?: string
   message?: string
+  requestBody?: Record<string, unknown>
   concurrency?: number
   onSuccess?: (result: TestModelFailoverResponse) => void
   /** Return `true` to indicate the failure has been handled; otherwise the composable sets `testResult`. */
@@ -61,6 +62,16 @@ export function useModelTest(options: UseModelTestOptions) {
     }
   }
 
+  async function refreshTraceSnapshot(reqId: string) {
+    try {
+      const trace = await requestTraceApi.getRequestTrace(reqId, { attemptedOnly: false })
+      if (requestId.value !== reqId) return
+      testTrace.value = trace
+    } catch (err: unknown) {
+      if (isAxiosError(err) && err.response?.status === 404) return
+    }
+  }
+
   function stopPolling(opts: { clearState?: boolean } = {}) {
     tracePollToken += 1
     if (tracePollTimer) {
@@ -128,6 +139,7 @@ export function useModelTest(options: UseModelTestOptions) {
         api_format: params.apiFormat,
         endpoint_id: params.endpointId,
         ...(normalizedMessage ? { message: normalizedMessage } : {}),
+        ...(params.requestBody ? { request_body: params.requestBody } : {}),
         request_id: reqId,
         concurrency: params.concurrency,
       }, {
@@ -135,6 +147,9 @@ export function useModelTest(options: UseModelTestOptions) {
       })
 
       if (result.success) {
+        await refreshTraceSnapshot(reqId)
+        stopPolling({ clearState: false })
+        testResult.value = result
         const successAttempt = result.attempts.find(a => a.status === 'success')
         const latency = successAttempt?.latency_ms != null ? ` (${successAttempt.latency_ms}ms)` : ''
         const mapped = successAttempt?.effective_model && successAttempt.effective_model !== params.modelName
@@ -142,10 +157,10 @@ export function useModelTest(options: UseModelTestOptions) {
           : ''
         params.onSuccess?.(result)
         showSuccess(`${params.displayLabel}${mapped} 测试成功${latency}`)
-        resetState()
         return
       }
 
+      await refreshTraceSnapshot(reqId)
       stopPolling({ clearState: false })
       const handled = params.onFailure?.(result)
       if (!handled) {

frontend/src/features/providers/components/EndpointFormDialog.vue

@@ -105,7 +105,7 @@
                   <div class="space-y-1.5">
                     <Label class="text-xs text-muted-foreground">自定义路径</Label>
                     <Input
-                      :model-value="getEndpointEditState(endpoint.id)?.path ?? (endpoint.custom_path || '')"
+                      :model-value="getDisplayedPath(endpoint)"
                       :placeholder="getDefaultPath(endpoint.api_format, endpoint.base_url) || '留空使用默认'"
                       :disabled="isFixedProvider"
                       @update:model-value="(v) => updateEndpointField(endpoint.id, 'path', v)"
@@ -1276,10 +1276,19 @@ async function preloadDefaultBodyRules(endpoints: ProviderEndpoint[]): Promise<v
 
 // 获取指定 API 格式的默认路径
 function getDefaultPath(apiFormat: string, baseUrl?: string): string {
+  const providerType = (props.provider?.provider_type || '').toLowerCase()
+  if (providerType === 'vertex_ai') {
+    if (apiFormat === 'gemini:chat') {
+      return '/v1/publishers/google/models/{model}:{action}'
+    }
+    if (apiFormat === 'claude:chat') {
+      return '/v1/projects/{project_id}/locations/{region}/publishers/anthropic/models/{model}:{action}'
+    }
+  }
+
   const format = apiFormats.value.find(f => f.value === apiFormat)
   const defaultPath = format?.default_path || ''
   // Codex 端点使用 /responses 而非 /v1/responses
-  const providerType = (props.provider?.provider_type || '').toLowerCase()
   const isCodex = providerType
     ? providerType === 'codex'
     : (!!baseUrl && isCodexUrl(baseUrl))
@@ -1289,6 +1298,13 @@ function getDefaultPath(apiFormat: string, baseUrl?: string): string {
   return defaultPath
 }
 
+function getDisplayedPath(endpoint: ProviderEndpoint): string {
+  if (isFixedProvider.value) {
+    return getDefaultPath(endpoint.api_format, endpoint.base_url)
+  }
+  return getEndpointEditState(endpoint.id)?.path ?? (endpoint.custom_path || '')
+}
+
 // 判断是否是 Codex OAuth 端点
 function isCodexUrl(baseUrl: string): boolean {
   const url = baseUrl.replace(/\/+$/, '')

frontend/src/features/providers/components/ProviderDetailDrawer.vue

@@ -937,6 +937,7 @@
                 ref="modelMappingTabRef"
                 :key="`mapping-${provider.id}`"
                 :provider="provider"
+                :endpoints="endpoints"
                 :provider-keys="providerKeys"
                 :models="providerModels"
                 :mapping-preview="providerMappingPreview"

frontend/src/features/providers/components/ProviderFormDialog.vue

@@ -45,8 +45,11 @@
                   <SelectItem value="vertex_ai">
                     Vertex AI
                   </SelectItem>
-                  <SelectItem value="claude_code">
+                  <SelectItem
-                    ClaudeCode
+                    value="claude_code"
+                    disabled
+                  >
+                    ClaudeCode（暂不可用）
                   </SelectItem>
                   <SelectItem value="codex">
                     Codex
@@ -432,6 +435,11 @@ watch(() => form.value.provider_type, () => {
 
 // 提交表单
 const handleSubmit = async () => {
+  if (!isEditMode.value && form.value.provider_type === 'claude_code') {
+    showError('ClaudeCode 提供商类型暂时禁用', '验证失败')
+    return
+  }
+
   // 月卡类型必须设置周期开始时间
   if (form.value.billing_type === 'monthly_quota' && !form.value.quota_last_reset_at) {
     showError('月卡类型必须设置周期开始时间', '验证失败')

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

@@ -314,13 +314,19 @@
     :result="modelTest.testResult.value"
     mode="direct"
     :selecting-model-name="testingModelName"
+    :endpoints="activeEndpoints"
+    :selected-endpoint="selectedTestEndpoint"
     :testing="modelTest.testing.value"
     :trace="modelTest.testTrace.value"
     :request-id="modelTest.requestId.value"
-    :message-draft="testMessageDraft"
+    :request-body-draft="testRequestBodyDraft"
+    :request-body-error="testRequestBodyError"
+    :start-disabled="!selectedTestEndpoint || !!testRequestBodyError"
     @close="handleTestDialogClose"
+    @back="handleTestDialogBack"
+    @select-endpoint="handleSelectTestEndpoint"
     @start="handleStartMappingTest"
-    @update:message-draft="testMessageDraft = $event"
+    @update:request-body-draft="testRequestBodyDraft = $event"
   />
 </template>
 
@@ -338,6 +344,7 @@ import ModelTestDialog from './ModelTestDialog.vue'
 import { useToast } from '@/composables/useToast'
 import {
   type Model,
+  type ProviderEndpoint,
   type ProviderModelAlias,
   type ProviderMappingPreviewResponse,
 } from '@/api/endpoints'
@@ -345,6 +352,10 @@ import { type EndpointAPIKey } from '@/api/endpoints/keys'
 import { updateModel } from '@/api/endpoints/models'
 import { parseApiError } from '@/utils/errorParser'
 import type { ProviderWithEndpointsSummary } from '@/api/endpoints'
+import {
+  buildDefaultModelTestRequestBody,
+  parseModelTestRequestBodyDraft,
+} from './model-test-request'
 
 interface MappingItem {
   name: string
@@ -372,6 +383,7 @@ interface CombinedMapping {
 
 const props = defineProps<{
   provider: ProviderWithEndpointsSummary
+  endpoints?: ProviderEndpoint[]
   providerKeys?: EndpointAPIKey[]
   models?: Model[]
   mappingPreview?: ProviderMappingPreviewResponse | null
@@ -396,7 +408,11 @@ const testingMapping = ref<string | null>(null)
 const pendingMappingKey = ref<string | null>(null)
 const testingModelName = ref<string | null>(null)
 const preselectedModelId = ref<string | null>(null)
-const testMessageDraft = ref('')
+const selectedTestEndpoint = ref<ProviderEndpoint | null>(null)
+const testRequestBodyDraft = ref('')
+const activeEndpoints = computed(() => (props.endpoints ?? []).filter(endpoint => endpoint.is_active))
+const parsedTestRequestBody = computed(() => parseModelTestRequestBodyDraft(testRequestBodyDraft.value))
+const testRequestBodyError = computed(() => parsedTestRequestBody.value.error)
 
 // 使用 props 传入的数据
 const models = computed(() => props.models ?? [])
@@ -634,31 +650,60 @@ function handleTestDialogClose() {
   pendingMappingKey.value = null
   testingModelName.value = null
   testingMapping.value = null
+  selectedTestEndpoint.value = null
+  testRequestBodyDraft.value = ''
+}
+
+function handleTestDialogBack() {
+  if (modelTest.testing.value) return
+  modelTest.testResult.value = null
+  modelTest.stopPolling()
+}
+
+function handleSelectTestEndpoint(endpointId: string) {
+  const endpoint = activeEndpoints.value.find(item => item.id === endpointId)
+  if (!endpoint) return
+  selectedTestEndpoint.value = endpoint
 }
 
 // 测试映射（直连测试，带故障转移和实时进度）
 function runMappingTest(testingKey: string, modelName: string) {
+  if (activeEndpoints.value.length === 0) {
+    showError('暂无可用于测试的活跃端点')
+    return
+  }
   pendingMappingKey.value = testingKey
   modelTest.testResult.value = null
   modelTest.dialogOpen.value = true
   testingMapping.value = null
   testingModelName.value = modelName
+  selectedTestEndpoint.value = activeEndpoints.value[0] ?? null
+  testRequestBodyDraft.value = buildDefaultModelTestRequestBody(modelName)
 }
 
 async function handleStartMappingTest() {
-  if (modelTest.testing.value || !testingModelName.value || !pendingMappingKey.value) return
+  if (modelTest.testing.value || !testingModelName.value) return
+  const endpoint = selectedTestEndpoint.value || activeEndpoints.value[0]
+  if (!endpoint) {
+    showError('请选择要测试的端点')
+    return
+  }
 
-  const currentMappingKey = pendingMappingKey.value
+  const { value: requestBody, error } = parsedTestRequestBody.value
-  testingMapping.value = currentMappingKey
+  if (!requestBody || error) {
+    showError(`测试请求体无效: ${error || '无效 JSON'}`)
+    return
+  }
+
+  const currentMappingKey = pendingMappingKey.value || testingModelName.value
+  testingMapping.value = pendingMappingKey.value ? currentMappingKey : null
   await modelTest.startTest({
     mode: 'direct',
     modelName: testingModelName.value,
-    displayLabel: `映射 "${testingModelName.value}"`,
+    displayLabel: `[${endpoint.api_format}] 映射 "${testingModelName.value}"`,
-    message: testMessageDraft.value,
+    apiFormat: endpoint.api_format,
-    onSuccess: () => {
+    endpointId: endpoint.id,
-      pendingMappingKey.value = null
+    requestBody,
-      testingModelName.value = null
-    },
   })
   if (pendingMappingKey.value === currentMappingKey) {
     pendingMappingKey.value = null

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

@@ -221,13 +221,14 @@
     :testing="modelTest.testing.value"
     :trace="modelTest.testTrace.value"
     :request-id="modelTest.requestId.value"
-    :show-endpoint-selector="activeEndpoints.length > 1"
+    :request-body-draft="testRequestBodyDraft"
-    :message-draft="testMessageDraft"
+    :request-body-error="testRequestBodyError"
+    :start-disabled="!selectedTestEndpoint || !!testRequestBodyError"
     @close="handleTestDialogClose"
     @back="handleTestDialogBack"
     @start="handleStartPendingTest"
     @select-endpoint="handleSelectTestEndpoint"
-    @update:message-draft="testMessageDraft = $event"
+    @update:request-body-draft="testRequestBodyDraft = $event"
   />
 </template>
 
@@ -250,6 +251,10 @@ import { parseApiError } from '@/utils/errorParser'
 import { formatApiFormat } from '@/api/endpoints/types/api-format'
 import type { ProviderWithEndpointsSummary } from '@/api/endpoints'
 import ModelTestDialog from './ModelTestDialog.vue'
+import {
+  buildDefaultModelTestRequestBody,
+  parseModelTestRequestBodyDraft,
+} from './model-test-request'
 
 const props = defineProps<{
   provider: ProviderWithEndpointsSummary
@@ -275,8 +280,10 @@ const localModels = ref<Model[]>([])
 const togglingModelId = ref<string | null>(null)
 const pendingTestModel = ref<Model | null>(null)
 const selectedTestEndpoint = ref<ProviderEndpoint | null>(null)
-const testMessageDraft = ref('')
+const testRequestBodyDraft = ref('')
 const activeEndpoints = computed(() => (props.endpoints ?? []).filter(endpoint => endpoint.is_active))
+const parsedTestRequestBody = computed(() => parseModelTestRequestBodyDraft(testRequestBodyDraft.value))
+const testRequestBodyError = computed(() => parsedTestRequestBody.value.error)
 const models = computed(() => props.models ?? localModels.value)
 // 按名称排序的模型列表
 const sortedModels = computed(() => {
@@ -442,19 +449,38 @@ function handleTestDialogClose() {
   modelTest.resetState()
   pendingTestModel.value = null
   selectedTestEndpoint.value = null
+  testRequestBodyDraft.value = ''
 }
 
 function handleTestDialogBack() {
   if (modelTest.testing.value) return
   modelTest.testResult.value = null
-  selectedTestEndpoint.value = null
+  modelTest.stopPolling()
 }
 
-async function handleSelectTestEndpoint(endpointId: string) {
+function handleSelectTestEndpoint(endpointId: string) {
-  if (!pendingTestModel.value) return
   const endpoint = activeEndpoints.value.find(item => item.id === endpointId)
   if (!endpoint) return
   selectedTestEndpoint.value = endpoint
+}
+
+async function handleStartPendingTest() {
+  if (modelTest.testing.value) return
+  if (!pendingTestModel.value) return
+
+  const endpoint = selectedTestEndpoint.value || activeEndpoints.value[0]
+  if (!endpoint) {
+    showError('请选择要测试的端点')
+    return
+  }
+
+  const { value: requestBody, error } = parsedTestRequestBody.value
+  if (!requestBody || error) {
+    showError(`测试请求体无效: ${error || '无效 JSON'}`)
+    return
+  }
+
+  selectedTestEndpoint.value = endpoint
   const model = pendingTestModel.value
   const modelName = model.global_model_name || model.provider_model_name
   const endpointPrefix = `[${formatApiFormat(endpoint.api_format)}] `
@@ -464,28 +490,16 @@ async function handleSelectTestEndpoint(endpointId: string) {
     displayLabel: `${endpointPrefix}${modelName}`,
     apiFormat: endpoint.api_format,
     endpointId: endpoint.id,
-    message: testMessageDraft.value,
+    requestBody,
     concurrency: 5,
-    onSuccess: () => {
-      pendingTestModel.value = null
-      selectedTestEndpoint.value = null
-    },
     onError: () => {
       if (activeEndpoints.value.length > 1) {
-        selectedTestEndpoint.value = null
         return true
       }
     },
   })
 }
 
-async function handleStartPendingTest() {
-  if (modelTest.testing.value) return
-  const endpoint = activeEndpoints.value[0]
-  if (!endpoint) return
-  await handleSelectTestEndpoint(endpoint.id)
-}
-
 async function testModelConnection(model: Model) {
   if (modelTest.testing.value) return
 
@@ -495,7 +509,10 @@ async function testModelConnection(model: Model) {
   }
 
   pendingTestModel.value = model
-  selectedTestEndpoint.value = null
+  selectedTestEndpoint.value = activeEndpoints.value[0] ?? null
+  testRequestBodyDraft.value = buildDefaultModelTestRequestBody(
+    model.global_model_name || model.provider_model_name,
+  )
   modelTest.testResult.value = null
   modelTest.dialogOpen.value = true
 }

frontend/src/features/providers/components/provider-tabs/model-test-request.ts

@@ -0,0 +1,47 @@
+const DEFAULT_MODEL_TEST_MESSAGE = 'Hello! This is a test message.'
+
+export function buildDefaultModelTestRequestBody(modelName: string): string {
+  return JSON.stringify({
+    model: modelName,
+    messages: [
+      {
+        role: 'user',
+        content: DEFAULT_MODEL_TEST_MESSAGE,
+      },
+    ],
+    max_tokens: 30,
+    temperature: 0.7,
+    stream: true,
+  }, null, 2)
+}
+
+export function parseModelTestRequestBodyDraft(
+  draft: string,
+): { value: Record<string, unknown> | null; error: string | null } {
+  const normalized = draft.trim()
+  if (!normalized) {
+    return {
+      value: null,
+      error: '测试请求体不能为空',
+    }
+  }
+
+  try {
+    const parsed = JSON.parse(normalized)
+    if (!parsed || typeof parsed !== 'object' || Array.isArray(parsed)) {
+      return {
+        value: null,
+        error: '测试请求体必须是 JSON 对象',
+      }
+    }
+    return {
+      value: parsed as Record<string, unknown>,
+      error: null,
+    }
+  } catch (error) {
+    return {
+      value: null,
+      error: error instanceof Error ? error.message : '无效的 JSON',
+    }
+  }
+}

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

@@ -240,6 +240,10 @@
                           -->{{ proxyTimingBreakdown(currentAttempt.extra_data.proxy) }}<!--
                         -->)</span>
                       </span>
+                      <code
+                        v-if="typeof currentAttempt.extra_data.proxy.node_id === 'string' && currentAttempt.extra_data.proxy.node_id"
+                        class="text-xs font-mono text-muted-foreground"
+                      >节点 Key {{ currentAttempt.extra_data.proxy.node_id }}</code>
                     </span>
                   </div>
                   <div
@@ -462,7 +466,7 @@ interface UsageData {
 }
 
 const props = defineProps<{
-  requestId: string
+  requestId?: string | null
   /** 外部传入的状态码，用于覆盖 trace.final_status 的判断 */
   overrideStatusCode?: number
   /** 请求侧 API 格式（客户端入口格式） */
@@ -471,6 +475,12 @@ const props = defineProps<{
   usageData?: UsageData | null
   /** 请求元数据（用于号池调度组装） */
   requestMetadata?: Record<string, unknown> | null
+  /** 已获取的追踪数据；传入时不再内部拉取 */
+  traceData?: RequestTrace | null
+}>()
+
+const emit = defineEmits<{
+  selectAttempt: [attempt: CandidateRecord | null]
 }>()
 
 // 用量数据（从 props 获取）
@@ -525,7 +535,8 @@ const getFinalStatusBadgeVariant = (status: string): BadgeVariant => {
 
 const loading = ref(false)
 const error = ref<string | null>(null)
-const trace = ref<RequestTrace | null>(null)
+const internalTrace = ref<RequestTrace | null>(null)
+const trace = computed(() => props.traceData ?? internalTrace.value)
 const selectedGroupIndex = ref(0)
 const selectedAttemptIndex = ref(0)
 const hoveredGroupIndex = ref<number | null>(null)
@@ -1045,6 +1056,10 @@ const currentAttempt = computed(() => {
   return selectedGroup.value.allAttempts[selectedAttemptIndex.value] || selectedGroup.value.primary
 })
 
+watch(currentAttempt, (attempt) => {
+  emit('selectAttempt', attempt ?? null)
+}, { immediate: true })
+
 const currentGroupTitle = computed(() => {
   if (!selectedGroup.value || !currentAttempt.value) return ''
   if (selectedGroup.value.isPoolGroup) {
@@ -1224,7 +1239,7 @@ const navigateGroup = (direction: number) => {
 // 加载请求追踪数据
 const isSilentRefresh = ref(false)
 const loadTrace = async (silent = false) => {
-  if (!props.requestId) return
+  if (!props.requestId || props.traceData) return
 
   isSilentRefresh.value = silent
 
@@ -1234,7 +1249,7 @@ const loadTrace = async (silent = false) => {
   error.value = null
 
   try {
-    trace.value = await requestTraceApi.getRequestTrace(props.requestId)
+    internalTrace.value = await requestTraceApi.getRequestTrace(props.requestId)
   } catch (err: unknown) {
     if (!silent) {
       error.value = parseApiError(err, '加载失败')
@@ -1304,12 +1319,31 @@ watch(groupedTimeline, (newGroups) => {
   selectedAttemptIndex.value = 0
 }, { immediate: true })
 
-// 监听 requestId 变化
+// 监听 requestId / 外部 trace 变化
-watch(() => props.requestId, () => {
+watch(
-  selectedGroupIndex.value = 0
+  [() => props.requestId, () => props.traceData],
-  selectedAttemptIndex.value = 0
+  () => {
-  loadTrace()
+    selectedGroupIndex.value = 0
-}, { immediate: true })
+    selectedAttemptIndex.value = 0
+
+    if (props.traceData) {
+      internalTrace.value = null
+      loading.value = false
+      error.value = null
+      return
+    }
+
+    if (!props.requestId) {
+      internalTrace.value = null
+      loading.value = false
+      error.value = null
+      return
+    }
+
+    void loadTrace()
+  },
+  { immediate: true },
+)
 
 defineExpose({ refresh: () => loadTrace(true) })
 

src/api/admin/provider_query.py

@@ -9,6 +9,7 @@ import asyncio
 import json
 import time
 from collections.abc import Awaitable, Callable
+from copy import deepcopy
 from datetime import datetime, timezone
 from typing import TYPE_CHECKING, Any
 from uuid import uuid4
@@ -46,6 +47,11 @@ from src.services.model.upstream_fetcher import (
 from src.services.provider.oauth_token import resolve_oauth_access_token
 from src.services.proxy_node.resolver import resolve_effective_proxy
 from src.services.request.candidate import RequestCandidateService
+from src.services.request.model_test_debug import (
+    get_model_test_debug_from_extra_data,
+    merge_model_test_debug,
+    set_candidate_model_test_debug,
+)
 from src.utils.auth_utils import get_current_user
 
 if TYPE_CHECKING:
@@ -239,6 +245,7 @@ class TestModelFailoverRequest(BaseModel):
     api_format: str | None = None  # 指定 API 格式（endpoint signature）
     endpoint_id: str | None = None  # 指定仅使用该端点测试
     message: str | None = None
+    request_body: dict[str, Any] | None = None
     request_id: str | None = None
     concurrency: int = Field(default=1, ge=1, le=20)
 
@@ -259,6 +266,11 @@ class TestAttemptDetail(BaseModel):
     error_message: str | None = None
     status_code: int | None = None
     latency_ms: int | None = None
+    request_url: str | None = None
+    request_headers: dict[str, Any] | None = None
+    request_body: Any = None
+    response_headers: dict[str, Any] | None = None
+    response_body: Any = None
 
 
 class TestModelFailoverResponse(BaseModel):
@@ -285,6 +297,38 @@ def _resolve_test_message(message: str | None) -> str:
     return normalized or DEFAULT_MODEL_TEST_MESSAGE
 
 
+def _build_test_request_payload(request: TestModelFailoverRequest) -> dict[str, Any]:
+    if isinstance(request.request_body, dict):
+        payload = deepcopy(request.request_body)
+        payload["model"] = request.model_name
+        return payload
+
+    return {
+        "model": request.model_name,
+        "messages": [{"role": "user", "content": _resolve_test_message(request.message)}],
+        "max_tokens": 30,
+        "temperature": 0.7,
+        "stream": True,
+    }
+
+
+def _extract_test_debug_payload(response: dict[str, Any]) -> dict[str, Any] | None:
+    debug = response.get("debug")
+    if not isinstance(debug, dict):
+        return None
+
+    payload: dict[str, Any] = {}
+    request_url = debug.get("request_url")
+    if isinstance(request_url, str) and request_url.strip():
+        payload["request_url"] = request_url.strip()
+
+    for key in ("request_headers", "request_body", "response_headers", "response_body"):
+        if key in debug and debug.get(key) is not None:
+            payload[key] = deepcopy(debug.get(key))
+
+    return payload or None
+
+
 def _test_check_response_has_error(resp: dict[str, Any]) -> bool:
     """快速判断 check_endpoint 结果是否失败。"""
     if resp.get("error"):
@@ -1800,6 +1844,7 @@ async def _run_concurrent_test(
         local_provider: Provider | None = None
         local_endpoint: ProviderEndpoint | None = None
         local_key: ProviderAPIKey | None = None
+        debug_payload: dict[str, Any] | None = None
 
         try:
             if success_event.is_set() or await is_cancelled():
@@ -1833,6 +1878,7 @@ async def _run_concurrent_test(
                 user=user,
                 db=None,
             )
+            debug_payload = _extract_test_debug_payload(response)
             elapsed_ms = max(0, int((time.perf_counter() - started_at) * 1000))
 
             with create_session() as parse_db:
@@ -1856,6 +1902,7 @@ async def _run_concurrent_test(
                     candidate_id=record_id,
                     status_code=200,
                     latency_ms=elapsed_ms,
+                    extra_data=merge_model_test_debug(None, debug_payload),
                 )
 
             if not success_event.is_set():
@@ -1898,6 +1945,7 @@ async def _run_concurrent_test(
                     ),
                     status_code=status_code,
                     latency_ms=elapsed_ms,
+                    extra_data=merge_model_test_debug(None, debug_payload),
                 )
             return {"status": "failed", "error": exc}
 
@@ -2137,6 +2185,9 @@ def _build_test_attempts_from_candidate_keys(
         meta = candidate_meta_by_pair.get((candidate_index, key_id)) or candidate_meta_by_index.get(
             candidate_index, {}
         )
+        debug_payload = get_model_test_debug_from_extra_data(
+            getattr(candidate_key, "extra_data", None)
+        )
 
         attempts.append(
             TestAttemptDetail(
@@ -2155,6 +2206,23 @@ def _build_test_attempts_from_candidate_keys(
                 error_message=getattr(candidate_key, "error_message", None),
                 status_code=getattr(candidate_key, "status_code", None),
                 latency_ms=getattr(candidate_key, "latency_ms", None),
+                request_url=(
+                    str(debug_payload.get("request_url"))
+                    if debug_payload and debug_payload.get("request_url")
+                    else None
+                ),
+                request_headers=(
+                    dict(debug_payload.get("request_headers"))
+                    if debug_payload and isinstance(debug_payload.get("request_headers"), dict)
+                    else None
+                ),
+                request_body=debug_payload.get("request_body") if debug_payload else None,
+                response_headers=(
+                    dict(debug_payload.get("response_headers"))
+                    if debug_payload and isinstance(debug_payload.get("response_headers"), dict)
+                    else None
+                ),
+                response_body=debug_payload.get("response_body") if debug_payload else None,
             )
         )
 
@@ -2285,13 +2353,7 @@ async def test_model_failover(
             error="No available candidates found for this model",
         ).model_dump()
 
-    request_payload = {
+    request_payload = _build_test_request_payload(request)
-        "model": request.model_name,
-        "messages": [{"role": "user", "content": _resolve_test_message(request.message)}],
-        "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()
@@ -2315,6 +2377,7 @@ async def test_model_failover(
             user=current_user,
             db=db,
         )
+        set_candidate_model_test_debug(candidate, _extract_test_debug_payload(response))
         return _extract_test_response_or_raise(
             response=response,
             endpoint=endpoint,

src/api/handlers/base/endpoint_checker.py

@@ -122,6 +122,18 @@ async def run_endpoint_check(
     if result.usage_data:
         response_data["usage"] = result.usage_data
 
+    response_data["debug"] = {
+        "request_url": request.url,
+        "request_headers": _redact_headers(request.headers),
+        "request_body": request.json_body,
+        "response_headers": _redact_headers(result.headers) if result.headers else None,
+        "response_body": (
+            result.raw_response_body
+            if result.raw_response_body is not None
+            else result.response_data
+        ),
+    }
+
     return response_data
 
 
@@ -600,6 +612,7 @@ class EndpointCheckResult:
     response_data: dict[str, Any] | None = None
     error_message: str | None = None
     usage_data: dict[str, Any] | None = None
+    raw_response_body: Any | None = None
 
 
 class HttpRequestExecutor:
@@ -665,6 +678,7 @@ class HttpRequestExecutor:
                             request_id=request_id,
                             response_data=None,
                             error_message=response_data.get("error"),
+                            raw_response_body=response_data.get("response_body"),
                         )
 
                     return EndpointCheckResult(
@@ -673,6 +687,7 @@ class HttpRequestExecutor:
                         response_time_ms=response_time_ms,
                         request_id=request_id,
                         response_data=response_data.get("final_response"),
+                        raw_response_body=response_data.get("final_response"),
                     )
                 else:
                     # 非流式请求：直接读取响应
@@ -705,6 +720,9 @@ class HttpRequestExecutor:
                             response_time_ms=response_time_ms,
                             request_id=request_id,
                             response_data=response_data,
+                            raw_response_body=(
+                                response_data if response_data is not None else response.text
+                            ),
                         )
                     else:
                         error_body = response.text[:500] if response.text else "(empty)"
@@ -743,10 +761,12 @@ class HttpRequestExecutor:
                 headers = dict(response.headers)
 
                 if response.status_code != 200:
+                    # 读取上限 16KB 用于 debug response_body；
+                    # error 字段仅取前 500 字符，避免日志/展示过长
                     error_body = ""
                     async for chunk in response.aiter_text():
                         error_body += chunk
-                        if len(error_body) > 500:
+                        if len(error_body) > 16384:
                             break
                     logger.debug(
                         "[{}] check_endpoint | stream error | {}",
@@ -757,6 +777,7 @@ class HttpRequestExecutor:
                         "error": f"HTTP {response.status_code}: {error_body[:500]}",
                         "status_code": response.status_code,
                         "headers": headers,
+                        "response_body": error_body,
                     }
 
                 # 收集 SSE 事件（兼容多种 API 格式）
@@ -841,7 +862,7 @@ class HttpRequestExecutor:
 
         except Exception as e:
             logger.warning("[{}] check_endpoint | stream error | {}", request.api_format, e)
-            return {"error": str(e), "status_code": 500, "headers": {}}
+            return {"error": str(e), "status_code": 500, "headers": {}, "response_body": None}
 
 
 class UsageCalculator:
@@ -1109,6 +1130,7 @@ class ErrorHandler:
                 "original_error": str(error),
                 "retryable": True,
             },
+            raw_response_body=None,
         )
 
     @staticmethod
@@ -1129,6 +1151,7 @@ class ErrorHandler:
                 "retryable": True,
                 "timeout_seconds": request.timeout,
             },
+            raw_response_body=None,
         )
 
     @staticmethod
@@ -1175,6 +1198,7 @@ class ErrorHandler:
                 "response_body": error.response.text[:500] if error.response.text else "",
                 "retryable": retryable,
             },
+            raw_response_body=error.response.text if error.response.text else None,
         )
 
     @staticmethod
@@ -1196,6 +1220,7 @@ class ErrorHandler:
                 "details": error.details,
                 "retryable": error.status_code >= 500 or error.status_code == 429,
             },
+            raw_response_body=None,
         )
 
     @staticmethod
@@ -1215,6 +1240,7 @@ class ErrorHandler:
                 "original_error": str(error),
                 "retryable": False,
             },
+            raw_response_body=None,
         )
 
     @staticmethod
@@ -1238,6 +1264,7 @@ class ErrorHandler:
                 "original_error": str(error),
                 "retryable": False,
             },
+            raw_response_body=None,
         )
 
 

src/api/handlers/base/handler_adapter_base.py

@@ -372,11 +372,17 @@ class HandlerAdapterBase(ApiAdapter):
         )
         from src.core.api_format.headers import HeaderBuilder
         from src.core.provider_types import ProviderType
+        from src.services.provider.adapters.vertex_ai.transport import is_vertex_ai_context
 
         validated_base_url = cls._validate_test_base_url(base_url)
         is_antigravity = provider_type == ProviderType.ANTIGRAVITY
         is_gemini_cli = provider_type == ProviderType.GEMINI_CLI
-        is_vertex = provider_type == ProviderType.VERTEX_AI
+        is_vertex = is_vertex_ai_context(
+            base_url=validated_base_url,
+            provider_type=provider_type,
+            endpoint=provider_endpoint,
+            key=provider_api_key,
+        )
         is_kiro = provider_type == ProviderType.KIRO
         is_oauth = auth_type == "oauth"
         vertex_auth_info: Any | None = None

src/api/handlers/gemini/adapter.py

@@ -253,6 +253,7 @@ class GeminiChatAdapter(ChatAdapterBase):
             evaluate_condition,
         )
         from src.core.api_format.headers import HeaderBuilder
+        from src.services.provider.adapters.vertex_ai.transport import is_vertex_ai_context
 
         # Gemini需要从request_data或model_name参数获取model名称
         effective_model_name = model_name or request_data.get("model", "")
@@ -264,7 +265,12 @@ class GeminiChatAdapter(ChatAdapterBase):
 
         is_antigravity = provider_type and provider_type.lower() == "antigravity"
         is_gemini_cli = provider_type and provider_type.lower() == "gemini_cli"
-        is_vertex = provider_type and provider_type.lower() == "vertex_ai"
+        is_vertex = is_vertex_ai_context(
+            base_url=base_url,
+            provider_type=provider_type,
+            endpoint=provider_endpoint,
+            key=provider_api_key,
+        )
         is_oauth = auth_type == "oauth"
         vertex_auth_info: Any | None = None
 

src/services/candidate/recorder.py

@@ -45,6 +45,7 @@ class CandidateRecorder:
                     error_message=getattr(row, "error_message", None),
                     status_code=getattr(row, "status_code", None),
                     latency_ms=getattr(row, "latency_ms", None),
+                    extra_data=getattr(row, "extra_data", None),
                 )
             )
 

src/services/candidate/schema.py

@@ -32,6 +32,7 @@ class CandidateKey:
     error_message: str | None = None
     status_code: int | None = None
     latency_ms: int | None = None
+    extra_data: dict[str, Any] | None = None
 
     def to_dict(self) -> dict[str, Any]:
         data: dict[str, Any] = {
@@ -52,6 +53,7 @@ class CandidateKey:
             "error_message": self.error_message,
             "status_code": self.status_code,
             "latency_ms": self.latency_ms,
+            "extra_data": self.extra_data,
         }
         # drop Nones for compact audit payload
         return {k: v for k, v in data.items() if v is not None}

src/services/model/fetch_scheduler.py

@@ -54,9 +54,7 @@ KEY_FETCH_TIMEOUT_SECONDS = 120
 MODEL_FETCH_HTTP_TIMEOUT = 10.0
 
 # 启动时首次自动获取开关与延迟
-MODEL_FETCH_STARTUP_ENABLED = (
+MODEL_FETCH_STARTUP_ENABLED = os.getenv("MODEL_FETCH_STARTUP_ENABLED", "true").lower() == "true"
-    os.getenv("MODEL_FETCH_STARTUP_ENABLED", "true").lower() == "true"
-)
 MODEL_FETCH_STARTUP_DELAY_SECONDS = max(
     0,
     int(os.getenv("MODEL_FETCH_STARTUP_DELAY_SECONDS", "10")),
@@ -487,13 +485,21 @@ class ModelFetchScheduler:
                 self._update_key_error(prepared.key_id, f"OAuth token resolution failed: {e}")
                 return "error"
         else:
-            try:
+            is_vertex_service_account = (
-                api_key_value = crypto_service.decrypt(prepared.encrypted_api_key)
+                prepared.provider_type.lower() == ProviderType.VERTEX_AI.value
-            except Exception:
+                and prepared.auth_type in ("service_account", "vertex_ai")
-                self._update_key_error(prepared.key_id, "Decrypt error")
+            )
-                return "error"
 
-            # Best-effort: decrypt auth_config if present (e.g. Antigravity project_id).
+            if is_vertex_service_account:
+                api_key_value = "__placeholder__"
+            else:
+                try:
+                    api_key_value = crypto_service.decrypt(prepared.encrypted_api_key)
+                except Exception:
+                    self._update_key_error(prepared.key_id, "Decrypt error")
+                    return "error"
+
+            # Best-effort: decrypt auth_config if present (e.g. Antigravity project_id / Vertex SA JSON).
             if prepared.encrypted_auth_config:
                 try:
                     parsed = json.loads(crypto_service.decrypt(prepared.encrypted_auth_config))
@@ -582,9 +588,13 @@ class ModelFetchScheduler:
                 db.commit()
                 return "error"
 
-            # Service Account 类型不支持自动获取模型（Vertex AI SA / 旧 vertex_ai auth_type）
             auth_type = getattr(key, "auth_type", "api_key") or "api_key"
-            if auth_type in ("service_account", "vertex_ai"):
+            provider_type = str(getattr(provider, "provider_type", "") or "")
+            is_vertex_service_account = (
+                provider_type.strip().lower() == ProviderType.VERTEX_AI.value
+                and auth_type in ("service_account", "vertex_ai")
+            )
+            if auth_type in ("service_account", "vertex_ai") and not is_vertex_service_account:
                 key.last_models_fetch_error = (
                     "auto_fetch_models 暂不支持 Service Account 类型的 Key"
                 )
@@ -610,9 +620,7 @@ class ModelFetchScheduler:
                 key.last_models_fetch_at = now
                 db.commit()
                 return "error"
-
             encrypted_auth_config = getattr(key, "auth_config", None)
-            provider_type = str(getattr(provider, "provider_type", "") or "")
 
             return PreparedModelsFetchContext(
                 key_id=key_id,

src/services/provider/adapters/vertex_ai/plugin.py

@@ -1,9 +1,9 @@
 """Vertex AI provider plugin — 统一注册入口。
 
 注册 Vertex AI 对各通用 registry / capability registry 的 hooks：
-- Transport Hook (URL 构建，支持 API Key / Service Account 双策略)
+- Transport Hook (URL 构建：Gemini 走 Express mode，Claude 走 Service Account)
-- Model Fetcher (专用上游模型获取链路，不走通用 /v1beta/models / /v1/models)
+- Model Fetcher (专用上游模型获取链路)
-- Provider Format Capability（跨格式支持：同一 Provider 同时访问 Gemini 和 Claude 模型）
+- Provider Format Capability（跨格式支持：同一 Provider 可配置 Gemini / Claude）
 """
 
 from __future__ import annotations
@@ -18,8 +18,6 @@ from src.services.provider.adapters.vertex_ai.transport import get_effective_for
 
 # Vertex AI 公共 API 根
 _VERTEX_API_BASE = "https://aiplatform.googleapis.com"
-# Gemini Developer API（API Key 场景兜底）
-_GEMINI_DEV_BASE = "https://generativelanguage.googleapis.com"
 
 _MODEL_PAGE_SIZE = 100
 _MODEL_MAX_PAGES = 20
@@ -261,6 +259,7 @@ async def _fetch_models_vertex_api_key(
     ctx: Any,
     auth_config: dict[str, Any] | None,
 ) -> tuple[list[dict[str, Any]], list[str], bool]:
+    """API Key 仅抓取 Vertex AI Express mode 的 Google publisher models。"""
     api_key = str(ctx.api_key_value or "").strip()
     if not api_key or api_key == "__placeholder__":
         return [], ["vertex_ai(api_key): missing api key"], False
@@ -275,7 +274,7 @@ async def _fetch_models_vertex_api_key(
         _build_google_publisher_list_url(base) for base in _iter_endpoint_base_urls(ctx)
     ]
 
-    # 1) Vertex API list (publisher=google)
+    # Vertex Express mode list (publisher=google)
     for url in vertex_list_urls:
         headers = {"Accept": "application/json", **endpoint_headers}
         models, err, success = await _fetch_models_from_url(
@@ -297,29 +296,6 @@ async def _fetch_models_vertex_api_key(
             continue
         all_models.extend(models)
 
-    # 2) 兜底：Gemini Developer API
-    if not all_models:
-        fallback_url = f"{_GEMINI_DEV_BASE}/v1beta/models"
-        headers = {"Accept": "application/json", **endpoint_headers}
-        models, err, success = await _fetch_models_from_url(
-            client,
-            url=fallback_url,
-            headers=headers,
-            params={"key": api_key, "pageSize": _MODEL_PAGE_SIZE},
-            auth_config=auth_config,
-            fallback_publisher="google",
-        )
-        if success:
-            has_success = True
-        if err:
-            labeled = f"{fallback_url}: {err}"
-            if _is_soft_not_found(err):
-                soft_errors.append(labeled)
-            else:
-                hard_errors.append(labeled)
-        else:
-            all_models.extend(models)
-
     deduped = _dedupe_models(all_models)
     if deduped:
         return deduped, hard_errors, has_success or True
@@ -338,6 +314,7 @@ async def _fetch_models_vertex_service_account(
     auth_config: dict[str, Any] | None,
     client_kwargs: dict[str, Any],
 ) -> tuple[list[dict[str, Any]], list[str], bool]:
+    """Service Account 抓取 Vertex AI Google + Anthropic publisher models。"""
     if not isinstance(auth_config, dict):
         return [], ["vertex_ai(service_account): missing auth_config"], False
 
@@ -420,8 +397,8 @@ async def fetch_models_vertex_ai(
 ) -> tuple[list[dict], list[str], bool, dict[str, Any] | None]:
     """Vertex AI 专用模型获取链路。
 
-    - API Key: 优先请求 Vertex publisher models，失败时兜底 Gemini Developer API
+    - API Key: 仅请求 Vertex AI Express mode 的 Gemini models
-    - Service Account: 使用 SA 凭证换取 Bearer Token，按 region + publisher 查询
+    - Service Account: 使用 SA 凭证换取 Bearer Token，按 region 查询 Gemini + Claude models
     """
     from src.services.proxy_node.resolver import build_proxy_client_kwargs
 

src/services/provider/adapters/vertex_ai/transport.py

@@ -1,12 +1,16 @@
 """Vertex AI URL 构建（Transport Hook）。
 
-根据 auth_type 选择两种完全不同的 URL 构建策略：
+Vertex AI Gemini / Imagen 支持两种认证路径：
 
-- API Key: 全局端点，简化路径
+- API Key + Gemini/Imagen (Express mode):
   https://aiplatform.googleapis.com/v1/publishers/google/models/{model}:{action}?key={API_KEY}
+- Service Account + Gemini/Imagen:
+  https://{region}-aiplatform.googleapis.com/v1/projects/{project_id}/locations/{region}/publishers/google/models/{model}:{action}
 
-- Service Account: 区域端点，完整路径
+Claude 仍走标准 Vertex AI Service Account 路径：
-  https://{region}-aiplatform.googleapis.com/v1/projects/{project_id}/locations/{region}/publishers/{publisher}/models/{model}:{action}
+
+- Service Account + Claude:
+  https://{region}-aiplatform.googleapis.com/v1/projects/{project_id}/locations/{region}/publishers/anthropic/models/{model}:{action}
 """
 
 from __future__ import annotations
@@ -16,6 +20,7 @@ from typing import Any
 from urllib.parse import urlencode
 
 from src.core.logger import logger
+from src.core.provider_types import ProviderType, normalize_provider_type
 from src.services.provider.adapters.vertex_ai.constants import (
     API_KEY_BASE_URL,
     DEFAULT_FORMAT,
@@ -23,7 +28,34 @@ from src.services.provider.adapters.vertex_ai.constants import (
     MODEL_FORMAT_MAPPING,
 )
 from src.services.provider.format import normalize_endpoint_signature
-from src.services.provider.transport import redact_url_for_log
+from src.services.provider.transport import looks_like_vertex_ai_host, redact_url_for_log
+
+
+def is_vertex_ai_context(
+    *,
+    base_url: str | None = None,
+    provider_type: Any = None,
+    endpoint: Any = None,
+    key: Any = None,
+) -> bool:
+    """Best-effort 判断当前测试/请求上下文是否应视为 Vertex AI。"""
+    if normalize_provider_type(provider_type) == ProviderType.VERTEX_AI.value:
+        return True
+
+    for obj in (endpoint, key):
+        provider = getattr(obj, "provider", None) if obj is not None else None
+        if normalize_provider_type(getattr(provider, "provider_type", None)) == (
+            ProviderType.VERTEX_AI.value
+        ):
+            return True
+
+    candidate_base_url = str(base_url or getattr(endpoint, "base_url", "") or "").strip()
+    if not candidate_base_url:
+        return False
+
+    endpoint_sig = str(getattr(endpoint, "api_format", "") or "").strip()
+    auth_type = str(getattr(key, "auth_type", "") or "").strip()
+    return looks_like_vertex_ai_host(candidate_base_url, endpoint_sig, auth_type)
 
 
 def get_effective_format(
@@ -95,28 +127,36 @@ def build_vertex_ai_url(
     key: Any = None,
     decrypted_auth_config: dict[str, Any] | None = None,
 ) -> str:
-    """Vertex AI transport hook — 统一 URL 构建入口。
+    """Vertex AI transport hook — 统一 URL 构建入口。"""
+    from src.core.exceptions import InvalidRequestException
 
-    根据 key.auth_type 分派到 API Key 或 Service Account 两种策略。
+    model = str((path_params or {}).get("model", "") or "").strip()
-    """
+    if not model:
-    auth_type = getattr(key, "auth_type", "api_key") if key else "api_key"
+        raise InvalidRequestException("Vertex AI 请求缺少 model 参数")
+
+    auth_type = str(getattr(key, "auth_type", "api_key") or "api_key").strip().lower()
+    is_claude_model = model.startswith("claude-")
 
     if auth_type == "api_key":
+        if is_claude_model:
+            raise InvalidRequestException(
+                "Vertex API Key 不支持 Claude 模型，请改用 Service Account 认证。"
+            )
         return _build_api_key_url(
             key=key,
             path_params=path_params,
             query_params=effective_query_params,
             is_stream=is_stream,
         )
-    else:
+
-        # service_account（以及向后兼容旧的 "vertex_ai" auth_type）
+    # service_account（以及向后兼容旧的 "vertex_ai" auth_type）
-        return _build_service_account_url(
+    return _build_service_account_url(
-            key=key,
+        key=key,
-            path_params=path_params,
+        path_params=path_params,
-            query_params=effective_query_params,
+        query_params=effective_query_params,
-            is_stream=is_stream,
+        is_stream=is_stream,
-            decrypted_auth_config=decrypted_auth_config,
+        decrypted_auth_config=decrypted_auth_config,
-        )
+    )
 
 
 def _build_api_key_url(
@@ -137,11 +177,6 @@ def _build_api_key_url(
     if not model:
         raise InvalidRequestException("Vertex AI 请求缺少 model 参数")
 
-    if str(model).startswith("claude-"):
-        raise InvalidRequestException(
-            "Vertex API Key 不支持 Claude 模型，请改用 Service Account 认证。"
-        )
-
     action = "streamGenerateContent" if is_stream else "generateContent"
     path = f"/v1/publishers/google/models/{model}:{action}"
     url = f"{API_KEY_BASE_URL}{path}"
@@ -177,10 +212,7 @@ def _build_service_account_url(
     is_stream: bool = False,
     decrypted_auth_config: dict[str, Any] | None = None,
 ) -> str:
-    """构建 Service Account 认证的区域端点 URL。
+    """构建 Service Account 认证的 Vertex AI 区域端点 URL。"""
-
-    格式: https://{region}-aiplatform.googleapis.com/v1/projects/{project_id}/locations/{region}/publishers/{publisher}/models/{model}:{action}
-    """
     from src.core.crypto import crypto_service
     from src.core.exceptions import InvalidRequestException
 
@@ -227,11 +259,7 @@ def _build_service_account_url(
     else:
         region = "global"
 
-    # 判断是 Claude 还是 Gemini 模型
+    if model.startswith("claude-"):
-    is_claude_model = model.startswith("claude-")
-
-    # 根据模型类型确定 publisher 和 action
-    if is_claude_model:
         publisher = "anthropic"
         action = "streamRawPredict" if is_stream else "rawPredict"
     else:
@@ -249,7 +277,7 @@ def _build_service_account_url(
     # 添加查询参数
     effective_query_params = dict(query_params) if query_params else {}
     # Gemini 流式请求使用 SSE 格式，Claude 不需要
-    if is_stream and not is_claude_model:
+    if is_stream and not model.startswith("claude-"):
         effective_query_params.setdefault("alt", "sse")
     # 移除不适用于 Vertex AI 的参数
     effective_query_params.pop("beta", None)

src/services/provider/transport.py

@@ -11,7 +11,7 @@ from __future__ import annotations
 
 import re
 from typing import TYPE_CHECKING, Any, Callable
-from urllib.parse import urlencode
+from urllib.parse import urlencode, urlparse
 
 from src.core.api_format import (
     EndpointKind,
@@ -72,6 +72,34 @@ def redact_url_for_log(url: str) -> str:
     return _SENSITIVE_QUERY_PARAMS_PATTERN.sub(r"\1\2=***", url)
 
 
+# Vertex AI host 白名单：aiplatform.googleapis.com 及其区域子域名
+_VERTEX_AI_HOST = "aiplatform.googleapis.com"
+_VERTEX_AI_ENDPOINT_SIGS = {"gemini:chat", "claude:chat"}
+_VERTEX_AI_AUTH_TYPES = {"api_key", "service_account", "vertex_ai"}
+
+
+def looks_like_vertex_ai_host(
+    base_url: str,
+    endpoint_sig: str = "",
+    auth_type: str = "",
+) -> bool:
+    """根据 base_url + endpoint_sig/auth_type 判断是否属于 Vertex AI。
+
+    用于历史数据兼容：部分旧记录缺少 provider_type，但 base_url 已固定到 aiplatform。
+    """
+    try:
+        host = (urlparse(base_url).netloc or "").strip().lower()
+    except Exception:
+        return False
+    if not host:
+        return False
+    if host != _VERTEX_AI_HOST and not host.endswith(f".{_VERTEX_AI_HOST}"):
+        return False
+    sig = endpoint_sig.strip().lower()
+    at = auth_type.strip().lower()
+    return sig in _VERTEX_AI_ENDPOINT_SIGS or at in _VERTEX_AI_AUTH_TYPES
+
+
 def _normalize_base_url(base_url: str, path: str) -> str:
     """
     规范化 base_url，去除末尾的斜杠和可能与 path 重复的版本前缀。
@@ -138,6 +166,16 @@ def _get_provider_type(
         if isinstance(pt, str) and pt.strip():
             return pt.strip().lower()
 
+    # Fallback: 历史 Vertex 数据可能缺少 provider_type，但 base_url 已固定到 aiplatform。
+    try:
+        base_url = str(getattr(endpoint, "base_url", "") or "").strip()
+        endpoint_sig = str(getattr(endpoint, "api_format", "") or "").strip()
+        auth_type = str(getattr(key, "auth_type", "") or "").strip() if key else ""
+        if base_url and looks_like_vertex_ai_host(base_url, endpoint_sig, auth_type):
+            return ProviderType.VERTEX_AI.value
+    except Exception:
+        pass
+
     return None
 
 

src/services/provider_keys/key_command_service.py

@@ -59,7 +59,7 @@ def _validate_vertex_api_formats(
     if auth_type == "api_key":
         allowed = {"gemini:chat"}
     elif auth_type in {"service_account", "vertex_ai"}:
-        allowed = {"gemini:chat", "claude:chat"}
+        allowed = {"claude:chat", "gemini:chat"}
     else:
         return
 
@@ -310,14 +310,16 @@ def _prepare_update_key_payload(
         exclude_key_id=key_id,
     )
 
-    # Vertex Provider: auth_type 与 api_formats 的组合必须合法
+    # Vertex Provider: 仅在 auth_type/api_formats 变更时校验组合，
-    provider = getattr(key, "provider", None)
+    # 避免历史旧数据在无关编辑时被强制阻断。
-    effective_api_formats = update_data.get("api_formats", key.api_formats)
+    if "auth_type" in update_data or "api_formats" in update_data:
-    _validate_vertex_api_formats(
+        provider = getattr(key, "provider", None)
-        getattr(provider, "provider_type", None),
+        effective_api_formats = update_data.get("api_formats", key.api_formats)
-        target_auth_type,
+        _validate_vertex_api_formats(
-        effective_api_formats,
+            getattr(provider, "provider_type", None),
-    )
+            target_auth_type,
+            effective_api_formats,
+        )
 
     if "api_key" in update_data:
         api_key_raw = update_data["api_key"]

src/services/request/executor.py

@@ -21,6 +21,10 @@ from src.services.provider.format import normalize_endpoint_signature
 from src.services.rate_limit.adaptive_reservation import get_adaptive_reservation_manager
 from src.services.rate_limit.adaptive_rpm import get_adaptive_rpm_manager
 from src.services.request.candidate import RequestCandidateService
+from src.services.request.model_test_debug import (
+    get_candidate_model_test_debug,
+    merge_model_test_debug,
+)
 
 
 @dataclass
@@ -219,6 +223,13 @@ class RequestExecutor:
                     _pi = await resolve_proxy_info_async(_eff_proxy)
                     if _pi:
                         _extra["proxy"] = _pi
+                    _extra = (
+                        merge_model_test_debug(
+                            _extra,
+                            get_candidate_model_test_debug(candidate),
+                        )
+                        or _extra
+                    )
                     RequestCandidateService.mark_candidate_success(
                         db=self.db,
                         candidate_id=candidate_id,

src/services/request/model_test_debug.py

@@ -0,0 +1,57 @@
+from __future__ import annotations
+
+from copy import deepcopy
+from typing import Any
+
+MODEL_TEST_DEBUG_KEY = "model_test_debug"
+MODEL_TEST_DEBUG_ATTR = "_model_test_debug"
+
+
+def normalize_model_test_debug_payload(debug_payload: Any) -> dict[str, Any] | None:
+    if not isinstance(debug_payload, dict):
+        return None
+
+    normalized: dict[str, Any] = {}
+    for key in (
+        "request_url",
+        "request_headers",
+        "request_body",
+        "response_headers",
+        "response_body",
+    ):
+        value = debug_payload.get(key)
+        if value is None:
+            continue
+        normalized[key] = deepcopy(value)
+
+    return normalized or None
+
+
+def set_candidate_model_test_debug(candidate: Any, debug_payload: Any) -> None:
+    normalized = normalize_model_test_debug_payload(debug_payload)
+    if normalized is None:
+        return
+    setattr(candidate, MODEL_TEST_DEBUG_ATTR, normalized)
+
+
+def get_candidate_model_test_debug(candidate: Any) -> dict[str, Any] | None:
+    return normalize_model_test_debug_payload(getattr(candidate, MODEL_TEST_DEBUG_ATTR, None))
+
+
+def merge_model_test_debug(
+    extra_data: dict[str, Any] | None,
+    debug_payload: Any,
+) -> dict[str, Any] | None:
+    normalized = normalize_model_test_debug_payload(debug_payload)
+    if normalized is None:
+        return extra_data
+
+    merged = dict(extra_data or {})
+    merged[MODEL_TEST_DEBUG_KEY] = normalized
+    return merged
+
+
+def get_model_test_debug_from_extra_data(extra_data: Any) -> dict[str, Any] | None:
+    if not isinstance(extra_data, dict):
+        return None
+    return normalize_model_test_debug_payload(extra_data.get(MODEL_TEST_DEBUG_KEY))

src/services/task/execute/error_handler.py

@@ -17,6 +17,10 @@ from src.core.exceptions import (
 from src.core.logger import logger
 from src.core.provider_types import ProviderType
 from src.services.request.candidate import RequestCandidateService
+from src.services.request.model_test_debug import (
+    get_candidate_model_test_debug,
+    merge_model_test_debug,
+)
 from src.services.task.execute.pool import TaskPoolOperationsService
 from src.services.task.request_state import RequestBodyState
 
@@ -206,6 +210,9 @@ class TaskErrorOperationsService:
         )
         _proxy_info = await resolve_proxy_info_async(_eff_proxy)
         _proxy_extra: dict[str, Any] | None = {"proxy": _proxy_info} if _proxy_info else None
+        _proxy_extra = merge_model_test_debug(
+            _proxy_extra, get_candidate_model_test_debug(candidate)
+        )
 
         if not isinstance(exec_err, ExecutionError):
             RequestCandidateService.mark_candidate_failed(
@@ -410,6 +417,13 @@ class TaskErrorOperationsService:
             }
             if _proxy_info:
                 serializable_extra_data["proxy"] = _proxy_info
+            serializable_extra_data = (
+                merge_model_test_debug(
+                    serializable_extra_data,
+                    get_candidate_model_test_debug(candidate),
+                )
+                or serializable_extra_data
+            )
 
             if isinstance(converted_error, ThinkingSignatureException):
                 action = self.handle_thinking_signature_error(

tests/services/test_model_fetch_scheduler.py

@@ -6,7 +6,9 @@ import pytest
 
 import src.services.model.fetch_scheduler as fetch_scheduler_module
 from src.services.model.fetch_scheduler import (
+    EndpointFetchConfig,
     ModelFetchScheduler,
+    PreparedModelsFetchContext,
     _aggregate_models_for_cache,
     _run_key_fetch_workers,
 )
@@ -108,3 +110,70 @@ async def test_perform_fetch_all_keys_scans_in_batches(monkeypatch: pytest.Monke
 
     assert batch_requests == [None, "b"]
     assert processed_batches == [["a", "b"], ["c"]]
+
+
+@pytest.mark.asyncio
+async def test_fetch_models_for_key_by_id_vertex_service_account_uses_auth_config(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    scheduler = ModelFetchScheduler()
+    captured: dict[str, object] = {}
+
+    prepared = PreparedModelsFetchContext(
+        key_id="key-vertex-sa",
+        provider_id="provider-vertex",
+        provider_name="Vertex",
+        provider_type="vertex_ai",
+        auth_type="service_account",
+        encrypted_api_key="ENC_PLACEHOLDER",
+        encrypted_auth_config="ENC_AUTH_CONFIG",
+        format_to_endpoint={
+            "gemini:chat": EndpointFetchConfig(base_url="https://aiplatform.googleapis.com"),
+        },
+        proxy_config=None,
+    )
+
+    monkeypatch.setattr(scheduler, "_prepare_fetch_context", lambda key_id: prepared)
+
+    async def fake_fetch_models_for_key(ctx, *, timeout_seconds: float):
+        captured["api_key_value"] = ctx.api_key_value
+        captured["auth_config"] = ctx.auth_config
+        captured["timeout_seconds"] = timeout_seconds
+        return ([], [], True, None)
+
+    async def fake_update_key_after_fetch(
+        key_id: str,
+        provider_id: str,
+        provider_name: str,
+        all_models: list[dict],
+        errors: list[str],
+        has_success: bool,
+        upstream_metadata=None,
+    ) -> str:
+        captured["update_key_id"] = key_id
+        captured["update_provider_id"] = provider_id
+        return "success"
+
+    def fake_decrypt(value: str) -> str:
+        if value == "ENC_AUTH_CONFIG":
+            return (
+                '{"project_id":"demo-project","client_email":"svc@example.com",'
+                '"private_key":"-----BEGIN PRIVATE KEY-----\\nTEST\\n-----END PRIVATE KEY-----\\n"}'
+            )
+        raise AssertionError(f"unexpected decrypt call for {value}")
+
+    monkeypatch.setattr(fetch_scheduler_module, "fetch_models_for_key", fake_fetch_models_for_key)
+    monkeypatch.setattr(scheduler, "_update_key_after_fetch", fake_update_key_after_fetch)
+    monkeypatch.setattr(fetch_scheduler_module.crypto_service, "decrypt", fake_decrypt)
+
+    result = await scheduler._fetch_models_for_key_by_id("key-vertex-sa")
+
+    assert result == "success"
+    assert captured["api_key_value"] == "__placeholder__"
+    assert captured["auth_config"] == {
+        "project_id": "demo-project",
+        "client_email": "svc@example.com",
+        "private_key": "-----BEGIN PRIVATE KEY-----\nTEST\n-----END PRIVATE KEY-----\n",
+    }
+    assert captured["update_key_id"] == "key-vertex-sa"
+    assert captured["update_provider_id"] == "provider-vertex"

tests/services/test_provider_keys_key_command_service.py

@@ -2,6 +2,7 @@ from __future__ import annotations
 
 import sys
 import types
+from contextlib import contextmanager
 from datetime import datetime, timezone
 from types import SimpleNamespace
 from typing import Any, cast
@@ -167,6 +168,42 @@ def test_prepare_update_payload_vertex_to_oauth_clears_auth_config(
     assert prepared.update_data["api_key"] == "ENC:__placeholder__"
 
 
+def test_validate_vertex_api_formats_api_key_allows_gemini_only() -> None:
+    command_module._validate_vertex_api_formats("vertex_ai", "api_key", ["gemini:chat"])
+
+    with pytest.raises(InvalidRequestException, match="claude:chat"):
+        command_module._validate_vertex_api_formats("vertex_ai", "api_key", ["claude:chat"])
+
+
+def test_validate_vertex_api_formats_service_account_allows_gemini_and_claude() -> None:
+    command_module._validate_vertex_api_formats("vertex_ai", "service_account", ["claude:chat"])
+    command_module._validate_vertex_api_formats("vertex_ai", "service_account", ["gemini:chat"])
+    command_module._validate_vertex_api_formats(
+        "vertex_ai", "service_account", ["gemini:chat", "claude:chat"]
+    )
+
+    with pytest.raises(InvalidRequestException, match="openai:chat"):
+        command_module._validate_vertex_api_formats("vertex_ai", "service_account", ["openai:chat"])
+
+
+def test_prepare_update_payload_allows_unrelated_update_for_legacy_vertex_combo() -> None:
+    key = _build_key(
+        auth_type="service_account",
+        api_formats=["gemini:chat"],
+        provider=SimpleNamespace(provider_type="vertex_ai"),
+    )
+    key_data = EndpointAPIKeyUpdate.model_validate({"name": "legacy-key"})
+
+    prepared = command_module._prepare_update_key_payload(
+        db=cast(Any, _NoQueryDB()),
+        key=cast(Any, key),
+        key_id="key-1",
+        key_data=key_data,
+    )
+
+    assert prepared.update_data["name"] == "legacy-key"
+
+
 def test_clear_oauth_invalid_response_invalidates_caches(
     monkeypatch: pytest.MonkeyPatch,
 ) -> None:
@@ -346,6 +383,12 @@ async def test_batch_delete_endpoint_keys_response_cleans_related_references(
     ]
     db = _FakeBatchDeleteDB(keys)
 
+    @contextmanager
+    def _fake_get_db_context() -> Any:
+        yield db
+
+    monkeypatch.setattr(command_module, "get_db_context", _fake_get_db_context)
+
     result = await command_module.batch_delete_endpoint_keys_response(
         cast(Any, db),
         ["key-1", "key-2"],

tests/services/test_provider_transport.py

@@ -1,4 +1,7 @@
 from dataclasses import dataclass
+from types import SimpleNamespace
+
+import pytest
 
 from src.services.provider.transport import build_provider_url
 
@@ -61,3 +64,52 @@ def test_gemini_non_stream_does_not_add_alt() -> None:
 
     assert url.endswith("/v1beta/models/gemini-1.5-pro:generateContent")
     assert "alt=" not in url
+
+
+def test_vertex_gemini_api_key_base_url_uses_vertex_transport_without_provider_type(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    endpoint = _DummyEndpoint(
+        base_url="https://aiplatform.googleapis.com",
+        api_format="gemini:chat",
+    )
+    key = SimpleNamespace(auth_type="api_key", api_key="enc-key")
+
+    from src.core.crypto import crypto_service
+
+    monkeypatch.setattr(crypto_service, "decrypt", lambda value: "test-key")
+
+    url = build_provider_url(
+        endpoint,  # type: ignore[arg-type] - test stub
+        path_params={"model": "gemini-3.1-pro-preview"},
+        is_stream=False,
+        key=key,  # type: ignore[arg-type] - test stub
+    )
+
+    assert (
+        url == "https://aiplatform.googleapis.com/v1/publishers/google/models/"
+        "gemini-3.1-pro-preview:generateContent?key=test-key"
+    )
+
+
+def test_vertex_gemini_service_account_base_url_uses_vertex_transport_without_provider_type() -> (
+    None
+):
+    endpoint = _DummyEndpoint(
+        base_url="https://aiplatform.googleapis.com",
+        api_format="gemini:chat",
+    )
+    key = SimpleNamespace(auth_type="service_account", auth_config={"project_id": "demo-project"})
+
+    url = build_provider_url(
+        endpoint,  # type: ignore[arg-type] - test stub
+        path_params={"model": "gemini-3.1-pro-preview"},
+        is_stream=False,
+        key=key,  # type: ignore[arg-type] - test stub
+        decrypted_auth_config={"project_id": "demo-project", "region": "global"},
+    )
+
+    assert (
+        url == "https://aiplatform.googleapis.com/v1/projects/demo-project/locations/global/"
+        "publishers/google/models/gemini-3.1-pro-preview:generateContent"
+    )

tests/services/test_upstream_fetcher_vertex_ai_models.py

@@ -52,7 +52,7 @@ async def test_fetch_models_for_key_vertex_api_key_custom_fetcher() -> None:
 
 
 @pytest.mark.asyncio
-async def test_fetch_models_for_key_vertex_service_account_ignores_soft_404_when_success() -> None:
+async def test_fetch_models_for_key_vertex_service_account_fetches_google_and_claude() -> None:
     auth_config = {
         "project_id": "demo-project",
         "client_email": "svc@example.iam.gserviceaccount.com",
@@ -74,16 +74,27 @@ async def test_fetch_models_for_key_vertex_service_account_ignores_soft_404_when
         (
             [
                 {
-                    "id": "gemini-2.0-flash",
+                    "id": "gemini-3.1-pro-preview",
                     "owned_by": "google",
-                    "display_name": "Gemini 2.0 Flash",
+                    "display_name": "Gemini 3.1 Pro Preview",
                     "api_format": "gemini:chat",
                 }
             ],
             None,
             True,
         ),
-        ([], "HTTP 404: not found", False),
+        (
+            [
+                {
+                    "id": "claude-3-7-sonnet@20250219",
+                    "owned_by": "anthropic",
+                    "display_name": "Claude 3.7 Sonnet",
+                    "api_format": "claude:chat",
+                }
+            ],
+            None,
+            True,
+        ),
     ]
 
     with (
@@ -111,7 +122,8 @@ async def test_fetch_models_for_key_vertex_service_account_ignores_soft_404_when
     assert errors == []
     assert meta is None
     ids = {m.get("id") for m in models}
-    assert "gemini-2.0-flash" in ids
+    assert "gemini-3.1-pro-preview" in ids
+    assert "claude-3-7-sonnet@20250219" in ids
 
 
 @pytest.mark.asyncio
@@ -132,7 +144,6 @@ async def test_fetch_models_for_key_vertex_api_key_returns_soft_404_when_all_fai
             AsyncMock(
                 side_effect=[
                     ([], "HTTP 404: not found", False),
-                    ([], "HTTP 404: not found", False),
                 ]
             ),
         ),

tests/services/test_vertex_ai_transport.py

@@ -0,0 +1,94 @@
+from __future__ import annotations
+
+from types import SimpleNamespace
+from urllib.parse import parse_qs, urlparse
+
+import pytest
+
+from src.core.exceptions import InvalidRequestException
+from src.services.provider.adapters.vertex_ai.transport import build_vertex_ai_url
+
+
+def test_build_vertex_ai_url_uses_express_mode_for_gemini_api_key(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    key = SimpleNamespace(auth_type="api_key", api_key="enc-key")
+
+    from src.core.crypto import crypto_service
+
+    monkeypatch.setattr(crypto_service, "decrypt", lambda value: "test-key")
+
+    url = build_vertex_ai_url(
+        SimpleNamespace(),
+        is_stream=True,
+        effective_query_params={"foo": "bar"},
+        path_params={"model": "gemini-2.5-pro"},
+        key=key,
+    )
+
+    parsed = urlparse(url)
+    assert parsed.scheme == "https"
+    assert parsed.netloc == "aiplatform.googleapis.com"
+    assert parsed.path == "/v1/publishers/google/models/gemini-2.5-pro:streamGenerateContent"
+    assert parse_qs(parsed.query) == {
+        "foo": ["bar"],
+        "key": ["test-key"],
+        "alt": ["sse"],
+    }
+
+
+def test_build_vertex_ai_url_uses_standard_vertex_path_for_gemini_service_account() -> None:
+    key = SimpleNamespace(auth_type="service_account")
+
+    url = build_vertex_ai_url(
+        SimpleNamespace(),
+        is_stream=False,
+        effective_query_params={"foo": "bar", "beta": "1"},
+        path_params={"model": "gemini-3.1-pro-preview"},
+        key=key,
+        decrypted_auth_config={"project_id": "demo-project", "region": "global"},
+    )
+
+    parsed = urlparse(url)
+    assert parsed.scheme == "https"
+    assert parsed.netloc == "aiplatform.googleapis.com"
+    assert (
+        parsed.path
+        == "/v1/projects/demo-project/locations/global/publishers/google/models/gemini-3.1-pro-preview:generateContent"
+    )
+    assert parse_qs(parsed.query) == {"foo": ["bar"]}
+
+
+def test_build_vertex_ai_url_rejects_claude_api_key() -> None:
+    key = SimpleNamespace(auth_type="api_key")
+
+    with pytest.raises(InvalidRequestException, match="Claude 模型"):
+        build_vertex_ai_url(
+            SimpleNamespace(),
+            is_stream=False,
+            effective_query_params={},
+            path_params={"model": "claude-3-7-sonnet@20250219"},
+            key=key,
+        )
+
+
+def test_build_vertex_ai_url_uses_standard_vertex_path_for_claude_service_account() -> None:
+    key = SimpleNamespace(auth_type="service_account")
+
+    url = build_vertex_ai_url(
+        SimpleNamespace(),
+        is_stream=False,
+        effective_query_params={"foo": "bar", "beta": "1"},
+        path_params={"model": "claude-3-7-sonnet@20250219"},
+        key=key,
+        decrypted_auth_config={"project_id": "demo-project", "region": "global"},
+    )
+
+    parsed = urlparse(url)
+    assert parsed.scheme == "https"
+    assert parsed.netloc == "aiplatform.googleapis.com"
+    assert (
+        parsed.path
+        == "/v1/projects/demo-project/locations/global/publishers/anthropic/models/claude-3-7-sonnet@20250219:rawPredict"
+    )
+    assert parse_qs(parsed.query) == {"foo": ["bar"]}

tests/unit/test_handler_adapter_base.py

@@ -1,9 +1,11 @@
+from types import SimpleNamespace
 from typing import Any
 
 import pytest
 
 from src.api.handlers.claude.adapter import ClaudeChatAdapter
 from src.api.handlers.gemini.adapter import GeminiChatAdapter
+from src.core.provider_auth_types import ProviderAuthInfo
 
 
 @pytest.mark.asyncio
@@ -132,3 +134,333 @@ async def test_gemini_check_endpoint_passes_original_body_to_body_rules(
     assert captured["original_body"] == captured["json_body"]
     assert result["status_code"] == 200
     assert result["json_body"] == captured["json_body"]
+
+
+@pytest.mark.asyncio
+async def test_gemini_check_endpoint_uses_provider_transport_for_vertex_ai(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    from src.api.handlers.base import endpoint_checker as endpoint_checker_module
+    from src.services.provider import auth as provider_auth_module
+    from src.services.provider import transport as provider_transport_module
+
+    captured: dict[str, Any] = {}
+
+    async def fake_run_endpoint_check(**kwargs: Any) -> dict[str, Any]:
+        captured["url"] = kwargs["url"]
+        captured["headers"] = kwargs["headers"]
+        captured["json_body"] = kwargs["json_body"]
+        return {"status_code": 200}
+
+    async def fake_get_provider_auth(endpoint: Any, key: Any) -> None:
+        captured["provider_auth_endpoint"] = endpoint
+        captured["provider_auth_key"] = key
+        return None
+
+    def fake_build_provider_url(
+        endpoint: Any,
+        *,
+        query_params: dict[str, Any] | None = None,
+        path_params: dict[str, Any] | None = None,
+        is_stream: bool = False,
+        key: Any = None,
+        decrypted_auth_config: dict[str, Any] | None = None,
+    ) -> str:
+        captured["build_provider_url"] = {
+            "endpoint": endpoint,
+            "query_params": query_params,
+            "path_params": path_params,
+            "is_stream": is_stream,
+            "key": key,
+            "decrypted_auth_config": decrypted_auth_config,
+        }
+        return (
+            "https://aiplatform.googleapis.com/v1/publishers/google/models/"
+            "gemini-2.5-pro:streamGenerateContent?key=test-key"
+        )
+
+    def fake_build_request_body(
+        cls: type[GeminiChatAdapter],
+        request_data: dict[str, Any] | None = None,
+        *,
+        base_url: str | None = None,
+        provider_type: str | None = None,
+    ) -> dict[str, Any]:
+        del cls, request_data, base_url, provider_type
+        return {
+            "contents": [{"role": "user", "parts": [{"text": "hello"}]}],
+            "generationConfig": {"temperature": 0.1},
+        }
+
+    monkeypatch.setattr(endpoint_checker_module, "run_endpoint_check", fake_run_endpoint_check)
+    monkeypatch.setattr(provider_auth_module, "get_provider_auth", fake_get_provider_auth)
+    monkeypatch.setattr(provider_transport_module, "build_provider_url", fake_build_provider_url)
+    monkeypatch.setattr(
+        GeminiChatAdapter, "build_request_body", classmethod(fake_build_request_body)
+    )
+
+    endpoint = SimpleNamespace(id="ep-vertex-gemini")
+    key = SimpleNamespace(id="key-vertex-gemini", auth_type="api_key")
+
+    result = await GeminiChatAdapter.check_endpoint(
+        client=None,  # type: ignore[arg-type]
+        base_url="https://generativelanguage.googleapis.com",
+        api_key="test-key",
+        request_data={"model": "gemini-2.5-pro", "stream": True},
+        provider_type="vertex_ai",
+        provider_endpoint=endpoint,
+        provider_api_key=key,
+        model_name="gemini-2.5-pro",
+    )
+
+    assert result["status_code"] == 200
+    assert (
+        captured["url"] == "https://aiplatform.googleapis.com/v1/publishers/google/models/"
+        "gemini-2.5-pro:streamGenerateContent?key=test-key"
+    )
+    assert captured["build_provider_url"]["path_params"] == {"model": "gemini-2.5-pro"}
+    assert captured["build_provider_url"]["is_stream"] is True
+    assert captured["build_provider_url"]["key"] is key
+    assert captured["headers"] == {}
+
+
+@pytest.mark.asyncio
+async def test_gemini_check_endpoint_infers_vertex_ai_from_base_url(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    from src.api.handlers.base import endpoint_checker as endpoint_checker_module
+    from src.services.provider import auth as provider_auth_module
+    from src.services.provider import transport as provider_transport_module
+
+    captured: dict[str, Any] = {}
+
+    async def fake_run_endpoint_check(**kwargs: Any) -> dict[str, Any]:
+        captured["url"] = kwargs["url"]
+        return {"status_code": 200}
+
+    async def fake_get_provider_auth(endpoint: Any, key: Any) -> None:
+        captured["provider_auth_endpoint"] = endpoint
+        captured["provider_auth_key"] = key
+        return None
+
+    def fake_build_provider_url(
+        endpoint: Any,
+        *,
+        query_params: dict[str, Any] | None = None,
+        path_params: dict[str, Any] | None = None,
+        is_stream: bool = False,
+        key: Any = None,
+        decrypted_auth_config: dict[str, Any] | None = None,
+    ) -> str:
+        captured["build_provider_url"] = {
+            "endpoint": endpoint,
+            "query_params": query_params,
+            "path_params": path_params,
+            "is_stream": is_stream,
+            "key": key,
+            "decrypted_auth_config": decrypted_auth_config,
+        }
+        return (
+            "https://aiplatform.googleapis.com/v1/publishers/google/models/"
+            "gemini-3.1-pro-preview:generateContent?key=test-key"
+        )
+
+    monkeypatch.setattr(endpoint_checker_module, "run_endpoint_check", fake_run_endpoint_check)
+    monkeypatch.setattr(provider_auth_module, "get_provider_auth", fake_get_provider_auth)
+    monkeypatch.setattr(provider_transport_module, "build_provider_url", fake_build_provider_url)
+
+    endpoint = SimpleNamespace(id="ep-aiplatform", api_format="gemini:chat")
+    key = SimpleNamespace(id="key-aiplatform", auth_type="api_key")
+
+    result = await GeminiChatAdapter.check_endpoint(
+        client=None,  # type: ignore[arg-type]
+        base_url="https://aiplatform.googleapis.com",
+        api_key="test-key",
+        request_data={"model": "gemini-3.1-pro-preview", "stream": False},
+        provider_endpoint=endpoint,
+        provider_api_key=key,
+        model_name="gemini-3.1-pro-preview",
+    )
+
+    assert result["status_code"] == 200
+    assert (
+        captured["url"] == "https://aiplatform.googleapis.com/v1/publishers/google/models/"
+        "gemini-3.1-pro-preview:generateContent?key=test-key"
+    )
+    assert captured["build_provider_url"]["path_params"] == {"model": "gemini-3.1-pro-preview"}
+    assert captured["build_provider_url"]["is_stream"] is False
+
+
+@pytest.mark.asyncio
+async def test_gemini_check_endpoint_uses_provider_transport_for_vertex_service_account(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    from src.api.handlers.base import endpoint_checker as endpoint_checker_module
+    from src.services.provider import auth as provider_auth_module
+    from src.services.provider import transport as provider_transport_module
+
+    captured: dict[str, Any] = {}
+    auth_info = ProviderAuthInfo(
+        auth_header="Authorization",
+        auth_value="Bearer vertex-token",
+        decrypted_auth_config={"project_id": "demo-project", "region": "global"},
+    )
+
+    async def fake_run_endpoint_check(**kwargs: Any) -> dict[str, Any]:
+        captured["url"] = kwargs["url"]
+        captured["headers"] = kwargs["headers"]
+        return {"status_code": 200}
+
+    async def fake_get_provider_auth(endpoint: Any, key: Any) -> ProviderAuthInfo:
+        captured["provider_auth_endpoint"] = endpoint
+        captured["provider_auth_key"] = key
+        return auth_info
+
+    def fake_build_provider_url(
+        endpoint: Any,
+        *,
+        query_params: dict[str, Any] | None = None,
+        path_params: dict[str, Any] | None = None,
+        is_stream: bool = False,
+        key: Any = None,
+        decrypted_auth_config: dict[str, Any] | None = None,
+    ) -> str:
+        captured["build_provider_url"] = {
+            "endpoint": endpoint,
+            "query_params": query_params,
+            "path_params": path_params,
+            "is_stream": is_stream,
+            "key": key,
+            "decrypted_auth_config": decrypted_auth_config,
+        }
+        return (
+            "https://aiplatform.googleapis.com/v1/projects/demo-project/locations/global/"
+            "publishers/google/models/gemini-3.1-pro-preview:generateContent"
+        )
+
+    monkeypatch.setattr(endpoint_checker_module, "run_endpoint_check", fake_run_endpoint_check)
+    monkeypatch.setattr(provider_auth_module, "get_provider_auth", fake_get_provider_auth)
+    monkeypatch.setattr(provider_transport_module, "build_provider_url", fake_build_provider_url)
+
+    endpoint = SimpleNamespace(id="ep-vertex-gemini-sa", api_format="gemini:chat")
+    key = SimpleNamespace(id="key-vertex-gemini-sa", auth_type="service_account")
+
+    result = await GeminiChatAdapter.check_endpoint(
+        client=None,  # type: ignore[arg-type]
+        base_url="https://aiplatform.googleapis.com",
+        api_key="ignored",
+        request_data={"model": "gemini-3.1-pro-preview", "stream": False},
+        provider_endpoint=endpoint,
+        provider_api_key=key,
+        model_name="gemini-3.1-pro-preview",
+    )
+
+    assert result["status_code"] == 200
+    assert (
+        captured["url"]
+        == "https://aiplatform.googleapis.com/v1/projects/demo-project/locations/global/"
+        "publishers/google/models/gemini-3.1-pro-preview:generateContent"
+    )
+    assert captured["build_provider_url"]["path_params"] == {"model": "gemini-3.1-pro-preview"}
+    assert captured["build_provider_url"]["is_stream"] is False
+    assert (
+        captured["build_provider_url"]["decrypted_auth_config"] == auth_info.decrypted_auth_config
+    )
+    assert captured["headers"]["Authorization"] == "Bearer vertex-token"
+
+
+@pytest.mark.asyncio
+async def test_claude_check_endpoint_uses_provider_transport_for_vertex_ai(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    from src.api.handlers.base import endpoint_checker as endpoint_checker_module
+    from src.services.provider import auth as provider_auth_module
+    from src.services.provider import transport as provider_transport_module
+
+    captured: dict[str, Any] = {}
+    auth_info = ProviderAuthInfo(
+        auth_header="Authorization",
+        auth_value="Bearer vertex-token",
+        decrypted_auth_config={"project_id": "demo-project", "region": "global"},
+    )
+
+    async def fake_run_endpoint_check(**kwargs: Any) -> dict[str, Any]:
+        captured["url"] = kwargs["url"]
+        captured["headers"] = kwargs["headers"]
+        captured["json_body"] = kwargs["json_body"]
+        return {"status_code": 200}
+
+    async def fake_get_provider_auth(endpoint: Any, key: Any) -> ProviderAuthInfo:
+        captured["provider_auth_endpoint"] = endpoint
+        captured["provider_auth_key"] = key
+        return auth_info
+
+    def fake_build_provider_url(
+        endpoint: Any,
+        *,
+        query_params: dict[str, Any] | None = None,
+        path_params: dict[str, Any] | None = None,
+        is_stream: bool = False,
+        key: Any = None,
+        decrypted_auth_config: dict[str, Any] | None = None,
+    ) -> str:
+        captured["build_provider_url"] = {
+            "endpoint": endpoint,
+            "query_params": query_params,
+            "path_params": path_params,
+            "is_stream": is_stream,
+            "key": key,
+            "decrypted_auth_config": decrypted_auth_config,
+        }
+        return (
+            "https://aiplatform.googleapis.com/v1/projects/demo-project/locations/global/"
+            "publishers/anthropic/models/claude-3-7-sonnet@20250219:rawPredict"
+        )
+
+    def fake_build_request_body(
+        cls: type[ClaudeChatAdapter],
+        request_data: dict[str, Any] | None = None,
+        *,
+        base_url: str | None = None,
+        provider_type: str | None = None,
+    ) -> dict[str, Any]:
+        del cls, request_data, base_url, provider_type
+        return {
+            "messages": [{"role": "user", "content": "hello"}],
+            "max_tokens": 32,
+        }
+
+    monkeypatch.setattr(endpoint_checker_module, "run_endpoint_check", fake_run_endpoint_check)
+    monkeypatch.setattr(provider_auth_module, "get_provider_auth", fake_get_provider_auth)
+    monkeypatch.setattr(provider_transport_module, "build_provider_url", fake_build_provider_url)
+    monkeypatch.setattr(
+        ClaudeChatAdapter, "build_request_body", classmethod(fake_build_request_body)
+    )
+
+    endpoint = SimpleNamespace(id="ep-vertex-claude")
+    key = SimpleNamespace(id="key-vertex-claude", auth_type="service_account")
+
+    result = await ClaudeChatAdapter.check_endpoint(
+        client=None,  # type: ignore[arg-type]
+        base_url="https://api.anthropic.com/v1",
+        api_key="ignored",
+        request_data={"model": "claude-3-7-sonnet@20250219", "stream": False},
+        provider_type="vertex_ai",
+        provider_endpoint=endpoint,
+        provider_api_key=key,
+        model_name="claude-3-7-sonnet@20250219",
+    )
+
+    assert result["status_code"] == 200
+    assert (
+        captured["url"]
+        == "https://aiplatform.googleapis.com/v1/projects/demo-project/locations/global/"
+        "publishers/anthropic/models/claude-3-7-sonnet@20250219:rawPredict"
+    )
+    assert captured["build_provider_url"]["path_params"] == {"model": "claude-3-7-sonnet@20250219"}
+    assert captured["build_provider_url"]["is_stream"] is False
+    assert captured["build_provider_url"]["key"] is key
+    assert (
+        captured["build_provider_url"]["decrypted_auth_config"] == auth_info.decrypted_auth_config
+    )
+    assert captured["headers"]["Authorization"] == "Bearer vertex-token"