refactor: 将 adapter 层的计费/模型抓取/行为变体能力下沉到 core.api_format 注册表

- 新增 core/api_format/capabilities.py，统一注册计费模板、模型抓取、
  total_input_context 计算和 provider behavior variant
- 新增 core/usage_tokens.py，抽取 cache token 解析逻辑到 core 层
- handler adapter 移除各自的 compute_total_input_context / fetch_models /
  BILLING_TEMPLATE 覆盖，改为委托 core 注册表解析
- provider/behavior.py 改为薄封装，底层委托 core registry
- 新增 tests/test_architecture_import_rules.py 架构导入约束测试
- 新增 tests/services/api_format/test_capabilities.py 能力注册表测试

Closes #207

Co-authored-by: AAEE86 <ppk0227@hotmail.com>

src/api/admin/provider_query.py

@@ -19,6 +19,8 @@ from pydantic import BaseModel, Field
 from sqlalchemy import update
 from sqlalchemy.orm import Session, joinedload, make_transient
 
+from src.api.handlers.base.chat_adapter_base import get_adapter_class
+from src.api.handlers.base.cli_adapter_base import get_cli_adapter_class
 from src.config.constants import TimeoutDefaults
 from src.core.api_format import get_extra_headers_from_endpoint
 from src.core.cache_service import CacheService
@@ -40,7 +42,6 @@ from src.services.model.upstream_fetcher import (
     UpstreamModelsFetcherRegistry,
     build_format_to_config,
     fetch_models_for_key,
-    get_adapter_for_format,
 )
 from src.services.provider.oauth_token import resolve_oauth_access_token
 from src.services.proxy_node.resolver import resolve_effective_proxy
