feat(codex): 拆分openai:compact为独立端点，简化Codex请求为透传模式

- 新增openai:compact端点类型(EndpointKind.COMPACT)，独立于openai:cli - OpenAICompactAdapter继承OpenAICliAdapter，自动标记compact模式 - Codex请求补丁改为纯透传：仅清理内部标记，不再修改客户端payload - stream_policy支持openai:compact独立策略，compact端点移除stream字段 - candidate_builder支持compact回退到cli端点 - auth_type: vertex_ai重命名为service_account，保持向后兼容 - Vertex Provider新增api_formats与auth_type组合校验 - KeyAllowedModels对话框改为从Provider获取模型，展示provider_model_name - Dialog内Select组件自动禁用Portal，修复层级遮挡问题 - 新增Codex compact端点回填迁移脚本
2026-09-02 01:10:23 +08:00 · 2026-03-01 23:55:26 +08:00
parent 4bf3a453e7
commit 97d42703da
37 changed files with 650 additions and 286 deletions
--- a/src/api/admin/models/routing.py
+++ b/src/api/admin/models/routing.py
@@ -422,6 +422,7 @@ class AdminGetModelRoutingPreviewAdapter(AdminApiAdapter):
            preferred_order = [
                "openai:chat",
                "openai:cli",
+                "openai:compact",
                "openai:video",
                "claude:chat",
                "claude:cli",
--- a/src/api/admin/system.py
+++ b/src/api/admin/system.py
@@ -838,15 +838,20 @@ class AdminGetApiFormatsAdapter(AdminApiAdapter):
        def _label_for(sig: str) -> str:
            fam, kind = (sig.split(":", 1) + [""])[:2]
            fam_title = {"claude": "Claude", "openai": "OpenAI", "gemini": "Gemini"}.get(fam, fam)
-            kind_title = {"chat": "Chat", "cli": "CLI", "video": "Video", "image": "Image"}.get(
-                kind, kind
-            )
+            kind_title = {
+                "chat": "Chat",
+                "cli": "CLI",
+                "compact": "Compact",
+                "video": "Video",
+                "image": "Image",
+            }.get(kind, kind)
            return f"{fam_title} {kind_title}".strip()

        endpoint_defs = list_endpoint_definitions()
        preferred_order = [
            "openai:chat",
            "openai:cli",
+            "openai:compact",
            "openai:video",
            "claude:chat",
            "claude:cli",
--- a/src/api/handlers/openai_cli/init.py
+++ b/src/api/handlers/openai_cli/init.py
@@ -2,10 +2,11 @@
 OpenAI CLI 透传处理器
 """

-from src.api.handlers.openai_cli.adapter import OpenAICliAdapter
+from src.api.handlers.openai_cli.adapter import OpenAICliAdapter, OpenAICompactAdapter
 from src.api.handlers.openai_cli.handler import OpenAICliMessageHandler

 __all__ = [
    "OpenAICliAdapter",
+    "OpenAICompactAdapter",
    "OpenAICliMessageHandler",
 ]
--- a/src/api/handlers/openai_cli/adapter.py
+++ b/src/api/handlers/openai_cli/adapter.py
@@ -15,7 +15,7 @@ from src.api.handlers.base.cli_adapter_base import CliAdapterBase, register_cli_
 from src.api.handlers.base.cli_handler_base import CliMessageHandlerBase
 from src.api.handlers.openai.adapter import OpenAIChatAdapter
 from src.config.settings import config
-from src.core.api_format import ApiFamily
+from src.core.api_format import ApiFamily, EndpointKind
 from src.utils.url_utils import is_codex_url


@@ -166,3 +166,18 @@ class OpenAICliAdapter(CliAdapterBase):


 __all__ = ["OpenAICliAdapter"]
+
+
+@register_cli_adapter
+class OpenAICompactAdapter(OpenAICliAdapter):
+    """OpenAI Compact Responses adapter (/v1/responses/compact)."""
+
+    FORMAT_ID = "openai:compact"
+    ENDPOINT_KIND = EndpointKind.COMPACT
+    name = "openai.compact"
+
+    def __init__(self, allowed_api_formats: list[str] | None = None):
+        super().__init__(allowed_api_formats=allowed_api_formats, compact=True)
+
+
+__all__.append("OpenAICompactAdapter")
--- a/src/api/public/models.py
+++ b/src/api/public/models.py
@@ -36,7 +36,7 @@ router = APIRouter(tags=["System Catalog"])

 # 各格式对应的 API 格式列表（包括对应的 CLI 格式）
 _CLAUDE_FORMATS = ["claude:chat", "claude:cli"]
-_OPENAI_FORMATS = ["openai:chat", "openai:cli"]
+_OPENAI_FORMATS = ["openai:chat", "openai:cli", "openai:compact"]
 _GEMINI_FORMATS = ["gemini:chat", "gemini:cli"]

 # 所有格式（用于格式转换时的查询）
--- a/src/api/public/openai.py
+++ b/src/api/public/openai.py
@@ -15,7 +15,7 @@ from sqlalchemy.orm import Session

 from src.api.base.pipeline import ApiRequestPipeline
 from src.api.handlers.openai import OpenAIChatAdapter
-from src.api.handlers.openai_cli import OpenAICliAdapter
+from src.api.handlers.openai_cli import OpenAICliAdapter, OpenAICompactAdapter
 from src.database import get_db

 router = APIRouter(tags=["OpenAI API"])
@@ -68,7 +68,7 @@ async def create_responses_compact(

    **认证方式**: Bearer Token（API Key 或 JWT Token）
    """
-    adapter = OpenAICliAdapter(compact=True)
+    adapter = OpenAICompactAdapter()
    return await pipeline.run(
        adapter=adapter,
        http_request=http_request,
--- a/src/api/user_me/routes.py
+++ b/src/api/user_me/routes.py
@@ -1263,6 +1263,7 @@ class ListAvailableModelsAdapter(AuthenticatedApiAdapter):
        all_formats = [
            "openai:chat",
            "openai:cli",
+            "openai:compact",
            "claude:chat",
            "claude:cli",
            "gemini:chat",
--- a/src/core/api_format/conversion/normalizers/openai_cli.py
+++ b/src/core/api_format/conversion/normalizers/openai_cli.py
@@ -199,24 +199,18 @@ class OpenAICliNormalizer(FormatNormalizer):

        return internal

-    # Codex 需要的 include 项
-    _CODEX_REQUIRED_INCLUDE = "reasoning.encrypted_content"
-
    def request_from_internal(
        self,
        internal: InternalRequest,
        *,
        target_variant: str | None = None,
    ) -> dict[str, Any]:
+        _ = target_variant
        openai_cli_extra = internal.extra.get("openai_cli", {})
-        is_compact = bool(openai_cli_extra.get("_aether_compact"))
-        is_codex = str(target_variant or "").lower() == "codex" and not is_compact

        result: dict[str, Any] = {
            "model": internal.model,
-            "input": self._internal_messages_to_input(
-                internal.messages, system_to_developer=is_codex
-            ),
+            "input": self._internal_messages_to_input(internal.messages, system_to_developer=False),
        }

        # 合并 instructions，如果没有则使用 system
@@ -229,20 +223,17 @@ class OpenAICliNormalizer(FormatNormalizer):
        # 统一添加该字段以确保兼容性
        result["instructions"] = instructions_text or ""

-        # max_output_tokens/temperature/top_p: Codex 不支持，标准 API 可选
-        if not is_codex:
-            if internal.max_tokens is not None:
-                # Responses API 使用 max_output_tokens
-                result["max_output_tokens"] = internal.max_tokens
-            if internal.temperature is not None:
-                result["temperature"] = internal.temperature
-            if internal.top_p is not None:
-                result["top_p"] = internal.top_p
+        if internal.max_tokens is not None:
+            # Responses API 使用 max_output_tokens
+            result["max_output_tokens"] = internal.max_tokens
+        if internal.temperature is not None:
+            result["temperature"] = internal.temperature
+        if internal.top_p is not None:
+            result["top_p"] = internal.top_p

        if internal.stop_sequences:
            result["stop"] = list(internal.stop_sequences)
-        # Codex 强制要求 stream=true；其他情况尊重客户端请求
-        result["stream"] = True if is_codex else bool(internal.stream)
+        result["stream"] = bool(internal.stream)

        if internal.tools:
            # Responses API 使用扁平结构: {type, name, description, parameters}
@@ -308,26 +299,10 @@ class OpenAICliNormalizer(FormatNormalizer):
            if key not in handled_keys and key not in result:
                result[key] = value

-        # 统一设置 store=false（Codex 强制要求，标准 API 兼容）
+        # 标准 Responses API 默认设置 store=false
        if "store" not in result:
            result["store"] = False

-        # Codex 特定设置（覆盖/删除不支持的字段）
-        if is_codex:
-            result["parallel_tool_calls"] = True
-            # 和 codex passthrough patch 保持一致：固定 include 列表
-            result["include"] = [self._CODEX_REQUIRED_INCLUDE]
-            # 删除 Codex 不支持的字段
-            for key in (
-                "previous_response_id",
-                "service_tier",
-                "max_completion_tokens",
-                "truncation",
-                "context_management",
-                "user",
-            ):
-                result.pop(key, None)
-
        return result

    # =========================
--- a/src/core/api_format/detection.py
+++ b/src/core/api_format/detection.py
@@ -62,6 +62,10 @@ def _detect_data_format(
            return EndpointSignature(api_family=ApiFamily.CLAUDE, endpoint_kind=EndpointKind.CLI)
        return EndpointSignature(api_family=ApiFamily.CLAUDE, endpoint_kind=EndpointKind.CHAT)

+    # OpenAI compact: /responses/compact
+    if "/responses/compact" in normalized:
+        return EndpointSignature(api_family=ApiFamily.OPENAI, endpoint_kind=EndpointKind.COMPACT)
+
    # OpenAI CLI: /responses
    if "/responses" in normalized:
        return EndpointSignature(api_family=ApiFamily.OPENAI, endpoint_kind=EndpointKind.CLI)
--- a/src/core/api_format/enums.py
+++ b/src/core/api_format/enums.py
@@ -36,6 +36,7 @@ class EndpointKind(str, Enum):

    CHAT = "chat"
    CLI = "cli"
+    COMPACT = "compact"
    VIDEO = "video"
    IMAGE = "image"

--- a/src/core/api_format/metadata.py
+++ b/src/core/api_format/metadata.py
@@ -123,6 +123,19 @@ _ENDPOINT_DEFINITIONS: dict[tuple[ApiFamily, EndpointKind], EndpointDefinition]
        protected_keys=frozenset({"authorization", "content-type"}),
        data_format_id="openai_responses",
    ),
+    (ApiFamily.OPENAI, EndpointKind.COMPACT): EndpointDefinition(
+        api_family=ApiFamily.OPENAI,
+        endpoint_kind=EndpointKind.COMPACT,
+        aliases=("openai_compact", "responses_compact"),
+        default_path="/v1/responses/compact",
+        auth_method=AuthMethod.BEARER,
+        auth_header="Authorization",
+        auth_type="bearer",
+        protected_keys=frozenset({"authorization", "content-type"}),
+        # compact endpoint is non-streaming by design.
+        stream_in_body=False,
+        data_format_id="openai_responses",
+    ),
    (ApiFamily.OPENAI, EndpointKind.VIDEO): EndpointDefinition(
        api_family=ApiFamily.OPENAI,
        endpoint_kind=EndpointKind.VIDEO,
--- a/src/services/health/endpoint.py
+++ b/src/services/health/endpoint.py
@@ -481,6 +481,7 @@ class EndpointHealthService:
        kind_label = {
            "chat": "Chat",
            "cli": "CLI",
+            "compact": "Compact",
            "video": "Video",
            "image": "Image",
        }.get(kind, kind)
--- a/src/services/model/upstream_fetcher.py
+++ b/src/services/model/upstream_fetcher.py
@@ -21,7 +21,7 @@ MAX_CONCURRENT_REQUESTS = 5

 # 模型获取格式优先级：同族内优先使用 chat 端点，若无则回退到 cli 端点
 MODEL_FETCH_FORMAT_PRIORITY: list[tuple[str, ...]] = [
-    ("openai:chat", "openai:cli"),
+    ("openai:chat", "openai:cli", "openai:compact"),
    ("claude:chat", "claude:cli"),
    ("gemini:chat", "gemini:cli"),
 ]
--- a/src/services/provider/adapters/codex/request_patching.py
+++ b/src/services/provider/adapters/codex/request_patching.py
@@ -1,21 +1,8 @@
-"""
-Codex provider request patching helpers (passthrough path).
+"""Codex provider request patching helpers (passthrough path).

-This is the **primary** Codex request transformation used by the normalizer's
-``patch_for_variant("codex")`` fast path.  It applies minimal, non-destructive
-patches directly on the original request dict -- no internal representation
-round-trip, so every field the client sent is preserved as-is unless explicitly
-modified here.
-
-Transformations applied:
- Force ``store=false``.
- Force ``stream=true`` (except compact requests).
- Force ``parallel_tool_calls=true``.
- Ensure ``instructions`` exists (empty string when absent).
- Convert ``role=system`` messages to ``role=developer``.
- Drop request parameters known to be rejected by Codex gateways.
- Force ``include`` to ``["reasoning.encrypted_content"]``.
- Drop compatibility-problematic fields (``context_management`` / ``user``).
+Codex requests are now treated as passthrough:
+- Do not mutate client payload fields.
+- Only strip internal sentinel fields that must never reach upstream.
 """

 from __future__ import annotations
@@ -24,20 +11,6 @@ from typing import Any

 from src.core.provider_types import ProviderType

-_REJECTED_PARAMS: frozenset[str] = frozenset(
-    {
-        "max_output_tokens",
-        "max_completion_tokens",
-        "temperature",
-        "top_p",
-        "service_tier",
-        "previous_response_id",
-        "truncation",
-    }
-)
-
-_REQUIRED_INCLUDE_ITEM = "reasoning.encrypted_content"
-

 def patch_openai_cli_request_for_codex(request_body: dict[str, Any]) -> dict[str, Any]:
    """
@@ -46,49 +19,8 @@ def patch_openai_cli_request_for_codex(request_body: dict[str, Any]) -> dict[str
    This function never mutates the input object.
    """
    out: dict[str, Any] = dict(request_body)
-
-    for k in _REJECTED_PARAMS:
-        out.pop(k, None)
-
-    # Codex gateways often reject/ignore persistence; be explicit.
-    out["store"] = False
-
-    # Codex compact endpoint is non-streaming; normal responses requires stream=true.
-    is_compact = bool(out.pop("_aether_compact", False))
-    if is_compact:
-        out.pop("stream", None)
-    else:
-        out["stream"] = True
-
-    # Codex expects parallel tool calls enabled.
-    out["parallel_tool_calls"] = True
-
-    # Ensure instructions exists (some gateways require it even if empty).
-    instructions = out.get("instructions")
-    if not isinstance(instructions, str):
-        out["instructions"] = ""
-
-    # Convert "system" role to "developer" (Codex behavior).
-    input_items = out.get("input")
-    if isinstance(input_items, list):
-        patched_items: list[Any] = []
-        for item in input_items:
-            if isinstance(item, dict):
-                patched = dict(item)
-                if patched.get("role") == "system":
-                    patched["role"] = "developer"
-                patched_items.append(patched)
-            else:
-                patched_items.append(item)
-        out["input"] = patched_items
-
-    # Keep codex behavior deterministic: force the exact include list.
-    out["include"] = [_REQUIRED_INCLUDE_ITEM]
-
-    # Codex upstream currently rejects these fields.
-    out.pop("context_management", None)
-    out.pop("user", None)
-
+    # Internal routing marker; never send upstream.
+    out.pop("_aether_compact", None)
    return out


@@ -108,7 +40,7 @@ def maybe_patch_request_for_codex(
    """
    if (provider_type or "").lower() != ProviderType.CODEX:
        return request_body
-    if (provider_api_format or "").lower() != "openai:cli":
+    if (provider_api_format or "").lower() not in {"openai:cli", "openai:compact"}:
        return request_body
    if not isinstance(request_body, dict):
        return request_body
--- a/src/services/provider/stream_policy.py
+++ b/src/services/provider/stream_policy.py
@@ -55,13 +55,15 @@ def get_upstream_stream_policy(

    Defaults:
    - Codex + openai:cli: FORCE_STREAM (Codex upstream requires stream=true).
+    - Codex + openai:compact: follow endpoint/client policy (no hard force).
    """

    provider_obj = getattr(endpoint, "provider", None)
    pt = str(provider_type or getattr(provider_obj, "provider_type", "") or "").strip().lower()
    sig = str(endpoint_sig or getattr(endpoint, "api_format", "") or "").strip().lower()
-    is_codex_compact = False
-    if pt == ProviderType.CODEX and sig == "openai:cli":
+    is_codex_cli = pt == ProviderType.CODEX and sig == "openai:cli"
+    is_codex_compact = pt == ProviderType.CODEX and sig == "openai:compact"
+    if is_codex_cli:
        try:
            from src.services.provider.adapters.codex.context import get_codex_request_context

@@ -82,8 +84,7 @@ def get_upstream_stream_policy(
        if parsed != UpstreamStreamPolicy.AUTO:
            # Codex upstream requires streaming; do not allow forcing non-stream.
            if (
-                pt == ProviderType.CODEX
-                and sig == "openai:cli"
+                is_codex_cli
                and parsed == UpstreamStreamPolicy.FORCE_NON_STREAM
                and not is_codex_compact
            ):
@@ -93,7 +94,7 @@ def get_upstream_stream_policy(
            return parsed

    # Safe-by-default: Codex Responses OAuth behaves like SSE-only.
-    if pt == ProviderType.CODEX and sig == "openai:cli":
+    if is_codex_cli:
        return (
            UpstreamStreamPolicy.FORCE_NON_STREAM
            if is_codex_compact
@@ -134,13 +135,30 @@ def enforce_stream_mode_for_upstream(
    meta = resolve_endpoint_definition(provider_api_format)
    provider_uses_stream = meta.stream_in_body if meta is not None else True

+    provider_fmt = str(provider_api_format or "").strip().lower()
+    # OpenAI compact endpoint: keep request body stream field absent.
+    if provider_fmt == "openai:compact":
+        request_body.pop("stream", None)
+        return request_body
+
+    # Backward compatibility: Codex compact routed through openai:cli + context marker.
+    if provider_fmt == "openai:cli":
+        try:
+            from src.services.provider.adapters.codex.context import get_codex_request_context
+
+            ctx = get_codex_request_context()
+            if ctx and ctx.is_compact:
+                request_body.pop("stream", None)
+                return request_body
+        except Exception:
+            pass
+
    if provider_uses_stream:
        request_body["stream"] = bool(upstream_is_stream)
    else:
        request_body.pop("stream", None)

    # OpenAI Chat Completions: request usage in streaming mode.
-    provider_fmt = str(provider_api_format or "").strip().lower()
    if upstream_is_stream and provider_fmt == "openai:chat":
        stream_options = request_body.get("stream_options")
        if not isinstance(stream_options, dict):
--- a/src/services/provider_keys/auth_type.py
+++ b/src/services/provider_keys/auth_type.py
@@ -6,7 +6,12 @@ Provider Key 认证类型相关规则。
 def normalize_auth_type(raw: str) -> str:
    """将数据库中的 auth_type 归一化为逻辑类型。

-    Kiro 在数据库中存储为 ``"kiro"`` 或 ``"oauth"``，统一映射为 ``"oauth"``。
+    - ``"kiro"`` -> ``"oauth"``  (Kiro 使用 OAuth 流程)
+    - ``"vertex_ai"`` -> ``"service_account"``  (旧的 Vertex AI auth_type 已重命名)
    """
    t = str(raw or "api_key").strip() or "api_key"
-    return "oauth" if t == "kiro" else t
+    if t == "kiro":  # TODO: 迁移稳定后清理，同步清理各处 in ("...", "kiro") 兼容检查
+        return "oauth"
+    if t == "vertex_ai":  # TODO: 迁移稳定后清理，同步清理各处 in ("...", "vertex_ai") 兼容检查
+        return "service_account"
+    return t
--- a/src/services/provider_keys/duplicate_check.py
+++ b/src/services/provider_keys/duplicate_check.py
@@ -26,14 +26,14 @@ def check_duplicate_key(

    对于不同的认证类型，使用不同的比较方式：
    - api_key: 比较 API Key 的哈希值
-    - vertex_ai: 比较 Service Account 的 client_email
+    - service_account: 比较 Service Account 的 client_email

    Args:
        db: 数据库会话
        provider_id: Provider ID
-        auth_type: 认证类型 (api_key, vertex_ai, oauth)
+        auth_type: 认证类型 (api_key, service_account, oauth)
        new_api_key: 新的 API Key（用于 api_key 类型）
-        new_auth_config: 新的认证配置（用于 vertex_ai 类型）
+        new_auth_config: 新的认证配置（用于 service_account 类型）
        exclude_key_id: 要排除的 Key ID（用于更新场景）
    """
    if auth_type == "api_key" and new_api_key:
@@ -66,7 +66,7 @@ def check_duplicate_key(
                # 解密失败时跳过该 Key
                continue

-    elif auth_type == "vertex_ai" and new_auth_config:
+    elif auth_type in ("service_account", "vertex_ai") and new_auth_config:
        new_client_email = (
            new_auth_config.get("client_email") if isinstance(new_auth_config, dict) else None
        )
@@ -76,7 +76,7 @@ def check_duplicate_key(
        # 仅查询同 auth_type 且有 auth_config 的 Keys
        query = db.query(ProviderAPIKey).filter(
            ProviderAPIKey.provider_id == provider_id,
-            ProviderAPIKey.auth_type == "vertex_ai",
+            ProviderAPIKey.auth_type.in_(["service_account", "vertex_ai"]),
            ProviderAPIKey.auth_config.isnot(None),
        )
        if exclude_key_id:
--- a/src/services/provider_keys/key_command_service.py
+++ b/src/services/provider_keys/key_command_service.py
@@ -16,6 +16,7 @@ from sqlalchemy.orm import Session
 from src.core.crypto import crypto_service
 from src.core.exceptions import InvalidRequestException, NotFoundException
 from src.core.logger import logger
+from src.core.provider_types import ProviderType
 from src.models.database import Provider, ProviderAPIKey
 from src.models.endpoint_models import (
    EndpointAPIKeyCreate,
@@ -32,6 +33,37 @@ from src.services.provider_keys.key_side_effects import (
 from src.services.provider_keys.response_builder import build_key_response


+def _validate_vertex_api_formats(
+    provider_type: str | None,
+    auth_type: str,
+    api_formats: list[str] | None,
+) -> None:
+    """校验 Vertex Provider 的 key.api_formats 与 auth_type 是否匹配。"""
+    if str(provider_type or "").strip().lower() != ProviderType.VERTEX_AI.value:
+        return
+
+    formats = [
+        str(fmt or "").strip().lower() for fmt in (api_formats or []) if str(fmt or "").strip()
+    ]
+    if not formats:
+        return
+
+    if auth_type == "api_key":
+        allowed = {"gemini:chat"}
+    elif auth_type in {"service_account", "vertex_ai"}:
+        allowed = {"gemini:chat", "claude:chat"}
+    else:
+        return
+
+    invalid = sorted({fmt for fmt in formats if fmt not in allowed})
+    if invalid:
+        allowed_text = ", ".join(sorted(allowed))
+        invalid_text = ", ".join(invalid)
+        raise InvalidRequestException(
+            f"Vertex {auth_type} 不支持以下 API 格式: {invalid_text}；允许: {allowed_text}"
+        )
+
+
@dataclass
 class _UpdateKeyPreparation:
    """更新 Key 前置准备结果。"""
@@ -154,14 +186,14 @@ def _prepare_update_key_payload(
            raise InvalidRequestException("API Key 认证模式下 api_key 不能为空")
        # 切换回 API Key：清理非本模式配置
        update_data["auth_config"] = None
-    elif target_auth_type == "vertex_ai":
+    elif target_auth_type == "service_account":
        if is_auth_type_switch and not update_data.get("auth_config"):
            raise InvalidRequestException(
-                "从 API Key 切换到 Vertex AI 认证模式时，必须提供 Service Account JSON"
+                "切换到 Service Account 认证模式时，必须提供 Service Account JSON"
            )
-        # Vertex AI 不允许手工写入 api_key，仅保留占位符
+        # Service Account 不允许手工写入 api_key，仅保留占位符
        if api_key_in_payload and api_key_value not in {None, "__placeholder__"}:
-            raise InvalidRequestException("Vertex AI 认证模式下不允许直接填写 api_key")
+            raise InvalidRequestException("Service Account 认证模式下不允许直接填写 api_key")
        if is_auth_type_switch or api_key_in_payload:
            update_data["api_key"] = "__placeholder__"
    elif target_auth_type == "oauth":
@@ -186,6 +218,15 @@ def _prepare_update_key_payload(
        exclude_key_id=key_id,
    )

+    # Vertex Provider: auth_type 与 api_formats 的组合必须合法
+    provider = getattr(key, "provider", None)
+    effective_api_formats = update_data.get("api_formats", key.api_formats)
+    _validate_vertex_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"]
        if api_key_raw is None:
@@ -261,7 +302,7 @@ def _prepare_create_key_payload(
    if auth_type == "api_key":
        if not key_data.api_key:
            raise InvalidRequestException("API Key 认证模式下 api_key 为必填字段")
-    elif auth_type == "vertex_ai":
+    elif auth_type == "service_account":
        if not key_data.auth_config:
            raise InvalidRequestException("Service Account 认证模式下 auth_config 为必填字段")
    elif auth_type == "oauth":
@@ -384,12 +425,19 @@ async def create_provider_key_response(
    if not key_data.api_formats:
        raise InvalidRequestException("api_formats 为必填字段")

-    _, new_key = _prepare_create_key_payload(
+    auth_type, new_key = _prepare_create_key_payload(
        db=db,
        provider_id=provider_id,
        key_data=key_data,
    )

+    # Vertex Provider: auth_type 与 api_formats 的组合必须合法
+    _validate_vertex_api_formats(
+        getattr(provider, "provider_type", None),
+        auth_type,
+        key_data.api_formats,
+    )
+
    db.add(new_key)
    db.commit()
    db.refresh(new_key)
--- a/src/services/provider_keys/key_query_service.py
+++ b/src/services/provider_keys/key_query_service.py
@@ -59,7 +59,7 @@ def get_keys_grouped_by_format(db: Session) -> dict:
            continue  # 跳过没有 API 格式的 Key

        auth_type = normalize_auth_type(getattr(key, "auth_type", "api_key"))
-        if auth_type == "vertex_ai":
+        if auth_type in ("service_account", "vertex_ai"):
            masked_key = "[Service Account]"
        elif auth_type == "oauth":
            masked_key = "[OAuth Token]"
@@ -166,15 +166,15 @@ def reveal_endpoint_key_payload(

    auth_type = normalize_auth_type(getattr(key, "auth_type", "api_key"))

-    # Vertex AI 类型返回 auth_config（需要解密）
-    if auth_type == "vertex_ai":
+    # Service Account 类型返回 auth_config（需要解密）
+    if auth_type in ("service_account", "vertex_ai"):
        encrypted_auth_config = getattr(key, "auth_config", None)
        if encrypted_auth_config:
            try:
                decrypted_config = crypto_service.decrypt(encrypted_auth_config)
                auth_config = json.loads(decrypted_config)
                logger.info(f"[REVEAL] 查看 Auth Config: ID={key_id}, Name={key.name}")
-                return {"auth_type": "vertex_ai", "auth_config": auth_config}
+                return {"auth_type": auth_type, "auth_config": auth_config}
            except Exception as e:
                logger.error(f"解密 Auth Config 失败: ID={key_id}, Error={e}")
                raise InvalidRequestException(
@@ -184,12 +184,11 @@ def reveal_endpoint_key_payload(
        # 兼容：auth_config 为空时尝试从 api_key 解密（仅对迁移前的旧数据有效）
        try:
            decrypted_key = crypto_service.decrypt(key.api_key)
-            # 检查是否是新格式的占位符（表示 auth_config 丢失）
            if decrypted_key == "__placeholder__":
-                logger.error(f"Vertex AI Key 缺少 auth_config: ID={key_id}")
+                logger.error(f"Service Account Key 缺少 auth_config: ID={key_id}")
                raise InvalidRequestException("认证配置丢失，请重新添加该密钥。")
-            logger.info(f"[REVEAL] 查看完整 Key (legacy vertex_ai): ID={key_id}, Name={key.name}")
-            return {"auth_type": "vertex_ai", "auth_config": decrypted_key}
+            logger.info(f"[REVEAL] 查看完整 Key (legacy SA): ID={key_id}, Name={key.name}")
+            return {"auth_type": auth_type, "auth_config": decrypted_key}
        except InvalidRequestException:
            raise
        except Exception as e:
--- a/src/services/provider_keys/response_builder.py
+++ b/src/services/provider_keys/response_builder.py
@@ -19,8 +19,8 @@ def build_key_response(
    """构建 Key 响应对象。"""
    auth_type = normalize_auth_type(getattr(key, "auth_type", "api_key"))

-    if auth_type == "vertex_ai":
-        # Vertex AI 使用 Service Account，不显示占位符
+    if auth_type in ("service_account", "vertex_ai"):
+        # Service Account 不显示占位符
        masked_key = "[Service Account]"
    elif auth_type == "oauth":
        masked_key = "[OAuth Token]"
--- a/src/services/scheduling/candidate_builder.py
+++ b/src/services/scheduling/candidate_builder.py
@@ -415,9 +415,12 @@ class CandidateBuilder:
                if isinstance(raw, int) and raw > 0:
                    output_limit = raw

-        # chat/cli 互相可回退（用于同协议族下的端点变体），video/image 等不跨类回退
+        # chat/cli 互相可回退（用于同协议族下的端点变体），compact 可回退到 cli。
+        # video/image 等不跨类回退。
        if client_kind in {EndpointKind.CHAT, EndpointKind.CLI}:
            allowed_kinds = {EndpointKind.CHAT, EndpointKind.CLI}
+        elif client_kind == EndpointKind.COMPACT:
+            allowed_kinds = {EndpointKind.COMPACT, EndpointKind.CLI}
        else:
            allowed_kinds = {client_kind}