@@ -77,6 +78,11 @@ async def _set_provider_upstream_models_cache(provider_id: str, models: list[dic
 _ANTIGRAVITY_TIER_PRIORITY: dict[str, int] = {"ultra": 3, "pro": 2, "free": 1}
 
 
+def _get_adapter_for_format(api_format: str) -> Any:
+    """按 api_format 获取 Chat/CLI adapter 类。"""
+    return get_adapter_class(api_format) or get_cli_adapter_class(api_format)
+
+
 def _antigravity_sort_keys(api_keys: list[Any]) -> list[Any]:
     """按 tier/可用性对 Antigravity Key 降序排列。
 
@@ -828,7 +834,7 @@ async def test_model(
 
     try:
         # 获取对应的 Adapter 类
-        adapter_class = get_adapter_for_format(endpoint.api_format)
+        adapter_class = _get_adapter_for_format(endpoint.api_format)
         if not adapter_class:
             return {
                 "success": False,
@@ -1497,7 +1503,7 @@ async def _execute_test_check(
         if account_id:
             extra_headers["chatgpt-account-id"] = str(account_id)
 
-    adapter_class = get_adapter_for_format(endpoint.api_format)
+    adapter_class = _get_adapter_for_format(endpoint.api_format)
     if not adapter_class:
         raise ValueError(f"Unknown API format: {endpoint.api_format}")
 

src/api/handlers/base/chat_adapter_base.py

@@ -7,6 +7,7 @@ Chat Adapter 通用基类
 - Handler 创建和调用
 
 公共逻辑（异常处理、计费、头部构建等）继承自 HandlerAdapterBase。
+计费策略、模型抓取与 provider 格式能力由 `core.api_format` 注册表统一提供。
 
 子类只需提供：
 - FORMAT_ID: API 格式标识

src/api/handlers/base/cli_adapter_base.py

@@ -7,12 +7,12 @@ CLI Adapter 通用基类
 - Handler 创建和调用
 
 公共逻辑（异常处理、计费、头部构建等）继承自 HandlerAdapterBase。
+计费策略、模型抓取与 provider 格式能力由 `core.api_format` 注册表统一提供。
 
 子类只需提供：
 - FORMAT_ID: API 格式标识
 - HANDLER_CLASS: 对应的 MessageHandler 类
 - 可选覆盖 _extract_message_count() 自定义消息计数逻辑
-- 可选覆盖 compute_total_input_context() 自定义总输入上下文计算
 """
 
 from __future__ import annotations

src/api/handlers/base/endpoint_checker.py

@@ -429,7 +429,7 @@ def _extract_tokens_from_response(
 
             # 尝试提取cache creation tokens
             try:
-                from src.api.handlers.base.utils import extract_cache_creation_tokens
+                from src.core.usage_tokens import extract_cache_creation_tokens
 
                 cache_creation_input_tokens = extract_cache_creation_tokens(usage_info)
             except Exception as e:
@@ -460,7 +460,7 @@ def _extract_tokens_from_response(
             output_tokens = usage_info.get("output_tokens", 0)
             cache_read_input_tokens = usage_info.get("cache_read_input_tokens", 0)
             try:
-                from src.api.handlers.base.utils import extract_cache_creation_tokens
+                from src.core.usage_tokens import extract_cache_creation_tokens
 
                 cache_creation_input_tokens = extract_cache_creation_tokens(usage_info)
             except Exception as e:

src/api/handlers/base/handler_adapter_base.py

@@ -4,11 +4,11 @@ Handler Adapter 公共基类
 从 ChatAdapterBase 和 CliAdapterBase 提取的共享逻辑：
 - API 格式与头部处理
 - 异常处理和错误响应
-- 计费策略
+- 通过 `core.api_format` 注册表解析计费模板与抓模能力
-- 模型列表查询和端点测试
+- 端点测试辅助
 - 路径参数合并
 
-子类（ChatAdapterBase / CliAdapterBase）只需关注各自的 handle() 流程差异。
+子类（ChatAdapterBase / CliAdapterBase）只需关注各自的 `handle()` 流程差异。
 """
 
 from __future__ import annotations
@@ -27,9 +27,12 @@ from src.core.api_format import (
     EndpointKind,
     build_adapter_base_headers_for_endpoint,
     build_adapter_headers_for_endpoint,
+    compute_total_input_context_for_api_format,
+    fetch_models_for_api_format,
     get_adapter_protected_keys_for_endpoint,
     get_auth_handler,
     get_default_auth_method_for_endpoint,
+    resolve_billing_template_for_api_format,
     resolve_header_name_case,
 )
 from src.core.exceptions import (
@@ -50,10 +53,10 @@ class HandlerAdapterBase(ApiAdapter):
     封装两者共享的逻辑：
     - API 格式与头部处理
     - 异常处理和错误响应
-    - 计费策略
+    - 通过 `core.api_format` 注册表解析计费模板与模型抓取能力
-    - 模型列表查询和端点测试
+    - 端点测试辅助
 
-    子类（ChatAdapterBase / CliAdapterBase）只需实现 handle() 和格式特有的方法。
+    子类（ChatAdapterBase / CliAdapterBase）只需实现 `handle()` 和格式特有的方法。
     """
 
     # 子类必须覆盖
@@ -63,7 +66,7 @@ class HandlerAdapterBase(ApiAdapter):
     API_FAMILY: ClassVar[ApiFamily | None] = None
     ENDPOINT_KIND: ClassVar[EndpointKind] = EndpointKind.CHAT
 
-    # 计费模板配置（子类可覆盖，如 "claude", "openai", "gemini"）
+    # 兼容性回退：若 api_format 注册表未声明计费模板，则使用该默认值。
     BILLING_TEMPLATE: str = "claude"
 
     def __init__(self, allowed_api_formats: list[str] | None = None):
@@ -242,18 +245,9 @@ class HandlerAdapterBase(ApiAdapter):
         )
 
     # =========================================================================
-    # 计费策略
+    # 计费能力委托
     # =========================================================================
 
-    def compute_total_input_context(
-        self,
-        input_tokens: int,
-        cache_read_input_tokens: int,
-        cache_creation_input_tokens: int = 0,
-    ) -> int:
-        """计算总输入上下文（用于阶梯计费判定）- 子类可覆盖"""
-        return input_tokens + cache_read_input_tokens
-
     def compute_cost(
         self,
         input_tokens: int,
@@ -269,8 +263,11 @@ class HandlerAdapterBase(ApiAdapter):
         cache_ttl_minutes: int | None = None,
     ) -> dict[str, Any]:
         """计算请求成本"""
-        total_input_context = self.compute_total_input_context(
+        total_input_context = compute_total_input_context_for_api_format(
-            input_tokens, cache_read_input_tokens, cache_creation_input_tokens
+            self.FORMAT_ID, input_tokens, cache_read_input_tokens, cache_creation_input_tokens
+        )
+        billing_template = (
+            resolve_billing_template_for_api_format(self.FORMAT_ID) or self.BILLING_TEMPLATE
         )
 
         return _calculate_request_cost(
@@ -286,11 +283,11 @@ class HandlerAdapterBase(ApiAdapter):
             tiered_pricing=tiered_pricing,
             cache_ttl_minutes=cache_ttl_minutes,
             total_input_context=total_input_context,
-            billing_template=self.BILLING_TEMPLATE,
+            billing_template=billing_template,
         )
 
     # =========================================================================
-    # 模型列表查询与端点测试
+    # 模型抓取委托与端点测试
     # =========================================================================
 
     @classmethod
@@ -301,8 +298,14 @@ class HandlerAdapterBase(ApiAdapter):
         api_key: str,
         extra_headers: dict[str, str] | None = None,
     ) -> tuple[list, str | None]:
-        """查询上游 API 支持的模型列表 - 子类应覆盖"""
+        """查询上游 API 支持的模型列表。"""
-        return [], f"{cls.FORMAT_ID} adapter does not implement fetch_models"
+        return await fetch_models_for_api_format(
+            client,
+            api_format=cls.FORMAT_ID,
+            base_url=base_url,
+            api_key=api_key,
+            extra_headers=extra_headers,
+        )
 
     @classmethod
     def build_request_body(

src/api/handlers/base/parsers.py

@@ -14,10 +14,10 @@ from src.api.handlers.base.response_parser import (
     ResponseParser,
     StreamStats,
 )
-from src.api.handlers.base.utils import extract_cache_creation_tokens
 
 # is_cli_format 权威定义在 core 层
 from src.core.api_format import is_cli_format
+from src.core.usage_tokens import extract_cache_creation_tokens
 
 
 def _check_nested_error(response: dict[str, Any]) -> tuple[bool, dict[str, Any] | None]:

src/api/handlers/base/utils.py

@@ -36,97 +36,6 @@ def get_format_converter_registry() -> FormatConversionRegistry:
     return format_conversion_registry
 
 
-def extract_cache_creation_tokens(usage: dict[str, Any]) -> int:
-    """
-    提取缓存创建 tokens（兼容三种格式）
-
-    根据 Anthropic API 文档，支持三种格式（按优先级）：
-
-    1. **嵌套格式（优先级最高）**：
-       usage.cache_creation.ephemeral_5m_input_tokens
-       usage.cache_creation.ephemeral_1h_input_tokens
-
-    2. **扁平新格式（优先级第二）**：
-       usage.claude_cache_creation_5_m_tokens
-       usage.claude_cache_creation_1_h_tokens
-
-    3. **旧格式（优先级第三）**：
-       usage.cache_creation_input_tokens
-
-    说明：
-    - 只要检测到新格式字段（嵌套/扁平），即视为权威来源：哪怕值为 0 也不回退到旧字段。
-    - 仅当新格式字段完全不存在时，才回退到旧字段。
-    - 扁平格式和嵌套格式互斥，按顺序检查。
-
-    Args:
-        usage: API 响应中的 usage 字典
-
-    Returns:
-        缓存创建 tokens 总数
-    """
-    # 1. 检查嵌套格式（最新格式）
-    cache_creation = usage.get("cache_creation")
-    has_nested_format = isinstance(cache_creation, dict) and (
-        "ephemeral_5m_input_tokens" in cache_creation
-        or "ephemeral_1h_input_tokens" in cache_creation
-    )
-
-    if has_nested_format:
-        cache_5m = int(cache_creation.get("ephemeral_5m_input_tokens", 0))
-        cache_1h = int(cache_creation.get("ephemeral_1h_input_tokens", 0))
-        total = cache_5m + cache_1h
-
-        logger.debug(f"Using nested cache_creation: 5m={cache_5m}, 1h={cache_1h}, total={total}")
-        return total
-
-    # 2. 检查扁平新格式
-    has_flat_format = (
-        "claude_cache_creation_5_m_tokens" in usage or "claude_cache_creation_1_h_tokens" in usage
-    )
-
-    if has_flat_format:
-        cache_5m = int(usage.get("claude_cache_creation_5_m_tokens", 0))
-        cache_1h = int(usage.get("claude_cache_creation_1_h_tokens", 0))
-        total = cache_5m + cache_1h
-
-        logger.debug(f"Using flat new format: 5m={cache_5m}, 1h={cache_1h}, total={total}")
-        return total
-
-    # 3. 回退到旧格式
-    old_format = int(usage.get("cache_creation_input_tokens", 0))
-    if old_format > 0:
-        logger.debug(f"Using old format: cache_creation_input_tokens={old_format}")
-    return old_format
-
-
-def extract_cache_creation_tokens_detail(usage: dict[str, Any]) -> tuple[int, int, int]:
-    """
-    提取缓存创建 tokens 细分（区分 5m 和 1h）
-
-    返回 (total, tokens_5m, tokens_1h) 三元组。
-    当无法区分时，tokens_5m 和 tokens_1h 均为 0，total 为合计值。
-    """
-    # 1. 嵌套格式
-    cache_creation = usage.get("cache_creation")
-    if isinstance(cache_creation, dict) and (
-        "ephemeral_5m_input_tokens" in cache_creation
-        or "ephemeral_1h_input_tokens" in cache_creation
-    ):
-        t5m = int(cache_creation.get("ephemeral_5m_input_tokens", 0))
-        t1h = int(cache_creation.get("ephemeral_1h_input_tokens", 0))
-        return t5m + t1h, t5m, t1h
-
-    # 2. 扁平新格式
-    if "claude_cache_creation_5_m_tokens" in usage or "claude_cache_creation_1_h_tokens" in usage:
-        t5m = int(usage.get("claude_cache_creation_5_m_tokens", 0))
-        t1h = int(usage.get("claude_cache_creation_1_h_tokens", 0))
-        return t5m + t1h, t5m, t1h
-
-    # 3. 旧格式：无法区分
-    old = int(usage.get("cache_creation_input_tokens", 0))
-    return old, 0, 0
-
-
 def build_sse_headers(extra_headers: dict[str, str] | None = None) -> dict[str, str]:
     """
     构建 SSE（text/event-stream）推荐响应头，用于减少代理缓冲带来的卡顿/成段输出。
@@ -205,9 +114,7 @@ def build_json_response_for_client(
     }
     cleaned_headers["Content-Encoding"] = "gzip"
 
-    existing_vary = next(
+    existing_vary = next((v for k, v in response_headers.items() if k.lower() == "vary"), "")
-        (v for k, v in response_headers.items() if k.lower() == "vary"), ""
-    )
     vary_values = [part.strip() for part in str(existing_vary).split(",") if part.strip()]
     if not any(part.lower() == "accept-encoding" for part in vary_values):
         vary_values.append("Accept-Encoding")

src/api/handlers/claude/adapter.py

@@ -8,7 +8,6 @@ from __future__ import annotations
 
 from typing import Any
 
-import httpx
 from fastapi import HTTPException, Request
 from fastapi.responses import JSONResponse
 
@@ -111,7 +110,6 @@ class ClaudeChatAdapter(ChatAdapterBase):
 
     FORMAT_ID = "claude:chat"
     API_FAMILY = ApiFamily.CLAUDE
-    BILLING_TEMPLATE = "claude"  # 使用 Claude 计费模板
     name = "claude.chat"
 
     @property
@@ -133,23 +131,6 @@ class ClaudeChatAdapter(ChatAdapterBase):
         """检测 Claude 请求中隐含的能力需求"""
         return ClaudeCapabilityDetector.detect_from_headers(headers, request_body)
 
-    # =========================================================================
-    # Claude 特定的计费逻辑
-    # =========================================================================
-
-    def compute_total_input_context(
-        self,
-        input_tokens: int,
-        cache_read_input_tokens: int,
-        cache_creation_input_tokens: int = 0,
-    ) -> int:
-        """
-        计算 Claude 的总输入上下文（用于阶梯计费判定）
-
-        Claude 的总输入 = input_tokens + cache_creation_input_tokens + cache_read_input_tokens
-        """
-        return input_tokens + cache_creation_input_tokens + cache_read_input_tokens
-
     def _validate_request_body(
         self, original_request_body: dict, path_params: dict | None = None
     ) -> None:
@@ -203,102 +184,6 @@ class ClaudeChatAdapter(ChatAdapterBase):
             "thinking_enabled": bool(request_obj.thinking),
         }
 
-    @classmethod
-    async def fetch_models(
-        cls,
-        client: httpx.AsyncClient,
-        base_url: str,
-        api_key: str,
-        extra_headers: dict[str, str] | None = None,
-    ) -> tuple[list, str | None]:
-        """查询 Claude API 支持的模型列表（兼容 x-api-key 和 Bearer 认证）"""
-        headers = cls.build_headers_with_extra(api_key, extra_headers)
-        # 兼容第三方提供商：同时发送 Authorization: Bearer 认证头
-        # 官方 Claude API 使用 x-api-key，第三方代理可能使用 Bearer Token
-        if "authorization" not in {k.lower() for k in headers}:
-            headers["Authorization"] = f"Bearer {api_key}"
-        return await cls._fetch_models_paginated(client, base_url, headers, cls.FORMAT_ID)
-
-    @staticmethod
-    async def _fetch_models_paginated(
-        client: httpx.AsyncClient,
-        base_url: str,
-        headers: dict[str, str],
-        format_id: str,
-    ) -> tuple[list, str | None]:
-        """Claude 模型列表分页获取核心逻辑
-
-        Anthropic 的 /v1/models 是分页接口（has_more/first_id/last_id），
-        默认只返回一页。这里做 best-effort 的全量拉取，确保管理端能展示完整模型列表。
-        """
-        # 构建 /v1/models URL
-        base_url = base_url.rstrip("/")
-        if base_url.endswith("/v1"):
-            models_url = f"{base_url}/models"
-        else:
-            models_url = f"{base_url}/v1/models"
-
-        try:
-            all_models: list[dict] = []
-            seen_ids: set[str] = set()
-
-            after_id: str | None = None
-            limit = 100  # Anthropic 支持 limit，尽量减少分页次数
-            max_pages = 20  # safety guard
-
-            for _ in range(max_pages):
-                params: dict[str, Any] = {"limit": limit}
-                if after_id:
-                    params["after_id"] = after_id
-
-                response = await client.get(models_url, headers=headers, params=params)
-                logger.debug(
-                    f"Claude models request to {models_url}: status={response.status_code}, after_id={after_id}"
-                )
-                if response.status_code != 200:
-                    error_body = response.text[:500] if response.text else "(empty)"
-                    error_msg = f"HTTP {response.status_code}: {error_body}"
-                    logger.warning(f"Claude models request to {models_url} failed: {error_msg}")
-                    return [], error_msg
-
-                data = response.json()
-                page_models: list[dict] = []
-                if isinstance(data, dict) and isinstance(data.get("data"), list):
-                    page_models = [m for m in data["data"] if isinstance(m, dict)]
-                elif isinstance(data, list):
-                    page_models = [m for m in data if isinstance(m, dict)]
-
-                for m in page_models:
-                    mid = m.get("id")
-                    if isinstance(mid, str) and mid and mid in seen_ids:
-                        continue
-                    if isinstance(mid, str) and mid:
-                        seen_ids.add(mid)
-                    m["api_format"] = format_id
-                    all_models.append(m)
-
-                # Pagination (Anthropic list response shape)
-                if not isinstance(data, dict):
-                    break
-
-                has_more = bool(data.get("has_more"))
-                last_id = data.get("last_id")
-                if not has_more:
-                    break
-                if not isinstance(last_id, str) or not last_id:
-                    break
-                if after_id == last_id:
-                    # Prevent infinite loops on unexpected upstream behavior.
-                    break
-                after_id = last_id
-
-            return all_models, None
-        except Exception as e:
-            error_msg = f"Request error: {str(e)}"
-            logger.warning(f"Failed to fetch Claude models from {models_url}: {e}")
-            return [], error_msg
-
-    @classmethod
     def build_endpoint_url(
         cls,
         base_url: str,

src/api/handlers/claude/handler.py

@@ -8,8 +8,8 @@ Claude Chat Handler - 基于通用 Chat Handler 基类的简化实现
 from typing import Any
 
 from src.api.handlers.base.chat_handler_base import ChatHandlerBase
-from src.api.handlers.base.utils import extract_cache_creation_tokens_detail
 from src.core.api_format import ApiFamily, EndpointKind
+from src.core.usage_tokens import extract_cache_creation_tokens_detail
 
 
 class ClaudeChatHandler(ChatHandlerBase):

src/api/handlers/claude/stream_parser.py

@@ -7,7 +7,7 @@ Claude SSE 流解析器
 import json
 from typing import Any
 
-from src.api.handlers.base.utils import extract_cache_creation_tokens
+from src.core.usage_tokens import extract_cache_creation_tokens
 
 
 class ClaudeStreamParser:

src/api/handlers/claude_cli/adapter.py

@@ -8,11 +8,9 @@ from __future__ import annotations
 
 from typing import Any
 
-import httpx
-
 from src.api.handlers.base.cli_adapter_base import CliAdapterBase, register_cli_adapter
 from src.api.handlers.base.cli_handler_base import CliMessageHandlerBase
-from src.api.handlers.claude.adapter import ClaudeCapabilityDetector, ClaudeChatAdapter
+from src.api.handlers.claude.adapter import ClaudeCapabilityDetector
 from src.config.settings import config
 from src.core.api_format import ApiFamily
 
@@ -27,7 +25,6 @@ class ClaudeCliAdapter(CliAdapterBase):
 
     FORMAT_ID = "claude:cli"
     API_FAMILY = ApiFamily.CLAUDE
-    BILLING_TEMPLATE = "claude"  # 使用 Claude 计费模板
     name = "claude.cli"
 
     @property
@@ -48,23 +45,6 @@ class ClaudeCliAdapter(CliAdapterBase):
         """检测 Claude CLI 请求中隐含的能力需求"""
         return ClaudeCapabilityDetector.detect_from_headers(headers, request_body)
 
-    # =========================================================================
-    # Claude CLI 特定的计费逻辑
-    # =========================================================================
-
-    def compute_total_input_context(
-        self,
-        input_tokens: int,
-        cache_read_input_tokens: int,
-        cache_creation_input_tokens: int = 0,
-    ) -> int:
-        """
-        计算 Claude CLI 的总输入上下文（用于阶梯计费判定）
-
-        Claude 的总输入 = input_tokens + cache_creation_input_tokens + cache_read_input_tokens
-        """
-        return input_tokens + cache_creation_input_tokens + cache_read_input_tokens
-
     def _extract_message_count(self, payload: dict[str, Any]) -> int:
         """Claude CLI 使用 messages 字段"""
         messages = payload.get("messages", [])
@@ -98,28 +78,6 @@ class ClaudeCliAdapter(CliAdapterBase):
             "system_present": bool(payload.get("system")),
         }
 
-    # =========================================================================
-    # 模型列表查询
-    # =========================================================================
-
-    @classmethod
-    async def fetch_models(
-        cls,
-        client: httpx.AsyncClient,
-        base_url: str,
-        api_key: str,
-        extra_headers: dict[str, str] | None = None,
-    ) -> tuple[list, str | None]:
-        """查询 Claude API 支持的模型列表（使用 CLI Bearer 认证）"""
-        cli_headers = {"User-Agent": config.internal_user_agent_claude_cli}
-        if extra_headers:
-            cli_headers.update(extra_headers)
-        # 使用 CLI adapter 自己的认证头（Authorization: Bearer），而非 Chat 的 x-api-key
-        headers = cls.build_headers_with_extra(api_key, cli_headers)
-        return await ClaudeChatAdapter._fetch_models_paginated(
-            client, base_url, headers, cls.FORMAT_ID
-        )
-
     @classmethod
     def build_endpoint_url(
         cls,

src/api/handlers/claude_cli/handler.py

@@ -10,8 +10,8 @@ from src.api.handlers.base.cli_handler_base import (
     CliMessageHandlerBase,
     StreamContext,
 )
-from src.api.handlers.base.utils import extract_cache_creation_tokens_detail
 from src.core.api_format import ApiFamily, EndpointKind
+from src.core.usage_tokens import extract_cache_creation_tokens_detail
 
 
 class ClaudeCliMessageHandler(CliMessageHandlerBase):

src/api/handlers/gemini/adapter.py

@@ -16,11 +16,9 @@ from src.api.handlers.base.chat_adapter_base import ChatAdapterBase, register_ad
 from src.api.handlers.base.chat_handler_base import ChatHandlerBase
 from src.core.api_format import ApiFamily, get_auth_handler, resolve_header_name_case
 from src.core.api_format.enums import AuthMethod
-from src.core.api_format.headers import BROWSER_FINGERPRINT_HEADERS
 from src.core.logger import logger
 from src.models.gemini import GeminiRequest
 from src.services.gemini_files_mapping import extract_file_names_from_request
-from src.services.provider.transport import redact_url_for_log
 
 
 class GeminiCapabilityDetector:
@@ -54,7 +52,6 @@ class GeminiChatAdapter(ChatAdapterBase):
 
     FORMAT_ID = "gemini:chat"
     API_FAMILY = ApiFamily.GEMINI
-    BILLING_TEMPLATE = "gemini"  # 使用 Gemini 计费模板
     name = "gemini.chat"
 
     @property
@@ -203,61 +200,6 @@ class GeminiChatAdapter(ChatAdapterBase):
             },
         )
 
-    @classmethod
-    async def fetch_models(
-        cls,
-        client: httpx.AsyncClient,
-        base_url: str,
-        api_key: str,
-        extra_headers: dict[str, str] | None = None,
-    ) -> tuple[list, str | None]:
-        """查询 Gemini API 支持的模型列表"""
-        # Gemini 使用 URL 参数传递 key，不需要 headers 中的认证
-        base_url_clean = base_url.rstrip("/")
-        if base_url_clean.endswith("/v1beta"):
-            models_url = f"{base_url_clean}/models?key={api_key}"
-        else:
-            models_url = f"{base_url_clean}/v1beta/models?key={api_key}"
-
-        headers: dict[str, str] = {**BROWSER_FINGERPRINT_HEADERS}
-        if extra_headers:
-            headers.update(extra_headers)
-
-        try:
-            response = await client.get(models_url, headers=headers)
-            logger.debug(
-                f"Gemini models request to {redact_url_for_log(models_url)}: status={response.status_code}"
-            )
-            if response.status_code == 200:
-                data = response.json()
-                if "models" in data:
-                    # 转换为统一格式
-                    return [
-                        {
-                            "id": m.get("name", "").replace("models/", ""),
-                            "owned_by": "google",
-                            "display_name": m.get("displayName", ""),
-                            "api_format": cls.FORMAT_ID,
-                        }
-                        for m in data["models"]
-                    ], None
-                return [], None
-            else:
-                error_body = response.text[:500] if response.text else "(empty)"
-                error_msg = f"HTTP {response.status_code}: {error_body}"
-                logger.warning(
-                    f"Gemini models request to {redact_url_for_log(models_url)} failed: {error_msg}"
-                )
-                return [], error_msg
-        except Exception as e:
-            # 异常信息可能包含带 key 参数的 URL，需要脱敏
-            sanitized_error = redact_url_for_log(str(e))
-            error_msg = f"Request error: {sanitized_error}"
-            logger.warning(
-                f"Failed to fetch Gemini models from {redact_url_for_log(models_url)}: {sanitized_error}"
-            )
-            return [], error_msg
-
     @classmethod
     def build_endpoint_url(
         cls,

src/api/handlers/gemini_cli/adapter.py

@@ -8,12 +8,11 @@ from __future__ import annotations
 
 from typing import Any
 
-import httpx
 from fastapi import Request
 
 from src.api.handlers.base.cli_adapter_base import CliAdapterBase, register_cli_adapter
 from src.api.handlers.base.cli_handler_base import CliMessageHandlerBase
-from src.api.handlers.gemini.adapter import GeminiCapabilityDetector, GeminiChatAdapter
+from src.api.handlers.gemini.adapter import GeminiCapabilityDetector
 from src.config.settings import config
 from src.core.api_format import ApiFamily, get_auth_handler
 from src.core.api_format.enums import AuthMethod
@@ -29,7 +28,6 @@ class GeminiCliAdapter(CliAdapterBase):
 
     FORMAT_ID = "gemini:cli"
     API_FAMILY = ApiFamily.GEMINI
-    BILLING_TEMPLATE = "gemini"  # 使用 Gemini 计费模板
     name = "gemini.cli"
 
     @property
@@ -119,29 +117,6 @@ class GeminiCliAdapter(CliAdapterBase):
             "safety_settings_count": len(payload.get("safety_settings") or []),
         }
 
-    # =========================================================================
-    # 模型列表查询
-    # =========================================================================
-
-    @classmethod
-    async def fetch_models(
-        cls,
-        client: httpx.AsyncClient,
-        base_url: str,
-        api_key: str,
-        extra_headers: dict[str, str] | None = None,
-    ) -> tuple[list, str | None]:
-        """查询 Gemini API 支持的模型列表（带 CLI User-Agent）"""
-        # 复用 GeminiChatAdapter 的实现，添加 CLI User-Agent
-        cli_headers = {"User-Agent": config.internal_user_agent_gemini_cli}
-        if extra_headers:
-            cli_headers.update(extra_headers)
-        models, error = await GeminiChatAdapter.fetch_models(client, base_url, api_key, cli_headers)
-        # 更新 api_format 为 CLI 格式
-        for m in models:
-            m["api_format"] = cls.FORMAT_ID
-        return models, error
-
     @classmethod
     def build_endpoint_url(
         cls,

src/api/handlers/openai/adapter.py

@@ -8,7 +8,6 @@ from __future__ import annotations
 
 from typing import Any
 
-import httpx
 from fastapi.responses import JSONResponse
 
 from src.api.handlers.base.chat_adapter_base import ChatAdapterBase, register_adapter
@@ -28,7 +27,6 @@ class OpenAIChatAdapter(ChatAdapterBase):
 
     FORMAT_ID = "openai:chat"
     API_FAMILY = ApiFamily.OPENAI
-    BILLING_TEMPLATE = "openai"  # 使用 OpenAI 计费模板
     name = "openai.chat"
 
     @property
@@ -105,48 +103,6 @@ class OpenAIChatAdapter(ChatAdapterBase):
             },
         )
 
-    @classmethod
-    async def fetch_models(
-        cls,
-        client: httpx.AsyncClient,
-        base_url: str,
-        api_key: str,
-        extra_headers: dict[str, str] | None = None,
-    ) -> tuple[list, str | None]:
-        """查询 OpenAI 兼容 API 支持的模型列表"""
-        headers = cls.build_headers_with_extra(api_key, extra_headers)
-
-        # 构建 /v1/models URL
-        base_url = base_url.rstrip("/")
-        if base_url.endswith("/v1"):
-            models_url = f"{base_url}/models"
-        else:
-            models_url = f"{base_url}/v1/models"
-
-        try:
-            response = await client.get(models_url, headers=headers)
-            logger.debug(f"OpenAI models request to {models_url}: status={response.status_code}")
-            if response.status_code == 200:
-                data = response.json()
-                models = []
-                if "data" in data:
-                    models = data["data"]
-                elif isinstance(data, list):
-                    models = data
-                # 为每个模型添加 api_format 字段
-                for m in models:
-                    m["api_format"] = cls.FORMAT_ID
-                return models, None
-            else:
-                error_body = response.text[:500] if response.text else "(empty)"
-                error_msg = f"HTTP {response.status_code}: {error_body}"
-                logger.warning(f"OpenAI models request to {models_url} failed: {error_msg}")
-                return [], error_msg
-        except Exception as e:
-            error_msg = f"Request error: {str(e)}"
-            logger.warning(f"Failed to fetch models from {models_url}: {e}")
-            return [], error_msg
-
     @classmethod
     def build_endpoint_url(
         cls,

src/api/handlers/openai_cli/adapter.py

@@ -8,12 +8,9 @@ from __future__ import annotations
 
 from typing import Any
 
-import httpx
-
 from src.api.base.context import ApiRequestContext
 from src.api.handlers.base.cli_adapter_base import CliAdapterBase, register_cli_adapter
 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, EndpointKind
 from src.core.provider_types import ProviderType
@@ -30,7 +27,6 @@ class OpenAICliAdapter(CliAdapterBase):
 
     FORMAT_ID = "openai:cli"
     API_FAMILY = ApiFamily.OPENAI
-    BILLING_TEMPLATE = "openai"  # 使用 OpenAI 计费模板
     name = "openai.cli"
 
     @property
@@ -67,29 +63,6 @@ class OpenAICliAdapter(CliAdapterBase):
             set_codex_request_context(CodexRequestContext(is_compact=True))
         return await super().handle(context)
 
-    # =========================================================================
-    # 模型列表查询
-    # =========================================================================
-
-    @classmethod
-    async def fetch_models(
-        cls,
-        client: httpx.AsyncClient,
-        base_url: str,
-        api_key: str,
-        extra_headers: dict[str, str] | None = None,
-    ) -> tuple[list, str | None]:
-        """查询 OpenAI 兼容 API 支持的模型列表（带 CLI User-Agent）"""
-        # 复用 OpenAIChatAdapter 的实现，添加 CLI User-Agent
-        cli_headers = {"User-Agent": config.internal_user_agent_openai_cli}
-        if extra_headers:
-            cli_headers.update(extra_headers)
-        models, error = await OpenAIChatAdapter.fetch_models(client, base_url, api_key, cli_headers)
-        # 更新 api_format 为 CLI 格式
-        for m in models:
-            m["api_format"] = cls.FORMAT_ID
-        return models, error
-
     @classmethod
     def build_endpoint_url(
         cls,

src/core/api_format/__init__.py

@@ -15,6 +15,25 @@ from src.core.api_format.auth import (
     get_auth_handler,
     get_default_auth_method_for_endpoint,
 )
+from src.core.api_format.capabilities import (
+    ApiFormatCapability,
+    ProviderFormatBehavior,
+    ProviderFormatCapability,
+    compute_total_input_context_for_api_format,
+    fetch_models_for_api_format,
+    get_api_format_capability,
+    get_provider_default_body_rules,
+    get_provider_default_body_rules_for_endpoint,
+    get_provider_format_behavior,
+    get_provider_format_capability,
+    list_api_format_capabilities,
+    register_api_format_capability,
+    register_provider_default_body_rules,
+    register_provider_format_behavior,
+    register_provider_format_capability,
+    resolve_billing_template_for_api_format,
+    resolve_provider_variants_for_endpoint,
+)
 from src.core.api_format.detection import (
     RequestContext,
     detect_cli_format_from_path,
@@ -145,4 +164,22 @@ __all__ = [
     "QueryKeyAuthHandler",
     "get_auth_handler",
     "get_default_auth_method_for_endpoint",
+    # Capabilities
+    "ApiFormatCapability",
+    "ProviderFormatBehavior",
+    "ProviderFormatCapability",
+    "get_api_format_capability",
+    "get_provider_default_body_rules",
+    "get_provider_default_body_rules_for_endpoint",
+    "get_provider_format_behavior",
+    "get_provider_format_capability",
+    "list_api_format_capabilities",
+    "register_api_format_capability",
+    "register_provider_default_body_rules",
+    "register_provider_format_behavior",
+    "register_provider_format_capability",
+    "resolve_billing_template_for_api_format",
+    "resolve_provider_variants_for_endpoint",
+    "compute_total_input_context_for_api_format",
+    "fetch_models_for_api_format",
 ]

src/core/api_format/capabilities.py

@@ -0,0 +1,654 @@
+from __future__ import annotations
+
+import re
+from copy import deepcopy
+from dataclasses import dataclass
+from typing import Any, Awaitable, Callable, Sequence
+
+import httpx
+
+from src.config.settings import config
+from src.core.api_format.enums import ApiFamily, EndpointKind
+from src.core.api_format.headers import (
+    BROWSER_FINGERPRINT_HEADERS,
+    build_adapter_headers_for_endpoint,
+)
+from src.core.api_format.signature import EndpointSignature, make_signature_key, parse_signature_key
+from src.core.logger import logger
+from src.core.provider_types import normalize_provider_type
+
+ModelFetcher = Callable[
+    [httpx.AsyncClient, str, str, str, dict[str, str] | None],
+    Awaitable[tuple[list[dict[str, Any]], str | None]],
+]
+TotalInputContextResolver = Callable[[int, int, int], int]
+
+_SENSITIVE_QUERY_PARAMS_PATTERN = re.compile(
+    r"([?&])(key|api_key|apikey|token|secret|password|credential)=([^&]*)",
+    re.IGNORECASE,
+)
+
+
+def _redact_url_for_log(url: str) -> str:
+    return _SENSITIVE_QUERY_PARAMS_PATTERN.sub(r"\1\2=***", url)
+
+
+def _default_total_input_context(
+    input_tokens: int,
+    cache_read_input_tokens: int,
+    _cache_creation_input_tokens: int = 0,
+) -> int:
+    return input_tokens + cache_read_input_tokens
+
+
+def _claude_total_input_context(
+    input_tokens: int,
+    cache_read_input_tokens: int,
+    cache_creation_input_tokens: int = 0,
+) -> int:
+    return input_tokens + cache_creation_input_tokens + cache_read_input_tokens
+
+
+@dataclass(frozen=True, slots=True)
+class ApiFormatCapability:
+    api_format: str
+    billing_template: str | None = None
+    total_input_context_resolver: TotalInputContextResolver = _default_total_input_context
+    model_fetcher: ModelFetcher | None = None
+
+
+@dataclass(frozen=True, slots=True)
+class ProviderFormatCapability:
+    provider_type: str
+    endpoint_sig: str = ""
+    same_format_variant: str | None = None
+    cross_format_variant: str | None = None
+    default_body_rules: tuple[dict[str, Any], ...] | None = None
+
+
+@dataclass(frozen=True, slots=True)
+class ProviderFormatBehavior:
+    provider_type: str
+    same_format_variant: str | None = None
+    cross_format_variant: str | None = None
+
+
+_registry: dict[str, ApiFormatCapability] = {}
+_provider_registry: dict[tuple[str, str], ProviderFormatCapability] = {}
+
+
+def _normalize_api_format(api_format: str | None) -> str:
+    return str(api_format or "").strip().lower()
+
+
+def _normalize_endpoint_sig(
+    value: str | EndpointSignature | tuple[ApiFamily, EndpointKind] | tuple[Any, Any],
+) -> str:
+    if isinstance(value, str):
+        try:
+            return parse_signature_key(value).key
+        except Exception:
+            return value.strip().lower()
+    if isinstance(value, EndpointSignature):
+        return value.key
+    if isinstance(value, tuple) and len(value) == 2:
+        return make_signature_key(value[0], value[1])
+    return str(value).strip().lower()
+
+
+def register_api_format_capability(capability: ApiFormatCapability) -> None:
+    """注册或覆盖 api_format 能力。"""
+    fmt = _normalize_api_format(capability.api_format)
+    if not fmt:
+        raise ValueError("api_format 不能为空")
+    _registry[fmt] = ApiFormatCapability(
+        api_format=fmt,
+        billing_template=capability.billing_template,
+        total_input_context_resolver=capability.total_input_context_resolver,
+        model_fetcher=capability.model_fetcher,
+    )
+
+
+def get_api_format_capability(api_format: str | None) -> ApiFormatCapability | None:
+    """按 api_format 获取能力定义。"""
+    return _registry.get(_normalize_api_format(api_format))
+
+
+def list_api_format_capabilities() -> list[ApiFormatCapability]:
+    """列出已注册能力。"""
+    return list(_registry.values())
+
+
+def register_provider_format_capability(
+    provider_type: str,
+    endpoint_sig: str | EndpointSignature | tuple[ApiFamily, EndpointKind] | tuple[Any, Any] = "",
+    *,
+    same_format_variant: str | None = None,
+    cross_format_variant: str | None = None,
+    default_body_rules: Sequence[dict[str, Any]] | None = None,
+) -> None:
+    """注册 provider + endpoint 维度的格式能力。"""
+    pt = normalize_provider_type(provider_type)
+    if not pt:
+        raise ValueError("provider_type 不能为空")
+    sig = _normalize_endpoint_sig(endpoint_sig)
+    current = _provider_registry.get((pt, sig))
+    _provider_registry[(pt, sig)] = ProviderFormatCapability(
+        provider_type=pt,
+        endpoint_sig=sig,
+        same_format_variant=(
+            same_format_variant
+            if same_format_variant is not None
+            else (current.same_format_variant if current else None)
+        ),
+        cross_format_variant=(
+            cross_format_variant
+            if cross_format_variant is not None
+            else (current.cross_format_variant if current else None)
+        ),
+        default_body_rules=(
+            tuple(deepcopy(list(default_body_rules)))
+            if default_body_rules is not None
+            else (current.default_body_rules if current else None)
+        ),
+    )
+
+
+def get_provider_format_capability(
+    provider_type: str | None,
+    endpoint_sig: str | EndpointSignature | tuple[ApiFamily, EndpointKind] | tuple[Any, Any] = "",
+) -> ProviderFormatCapability | None:
+    """获取 provider + endpoint 维度能力，未命中时回退 provider 级默认能力。"""
+    pt = normalize_provider_type(provider_type)
+    if not pt:
+        return None
+    sig = _normalize_endpoint_sig(endpoint_sig)
+    return _provider_registry.get((pt, sig)) or _provider_registry.get((pt, ""))
+
+
+def register_provider_behavior_variant(
+    provider_type: str,
+    *,
+    same_format: bool = False,
+    cross_format: bool = False,
+) -> None:
+    """注册 provider 维度的格式变体标志。"""
+    pt = normalize_provider_type(provider_type)
+    current = get_provider_format_capability(pt)
+    register_provider_format_capability(
+        pt,
+        same_format_variant=(
+            pt if same_format else (current.same_format_variant if current else None)
+        ),
+        cross_format_variant=(
+            pt if cross_format else (current.cross_format_variant if current else None)
+        ),
+    )
+
+
+def register_provider_format_behavior(
+    provider_type: str,
+    *,
+    same_format_variant: str | None = None,
+    cross_format_variant: str | None = None,
+) -> None:
+    """兼容接口：按显式 variant 名称注册 provider 行为。"""
+    register_provider_format_capability(
+        provider_type,
+        same_format_variant=same_format_variant,
+        cross_format_variant=cross_format_variant,
+    )
+
+
+def get_provider_format_behavior(provider_type: str | None) -> ProviderFormatBehavior | None:
+    """兼容接口：获取 provider 维度的格式变体能力。"""
+    capability = get_provider_format_capability(provider_type)
+    if capability is None:
+        return None
+    return ProviderFormatBehavior(
+        provider_type=capability.provider_type,
+        same_format_variant=capability.same_format_variant,
+        cross_format_variant=capability.cross_format_variant,
+    )
+
+
+def get_provider_behavior_variants(
+    provider_type: str | None,
+    endpoint_sig: str | EndpointSignature | tuple[ApiFamily, EndpointKind] | tuple[Any, Any] = "",
+) -> tuple[str | None, str | None]:
+    capability = get_provider_format_capability(provider_type, endpoint_sig)
+    if capability is None:
+        return None, None
+    return capability.same_format_variant, capability.cross_format_variant
+
+
+def resolve_provider_variants_for_endpoint(
+    provider_type: str | None,
+    endpoint_sig: str | EndpointSignature | tuple[ApiFamily, EndpointKind] | tuple[Any, Any] = "",
+) -> tuple[str | None, str | None]:
+    return get_provider_behavior_variants(provider_type, endpoint_sig)
+
+
+def register_provider_default_body_rules(
+    provider_type: str,
+    endpoint_sig: str | EndpointSignature | tuple[ApiFamily, EndpointKind] | tuple[Any, Any],
+    rules: Sequence[dict[str, Any]],
+) -> None:
+    """注册 provider + endpoint 维度的默认 body_rules。"""
+    register_provider_format_capability(
+        provider_type,
+        endpoint_sig,
+        default_body_rules=rules,
+    )
+
+
+def get_provider_default_body_rules(
+    provider_type: str | None,
+    endpoint_sig: str | EndpointSignature | tuple[ApiFamily, EndpointKind] | tuple[Any, Any],
+) -> list[dict[str, Any]] | None:
+    """获取 provider + endpoint 维度默认 body_rules。"""
+    capability = get_provider_format_capability(provider_type, endpoint_sig)
+    if capability is None or capability.default_body_rules is None:
+        return None
+    return deepcopy(list(capability.default_body_rules))
+
+
+def get_provider_default_body_rules_for_endpoint(
+    provider_type: str | None,
+    endpoint_sig: str | EndpointSignature | tuple[ApiFamily, EndpointKind] | tuple[Any, Any] = "",
+) -> list[dict[str, Any]] | None:
+    return get_provider_default_body_rules(provider_type, endpoint_sig)
+
+
+def resolve_billing_template_for_api_format(api_format: str | None) -> str | None:
+    """解析 api_format 对应的计费模板。"""
+    capability = get_api_format_capability(api_format)
+    if capability and capability.billing_template:
+        return capability.billing_template
+
+    family = _normalize_api_format(api_format).split(":", 1)[0]
+    if family in {"claude", "openai", "gemini"}:
+        return family
+    return None
+
+
+def compute_total_input_context_for_api_format(
+    api_format: str | None,
+    input_tokens: int,
+    cache_read_input_tokens: int,
+    cache_creation_input_tokens: int = 0,
+) -> int:
+    """按 api_format 计算阶梯计费口径中的总输入上下文。"""
+    capability = get_api_format_capability(api_format)
+    if capability is not None:
+        return capability.total_input_context_resolver(
+            input_tokens,
+            cache_read_input_tokens,
+            cache_creation_input_tokens,
+        )
+
+    if resolve_billing_template_for_api_format(api_format) == "claude":
+        return _claude_total_input_context(
+            input_tokens,
+            cache_read_input_tokens,
+            cache_creation_input_tokens,
+        )
+
+    return _default_total_input_context(
+        input_tokens,
+        cache_read_input_tokens,
+        cache_creation_input_tokens,
+    )
+
+
+def _build_v1_models_url(base_url: str) -> str:
+    base_url = str(base_url or "").rstrip("/")
+    if base_url.endswith("/v1"):
+        return f"{base_url}/models"
+    return f"{base_url}/v1/models"
+
+
+async def _fetch_openai_models(
+    client: httpx.AsyncClient,
+    base_url: str,
+    api_key: str,
+    api_format: str,
+    extra_headers: dict[str, str] | None,
+) -> tuple[list[dict[str, Any]], str | None]:
+    headers = build_adapter_headers_for_endpoint(api_format, api_key, extra_headers)
+    models_url = _build_v1_models_url(base_url)
+
+    try:
+        response = await client.get(models_url, headers=headers)
+        logger.debug("OpenAI models request to {}: status={}", models_url, response.status_code)
+        if response.status_code == 200:
+            data = response.json()
+            models: list[dict[str, Any]] = []
+            if isinstance(data, dict) and isinstance(data.get("data"), list):
+                models = [m for m in data["data"] if isinstance(m, dict)]
+            elif isinstance(data, list):
+                models = [m for m in data if isinstance(m, dict)]
+
+            for model in models:
+                model.setdefault("api_format", api_format)
+            return models, None
+
+        error_body = response.text[:500] if response.text else "(empty)"
+        error_msg = f"HTTP {response.status_code}: {error_body}"
+        logger.warning("OpenAI models request to {} failed: {}", models_url, error_msg)
+        return [], error_msg
+    except Exception as exc:
+        error_msg = f"Request error: {str(exc)}"
+        logger.warning("Failed to fetch models from {}: {}", models_url, exc)
+        return [], error_msg
+
+
+async def _fetch_openai_cli_models(
+    client: httpx.AsyncClient,
+    base_url: str,
+    api_key: str,
+    api_format: str,
+    extra_headers: dict[str, str] | None,
+) -> tuple[list[dict[str, Any]], str | None]:
+    headers = {"User-Agent": config.internal_user_agent_openai_cli}
+    if extra_headers:
+        headers.update(extra_headers)
+    return await _fetch_openai_models(client, base_url, api_key, api_format, headers)
+
+
+async def _fetch_claude_models_paginated(
+    client: httpx.AsyncClient,
+    base_url: str,
+    headers: dict[str, str],
+    api_format: str,
+) -> tuple[list[dict[str, Any]], str | None]:
+    models_url = _build_v1_models_url(base_url)
+
+    try:
+        all_models: list[dict[str, Any]] = []
+        seen_ids: set[str] = set()
+        after_id: str | None = None
+        limit = 100
+        max_pages = 20
+
+        for _ in range(max_pages):
+            params: dict[str, Any] = {"limit": limit}
+            if after_id:
+                params["after_id"] = after_id
+
+            response = await client.get(models_url, headers=headers, params=params)
+            logger.debug(
+                "Claude models request to {}: status={}, after_id={}",
+                models_url,
+                response.status_code,
+                after_id,
+            )
+            if response.status_code != 200:
+                error_body = response.text[:500] if response.text else "(empty)"
+                error_msg = f"HTTP {response.status_code}: {error_body}"
+                logger.warning("Claude models request to {} failed: {}", models_url, error_msg)
+                return [], error_msg
+
+            data = response.json()
+            page_models: list[dict[str, Any]] = []
+            if isinstance(data, dict) and isinstance(data.get("data"), list):
+                page_models = [m for m in data["data"] if isinstance(m, dict)]
+            elif isinstance(data, list):
+                page_models = [m for m in data if isinstance(m, dict)]
+
+            for model in page_models:
+                model_id = model.get("id")
+                if isinstance(model_id, str) and model_id and model_id in seen_ids:
+                    continue
+                if isinstance(model_id, str) and model_id:
+                    seen_ids.add(model_id)
+                model.setdefault("api_format", api_format)
+                all_models.append(model)
+
+            if not isinstance(data, dict):
+                break
+
+            has_more = bool(data.get("has_more"))
+            last_id = data.get("last_id")
+            if not has_more:
+                break
+            if not isinstance(last_id, str) or not last_id:
+                break
+            if after_id == last_id:
+                break
+            after_id = last_id
+
+        return all_models, None
+    except Exception as exc:
+        error_msg = f"Request error: {str(exc)}"
+        logger.warning("Failed to fetch Claude models from {}: {}", models_url, exc)
+        return [], error_msg
+
+
+async def _fetch_claude_models(
+    client: httpx.AsyncClient,
+    base_url: str,
+    api_key: str,
+    api_format: str,
+    extra_headers: dict[str, str] | None,
+    *,
+    force_bearer_fallback: bool,
+) -> tuple[list[dict[str, Any]], str | None]:
+    headers = build_adapter_headers_for_endpoint(api_format, api_key, extra_headers)
+    if force_bearer_fallback and "authorization" not in {k.lower() for k in headers}:
+        headers["Authorization"] = f"Bearer {api_key}"
+    return await _fetch_claude_models_paginated(client, base_url, headers, api_format)
+
+
+async def _fetch_claude_chat_models(
+    client: httpx.AsyncClient,
+    base_url: str,
+    api_key: str,
+    api_format: str,
+    extra_headers: dict[str, str] | None,
+) -> tuple[list[dict[str, Any]], str | None]:
+    return await _fetch_claude_models(
+        client,
+        base_url,
+        api_key,
+        api_format,
+        extra_headers,
+        force_bearer_fallback=True,
+    )
+
+
+async def _fetch_claude_cli_models(
+    client: httpx.AsyncClient,
+    base_url: str,
+    api_key: str,
+    api_format: str,
+    extra_headers: dict[str, str] | None,
+) -> tuple[list[dict[str, Any]], str | None]:
+    headers = {"User-Agent": config.internal_user_agent_claude_cli}
+    if extra_headers:
+        headers.update(extra_headers)
+    return await _fetch_claude_models(
+        client,
+        base_url,
+        api_key,
+        api_format,
+        headers,
+        force_bearer_fallback=False,
+    )
+
+
+async def _fetch_gemini_models(
+    client: httpx.AsyncClient,
+    base_url: str,
+    api_key: str,
+    api_format: str,
+    extra_headers: dict[str, str] | None,
+) -> tuple[list[dict[str, Any]], str | None]:
+    base_url_clean = str(base_url or "").rstrip("/")
+    if base_url_clean.endswith("/v1beta"):
+        models_url = f"{base_url_clean}/models?key={api_key}"
+    else:
+        models_url = f"{base_url_clean}/v1beta/models?key={api_key}"
+
+    headers: dict[str, str] = {**BROWSER_FINGERPRINT_HEADERS}
+    if extra_headers:
+        headers.update(extra_headers)
+
+    try:
+        response = await client.get(models_url, headers=headers)
+        logger.debug(
+            "Gemini models request to {}: status={}",
+            _redact_url_for_log(models_url),
+            response.status_code,
+        )
+        if response.status_code == 200:
+            data = response.json()
+            if isinstance(data, dict) and isinstance(data.get("models"), list):
+                out: list[dict[str, Any]] = []
+                for model in data["models"]:
+                    if not isinstance(model, dict):
+                        continue
+                    out.append(
+                        {
+                            "id": str(model.get("name", "")).replace("models/", ""),
+                            "owned_by": "google",
+                            "display_name": model.get("displayName", ""),
+                            "api_format": api_format,
+                        }
+                    )
+                return out, None
+            return [], None
+
+        error_body = response.text[:500] if response.text else "(empty)"
+        error_msg = f"HTTP {response.status_code}: {error_body}"
+        logger.warning(
+            "Gemini models request to {} failed: {}",
+            _redact_url_for_log(models_url),
+            error_msg,
+        )
+        return [], error_msg
+    except Exception as exc:
+        sanitized_error = _redact_url_for_log(str(exc))
+        error_msg = f"Request error: {sanitized_error}"
+        logger.warning(
+            "Failed to fetch Gemini models from {}: {}",
+            _redact_url_for_log(models_url),
+            sanitized_error,
+        )
+        return [], error_msg
+
+
+async def _fetch_gemini_cli_models(
+    client: httpx.AsyncClient,
+    base_url: str,
+    api_key: str,
+    api_format: str,
+    extra_headers: dict[str, str] | None,
+) -> tuple[list[dict[str, Any]], str | None]:
+    headers = {"User-Agent": config.internal_user_agent_gemini_cli}
+    if extra_headers:
+        headers.update(extra_headers)
+    return await _fetch_gemini_models(client, base_url, api_key, api_format, headers)
+
+
+async def fetch_models_for_api_format(
+    client: httpx.AsyncClient,
+    *,
+    api_format: str,
+    base_url: str,
+    api_key: str,
+    extra_headers: dict[str, str] | None = None,
+) -> tuple[list[dict[str, Any]], str | None]:
+    """按 api_format 获取模型列表。"""
+    normalized_api_format = _normalize_api_format(api_format)
+    capability = get_api_format_capability(normalized_api_format)
+    if capability is None or capability.model_fetcher is None:
+        return [], f"Unknown API format: {api_format}"
+
+    return await capability.model_fetcher(
+        client,
+        base_url,
+        api_key,
+        normalized_api_format,
+        extra_headers,
+    )
+
+
+def _register_builtin_capabilities() -> None:
+    register_api_format_capability(
+        ApiFormatCapability(
+            api_format="openai:chat",
+            billing_template="openai",
+            model_fetcher=_fetch_openai_models,
+        )
+    )
+    register_api_format_capability(
+        ApiFormatCapability(
+            api_format="openai:cli",
+            billing_template="openai",
+            model_fetcher=_fetch_openai_cli_models,
+        )
+    )
+    register_api_format_capability(
+        ApiFormatCapability(
+            api_format="openai:compact",
+            billing_template="openai",
+            model_fetcher=_fetch_openai_cli_models,
+        )
+    )
+    register_api_format_capability(
+        ApiFormatCapability(
+            api_format="claude:chat",
+            billing_template="claude",
+            total_input_context_resolver=_claude_total_input_context,
+            model_fetcher=_fetch_claude_chat_models,
+        )
+    )
+    register_api_format_capability(
+        ApiFormatCapability(
+            api_format="claude:cli",
+            billing_template="claude",
+            total_input_context_resolver=_claude_total_input_context,
+            model_fetcher=_fetch_claude_cli_models,
+        )
+    )
+    register_api_format_capability(
+        ApiFormatCapability(
+            api_format="gemini:chat",
+            billing_template="gemini",
+            model_fetcher=_fetch_gemini_models,
+        )
+    )
+    register_api_format_capability(
+        ApiFormatCapability(
+            api_format="gemini:cli",
+            billing_template="gemini",
+            model_fetcher=_fetch_gemini_cli_models,
+        )
+    )
+
+
+_register_builtin_capabilities()
+
+
+__all__ = [
+    "ApiFormatCapability",
+    "ProviderFormatBehavior",
+    "ProviderFormatCapability",
+    "compute_total_input_context_for_api_format",
+    "fetch_models_for_api_format",
+    "get_api_format_capability",
+    "get_provider_behavior_variants",
+    "get_provider_default_body_rules",
+    "get_provider_default_body_rules_for_endpoint",
+    "get_provider_format_behavior",
+    "get_provider_format_capability",
+    "list_api_format_capabilities",
+    "register_api_format_capability",
+    "register_provider_behavior_variant",
+    "register_provider_default_body_rules",
+    "register_provider_format_behavior",
+    "register_provider_format_capability",
+    "resolve_billing_template_for_api_format",
+    "resolve_provider_variants_for_endpoint",
+]

src/core/api_format/conversion/normalizers/openai_cli.py

@@ -120,11 +120,10 @@ class OpenAICliNormalizer(FormatNormalizer):
         """Codex 同格式透传：直接在原始请求体上做最小补丁，跳过 internal 转换。"""
         if variant.lower() != "codex":
             return None
-        from src.services.provider.adapters.codex.request_patching import (
+        out: dict[str, Any] = dict(request)
-            patch_openai_cli_request_for_codex,
+        # 内部路由标记：绝不能透传到上游。
-        )
+        out.pop("_aether_compact", None)
-
+        return out
-        return patch_openai_cli_request_for_codex(request)
 
     def request_to_internal(self, request: dict[str, Any]) -> InternalRequest:
         model = str(request.get("model") or "")

src/core/api_format/metadata.py

@@ -68,6 +68,20 @@ class EndpointDefinition:
                 yield value
 
 
+CODEX_DEFAULT_BODY_RULES: tuple[dict[str, Any], ...] = (
+    {"action": "drop", "path": "max_output_tokens"},
+    {"action": "drop", "path": "temperature"},
+    {"action": "drop", "path": "top_p"},
+    {"action": "set", "path": "store", "value": False},
+    {
+        "action": "set",
+        "path": "instructions",
+        "value": "You are GPT-5.",
+        "condition": {"path": "instructions", "op": "not_exists"},
+    },
+)
+
+
 _ENDPOINT_DEFINITIONS: dict[tuple[ApiFamily, EndpointKind], EndpointDefinition] = {
     # Claude
     (ApiFamily.CLAUDE, EndpointKind.CHAT): EndpointDefinition(
@@ -138,6 +152,7 @@ _ENDPOINT_DEFINITIONS: dict[tuple[ApiFamily, EndpointKind], EndpointDefinition]
         # compact endpoint is non-streaming by design.
         stream_in_body=False,
         data_format_id="openai_responses",
+        default_body_rules=CODEX_DEFAULT_BODY_RULES,
     ),
     (ApiFamily.OPENAI, EndpointKind.VIDEO): EndpointDefinition(
         api_family=ApiFamily.OPENAI,
@@ -302,16 +317,17 @@ def get_default_body_rules_for_endpoint(
 ) -> list[dict[str, Any]]:
     """获取端点的默认 body_rules。
 
-    优先查找 provider_type 维度的注册规则（如 Codex 对 openai:cli 的定制规则），
+    优先查找 unified api_format capability registry 中的 provider 维度规则（如 Codex 对 openai:cli 的定制规则），
     找不到时回退到 EndpointDefinition 上的通用默认规则。
     """
-    # 确保 provider plugins 已注册（填充 _provider_default_body_rules）
+    # 确保 provider plugins 已注册（填充 capabilities 中的 provider registry）
     # ensure_providers_bootstrapped 是幂等的，重复调用无副作用
     if provider_type:
         try:
-            from src.services.provider.envelope import ensure_providers_bootstrapped
+            import importlib
 
-            ensure_providers_bootstrapped()
+            envelope = importlib.import_module("src.services.provider.envelope")
+            getattr(envelope, "ensure_providers_bootstrapped")()
         except Exception:
             pass
 
@@ -319,9 +335,11 @@ def get_default_body_rules_for_endpoint(
     if provider_type:
         pt = provider_type.strip().lower()
         sig = _normalize_sig_key(value)
-        provider_rules = _provider_default_body_rules.get((pt, sig))
+        from src.core.api_format.capabilities import get_provider_default_body_rules
+
+        provider_rules = get_provider_default_body_rules(pt, sig)
         if provider_rules is not None:
-            return deepcopy(list(provider_rules))
+            return provider_rules
 
     # 2) 回退到 EndpointDefinition 上的通用默认规则
     definition = resolve_endpoint_definition(value)
@@ -331,10 +349,8 @@ def get_default_body_rules_for_endpoint(
 
 
 # ---------------------------------------------------------------------------
-# Provider-scoped default body rules registry
+# Provider-scoped default body rules compatibility wrappers
 # ---------------------------------------------------------------------------
-# key: (provider_type, endpoint_sig_key)  e.g. ("codex", "openai:cli")
-_provider_default_body_rules: dict[tuple[str, str], Sequence[dict[str, Any]]] = {}
 
 
 def register_provider_default_body_rules(
@@ -342,10 +358,14 @@ def register_provider_default_body_rules(
     endpoint_sig: str,
     rules: Sequence[dict[str, Any]],
 ) -> None:
-    """注册特定 provider_type + endpoint_sig 的默认 body_rules。"""
+    """兼容入口：注册特定 provider_type + endpoint_sig 的默认 body_rules，真实存储位于 core registry。"""
     pt = provider_type.strip().lower()
     sig = _normalize_sig_key(endpoint_sig)
-    _provider_default_body_rules[(pt, sig)] = tuple(rules)
+    from src.core.api_format.capabilities import (
+        register_provider_default_body_rules as register_provider_default_body_rules_in_registry,
+    )
+
+    register_provider_default_body_rules_in_registry(pt, sig, rules)
 
 
 def _normalize_sig_key(
@@ -397,6 +417,7 @@ def make_endpoint_signature(api_family: str, endpoint_kind: str) -> str:
 
 
 __all__ = [
+    "CODEX_DEFAULT_BODY_RULES",
     "EndpointDefinition",
     "ENDPOINT_DEFINITIONS",
     "list_endpoint_definitions",

src/core/modules/registry.py

@@ -6,6 +6,7 @@
 
 from __future__ import annotations
 
+import importlib
 import importlib.util
 import os
 from typing import TYPE_CHECKING, Any, Protocol
@@ -33,6 +34,19 @@ class ConfigBackend(Protocol):
     def set_config(self, db: Any, key: str, value: Any, description: Any = None) -> Any: ...
 
 
+class _DefaultConfigBackend:
+    """默认配置后端：始终返回 default（用于独立脚本/极简测试场景）。"""
+
+    def get_config(self, _db: Any, _key: str, default: Any = None) -> Any:
+        return default
+
+    def set_config(self, _db: Any, _key: str, _value: Any, _description: Any = None) -> Any:
+        return None
+
+
+_DEFAULT_CONFIG_BACKEND: ConfigBackend = _DefaultConfigBackend()
+
+
 class ModuleRegistry:
     """
     模块注册中心 - 单例模式
@@ -142,13 +156,21 @@ class ModuleRegistry:
     # ========== 启用状态检查（运行级）==========
 
     def _get_config_backend(self) -> ConfigBackend:
-        """获取配置后端（优先使用已注入的，兜底 lazy import）"""
+        """获取配置后端（优先使用已注入的）。"""
         if self._config_backend is not None:
             return self._config_backend
-        # 兜底: 未注入时使用 lazy import（向后兼容独立脚本/测试场景）
-        from src.services.system.config import SystemConfigService  # noqa: lazy fallback
 
-        return SystemConfigService  # type: ignore[return-value]
+        # 兜底：best-effort 动态加载（避免 core→services 的静态依赖）。
+        try:
+            module = importlib.import_module("src.services.system.config")
+            backend = getattr(module, "SystemConfigService", None)
+            if backend is not None:
+                return backend  # type: ignore[return-value]
+        except Exception:
+            pass
+
+        # 最终兜底：未注入且无法动态加载时，使用默认后端（始终返回 default）。
+        return _DEFAULT_CONFIG_BACKEND
 
     def is_enabled(self, name: str, db: Session) -> bool:
         """

src/core/provider_oauth_utils.py

@@ -3,7 +3,7 @@ from __future__ import annotations
 import asyncio
 import json
 from typing import Any, Awaitable, Callable
-from urllib.parse import urlsplit, urlunsplit
+from urllib.parse import quote, urlsplit, urlunsplit
 
 import httpx
 import jwt
@@ -17,17 +17,75 @@ _GOOGLE_USERINFO_URL = "https://www.googleapis.com/oauth2/v1/userinfo?alt=json"
 
 
 def _coerce_proxy_url(proxy_config: dict[str, Any] | None) -> str | None:
-    if not proxy_config:
+    """为 tls-client 构建可用的代理 URL（best-effort）。
-        return None
-    try:
-        if not proxy_config.get("enabled", True):
-            return None
-        from src.services.proxy_node.resolver import build_proxy_url  # lazy: core→services
 
-        return build_proxy_url(proxy_config)
+    说明：
-    except Exception:
+    - core 层不解析 ProxyNode（node_id）模式，避免 core→services 反向依赖。
+    - 仅支持手工 URL 模式：{url, username, password, enabled}。
+    - node_id / tunnel 等复杂模式由 httpx 路径（HTTPClientPool）处理。
+    """
+    if not proxy_config or not proxy_config.get("enabled", True):
         return None
 
+    raw_url = proxy_config.get("url")
+    if not isinstance(raw_url, str) or not raw_url.strip():
+        return None
+
+    proxy_url = raw_url.strip()
+    username = proxy_config.get("username")
+    password = proxy_config.get("password")
+    if isinstance(username, str) and username.strip():
+        return _inject_auth_into_url(
+            proxy_url, username.strip(), str(password) if password else None
+        )
+    return proxy_url
+
+
+def _inject_auth_into_url(url: str, username: str, password: str | None = None) -> str:
+    """将用户名密码注入 URL（仅用于 tls-client 同步请求）。"""
+    try:
+        parsed = urlsplit(url)
+        if not parsed.scheme or not parsed.hostname:
+            return url
+
+        encoded_username = quote(username, safe="")
+        encoded_password = quote(password, safe="") if password else ""
+        host_part = parsed.hostname
+        if parsed.port:
+            host_part = f"{host_part}:{parsed.port}"
+        auth_part = (
+            f"{encoded_username}:{encoded_password}" if encoded_password else encoded_username
+        )
+        netloc = f"{auth_part}@{host_part}"
+
+        return urlunsplit((parsed.scheme, netloc, parsed.path, parsed.query, parsed.fragment))
+    except Exception:
+        return url
+
+
+def _proxy_display(proxy_config: dict[str, Any] | None) -> str | None:
+    """生成用于日志输出的 proxy 摘要（不泄露认证信息）。"""
+    if not proxy_config or not proxy_config.get("enabled", True):
+        return None
+
+    node_id = proxy_config.get("node_id")
+    if isinstance(node_id, str) and node_id.strip():
+        return f"node_id:{node_id.strip()}"
+
+    proxy_url = _coerce_proxy_url(proxy_config)
+    if not proxy_url:
+        return None
+
+    try:
+        parts = urlsplit(proxy_url)
+        host = parts.hostname or ""
+        if parts.port:
+            host = f"{host}:{parts.port}"
+        # 仅保留 scheme + host + path，移除 userinfo/query/fragment
+        return urlunsplit((parts.scheme, host, parts.path, "", ""))
+    except Exception:
+        return "<invalid_proxy>"
+
 
 def _redact_url(url: str) -> str:
     """Remove query and fragment to avoid leaking secrets in logs."""
@@ -58,7 +116,7 @@ async def _httpx_post(
     timeout_seconds: float,
 ) -> httpx.Response:
     client = await HTTPClientPool.get_proxy_client(proxy_config)
-    proxy_url = _coerce_proxy_url(proxy_config)
+    proxy_url = _proxy_display(proxy_config)
     safe_url = _redact_url(url)
 
     last_exc: Exception | None = None

src/core/usage_tokens.py

@@ -0,0 +1,113 @@
+"""
+Usage 相关的 token 解析工具。
+
+该模块用于从不同上游的 usage 结构中提取缓存 token 信息（兼容多种字段命名）。
+放在 core 层，便于 services/api 共用，避免跨层反向依赖。
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+from src.core.logger import logger
+
+
+def extract_cache_creation_tokens(usage: dict[str, Any]) -> int:
+    """
+    提取缓存创建 tokens（兼容三种格式）
+
+    根据 Anthropic API 文档，支持三种格式（按优先级）：
+
+    1. **嵌套格式（优先级最高）**：
+       usage.cache_creation.ephemeral_5m_input_tokens
+       usage.cache_creation.ephemeral_1h_input_tokens
+
+    2. **扁平新格式（优先级第二）**：
+       usage.claude_cache_creation_5_m_tokens
+       usage.claude_cache_creation_1_h_tokens
+
+    3. **旧格式（优先级第三）**：
+       usage.cache_creation_input_tokens
+
+    说明：
+    - 只要检测到新格式字段（嵌套/扁平），即视为权威来源：哪怕值为 0 也不回退到旧字段。
+    - 仅当新格式字段完全不存在时，才回退到旧字段。
+
+    Args:
+        usage: API 响应中的 usage 字典
+
+    Returns:
+        缓存创建 tokens 总数
+    """
+    # 1. 检查嵌套格式（最新格式）
+    cache_creation = usage.get("cache_creation")
+    has_nested_format = isinstance(cache_creation, dict) and (
+        "ephemeral_5m_input_tokens" in cache_creation
+        or "ephemeral_1h_input_tokens" in cache_creation
+    )
+
+    if has_nested_format:
+        cache_5m = int(cache_creation.get("ephemeral_5m_input_tokens", 0))
+        cache_1h = int(cache_creation.get("ephemeral_1h_input_tokens", 0))
+        total = cache_5m + cache_1h
+
+        logger.debug(
+            "Using nested cache_creation: 5m={}, 1h={}, total={}",
+            cache_5m,
+            cache_1h,
+            total,
+        )
+        return total
+
+    # 2. 检查扁平新格式
+    has_flat_format = (
+        "claude_cache_creation_5_m_tokens" in usage or "claude_cache_creation_1_h_tokens" in usage
+    )
+
+    if has_flat_format:
+        cache_5m = int(usage.get("claude_cache_creation_5_m_tokens", 0))
+        cache_1h = int(usage.get("claude_cache_creation_1_h_tokens", 0))
+        total = cache_5m + cache_1h
+
+        logger.debug("Using flat new format: 5m={}, 1h={}, total={}", cache_5m, cache_1h, total)
+        return total
+
+    # 3. 回退到旧格式
+    old_format = int(usage.get("cache_creation_input_tokens", 0))
+    if old_format > 0:
+        logger.debug("Using old format: cache_creation_input_tokens={}", old_format)
+    return old_format
+
+
+def extract_cache_creation_tokens_detail(usage: dict[str, Any]) -> tuple[int, int, int]:
+    """
+    提取缓存创建 tokens 细分（区分 5m 和 1h）
+
+    返回 (total, tokens_5m, tokens_1h) 三元组。
+    当无法区分时，tokens_5m 和 tokens_1h 均为 0，total 为合计值。
+    """
+    # 1. 嵌套格式
+    cache_creation = usage.get("cache_creation")
+    if isinstance(cache_creation, dict) and (
+        "ephemeral_5m_input_tokens" in cache_creation
+        or "ephemeral_1h_input_tokens" in cache_creation
+    ):
+        t5m = int(cache_creation.get("ephemeral_5m_input_tokens", 0))
+        t1h = int(cache_creation.get("ephemeral_1h_input_tokens", 0))
+        return t5m + t1h, t5m, t1h
+
+    # 2. 扁平新格式
+    if "claude_cache_creation_5_m_tokens" in usage or "claude_cache_creation_1_h_tokens" in usage:
+        t5m = int(usage.get("claude_cache_creation_5_m_tokens", 0))
+        t1h = int(usage.get("claude_cache_creation_1_h_tokens", 0))
+        return t5m + t1h, t5m, t1h
+
+    # 3. 旧格式：无法区分
+    old = int(usage.get("cache_creation_input_tokens", 0))
+    return old, 0, 0
+
+
+__all__ = [
+    "extract_cache_creation_tokens",
+    "extract_cache_creation_tokens_detail",
+]

src/services/model/cost.py

@@ -4,8 +4,8 @@
 支持固定价格、按次计费和阶梯计费三种模式。
 
 计费策略：
-- 不同 API format 可以有不同的计费逻辑
+- 价格来源仍由 ModelCostService 解析
-- 通过 PricingStrategy 抽象，支持自定义总输入上下文计算、缓存 TTL 差异化等
+- 格式相关口径（计费模板、总输入上下文）统一由 core.api_format.capabilities 提供
 """
 
 from __future__ import annotations
@@ -14,8 +14,13 @@ from dataclasses import dataclass
 
 from sqlalchemy.orm import Session
 
+from src.core.api_format.capabilities import (
+    compute_total_input_context_for_api_format,
+    resolve_billing_template_for_api_format,
+)
 from src.core.logger import logger
 from src.models.database import GlobalModel, Model, Provider
+from src.services.billing import calculate_request_cost
 
 ProviderRef = str | Provider | None
 
@@ -902,7 +907,7 @@ class ModelCostService:
         """
         使用计费策略计算成本（异步版本）
 
-        根据 api_format 选择对应的 Adapter 计费逻辑，支持阶梯计费和 TTL 差异化。
+        根据 core.api_format.capabilities 解析计费模板与总输入上下文，支持阶梯计费和 TTL 差异化。
 
         Args:
             provider: Provider 对象或提供商名称
@@ -911,7 +916,7 @@ class ModelCostService:
             output_tokens: 输出 token 数
             cache_creation_input_tokens: 缓存创建 token 数
             cache_read_input_tokens: 缓存读取 token 数
-            api_format: API 格式（用于选择计费策略）
+            api_format: API 格式（用于解析格式相关计费口径）
             cache_ttl_minutes: 缓存时长（分钟），用于 TTL 差异化定价
 
         Returns:
@@ -926,21 +931,17 @@ class ModelCostService:
         request_price = await self.get_request_price_async(provider, model)
         tiered_pricing = await self.get_tiered_pricing_async(provider, model)
 
-        # 获取对应 API 格式的 Adapter 实例来计算成本
+        billing_template = resolve_billing_template_for_api_format(api_format) or ""
-        # 优先检查 Chat Adapter，然后检查 CLI Adapter
-        # TODO(arch): 引入 adapter 能力注册表，消除 services->api 依赖
-        from src.api.handlers.base.chat_adapter_base import get_adapter_instance
-        from src.api.handlers.base.cli_adapter_base import get_cli_adapter_instance
 
-        adapter = None
+        if billing_template:
-        if api_format:
+            total_input_context = compute_total_input_context_for_api_format(
-            adapter = get_adapter_instance(api_format)
+                api_format,
-            if adapter is None:
+                input_tokens,
-                adapter = get_cli_adapter_instance(api_format)
+                cache_read_input_tokens,
+                cache_creation_input_tokens,
+            )
 
-        if adapter:
+            result = calculate_request_cost(
-            # 使用 Adapter 的计费方法
-            result = adapter.compute_cost(
                 input_tokens=input_tokens,
                 output_tokens=output_tokens,
                 cache_creation_input_tokens=cache_creation_input_tokens,
@@ -952,6 +953,8 @@ class ModelCostService:
                 price_per_request=request_price,
                 tiered_pricing=tiered_pricing,
                 cache_ttl_minutes=cache_ttl_minutes,
+                total_input_context=total_input_context,
+                billing_template=billing_template,
             )
             return (
                 result["input_cost"],
@@ -964,7 +967,7 @@ class ModelCostService:
                 result["tier_index"],
             )
         else:
-            # 回退到默认计算逻辑（无 Adapter 时使用静态方法）
+            # 回退到默认计算逻辑（无显式格式能力时使用静态方法）
             return self.compute_cost_with_tiered_pricing(
                 input_tokens=input_tokens,
                 output_tokens=output_tokens,

src/services/model/fetch_scheduler.py

@@ -5,7 +5,7 @@
 
 功能:
 - 扫描所有启用了 auto_fetch_models 的 ProviderAPIKey
-- 调用 Adapter.fetch_models() 获取模型列表
+- 调用 core.api_format 注册表获取模型列表
 - 更新 Key 的 allowed_models（保留 locked_models 中的模型）
 - 支持包含/排除规则过滤模型
 - 记录获取结果和错误信息

src/services/model/pricing_strategy.py

@@ -2,7 +2,7 @@
 计费相关数据类
 
 定义计费计算所需的数据结构。
-实际的计费逻辑已移至 ChatAdapterBase，每种 API 格式可以覆盖计费方法。
+实际的计费能力已收敛到 core.api_format 注册表，避免依赖 API Adapter。
 
 数据类：
 - UsageTokens: 请求的 token 使用量

src/services/model/upstream_fetcher.py

@@ -1,7 +1,7 @@
 """
 上游模型获取公共模块
 
-提供从上游 API 获取模型列表的公共函数，供以下场景使用：
+提供从上游 API 获取模型列表的公共函数；通用 api_format 抓取能力统一来自 core.api_format.capabilities，供以下场景使用：
 - 定时任务自动获取（fetch_scheduler.py）
 - 管理后台手动查询（provider_query.py）
 """
@@ -155,19 +155,6 @@ def merge_upstream_metadata(
     return merged
 
 
-# Provider-specific fetchers are registered by plugin.register_all()
-# (called from envelope.py bootstrap)
-
-
-def get_adapter_for_format(api_format: str) -> type | None:
-    """根据 API 格式获取对应的 Adapter 类"""
-    # TODO(arch): 引入 adapter 能力注册表，消除 services->api 依赖
-    from src.api.handlers.base.chat_adapter_base import get_adapter_class
-    from src.api.handlers.base.cli_adapter_base import get_cli_adapter_class
-
-    return get_adapter_class(api_format) or get_cli_adapter_class(api_format)
-
-
 def build_all_format_configs(
     api_key_value: str,
     format_to_endpoint: dict[str, EndpointFetchConfig],
@@ -252,13 +239,15 @@ async def fetch_models_from_endpoints(
         extra_headers = config.get("extra_headers")
 
         try:
-            adapter_class = get_adapter_for_format(api_format)
-            if not adapter_class:
-                return [], f"Unknown API format: {api_format}", False
-
             async with semaphore:
-                models, error = await adapter_class.fetch_models(  # type: ignore[attr-defined]
+                from src.core.api_format.capabilities import fetch_models_for_api_format
-                    client, base_url, api_key_value, extra_headers
+
+                models, error = await fetch_models_for_api_format(
+                    client,
+                    api_format=api_format,
+                    base_url=base_url,
+                    api_key=api_key_value,
+                    extra_headers=extra_headers,
                 )
 
             for m in models:

src/services/orchestration/error_handler.py

@@ -434,7 +434,7 @@ class ErrorHandlerService:
             return
 
         async def _cleanup() -> None:
-            from src.api.base.models_service import invalidate_models_list_cache
+            from src.services.cache.model_list_cache import invalidate_models_list_cache
             from src.services.cache.provider_cache import ProviderCacheService
             from src.services.provider.pool import redis_ops as pool_redis
 

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

@@ -1,11 +1,11 @@
 """Antigravity provider plugin — 统一注册入口。
 
-将 Antigravity 对各通用 registry 的注册集中在一个文件中：
+将 Antigravity 对各通用 registry / capability registry 的注册集中在一个文件中：
 - Envelope (v1internal 信封)
 - Transport Hook (URL 构建)
 - Auth Enricher (OAuth enrichment)
 - Model Fetcher (模型获取)
-- Behavior Variants (格式变体)
+- Provider Format Capability（跨格式变体）
 
 新增 provider 时参照此文件创建对应的 plugin.py 即可。
 """
@@ -353,10 +353,10 @@ def antigravity_export_builder(
 
 def register_all() -> None:
     """一次性注册 Antigravity 的所有 hooks 到各通用 registry。"""
+    from src.core.api_format.capabilities import register_provider_behavior_variant
     from src.core.provider_oauth_utils import register_auth_enricher
     from src.services.model.upstream_fetcher import UpstreamModelsFetcherRegistry
     from src.services.provider.adapters.antigravity.envelope import antigravity_v1internal_envelope
-    from src.services.provider.behavior import register_behavior_variant
     from src.services.provider.envelope import register_envelope
     from src.services.provider.export import register_export_builder
     from src.services.provider.transport import register_transport_hook
@@ -384,5 +384,5 @@ def register_all() -> None:
         fetcher=fetch_models_antigravity,
     )
 
-    # Behavior
+    # Provider Format Capability
-    register_behavior_variant("antigravity", cross_format=True)
+    register_provider_behavior_variant("antigravity", cross_format=True)

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

@@ -1,10 +1,10 @@
 """Codex provider plugin — 统一注册入口。
 
-将 Codex 对各通用 registry 的注册集中在一个文件中：
+将 Codex 对各通用 registry / capability registry 的注册集中在一个文件中：
 - Envelope (OAuth headers)
 - Transport Hook (URL 构建)
 - Auth Enricher (OAuth enrichment)
-- Behavior Variants (格式变体)
+- Provider Format Capability（格式变体 + 默认 body_rules）
 - Model Fetcher (fixed catalog — Codex has no /v1/models endpoint)
 
 新增 provider 时参照此文件创建对应的 plugin.py 即可。
@@ -154,10 +154,13 @@ async def enrich_codex(
 
 def register_all() -> None:
     """一次性注册 Codex 的所有 hooks 到各通用 registry。"""
+    from src.core.api_format.capabilities import (
+        register_provider_behavior_variant,
+        register_provider_default_body_rules,
+    )
     from src.core.provider_oauth_utils import register_auth_enricher
     from src.services.model.upstream_fetcher import UpstreamModelsFetcherRegistry
     from src.services.provider.adapters.codex.envelope import codex_oauth_envelope
-    from src.services.provider.behavior import register_behavior_variant
     from src.services.provider.envelope import register_envelope
     from src.services.provider.transport import register_transport_hook
 
@@ -173,25 +176,11 @@ def register_all() -> None:
     # Auth
     register_auth_enricher("codex", enrich_codex)
 
-    # Behavior
+    # Provider Format Capability：格式变体 + 默认 body_rules
-    register_behavior_variant("codex", same_format=True, cross_format=True)
+    from src.core.api_format.metadata import CODEX_DEFAULT_BODY_RULES
 
-    # Default Body Rules (Codex-specific, not format-wide)
+    register_provider_behavior_variant("codex", same_format=True, cross_format=True)
-    from src.core.api_format.metadata import register_provider_default_body_rules
+    register_provider_default_body_rules("codex", "openai:cli", CODEX_DEFAULT_BODY_RULES)
-
-    _codex_body_rules = (
-        {"action": "drop", "path": "max_output_tokens"},
-        {"action": "drop", "path": "temperature"},
-        {"action": "drop", "path": "top_p"},
-        {"action": "set", "path": "store", "value": False},
-        {
-            "action": "set",
-            "path": "instructions",
-            "value": "You are GPT-5.",
-            "condition": {"path": "instructions", "op": "not_exists"},
-        },
-    )
-    register_provider_default_body_rules("codex", "openai:cli", _codex_body_rules)
 
     # Export: Codex uses the default export builder (strip null + temp fields)
     # No need to register a custom one — the default in export.py suffices.

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

@@ -1,9 +1,9 @@
 """Vertex AI provider plugin — 统一注册入口。
 
-注册 Vertex AI 对各通用 registry 的 hooks：
+注册 Vertex AI 对各通用 registry / capability registry 的 hooks：
 - Transport Hook (URL 构建，支持 API Key / Service Account 双策略)
 - Model Fetcher (专用上游模型获取链路，不走通用 /v1beta/models / /v1/models)
-- Behavior Variants (跨格式支持：同一 Provider 同时访问 Gemini 和 Claude 模型)
+- Provider Format Capability（跨格式支持：同一 Provider 同时访问 Gemini 和 Claude 模型）
 """
 
 from __future__ import annotations
@@ -452,9 +452,9 @@ async def fetch_models_vertex_ai(
 
 def register_all() -> None:
     """一次性注册 Vertex AI 的所有 hooks 到各通用 registry。"""
+    from src.core.api_format.capabilities import register_provider_behavior_variant
     from src.services.model.upstream_fetcher import UpstreamModelsFetcherRegistry
     from src.services.provider.adapters.vertex_ai.transport import build_vertex_ai_url
-    from src.services.provider.behavior import register_behavior_variant
     from src.services.provider.transport import register_transport_hook
 
     # Transport: Vertex AI 同时支持 gemini:chat 和 claude:chat 格式
@@ -467,8 +467,8 @@ def register_all() -> None:
         fetcher=fetch_models_vertex_ai,
     )
 
-    # Behavior: 跨格式支持（同一 Vertex AI Provider 可同时访问 Gemini 和 Claude 模型）
+    # Provider Format Capability：跨格式支持（同一 Vertex AI Provider 可同时访问 Gemini 和 Claude 模型）
-    register_behavior_variant("vertex_ai", cross_format=True)
+    register_provider_behavior_variant("vertex_ai", cross_format=True)
 
 
 __all__ = ["fetch_models_vertex_ai", "register_all"]

src/services/provider/behavior.py

@@ -1,42 +1,19 @@
-"""Provider behavior resolver.
+"""Provider behavior 薄封装。
 
-Keep provider-specific quirks centralized so handler code stays generic.
+对外保持既有调用接口，内部统一委托给 core.api_format.capabilities 中的 provider registry。
-
-Concepts:
-- envelope: wire-level request/response wrappers and transport side-effects
-- same_format_variant: subtle same-format differences (e.g. Codex)
-- cross_format_variant: cross-format conversion tweaks (e.g. Antigravity thinking blocks)
 """
 
 from __future__ import annotations
 
 from dataclasses import dataclass
 
+from src.core.api_format.capabilities import (
+    get_provider_behavior_variants,
+    register_provider_behavior_variant,
+)
 from src.core.provider_types import normalize_provider_type
 from src.services.provider.envelope import ProviderEnvelope, get_provider_envelope
 
-# --- Behavior Variant Registries ---
-_same_format_variants: set[str] = set()
-_cross_format_variants: set[str] = set()
-
-
-def register_behavior_variant(
-    provider_type: str,
-    *,
-    same_format: bool = False,
-    cross_format: bool = False,
-) -> None:
-    """注册 provider 的格式变体标志。
-
-    - same_format: 同格式下有微妙差异（如 Codex 的 OpenAI Responses 变体）
-    - cross_format: 跨格式转换时有特殊处理（如 Antigravity thinking blocks）
-    """
-    pt = normalize_provider_type(provider_type)
-    if same_format:
-        _same_format_variants.add(pt)
-    if cross_format:
-        _cross_format_variants.add(pt)
-
 
 @dataclass(frozen=True, slots=True)
 class ProviderBehavior:
@@ -46,6 +23,20 @@ class ProviderBehavior:
     cross_format_variant: str | None
 
 
+def register_behavior_variant(
+    provider_type: str,
+    *,
+    same_format: bool = False,
+    cross_format: bool = False,
+) -> None:
+    """兼容入口：注册 provider 的格式变体标志，真实存储位于 core registry。"""
+    register_provider_behavior_variant(
+        provider_type,
+        same_format=same_format,
+        cross_format=cross_format,
+    )
+
+
 def get_provider_behavior(
     *,
     provider_type: str | None,
@@ -53,9 +44,10 @@ def get_provider_behavior(
 ) -> ProviderBehavior:
     pt = normalize_provider_type(provider_type)
     envelope = get_provider_envelope(provider_type=pt, endpoint_sig=endpoint_sig)
-
+    same_format_variant, cross_format_variant = get_provider_behavior_variants(
-    same_format_variant = pt if pt in _same_format_variants else None
+        provider_type=pt,
-    cross_format_variant = pt if pt in _cross_format_variants else None
+        endpoint_sig=endpoint_sig or "",
+    )
 
     return ProviderBehavior(
         provider_type=pt,
@@ -65,7 +57,4 @@ def get_provider_behavior(
     )
 
 
-# Behavior variants are registered by provider plugin.register_all()
-# (called from envelope.py bootstrap)
-
 __all__ = ["ProviderBehavior", "get_provider_behavior", "register_behavior_variant"]

src/services/provider_keys/key_command_service.py

@@ -110,7 +110,7 @@ def _run_async_with_fallback(coro: Any) -> None:
 
 async def _invalidate_cache_after_clear_oauth_invalid(key_id: str) -> None:
     """清除 OAuth 失效标记后同步失效相关缓存。"""
-    from src.api.base.models_service import invalidate_models_list_cache
+    from src.services.cache.model_list_cache import invalidate_models_list_cache
     from src.services.cache.provider_cache import ProviderCacheService
 
     await ProviderCacheService.invalidate_provider_api_key_cache(key_id)

src/services/provider_keys/key_side_effects.py

@@ -7,9 +7,9 @@ from __future__ import annotations
 from sqlalchemy import delete as sa_delete
 from sqlalchemy.orm import Session
 
-from src.api.base.models_service import invalidate_models_list_cache
 from src.core.logger import logger
 from src.models.database import GeminiFileMapping, ProviderAPIKey, RequestCandidate, VideoTask
+from src.services.cache.model_list_cache import invalidate_models_list_cache
 from src.services.cache.provider_cache import ProviderCacheService
 
 _SQLITE_BATCH_SIZE = 900

src/services/provider_keys/quota_refresh/antigravity_refresher.py

@@ -9,9 +9,9 @@ from datetime import datetime, timezone
 
 from sqlalchemy.orm import Session
 
-from src.api.handlers.base.request_builder import get_provider_auth
 from src.core.logger import logger
 from src.models.database import Provider, ProviderAPIKey, ProviderEndpoint
+from src.services.provider.auth import get_provider_auth
 
 
 async def refresh_antigravity_key_quota(

src/services/provider_keys/quota_refresh/codex_refresher.py

@@ -12,14 +12,14 @@ from typing import Any
 import httpx
 from sqlalchemy.orm import Session
 
-from src.api.handlers.base.request_builder import get_provider_auth
 from src.core.crypto import crypto_service
 from src.models.database import Provider, ProviderAPIKey, ProviderEndpoint
-from src.services.provider_keys.auth_type import normalize_auth_type
+from src.services.provider.auth import get_provider_auth
 from src.services.provider.pool.account_state import (
     OAUTH_ACCOUNT_BLOCK_PREFIX,
     OAUTH_EXPIRED_PREFIX,
 )
+from src.services.provider_keys.auth_type import normalize_auth_type
 from src.services.provider_keys.codex_usage_parser import (
     parse_codex_usage_headers,
     parse_codex_wham_usage_response,
@@ -68,37 +68,42 @@ def _extract_error_message_from_response(response: httpx.Response) -> str:
 
 
 def _looks_like_token_invalidated(message: str | None) -> bool:
-    lowered = str(message or '').strip().lower()
+    lowered = str(message or "").strip().lower()
-    return 'authentication token has been invalidated' in lowered or 'token has been invalidated' in lowered
+    return (
+        "authentication token has been invalidated" in lowered
+        or "token has been invalidated" in lowered
+    )
 
 
 def _looks_like_account_deactivated(message: str | None) -> bool:
-    lowered = str(message or '').strip().lower()
+    lowered = str(message or "").strip().lower()
-    return 'account has been deactivated' in lowered or 'account deactivated' in lowered
+    return "account has been deactivated" in lowered or "account deactivated" in lowered
 
 
 def _looks_like_workspace_deactivated(message: str | None) -> bool:
-    lowered = str(message or '').strip().lower()
+    lowered = str(message or "").strip().lower()
-    return 'deactivated_workspace' in lowered or ('workspace' in lowered and 'deactivated' in lowered)
+    return "deactivated_workspace" in lowered or (
+        "workspace" in lowered and "deactivated" in lowered
+    )
 
 
 def _build_structured_invalid_reason(*, status_code: int, upstream_message: str | None) -> str:
-    message = str(upstream_message or '').strip()
+    message = str(upstream_message or "").strip()
 
     if status_code == 402 and _looks_like_workspace_deactivated(message):
-        return f'{OAUTH_ACCOUNT_BLOCK_PREFIX}工作区已停用 (deactivated_workspace)'
+        return f"{OAUTH_ACCOUNT_BLOCK_PREFIX}工作区已停用 (deactivated_workspace)"
 
     if _looks_like_account_deactivated(message):
-        detail = message or 'OpenAI 账号已停用'
+        detail = message or "OpenAI 账号已停用"
-        return f'{OAUTH_ACCOUNT_BLOCK_PREFIX}{detail}'
+        return f"{OAUTH_ACCOUNT_BLOCK_PREFIX}{detail}"
 
     if status_code == 401:
-        detail = message or 'Codex Token 无效或已过期 (401)'
+        detail = message or "Codex Token 无效或已过期 (401)"
-        return f'{OAUTH_EXPIRED_PREFIX}{detail}'
+        return f"{OAUTH_EXPIRED_PREFIX}{detail}"
 
     if status_code == 403:
-        detail = message or 'Codex 账户访问受限 (403)'
+        detail = message or "Codex 账户访问受限 (403)"
-        return f'{OAUTH_ACCOUNT_BLOCK_PREFIX}{detail}'
+        return f"{OAUTH_ACCOUNT_BLOCK_PREFIX}{detail}"
 
     return message
 
@@ -202,32 +207,32 @@ async def refresh_codex_key_quota(
 
         if status_code == 402:
             if _looks_like_workspace_deactivated(err_msg):
-                codex_meta = metadata_updates.get(key.id, {}).get('codex')
+                codex_meta = metadata_updates.get(key.id, {}).get("codex")
                 if not isinstance(codex_meta, dict):
                     codex_meta = {}
                 codex_meta = {
                     **codex_meta,
-                    'updated_at': int(time.time()),
+                    "updated_at": int(time.time()),
-                    'account_disabled': True,
+                    "account_disabled": True,
-                    'reason': 'deactivated_workspace',
+                    "reason": "deactivated_workspace",
-                    'message': err_msg or 'deactivated_workspace',
+                    "message": err_msg or "deactivated_workspace",
                 }
-                if oauth_plan_type and not codex_meta.get('plan_type'):
+                if oauth_plan_type and not codex_meta.get("plan_type"):
-                    codex_meta['plan_type'] = oauth_plan_type
+                    codex_meta["plan_type"] = oauth_plan_type
-                metadata_updates[key.id] = {'codex': codex_meta}
+                metadata_updates[key.id] = {"codex": codex_meta}
                 state_updates[key.id] = {
-                    'oauth_invalid_at': datetime.now(timezone.utc),
+                    "oauth_invalid_at": datetime.now(timezone.utc),
-                    'oauth_invalid_reason': _build_structured_invalid_reason(
+                    "oauth_invalid_reason": _build_structured_invalid_reason(
                         status_code=402,
                         upstream_message=err_msg,
                     ),
                 }
                 return {
-                    'key_id': key.id,
+                    "key_id": key.id,
-                    'key_name': key.name,
+                    "key_name": key.name,
-                    'status': 'workspace_deactivated',
+                    "status": "workspace_deactivated",
-                    'message': f"wham/usage API 返回状态码 402{f': {err_msg}' if err_msg else ''}",
+                    "message": f"wham/usage API 返回状态码 402{f': {err_msg}' if err_msg else ''}",
-                    'status_code': 402,
+                    "status_code": 402,
                 }
 
             if key.id not in metadata_updates:

src/services/provider_ops/actions/balance.py

@@ -201,7 +201,7 @@ class BalanceAction(ProviderAction):
         缓存 key 使用 host（同一站点多个 provider 只需签到一次），TTL 6 小时。
         签到失败或 cookie_expired 不写入缓存，允许下次重试。
         """
-        host = client.base_url.host or client.base_url.netloc or str(client.base_url)
+        host = str(client.base_url.host or client.base_url.netloc or client.base_url)
         cache_key = f"provider_ops:checkin:{host}"
 
         # 检查缓存

src/services/system/cache_warmup.py

@@ -10,6 +10,7 @@
 """
 
 import asyncio
+import importlib
 import time
 from dataclasses import dataclass, field
 from typing import Any
@@ -34,6 +35,16 @@ class WarmupContext:
         self.audit_metadata.update(kwargs)
 
 
+def _lazy_create_adapter(class_name: str, **kwargs: Any) -> Any:
+    """从 src.api.dashboard.routes 动态加载 Adapter 类并实例化。
+
+    避免 services→api 的静态 import 依赖。
+    """
+    routes = importlib.import_module("src.api.dashboard.routes")
+    adapter_cls = getattr(routes, class_name)
+    return adapter_cls(**kwargs)
+
+
 class CacheWarmupService:
     """缓存预热服务"""
 
@@ -63,19 +74,20 @@ class CacheWarmupService:
 
         if error_count > 0:
             logger.warning(
-                f"缓存预热完成: {success_count}/3 成功, {error_count} 失败, 耗时 {elapsed:.2f}s"
+                "缓存预热完成: {}/{} 成功, {} 失败, 耗时 {:.2f}s",
+                success_count,
+                3,
+                error_count,
+                elapsed,
             )
         else:
-            logger.info(f"缓存预热完成: {success_count}/3 成功, 耗时 {elapsed:.2f}s")
+            logger.info("缓存预热完成: {}/{} 成功, 耗时 {:.2f}s", success_count, 3, elapsed)
 
     @classmethod
     async def _warmup_admin_dashboard_stats(cls) -> bool:
         """预热管理员仪表盘统计缓存"""
         db = None
         try:
-            from src.api.dashboard.routes import (  # TODO(arch): 提取 dashboard 统计计算到 services 层
-                AdminDashboardStatsAdapter,
-            )
             from src.models.database import User as DBUser
 
             db = create_session()
@@ -87,14 +99,14 @@ class CacheWarmupService:
                 return True
 
             context = WarmupContext(db=db, user=admin_user)
-            adapter = AdminDashboardStatsAdapter()
+            adapter = _lazy_create_adapter("AdminDashboardStatsAdapter")
             await adapter.handle(context)
 
             logger.debug("缓存预热: 管理员仪表盘统计已预热")
             return True
 
         except Exception as e:
-            logger.warning(f"缓存预热失败 (仪表盘统计): {e}")
+            logger.warning("缓存预热失败 (仪表盘统计): {}", e)
             return False
         finally:
             if db:
@@ -120,7 +132,7 @@ class CacheWarmupService:
             return True
 
         except Exception as e:
-            logger.warning(f"缓存预热失败 (热力图): {e}")
+            logger.warning("缓存预热失败 (热力图): {}", e)
             return False
         finally:
             if db:
@@ -131,9 +143,6 @@ class CacheWarmupService:
         """预热每日统计缓存"""
         db = None
         try:
-            from src.api.dashboard.routes import (  # TODO(arch): 提取 dashboard 统计计算到 services 层
-                DashboardDailyStatsAdapter,
-            )
             from src.models.database import User as DBUser
 
             db = create_session()
@@ -147,14 +156,14 @@ class CacheWarmupService:
             context = WarmupContext(db=db, user=admin_user)
 
             # 预热 7 天的每日统计
-            adapter = DashboardDailyStatsAdapter(days=7)
+            adapter = _lazy_create_adapter("DashboardDailyStatsAdapter", days=7)
             await adapter.handle(context)
 
             logger.debug("缓存预热: 每日统计已预热")
             return True
 
         except Exception as e:
-            logger.warning(f"缓存预热失败 (每日统计): {e}")
+            logger.warning("缓存预热失败 (每日统计): {}", e)
             return False
         finally:
             if db:

src/services/usage/stream.py

@@ -480,7 +480,7 @@ class StreamUsageTracker:
         """
         import time
 
-        from src.api.handlers.base.utils import extract_cache_creation_tokens_detail
+        from src.core.usage_tokens import extract_cache_creation_tokens_detail
 
         self.start_time = time.time()
         self.request_data = request_data  # 保存请求数据

tests/api/handlers/base/test_utils.py

@@ -8,11 +8,11 @@ import pytest
 from src.api.handlers.base.utils import (
     build_json_response_for_client,
     build_sse_headers,
-    extract_cache_creation_tokens,
     filter_proxy_response_headers,
     resolve_client_accept_encoding,
     resolve_client_content_encoding,
 )
+from src.core.usage_tokens import extract_cache_creation_tokens
 
 
 class TestExtractCacheCreationTokens:
@@ -60,7 +60,7 @@ class TestExtractCacheCreationTokens:
 
     def test_empty_usage(self) -> None:
         """测试空字典"""
-        usage = {}
+        usage: dict[str, int] = {}
         assert extract_cache_creation_tokens(usage) == 0
 
     def test_all_zeros(self) -> None:

tests/services/api_format/test_capabilities.py

@@ -0,0 +1,109 @@
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+from types import SimpleNamespace
+from unittest.mock import AsyncMock
+
+import pytest
+
+ROOT = Path(__file__).resolve().parents[3]
+if str(ROOT) not in sys.path:
+    sys.path.insert(0, str(ROOT))
+
+from src.core.api_format.capabilities import (
+    compute_total_input_context_for_api_format,
+    fetch_models_for_api_format,
+    get_provider_default_body_rules,
+    register_provider_default_body_rules,
+    register_provider_format_behavior,
+    resolve_billing_template_for_api_format,
+)
+from src.core.api_format.metadata import get_default_body_rules_for_endpoint
+from src.services.provider.behavior import get_provider_behavior
+
+
+class _DummyResp:
+    def __init__(self, status_code: int, payload: object, text: str = "") -> None:
+        self.status_code = status_code
+        self.payload = payload
+        self.text = text
+
+    def json(self) -> object:
+        return self.payload
+
+
+def test_api_format_capability_billing_mapping() -> None:
+    assert resolve_billing_template_for_api_format("openai:cli") == "openai"
+    assert resolve_billing_template_for_api_format("claude:chat") == "claude"
+    assert resolve_billing_template_for_api_format("gemini:cli") == "gemini"
+
+
+def test_provider_default_body_rules_use_unified_registry() -> None:
+    provider_type = "unit_test_provider_rules"
+    expected = [{"action": "drop", "path": "foo"}]
+    register_provider_default_body_rules(provider_type, "openai:cli", expected)
+
+    rules = get_provider_default_body_rules(provider_type, "openai:cli")
+    assert rules == expected
+    assert (
+        get_default_body_rules_for_endpoint("openai:cli", provider_type=provider_type) == expected
+    )
+
+
+def test_provider_behavior_variants_use_unified_registry() -> None:
+    provider_type = "unit_test_variant_provider"
+    register_provider_format_behavior(
+        provider_type,
+        same_format_variant="same-unit",
+        cross_format_variant="cross-unit",
+    )
+
+    behavior = get_provider_behavior(provider_type=provider_type, endpoint_sig="openai:cli")
+    assert behavior.same_format_variant == "same-unit"
+    assert behavior.cross_format_variant == "cross-unit"
+
+
+def test_api_format_capability_total_input_context() -> None:
+    assert compute_total_input_context_for_api_format("openai:chat", 100, 20, 30) == 120
+    assert compute_total_input_context_for_api_format("claude:chat", 100, 20, 30) == 150
+
+
+@pytest.mark.asyncio
+async def test_fetch_models_uses_registered_claude_strategy() -> None:
+    client = SimpleNamespace(
+        get=AsyncMock(
+            side_effect=[
+                _DummyResp(
+                    status_code=200,
+                    payload={
+                        "data": [{"id": "m1"}, {"id": "m2"}],
+                        "has_more": True,
+                        "last_id": "m2",
+                    },
+                ),
+                _DummyResp(
+                    status_code=200,
+                    payload={
+                        "data": [{"id": "m3"}],
+                        "has_more": False,
+                        "last_id": "m3",
+                    },
+                ),
+            ]
+        )
+    )
+
+    models, err = await fetch_models_for_api_format(
+        client,  # type: ignore[arg-type]
+        api_format="claude:chat",
+        base_url="https://api.anthropic.com",
+        api_key="k",
+    )
+
+    assert err is None
+    assert [m.get("id") for m in models] == ["m1", "m2", "m3"]
+    assert all(m.get("api_format") == "claude:chat" for m in models)
+    assert client.get.call_count == 2
+    _, kwargs2 = client.get.call_args_list[1]
+    assert kwargs2.get("params", {}).get("after_id") == "m2"

tests/services/test_provider_keys_key_command_service.py

@@ -178,11 +178,13 @@ def test_clear_oauth_invalid_response_invalidates_caches(
         sys.modules, "src.services.cache.provider_cache", fake_provider_cache_module
     )
 
-    fake_models_service_module = types.ModuleType("src.api.base.models_service")
+    fake_models_service_module = types.ModuleType("src.services.cache.model_list_cache")
     setattr(
         fake_models_service_module, "invalidate_models_list_cache", _fake_invalidate_models_cache
     )
-    monkeypatch.setitem(sys.modules, "src.api.base.models_service", fake_models_service_module)
+    monkeypatch.setitem(
+        sys.modules, "src.services.cache.model_list_cache", fake_models_service_module
+    )
 
     key = SimpleNamespace(
         oauth_invalid_at=datetime.now(timezone.utc),

tests/test_architecture_import_rules.py

@@ -0,0 +1,78 @@
+from __future__ import annotations
+
+import ast
+from dataclasses import dataclass
+from pathlib import Path
+
+
+@dataclass(frozen=True, slots=True)
+class _Violation:
+    file: Path
+    line: int
+    imported: str
+
+
+def _repo_root() -> Path:
+    return Path(__file__).resolve().parents[1]
+
+
+def _iter_python_files(root: Path) -> list[Path]:
+    files: list[Path] = []
+    for path in root.rglob("*.py"):
+        if "__pycache__" in path.parts:
+            continue
+        files.append(path)
+    return files
+
+
+def _scan_imports(py_file: Path) -> list[tuple[int, str]]:
+    """返回 (lineno, module_name) 列表。"""
+    text = py_file.read_text(encoding="utf-8").lstrip("\ufeff")
+    tree = ast.parse(text, filename=str(py_file))
+
+    imports: list[tuple[int, str]] = []
+    for node in ast.walk(tree):
+        if isinstance(node, ast.Import):
+            for alias in node.names:
+                imports.append((int(getattr(node, "lineno", 0) or 0), str(alias.name)))
+        elif isinstance(node, ast.ImportFrom):
+            if node.level:
+                continue
+            if node.module:
+                imports.append((int(getattr(node, "lineno", 0) or 0), str(node.module)))
+    return imports
+
+
+def _scan_forbidden_imports(*, scope_dir: Path, forbidden_prefix: str) -> list[_Violation]:
+    violations: list[_Violation] = []
+    for py_file in _iter_python_files(scope_dir):
+        for line, imported in _scan_imports(py_file):
+            if imported == forbidden_prefix or imported.startswith(f"{forbidden_prefix}."):
+                violations.append(_Violation(file=py_file, line=line, imported=imported))
+    return violations
+
+
+def _format_violations(title: str, violations: list[_Violation]) -> str:
+    lines = [title, ""]
+    for v in sorted(violations, key=lambda x: (str(x.file), x.line, x.imported)):
+        rel = v.file.resolve().relative_to(_repo_root())
+        lines.append(f"- {rel}:{v.line} -> {v.imported}")
+    return "\n".join(lines)
+
+
+def test_services_should_not_import_api() -> None:
+    repo = _repo_root()
+    violations = _scan_forbidden_imports(
+        scope_dir=repo / "src" / "services",
+        forbidden_prefix="src.api",
+    )
+    assert not violations, _format_violations("services 层禁止 import api 层：", violations)
+
+
+def test_core_should_not_import_services() -> None:
+    repo = _repo_root()
+    violations = _scan_forbidden_imports(
+        scope_dir=repo / "src" / "core",
+        forbidden_prefix="src.services",
+    )
+    assert not violations, _format_violations("core 层禁止 import services 层：", violations)