refactor: Antigravity/Codex 服务重构为插件化适配器架构

- 将 Antigravity 和 Codex 从独立模块迁移至 src/services/provider/adapters/ 插件体系
- 新增 provider_types 和 oauth_token 模块，移除 maintenance_scheduler 中的 OAuth 定时刷新
- 增强 admin API：扩展 keys 和 provider_query 端点，新增 dashboard 路由
- 大幅增强 ProviderDetailDrawer 组件，新增 AntigravityQuotaDialog
- 改进 handler 基类（chat/cli）和错误分类器
- 优化 fetch_scheduler 和 upstream_fetcher
- 前端 UI 组件清理和优化
- 更新测试以匹配新模块结构

src/services/antigravity/client.py

@@ -1,75 +0,0 @@
-"""Antigravity API 客户端（最小封装）。"""
-
-from __future__ import annotations
-
-from typing import Any
-
-from src.clients.http_client import HTTPClientPool
-from src.services.antigravity.constants import (
-    DAILY_BASE_URL,
-    HTTP_USER_AGENT,
-    PROD_BASE_URL,
-)
-from src.services.antigravity.url_availability import url_availability
-
-
-async def load_code_assist(
-    access_token: str,
-    proxy_config: dict[str, Any] | None = None,
-    *,
-    timeout_seconds: float = 10.0,
-) -> dict[str, Any]:
-    """调用 /v1internal:loadCodeAssist 获取账户信息。
-
-    注意：
-    - email 需通过 Google userinfo API 获取（由 enrich_auth_config 复用已有逻辑）
-    - 这里仅负责 project_id / tier 等信息
-    - 使用 url_availability 决定优先尝试的 URL
-    """
-    if not access_token:
-        raise ValueError("missing access_token")
-
-    client = await HTTPClientPool.get_proxy_client(proxy_config)
-
-    headers = {
-        "Authorization": f"Bearer {access_token}",
-        "User-Agent": HTTP_USER_AGENT,
-        "Content-Type": "application/json",
-        "Accept": "application/json",
-    }
-    body = {"metadata": {"ideType": "ANTIGRAVITY"}}
-
-    # 使用可用性排序（prod 优先，但会参考历史成功/失败记录）
-    urls = url_availability.get_ordered_urls(prefer_daily=False)
-    if not urls:
-        urls = [PROD_BASE_URL, DAILY_BASE_URL]
-    last_exc: Exception | None = None
-
-    for base_url in urls:
-        try:
-            resp = await client.post(
-                f"{base_url}/v1internal:loadCodeAssist",
-                json=body,
-                headers=headers,
-                timeout=timeout_seconds,
-            )
-            if 200 <= resp.status_code < 300:
-                url_availability.mark_success(base_url)
-                data = resp.json()
-                return data if isinstance(data, dict) else {}
-
-            # 非 2xx：标记不可用并继续 fallback
-            if resp.status_code in (429, 500, 502, 503, 504):
-                url_availability.mark_unavailable(base_url)
-            last_exc = RuntimeError(
-                f"loadCodeAssist failed: status={resp.status_code} base_url={base_url}"
-            )
-        except Exception as e:
-            url_availability.mark_unavailable(base_url)
-            last_exc = e
-            continue
-
-    raise last_exc or RuntimeError("loadCodeAssist failed")
-
-
-__all__ = ["load_code_assist"]

src/services/antigravity/constants.py

@@ -1,40 +0,0 @@
-"""Antigravity 全局常量定义。
-
-注意：这里的 PROVIDER_TYPE 指的是 Provider.provider_type（用于路由与特判），
-不是 endpoint signature（family:kind）。
-"""
-
-from __future__ import annotations
-
-# ============== Provider 标识 ==============
-PROVIDER_TYPE = "antigravity"
-
-# ============== API 端点 ==============
-PROD_BASE_URL = "https://cloudcode-pa.googleapis.com"
-DAILY_BASE_URL = "https://daily-cloudcode-pa.sandbox.googleapis.com"
-
-# ============== User-Agent ==============
-# HTTP Header
-HTTP_USER_AGENT = "antigravity/1.15.8 windows/amd64"
-# V1InternalRequest.userAgent 字段
-REQUEST_USER_AGENT = "antigravity"
-
-# ============== URL 可用性 ==============
-URL_UNAVAILABLE_TTL_SECONDS = 300  # 5 分钟
-
-# ============== Thinking Signature ==============
-DUMMY_THOUGHT_SIGNATURE = "skip_thought_signature_validator"
-
-# ============== v1internal 路径 ==============
-V1INTERNAL_PATH_TEMPLATE = "/v1internal:{action}"
-
-__all__ = [
-    "DAILY_BASE_URL",
-    "DUMMY_THOUGHT_SIGNATURE",
-    "HTTP_USER_AGENT",
-    "PROD_BASE_URL",
-    "PROVIDER_TYPE",
-    "REQUEST_USER_AGENT",
-    "URL_UNAVAILABLE_TTL_SECONDS",
-    "V1INTERNAL_PATH_TEMPLATE",
-]

src/services/antigravity/envelope.py

@@ -1,177 +0,0 @@
-"""Antigravity v1internal request/response envelope helpers.
-
-Antigravity reuses the `gemini:cli` endpoint signature but wraps the actual
-wire format:
-- Request: V1InternalRequest (top-level metadata + nested GeminiRequest)
-- Response: V1InternalResponse (top-level responseId + nested GeminiResponse)
-
-We keep this logic isolated so other providers can reuse the same envelope hook
-pattern.
-"""
-
-from __future__ import annotations
-
-import uuid
-from typing import Any
-
-from src.services.antigravity.constants import HTTP_USER_AGENT as ANTIGRAVITY_HTTP_USER_AGENT
-from src.services.antigravity.constants import REQUEST_USER_AGENT as ANTIGRAVITY_REQUEST_USER_AGENT
-from src.services.antigravity.url_availability import url_availability
-from src.services.provider.request_context import get_selected_base_url
-
-
-def wrap_v1internal_request(
-    gemini_request: dict[str, Any],
-    *,
-    project_id: str,
-    model: str,
-) -> dict[str, Any]:
-    """Wrap a GeminiRequest into Antigravity V1InternalRequest.
-
-    Note: Antigravity expects `model` at top-level; the nested request must not
-    include `model` again.
-    """
-
-    inner_request = dict(gemini_request)
-    inner_request.pop("model", None)
-
-    return {
-        "project": project_id,
-        "requestId": str(uuid.uuid4()),
-        "userAgent": ANTIGRAVITY_REQUEST_USER_AGENT,
-        "requestType": "agent",
-        "model": model,
-        "request": inner_request,
-    }
-
-
-def unwrap_v1internal_response(response: dict[str, Any]) -> dict[str, Any]:
-    """Unwrap Antigravity V1InternalResponse into a GeminiResponse-like dict."""
-
-    inner = response.get("response")
-    if isinstance(inner, dict):
-        unwrapped = dict(inner)
-        resp_id = response.get("responseId")
-        if resp_id is not None:
-            unwrapped["_v1internal_response_id"] = resp_id
-        return unwrapped
-    return response
-
-
-def cache_thought_signatures(model: str, response: dict[str, Any]) -> None:
-    """Best-effort cache for Antigravity thought signatures."""
-
-    try:
-        from src.services.antigravity.signature_cache import signature_cache
-    except Exception:
-        return
-
-    try:
-        candidates = response.get("candidates")
-        if not isinstance(candidates, list):
-            return
-
-        for cand in candidates:
-            if not isinstance(cand, dict):
-                continue
-            content = cand.get("content")
-            if not isinstance(content, dict):
-                continue
-            parts = content.get("parts")
-            if not isinstance(parts, list):
-                continue
-
-            for part in parts:
-                if not isinstance(part, dict):
-                    continue
-                text = part.get("text")
-                if not isinstance(text, str) or not text:
-                    continue
-                sig = (
-                    part.get("thoughtSignature")
-                    or part.get("thought_signature")
-                    or part.get("signature")
-                )
-                if not isinstance(sig, str) or not sig:
-                    continue
-                signature_cache.cache(model, text, sig)
-    except Exception:
-        # Never fail request path due to cache issues.
-        return
-
-
-class AntigravityV1InternalEnvelope:
-    """Provider envelope hooks for Antigravity v1internal wrapper."""
-
-    name = "antigravity:v1internal"
-
-    def extra_headers(self) -> dict[str, str] | None:
-        return {"User-Agent": ANTIGRAVITY_HTTP_USER_AGENT}
-
-    def wrap_request(
-        self,
-        request_body: dict[str, Any],
-        *,
-        model: str,
-        url_model: str | None,
-        decrypted_auth_config: dict[str, Any] | None,
-    ) -> tuple[dict[str, Any], str | None]:
-        project_id = (decrypted_auth_config or {}).get("project_id")
-        if not isinstance(project_id, str) or not project_id:
-            from src.core.exceptions import ProviderNotAvailableException
-
-            raise ProviderNotAvailableException(
-                "Antigravity OAuth 配置缺少 project_id，请重新授权",
-                provider_name="antigravity",
-                upstream_response="missing auth_config.project_id",
-            )
-
-        wrapped = wrap_v1internal_request(
-            request_body,
-            project_id=project_id,
-            model=model,
-        )
-
-        # Antigravity's model lives in the request body, not the URL path.
-        return wrapped, None
-
-    def unwrap_response(self, data: Any) -> Any:
-        if isinstance(data, dict):
-            return unwrap_v1internal_response(data)
-        return data
-
-    def postprocess_unwrapped_response(self, *, model: str, data: Any) -> None:
-        if isinstance(data, dict):
-            cache_thought_signatures(model, data)
-
-    def capture_selected_base_url(self) -> str | None:
-        return get_selected_base_url()
-
-    def on_http_status(self, *, base_url: str | None, status_code: int) -> None:
-        if not base_url:
-            return
-        if status_code == 200:
-            url_availability.mark_success(base_url)
-        elif status_code in (429, 500, 502, 503, 504):
-            url_availability.mark_unavailable(base_url)
-
-    def on_connection_error(self, *, base_url: str | None, exc: Exception) -> None:  # noqa: ARG002
-        if not base_url:
-            return
-        url_availability.mark_unavailable(base_url)
-
-    def force_stream_rewrite(self) -> bool:
-        # Streaming must be rewritten even when endpoint signature matches, because
-        # Antigravity wraps chunks in v1internal envelope.
-        return True
-
-
-antigravity_v1internal_envelope = AntigravityV1InternalEnvelope()
-
-__all__ = [
-    "AntigravityV1InternalEnvelope",
-    "antigravity_v1internal_envelope",
-    "cache_thought_signatures",
-    "unwrap_v1internal_response",
-    "wrap_v1internal_request",
-]

src/services/antigravity/signature_cache.py

@@ -1,58 +0,0 @@
-"""Antigravity thinking block signature cache (minimal)."""
-
-from __future__ import annotations
-
-import hashlib
-import threading
-
-from src.services.antigravity.constants import DUMMY_THOUGHT_SIGNATURE
-
-
-class ThinkingSignatureCache:
-    """缓存 thinking block 签名。
-
-    说明：
-    - 优先使用缓存（比客户端透传更可靠）
-    - Gemini 模型允许使用 dummy signature 作为兜底（跳过验证）
-    - 线程安全
-    """
-
-    def __init__(self, maxsize: int = 1000) -> None:
-        self._cache: dict[str, str] = {}
-        self._maxsize = maxsize
-        self._lock = threading.Lock()
-
-    def get_or_dummy(self, model: str, thinking_text: str) -> str | None:
-        key = self._key(model, thinking_text)
-        with self._lock:
-            cached = self._cache.get(key)
-        if cached:
-            return cached
-        if str(model).startswith("gemini-"):
-            return DUMMY_THOUGHT_SIGNATURE
-        return None
-
-    def cache(self, model: str, thinking_text: str, signature: str) -> None:
-        key = self._key(model, thinking_text)
-
-        with self._lock:
-            if key in self._cache:
-                self._cache[key] = signature
-                return
-
-            if len(self._cache) >= self._maxsize:
-                # 简单 FIFO 清理（dict 保持插入顺序）
-                evict_n = max(1, self._maxsize // 4)
-                for k in list(self._cache.keys())[:evict_n]:
-                    self._cache.pop(k, None)
-
-            self._cache[key] = signature
-
-    def _key(self, model: str, thinking_text: str) -> str:
-        content = f"{model}:{thinking_text}"
-        return hashlib.sha256(content.encode("utf-8")).hexdigest()[:32]
-
-
-signature_cache = ThinkingSignatureCache()
-
-__all__ = ["ThinkingSignatureCache", "signature_cache"]

src/services/model/fetch_scheduler.py

@@ -13,20 +13,25 @@
 
 import asyncio
 import fnmatch
+import json
 import os
+from dataclasses import dataclass
 from datetime import datetime, timezone
+from typing import Any
 
 from sqlalchemy.orm import Session, joinedload
 
 from src.core.cache_service import CacheService
 from src.core.crypto import crypto_service
 from src.core.logger import logger
+from src.core.provider_types import ProviderType
 from src.database import create_session
 from src.models.database import Provider, ProviderAPIKey
 from src.services.model.upstream_fetcher import (
-    build_all_format_configs,
-    fetch_models_from_endpoints,
+    UpstreamModelsFetchContext,
+    fetch_models_for_key,
 )
+from src.services.provider.oauth_token import resolve_oauth_access_token
 from src.services.system.scheduler import get_scheduler
 
 # 从环境变量读取间隔，默认 1440 分钟（1 天），限制在 60-10080 分钟之间
@@ -47,6 +52,19 @@ MODEL_FETCH_HTTP_TIMEOUT = 10.0
 UPSTREAM_MODELS_CACHE_TTL_SECONDS = MODEL_FETCH_INTERVAL_MINUTES * 60
 
 
+@dataclass(frozen=True, slots=True)
+class PreparedModelsFetchContext:
+    key_id: str
+    provider_id: str
+    provider_name: str
+    provider_type: str
+    auth_type: str
+    encrypted_api_key: str
+    encrypted_auth_config: str | None
+    format_to_endpoint: dict[str, Any]
+    proxy_config: dict[str, Any] | None
+
+
 def _match_pattern(model_id: str, pattern: str) -> bool:
     """
     检查模型 ID 是否匹配模式
@@ -270,38 +288,82 @@ class ModelFetchScheduler:
         优化：分两个阶段处理，HTTP 请求期间不持有数据库连接，避免阻塞其他请求
         """
         # ========== 阶段 1：准备数据（短暂持有连接）==========
-        fetch_context = self._prepare_fetch_context(key_id)
-        if fetch_context is None:
+        prepared = self._prepare_fetch_context(key_id)
+        if prepared is None:
             return "skip"
-        if isinstance(fetch_context, str):
-            return fetch_context  # "error" or "skip"
+        if isinstance(prepared, str):
+            return prepared  # "error" or "skip"
 
-        key_id, provider_id, provider_name, api_key_value, endpoint_configs = fetch_context
+        # Resolve auth (incl. lazy OAuth refresh) without holding a DB session.
+        api_key_value: str = ""
+        auth_config: dict[str, Any] | None = None
+
+        if prepared.auth_type == "oauth":
+            # Use request_builder's lazy refresh logic and persist refreshed token back to DB.
+            # Endpoint signature is only used for tracing/debug; auth logic doesn't depend on it.
+            endpoint_api_format = (
+                "gemini:cli" if prepared.provider_type.lower() == ProviderType.ANTIGRAVITY else None
+            )
+            try:
+                resolved = await resolve_oauth_access_token(
+                    key_id=prepared.key_id,
+                    encrypted_api_key=prepared.encrypted_api_key,
+                    encrypted_auth_config=prepared.encrypted_auth_config,
+                    provider_proxy_config=prepared.proxy_config,
+                    endpoint_api_format=endpoint_api_format,
+                )
+                api_key_value = resolved.access_token
+                auth_config = resolved.decrypted_auth_config
+            except Exception as e:
+                self._update_key_error(prepared.key_id, f"OAuth token resolution failed: {e}")
+                return "error"
+        else:
+            try:
+                api_key_value = crypto_service.decrypt(prepared.encrypted_api_key)
+            except Exception:
+                self._update_key_error(prepared.key_id, "Decrypt error")
+                return "error"
+
+            # Best-effort: decrypt auth_config if present (e.g. Antigravity project_id).
+            if prepared.encrypted_auth_config:
+                try:
+                    parsed = json.loads(crypto_service.decrypt(prepared.encrypted_auth_config))
+                    auth_config = parsed if isinstance(parsed, dict) else None
+                except Exception:
+                    auth_config = None
+
+        fetch_ctx = UpstreamModelsFetchContext(
+            provider_type=prepared.provider_type,
+            api_key_value=api_key_value,
+            format_to_endpoint=prepared.format_to_endpoint,
+            proxy_config=prepared.proxy_config,
+            auth_config=auth_config,
+        )
 
         # ========== 阶段 2：HTTP 请求（不持有数据库连接）==========
         # 使用较短的超时时间（10秒），避免长时间阻塞
-        all_models, errors, has_success = await fetch_models_from_endpoints(
-            endpoint_configs, timeout=MODEL_FETCH_HTTP_TIMEOUT
+        all_models, errors, has_success, upstream_metadata = await fetch_models_for_key(
+            fetch_ctx,
+            timeout_seconds=MODEL_FETCH_HTTP_TIMEOUT,
         )
 
         # ========== 阶段 3：更新数据库（获取新连接）==========
         return await self._update_key_after_fetch(
-            key_id=key_id,
-            provider_id=provider_id,
-            provider_name=provider_name,
+            key_id=prepared.key_id,
+            provider_id=prepared.provider_id,
+            provider_name=prepared.provider_name,
             all_models=all_models,
             errors=errors,
             has_success=has_success,
+            upstream_metadata=upstream_metadata,
         )
 
-    def _prepare_fetch_context(
-        self, key_id: str
-    ) -> tuple[str, str, str, str, list[dict]] | str | None:
+    def _prepare_fetch_context(self, key_id: str) -> PreparedModelsFetchContext | str | None:
         """
         准备获取模型所需的上下文数据
 
         Returns:
-            - tuple: (key_id, provider_id, provider_name, api_key_value, endpoint_configs)
+            - PreparedModelsFetchContext: 准备好的上下文（不包含解密后的 token）
             - "skip": 跳过该 Key
             - "error": 出错
             - None: Key 不存在
@@ -349,7 +411,7 @@ class ModelFetchScheduler:
                 logger.info(f"Key {key.id} 为 Vertex AI 类型，跳过自动获取模型")
                 return "skip"
 
-            # 解密 API Key
+            # 基础校验：必须有 api_key（OAuth: 加密 access_token；API Key: 加密 key）
             if not key.api_key:
                 logger.warning(f"Key {key.id} 没有 API Key，跳过")
                 key.last_models_fetch_error = "No API key configured"
@@ -357,17 +419,8 @@ class ModelFetchScheduler:
                 db.commit()
                 return "error"
 
-            try:
-                api_key_value = crypto_service.decrypt(key.api_key)
-            except Exception:
-                logger.error(f"解密 Key {key.id} 失败")
-                key.last_models_fetch_error = "Decrypt error"
-                key.last_models_fetch_at = now
-                db.commit()
-                return "error"
-
             # 构建 api_format -> endpoint 映射
-            format_to_endpoint: dict[str, object] = {}
+            format_to_endpoint: dict[str, Any] = {}
             for endpoint in provider.endpoints:  # type: ignore[attr-defined]
                 if endpoint.is_active:
                     format_to_endpoint[endpoint.api_format] = endpoint
@@ -379,10 +432,22 @@ class ModelFetchScheduler:
                 db.commit()
                 return "error"
 
-            # 使用公共函数构建所有格式的端点配置
-            endpoint_configs = build_all_format_configs(api_key_value, format_to_endpoint)  # type: ignore[arg-type]
+            encrypted_auth_config = getattr(key, "auth_config", None)
+            provider_type = str(getattr(provider, "provider_type", "") or "")
 
-            return (key_id, provider_id, provider.name, api_key_value, endpoint_configs)
+            return PreparedModelsFetchContext(
+                key_id=key_id,
+                provider_id=provider_id,
+                provider_name=provider.name,
+                provider_type=provider_type,
+                auth_type=auth_type,
+                encrypted_api_key=str(key.api_key),
+                encrypted_auth_config=(
+                    encrypted_auth_config if isinstance(encrypted_auth_config, str) else None
+                ),
+                format_to_endpoint=format_to_endpoint,
+                proxy_config=getattr(provider, "proxy", None),
+            )
 
     async def _update_key_after_fetch(
         self,
@@ -392,6 +457,7 @@ class ModelFetchScheduler:
         all_models: list[dict],
         errors: list[str],
         has_success: bool,
+        upstream_metadata: dict[str, Any] | None = None,
     ) -> str:
         """
         HTTP 请求完成后更新数据库
@@ -423,6 +489,16 @@ class ModelFetchScheduler:
             # 有成功的响应，清除错误状态
             key.last_models_fetch_error = None
 
+            # 最佳努力：保存上游元数据（如 Antigravity 配额信息）
+            if upstream_metadata and isinstance(upstream_metadata, dict):
+                # NOTE: upstream_metadata is a plain JSON column (not MutableDict),
+                # so in-place mutation won't be persisted reliably. Always assign
+                # a new dict object to mark the column as dirty.
+                current = key.upstream_metadata
+                merged: dict[str, Any] = dict(current) if isinstance(current, dict) else {}
+                merged.update(upstream_metadata)
+                key.upstream_metadata = merged
+
             # 去重获取模型 ID 列表
             fetched_model_ids: set[str] = set()
             for model in all_models:

src/services/model/upstream_fetcher.py

@@ -7,6 +7,8 @@
 """
 
 import asyncio
+from dataclasses import dataclass
+from typing import Any, Awaitable, Callable
 
 import httpx
 
@@ -21,8 +23,70 @@ MAX_CONCURRENT_REQUESTS = 5
 # 只对这些基础 endpoint signature 获取模型列表，CLI 使用相同的上游 API
 MODEL_FETCH_FORMATS = ["openai:chat", "claude:chat", "gemini:chat"]
 
+# Return tuple signature:
+# (models, errors, has_success, upstream_metadata)
+_ModelsFetcher = Callable[
+    ["UpstreamModelsFetchContext", float],
+    Awaitable[tuple[list[dict], list[str], bool, dict[str, Any] | None]],
+]
 
-def _get_adapter_for_format(api_format: str) -> type | None:
+
+@dataclass(frozen=True)
+class UpstreamModelsFetchContext:
+    """上游模型获取上下文（Key 级别）。"""
+
+    provider_type: str
+    api_key_value: str
+    format_to_endpoint: dict[str, Any]
+    proxy_config: dict[str, Any] | None = None
+    auth_config: dict[str, Any] | None = None
+
+
+class UpstreamModelsFetcherRegistry:
+    """按 provider_type 注册上游模型获取策略，避免到处写特判。"""
+
+    _fetchers: dict[str, _ModelsFetcher] = {}
+
+    @classmethod
+    def register(cls, *, provider_types: list[str], fetcher: _ModelsFetcher) -> None:
+        for pt in provider_types:
+            if not pt:
+                continue
+            cls._fetchers[pt.lower()] = fetcher
+
+    @classmethod
+    def get(cls, provider_type: str) -> _ModelsFetcher | None:
+        if not provider_type:
+            return None
+        return cls._fetchers.get(provider_type.lower())
+
+
+async def _fetch_models_default(
+    ctx: UpstreamModelsFetchContext,
+    timeout_seconds: float,
+) -> tuple[list[dict], list[str], bool, dict[str, Any] | None]:
+    endpoint_configs = build_all_format_configs(ctx.api_key_value, ctx.format_to_endpoint)
+    models, errors, has_success = await fetch_models_from_endpoints(
+        endpoint_configs, timeout=timeout_seconds
+    )
+    return models, errors, has_success, None
+
+
+async def fetch_models_for_key(
+    ctx: UpstreamModelsFetchContext,
+    *,
+    timeout_seconds: float = 30.0,
+) -> tuple[list[dict], list[str], bool, dict[str, Any] | None]:
+    """统一入口：按 provider_type 选择策略获取模型列表（可附带 upstream_metadata）。"""
+    fetcher = UpstreamModelsFetcherRegistry.get(ctx.provider_type) or _fetch_models_default
+    return await fetcher(ctx, timeout_seconds)
+
+
+# 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 类"""
     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
@@ -125,7 +189,7 @@ async def fetch_models_from_endpoints(
         extra_headers = config.get("extra_headers")
 
         try:
-            adapter_class = _get_adapter_for_format(api_format)
+            adapter_class = get_adapter_for_format(api_format)
             if not adapter_class:
                 return [], f"Unknown API format: {api_format}", False
 

src/services/orchestration/error_classifier.py

@@ -14,6 +14,7 @@ import httpx
 from sqlalchemy.orm import Session
 
 from src.core.api_format.signature import make_signature_key
+from src.core.crypto import CryptoService
 from src.core.exceptions import (
     ConcurrencyLimitError,
     ProviderAuthException,
@@ -117,6 +118,37 @@ class ErrorClassifier:
         self.adaptive_manager = adaptive_manager or get_adaptive_rpm_manager()
         self.cache_scheduler = cache_scheduler
 
+    def _extract_oauth_email(self, key: ProviderAPIKey | None) -> str | None:
+        if not key or str(getattr(key, "auth_type", "") or "").lower() != "oauth":
+            return None
+        encrypted_auth_config = getattr(key, "auth_config", None)
+        if not encrypted_auth_config:
+            return None
+        try:
+            decrypted = CryptoService().decrypt(encrypted_auth_config, silent=True)
+            auth_config = json.loads(decrypted) if decrypted else {}
+        except Exception:
+            return None
+        email = auth_config.get("email")
+        if isinstance(email, str):
+            email = email.strip()
+            if email:
+                return email
+        return None
+
+    def _format_key_display(self, key: ProviderAPIKey | None) -> str:
+        if not key:
+            return "key=unknown"
+        key_id = str(getattr(key, "id", "") or "")[:8] or "unknown"
+        name = str(getattr(key, "name", "") or "").strip()
+        email = self._extract_oauth_email(key)
+        parts = [f"key={key_id}"]
+        if email:
+            parts.append(f"email={email}")
+        if name and name != email:
+            parts.append(f"name={name}")
+        return " ".join(parts)
+
     # 表示客户端错误的 error type（不区分大小写）
     # 这些 type 表明是请求本身的问题，不应重试
     CLIENT_ERROR_TYPES: tuple[str, ...] = (
@@ -344,6 +376,38 @@ class ErrorClassifier:
         search_text = error_text.lower()
         return any(p.lower() in search_text for p in self.THINKING_ERROR_PATTERNS)
 
+    def _is_account_validation_required(self, error_text: str | None) -> bool:
+        """
+        检测 403 错误是否为 Google 账号验证要求 (VALIDATION_REQUIRED)
+
+        Google 会在某些情况下要求账号所有者手动完成人机验证，
+        此时所有 API 请求都会返回 403 + VALIDATION_REQUIRED。
+        这是账号级别的永久性错误，重试无法修复，需要人工干预。
+
+        匹配条件（满足任一即可）：
+        - error.details 中包含 reason=VALIDATION_REQUIRED
+        - error.status 为 PERMISSION_DENIED 且 message 包含 "verify your account"
+        - error.message 包含 "verify your account"
+
+        Args:
+            error_text: 错误响应文本
+
+        Returns:
+            是否为账号验证要求错误
+        """
+        if not error_text:
+            return False
+
+        search_text = error_text.lower()
+
+        # 快速路径：关键词匹配
+        if "validation_required" in search_text:
+            return True
+        if "verify your account" in search_text and "permission_denied" in search_text:
+            return True
+
+        return False
+
     def _extract_error_message(self, error_text: str | None) -> str | None:
         """
         从错误响应中提取错误消息
@@ -393,7 +457,10 @@ class ErrorClassifier:
             return ErrorAction.BREAK
 
         if isinstance(error, httpx.HTTPStatusError):
-            # HTTP 错误根据状态码决定
+            status_code = int(getattr(error.response, "status_code", 0) or 0)
+            # 401/403 是认证/权限错误，在同一个 key 上重试无意义，直接跳到下一个候选
+            if status_code in (401, 403):
+                return ErrorAction.BREAK
             return ErrorAction.CONTINUE if has_retry_left else ErrorAction.BREAK
 
         if isinstance(error, self.RETRIABLE_ERRORS):
@@ -500,6 +567,12 @@ class ErrorClassifier:
         if status == 401:
             return ProviderAuthException(provider_name=provider_name)
 
+        # 403: 检查是否为 Google VALIDATION_REQUIRED（账号需要手动验证）
+        # 这类错误是永久性的，重试同一个 key 无意义，应视为认证错误
+        if status == 403 and self._is_account_validation_required(error_response_text):
+            logger.warning("检测到 Google 账号验证要求 (VALIDATION_REQUIRED): {}", provider_name)
+            return ProviderAuthException(provider_name=provider_name)
+
         if status == 429:
             return ProviderRateLimitException(
                 message="请求过于频繁，请稍后重试",
@@ -651,6 +724,32 @@ class ErrorClassifier:
                     api_format=provider_format_str,
                     error_type="ProviderAuthException",
                 )
+            # 403 VALIDATION_REQUIRED → 标记 OAuth key 为账号级别封禁
+            # 这与 test-model 端点的行为对齐（provider_query.py 第 669-690 行）
+            status_code = http_error.response.status_code if http_error.response else None
+            if (
+                status_code == 403
+                and key
+                and str(getattr(key, "auth_type", "") or "").lower() == "oauth"
+                and self._is_account_validation_required(error_response_text)
+            ):
+                try:
+                    from datetime import datetime, timezone
+
+                    from src.services.provider.oauth_token import (
+                        OAUTH_ACCOUNT_BLOCK_PREFIX,
+                    )
+
+                    key.oauth_invalid_at = datetime.now(timezone.utc)
+                    key.oauth_invalid_reason = f"{OAUTH_ACCOUNT_BLOCK_PREFIX}Google 要求验证账号"
+                    self.db.commit()
+                    logger.warning(
+                        "  [{}] {} 因 403 VALIDATION_REQUIRED 已标记为账号异常",
+                        request_id,
+                        self._format_key_display(key),
+                    )
+                except Exception as mark_exc:
+                    logger.debug("  [{}] 标记 oauth_invalid 失败: {}", request_id, mark_exc)
             return extra_data
 
         # 处理限流错误

src/services/provider/adapters/__init__.py

@@ -0,0 +1 @@
+"""Provider-specific adapters (antigravity, codex, etc.)."""

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

@@ -0,0 +1,524 @@
+"""Antigravity API 客户端（与 Antigravity-Manager 对齐）。"""
+
+from __future__ import annotations
+
+import asyncio
+import json
+import random
+import re
+from typing import Any
+
+from src.clients.http_client import HTTPClientPool
+from src.core.logger import logger
+from src.services.provider.adapters.antigravity.constants import (
+    DAILY_BASE_URL,
+    PROD_BASE_URL,
+    SANDBOX_BASE_URL,
+    VERSION_FETCH_URL,
+    get_http_user_agent,
+    parse_version_string,
+    update_user_agent_version,
+)
+from src.services.provider.adapters.antigravity.url_availability import url_availability
+
+# loadCodeAssist 请求体 metadata
+_CODE_ASSIST_METADATA = {
+    "ideType": "ANTIGRAVITY",
+}
+
+# Duration 解析正则（与 AM 的 retry.rs 对齐）
+_DURATION_RE = re.compile(r"([\d.]+)\s*(ms|s|m|h)")
+
+
+# ---------------------------------------------------------------------------
+# Retry-After / Duration 解析工具（对齐 AM upstream/retry.rs）
+# ---------------------------------------------------------------------------
+
+
+def _parse_duration_ms(duration_str: str) -> int | None:
+    """解析 Duration 字符串 (e.g. '1.5s', '200ms', '1h16m0.667s')，返回毫秒。"""
+    total_ms = 0.0
+    matched = False
+    for m in _DURATION_RE.finditer(duration_str):
+        matched = True
+        value = float(m.group(1))
+        unit = m.group(2)
+        if unit == "ms":
+            total_ms += value
+        elif unit == "s":
+            total_ms += value * 1000
+        elif unit == "m":
+            total_ms += value * 60_000
+        elif unit == "h":
+            total_ms += value * 3_600_000
+    return round(total_ms) if matched else None
+
+
+def parse_retry_delay(error_body: str | bytes | dict[str, Any]) -> float | None:
+    """从 429 错误响应体中提取 retry delay（秒）。
+
+    支持两种格式（与 AM 对齐）：
+    1. error.details[].@type="...RetryInfo" → retryDelay
+    2. error.details[].metadata.quotaResetDelay
+    """
+    try:
+        data: dict[str, Any]
+        if isinstance(error_body, (str, bytes)):
+            data = json.loads(error_body)
+        elif isinstance(error_body, dict):
+            data = error_body
+        else:
+            return None
+
+        details = data.get("error", {}).get("details", [])
+        if not isinstance(details, list):
+            return None
+
+        # 方式 1: RetryInfo.retryDelay
+        for detail in details:
+            if not isinstance(detail, dict):
+                continue
+            type_str = detail.get("@type", "")
+            if isinstance(type_str, str) and "RetryInfo" in type_str:
+                retry_delay = detail.get("retryDelay")
+                if isinstance(retry_delay, str):
+                    ms = _parse_duration_ms(retry_delay)
+                    if ms is not None:
+                        return min((ms + 200) / 1000.0, 30.0)
+
+        # 方式 2: metadata.quotaResetDelay
+        for detail in details:
+            if not isinstance(detail, dict):
+                continue
+            metadata = detail.get("metadata")
+            if not isinstance(metadata, dict):
+                continue
+            quota_delay = metadata.get("quotaResetDelay")
+            if isinstance(quota_delay, str):
+                ms = _parse_duration_ms(quota_delay)
+                if ms is not None:
+                    return min((ms + 200) / 1000.0, 30.0)
+    except Exception:
+        pass
+    return None
+
+
+# ---------------------------------------------------------------------------
+# Fallback 判断（对齐 AM upstream/client.rs: should_try_next_endpoint）
+# ---------------------------------------------------------------------------
+
+
+def _should_fallback_status(status_code: int) -> bool:
+    """判断是否应该 fallback 到下一个端点。
+
+    与 AM 对齐：429 + 408(超时) + 404 + 所有 5xx。
+    4xx 客户端错误（400/401/403 等）不 fallback——换 URL 也不会成功。
+    """
+    if status_code == 429:
+        return True
+    if status_code in (404, 408):
+        return True
+    if 500 <= status_code < 600:  # noqa: PLR2004
+        return True
+    return False
+
+
+# ---------------------------------------------------------------------------
+# 从 loadCodeAssist 响应中提取信息
+# ---------------------------------------------------------------------------
+
+
+def extract_tier_id(data: dict[str, Any]) -> str:
+    """从 loadCodeAssist 响应中提取 tier ID。
+
+    优先选 allowedTiers 中 isDefault=true 的，fallback 到第一个，最终 fallback 到 "LEGACY"。
+    """
+    allowed_tiers = data.get("allowedTiers")
+    if not isinstance(allowed_tiers, list):
+        return "LEGACY"
+
+    # 第一轮：找 isDefault
+    for tier in allowed_tiers:
+        if isinstance(tier, dict) and tier.get("isDefault") is True:
+            tier_id = tier.get("id", "")
+            if isinstance(tier_id, str) and tier_id.strip():
+                return tier_id.strip()
+
+    # 第二轮：取第一个有 id 的
+    for tier in allowed_tiers:
+        if isinstance(tier, dict):
+            tier_id = tier.get("id", "")
+            if isinstance(tier_id, str) and tier_id.strip():
+                return tier_id.strip()
+
+    return "LEGACY"
+
+
+def extract_project_id(data: dict[str, Any]) -> str:
+    """从响应中提取 project_id，兼容 string 和 {"id": "..."} 两种格式。"""
+    raw = data.get("cloudaicompanionProject")
+    if isinstance(raw, str) and raw.strip():
+        return raw.strip()
+    if isinstance(raw, dict):
+        pid = raw.get("id", "")
+        if isinstance(pid, str) and pid.strip():
+            return pid.strip()
+    return ""
+
+
+# ---------------------------------------------------------------------------
+# 核心 API 客户端函数
+# ---------------------------------------------------------------------------
+
+
+async def load_code_assist(
+    access_token: str,
+    proxy_config: dict[str, Any] | None = None,
+    *,
+    timeout_seconds: float = 10.0,
+) -> dict[str, Any]:
+    """调用 /v1internal:loadCodeAssist 获取账户信息。
+
+    对齐 AM project_resolver.rs：Sandbox 优先，避免 Prod 429。
+
+    注意：
+    - email 需通过 Google userinfo API 获取（由 enrich_auth_config 复用已有逻辑）
+    - 这里仅负责 project_id / tier 等信息
+    """
+    if not access_token:
+        raise ValueError("missing access_token")
+
+    client = await HTTPClientPool.get_proxy_client(proxy_config)
+
+    headers = {
+        "Authorization": f"Bearer {access_token}",
+        "User-Agent": get_http_user_agent(),
+        "Content-Type": "application/json",
+    }
+    body = {"metadata": _CODE_ASSIST_METADATA}
+
+    # Sandbox 优先（与 AM 对齐：避免 Prod 429）
+    urls = url_availability.get_ordered_urls(prefer_daily=True)
+    if not urls:
+        urls = [SANDBOX_BASE_URL, DAILY_BASE_URL, PROD_BASE_URL]
+    last_exc: Exception | None = None
+
+    for base_url in urls:
+        try:
+            resp = await client.post(
+                f"{base_url}/v1internal:loadCodeAssist",
+                json=body,
+                headers=headers,
+                timeout=timeout_seconds,
+            )
+            if 200 <= resp.status_code < 300:
+                url_availability.mark_success(base_url)
+                data = resp.json()
+                return data if isinstance(data, dict) else {}
+
+            # 可 fallback 的错误：标记不可用，尝试下一个 URL
+            if _should_fallback_status(resp.status_code):
+                url_availability.mark_unavailable(base_url)
+
+                # 429：尝试解析 Retry-After 并等待
+                if resp.status_code == 429:
+                    delay = parse_retry_delay(resp.text)
+                    if delay and delay > 0:
+                        logger.debug(
+                            "[antigravity] loadCodeAssist 429, waiting {:.1f}s before fallback",
+                            delay,
+                        )
+                        await asyncio.sleep(min(delay, 5.0))
+
+                last_exc = RuntimeError(
+                    f"loadCodeAssist failed: status={resp.status_code} base_url={base_url}"
+                )
+                continue
+
+            # 不可 fallback 的 4xx（400/401/403 等）：直接抛出，换 URL 也不会成功
+            raise RuntimeError(
+                f"loadCodeAssist failed: status={resp.status_code} base_url={base_url} "
+                f"body={resp.text[:200] if resp.text else ''}"
+            )
+        except RuntimeError:
+            raise
+        except Exception as e:
+            url_availability.mark_unavailable(base_url)
+            last_exc = e
+            continue
+
+    raise last_exc or RuntimeError("loadCodeAssist failed: all endpoints exhausted")
+
+
+async def onboard_user(
+    access_token: str,
+    tier_id: str = "LEGACY",
+    proxy_config: dict[str, Any] | None = None,
+    *,
+    max_attempts: int = 5,
+    poll_interval: float = 2.0,
+    timeout_seconds: float = 30.0,
+) -> str:
+    """调用 /v1internal:onboardUser 激活账号并获取 project_id。
+
+    当 loadCodeAssist 返回 allowedTiers 但没有 cloudaicompanionProject 时，
+    需要先通过 onboardUser 选择 tier 来分配 project。
+
+    改进（对齐 AM）：
+    - 使用 url_availability 做多 URL fallback
+    - 轮询期间的临时网络错误会 continue 而非直接终止
+
+    Args:
+        access_token: OAuth access token
+        tier_id: 要选择的 tier ID（从 allowedTiers 中提取）
+        proxy_config: 代理配置
+        max_attempts: 最大轮询次数（onboardUser 是异步操作）
+        poll_interval: 轮询间隔（秒）
+        timeout_seconds: 单次请求超时
+
+    Returns:
+        project_id
+
+    Raises:
+        RuntimeError: 激活失败
+    """
+    if not access_token:
+        raise ValueError("missing access_token")
+
+    client = await HTTPClientPool.get_proxy_client(proxy_config)
+
+    headers = {
+        "Authorization": f"Bearer {access_token}",
+        "User-Agent": get_http_user_agent(),
+        "Content-Type": "application/json",
+    }
+    body = {
+        "tierId": tier_id,
+        "metadata": _CODE_ASSIST_METADATA,
+    }
+
+    # 使用 url_availability 选择端点（与其他函数一致）
+    urls = url_availability.get_ordered_urls(prefer_daily=True)
+    if not urls:
+        urls = [SANDBOX_BASE_URL, DAILY_BASE_URL, PROD_BASE_URL]
+    base_url = urls[0]
+
+    logger.info("[antigravity] onboardUser: 开始激活账号, tier={}, endpoint={}", tier_id, base_url)
+
+    for attempt in range(1, max_attempts + 1):
+        try:
+            resp = await client.post(
+                f"{base_url}/v1internal:onboardUser",
+                json=body,
+                headers=headers,
+                timeout=timeout_seconds,
+            )
+            if resp.status_code < 200 or resp.status_code >= 300:
+                # 标记可用性
+                if _should_fallback_status(resp.status_code):
+                    url_availability.mark_unavailable(base_url)
+                raise RuntimeError(
+                    f"onboardUser failed: status={resp.status_code}, "
+                    f"body={resp.text[:200] if resp.text else ''}"
+                )
+
+            url_availability.mark_success(base_url)
+            data = resp.json()
+            if not isinstance(data, dict):
+                raise RuntimeError(f"onboardUser: unexpected response type: {type(data)}")
+
+            done = data.get("done")
+            if done is True:
+                # 从 response.cloudaicompanionProject 提取 project_id
+                response_data = data.get("response")
+                if isinstance(response_data, dict):
+                    project_id = extract_project_id(response_data)
+                    if project_id:
+                        logger.info(
+                            "[antigravity] onboardUser: 激活成功, project_id={}",
+                            project_id[:8] + "...",
+                        )
+                        return project_id
+
+                # done=true 但 project_id 为空（常见：cloudaicompanionProject: {}）
+                # 返回空串，由调用方 fallback 到随机 project_id
+                logger.debug("[antigravity] onboardUser: done=true 但 project_id 为空")
+                return ""
+
+            # done != true，继续轮询
+            logger.debug(
+                "[antigravity] onboardUser: 轮询 {}/{}, 等待完成...",
+                attempt,
+                max_attempts,
+            )
+            if attempt < max_attempts:
+                await asyncio.sleep(poll_interval)
+
+        except RuntimeError:
+            raise
+        except Exception as e:
+            # 临时网络错误：记录并继续轮询（而非直接终止）
+            logger.warning(
+                "[antigravity] onboardUser: 轮询 {}/{} 网络错误: {}, 继续重试...",
+                attempt,
+                max_attempts,
+                e,
+            )
+            if attempt < max_attempts:
+                await asyncio.sleep(poll_interval)
+                continue
+            raise RuntimeError(
+                f"onboardUser request failed after {max_attempts} attempts: {e}"
+            ) from e
+
+    raise RuntimeError(f"onboardUser: 超时，已轮询 {max_attempts} 次仍未完成")
+
+
+async def fetch_available_models(
+    access_token: str,
+    *,
+    project_id: str,
+    proxy_config: dict[str, Any] | None = None,
+    timeout_seconds: float = 10.0,
+) -> dict[str, Any]:
+    """调用 /v1internal:fetchAvailableModels 获取可用模型（包含配额信息）。
+
+    Antigravity 的此接口会返回类似：
+        {"models": {"claude-sonnet-4": {"displayName": "...", "quotaInfo": {...}}, ...}}
+
+    对齐 AM：Sandbox 优先 + 正确的 fallback 逻辑。
+    """
+    if not access_token:
+        raise ValueError("missing access_token")
+    if not project_id:
+        raise ValueError("missing project_id")
+
+    client = await HTTPClientPool.get_proxy_client(proxy_config)
+
+    headers = {
+        "Authorization": f"Bearer {access_token}",
+        "User-Agent": get_http_user_agent(),
+        "Content-Type": "application/json",
+        "Accept": "application/json",
+    }
+    body = {"project": project_id}
+
+    # Sandbox 优先
+    urls = url_availability.get_ordered_urls(prefer_daily=True)
+    if not urls:
+        urls = [SANDBOX_BASE_URL, DAILY_BASE_URL, PROD_BASE_URL]
+    last_exc: Exception | None = None
+
+    for base_url in urls:
+        try:
+            resp = await client.post(
+                f"{base_url}/v1internal:fetchAvailableModels",
+                json=body,
+                headers=headers,
+                timeout=timeout_seconds,
+            )
+            if 200 <= resp.status_code < 300:
+                url_availability.mark_success(base_url)
+                data = resp.json()
+                return data if isinstance(data, dict) else {}
+
+            # 可 fallback 的错误
+            if _should_fallback_status(resp.status_code):
+                url_availability.mark_unavailable(base_url)
+
+                # 429：尝试等待
+                if resp.status_code == 429:
+                    delay = parse_retry_delay(resp.text)
+                    if delay and delay > 0:
+                        logger.debug(
+                            "[antigravity] fetchAvailableModels 429, waiting {:.1f}s",
+                            delay,
+                        )
+                        await asyncio.sleep(min(delay, 5.0))
+
+                last_exc = RuntimeError(
+                    f"fetchAvailableModels failed: status={resp.status_code} base_url={base_url}"
+                )
+                continue
+
+            # 不可 fallback 的 4xx：直接抛出
+            raise RuntimeError(
+                f"fetchAvailableModels failed: status={resp.status_code} base_url={base_url} "
+                f"body={resp.text[:200] if resp.text else ''}"
+            )
+        except RuntimeError:
+            raise
+        except Exception as e:
+            url_availability.mark_unavailable(base_url)
+            last_exc = e
+            continue
+
+    raise last_exc or RuntimeError("fetchAvailableModels failed: all endpoints exhausted")
+
+
+# ---------------------------------------------------------------------------
+# User-Agent 版本号动态更新
+# ---------------------------------------------------------------------------
+
+
+async def refresh_user_agent(
+    proxy_config: dict[str, Any] | None = None,
+    *,
+    timeout_seconds: float = 5.0,
+) -> str | None:
+    """从远程 API 获取最新 Antigravity 版本号并更新 User-Agent。
+
+    对齐 AM constants.rs：
+    1. 尝试 VERSION_FETCH_URL
+    2. Fallback 到 changelog 页面
+    3. 最终 fallback 到 _FALLBACK_VERSION
+
+    Returns:
+        更新后的版本号，或 None（如果获取失败但 fallback 到默认版本）。
+    """
+    try:
+        client = await HTTPClientPool.get_proxy_client(proxy_config)
+        resp = await client.get(VERSION_FETCH_URL, timeout=timeout_seconds)
+        if 200 <= resp.status_code < 300:
+            version = parse_version_string(resp.text)
+            if version:
+                update_user_agent_version(version)
+                logger.info("[antigravity] User-Agent 版本已更新: {}", version)
+                return version
+    except Exception as e:
+        logger.debug("[antigravity] 获取远程版本失败: {}", e)
+
+    return None
+
+
+# ---------------------------------------------------------------------------
+# Fallback Project ID 生成
+# ---------------------------------------------------------------------------
+
+_ADJECTIVES = ("useful", "bright", "swift", "calm", "bold")
+_NOUNS = ("fuze", "wave", "spark", "flow", "core")
+
+
+def generate_fallback_project_id() -> str:
+    """生成随机 project_id（与 AM project_resolver.rs 的 generateProjectID 一致）。
+
+    格式: {adjective}-{noun}-{5位随机字符(base36)}
+    Antigravity API 不严格校验 project 字段，当所有正常获取途径都失败时用作 fallback。
+    """
+    adj = random.choice(_ADJECTIVES)  # noqa: S311
+    noun = random.choice(_NOUNS)  # noqa: S311
+    chars = "abcdefghijklmnopqrstuvwxyz0123456789"
+    rand_part = "".join(random.choice(chars) for _ in range(5))  # noqa: S311
+    return f"{adj}-{noun}-{rand_part}"
+
+
+__all__ = [
+    "extract_project_id",
+    "extract_tier_id",
+    "fetch_available_models",
+    "generate_fallback_project_id",
+    "load_code_assist",
+    "onboard_user",
+    "parse_retry_delay",
+    "refresh_user_agent",
+]

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

@@ -0,0 +1,143 @@
+"""Antigravity 全局常量定义。
+
+注意：这里的 PROVIDER_TYPE 指的是 Provider.provider_type（用于路由与特判），
+不是 endpoint signature（family:kind）。
+"""
+
+from __future__ import annotations
+
+import platform
+import re
+import threading
+
+# ============== API 端点 ==============
+PROD_BASE_URL = "https://cloudcode-pa.googleapis.com"
+DAILY_BASE_URL = "https://daily-cloudcode-pa.googleapis.com"
+SANDBOX_BASE_URL = "https://daily-cloudcode-pa.sandbox.googleapis.com"
+
+# ============== User-Agent ==============
+VERSION_FETCH_URL = "https://antigravity-auto-updater-974169037036.us-central1.run.app"
+_FALLBACK_VERSION = "1.15.8"
+_VERSION_RE = re.compile(r"\d+\.\d+\.\d+")
+
+
+def _detect_platform_tag() -> str:
+    """检测当前运行平台，格式与 Go runtime 保持一致。"""
+    os_name = platform.system().lower()  # linux, darwin, windows
+    arch = platform.machine().lower()
+    if arch in ("x86_64", "amd64"):
+        arch = "amd64"
+    elif arch in ("aarch64", "arm64"):
+        arch = "arm64"
+    return f"{os_name}/{arch}"
+
+
+_PLATFORM_TAG = _detect_platform_tag()
+
+# HTTP Header User-Agent（向后兼容：模块级常量保留，新代码应使用 get_http_user_agent()）
+HTTP_USER_AGENT = f"antigravity/{_FALLBACK_VERSION} {_PLATFORM_TAG}"
+
+# V1InternalRequest.userAgent 字段（固定值）
+REQUEST_USER_AGENT = "antigravity"
+
+# --- 动态 User-Agent 支持 ---
+_ua_lock = threading.Lock()
+_ua_version: str = _FALLBACK_VERSION
+
+
+def get_http_user_agent() -> str:
+    """返回当前 HTTP User-Agent 字符串（支持动态版本号更新）。"""
+    with _ua_lock:
+        return f"antigravity/{_ua_version} {_PLATFORM_TAG}"
+
+
+def update_user_agent_version(version: str) -> None:
+    """更新 User-Agent 中的版本号（由 refresh_user_agent 调用）。"""
+    global _ua_version  # noqa: PLW0603
+    with _ua_lock:
+        _ua_version = version
+
+
+def parse_version_string(text: str) -> str | None:
+    """从任意文本中提取 X.Y.Z 格式的版本号。"""
+    m = _VERSION_RE.search(text)
+    return m.group(0) if m else None
+
+
+# ============== URL 可用性 ==============
+URL_UNAVAILABLE_TTL_SECONDS = 300  # 5 分钟
+
+# ============== Thinking Signature ==============
+DUMMY_THOUGHT_SIGNATURE = "skip_thought_signature_validator"
+MIN_SIGNATURE_LENGTH = 50  # 与 Antigravity-Manager 对齐
+
+# ============== Thinking Budget ==============
+THINKING_BUDGET_AUTO_CAP = 24576
+THINKING_BUDGET_DEFAULT_INJECT = 16000
+# 包含这些关键字的模型会自动注入 thinkingConfig（如果缺失）
+THINKING_MODELS_AUTO_INJECT_KEYWORDS = ("thinking", "gemini-2.0-pro", "gemini-3-pro")
+
+# ============== Retry ==============
+RETRY_429_BASE_SECONDS = 5.0
+RETRY_503_BASE_SECONDS = 10.0
+RETRY_503_MAX_SECONDS = 60.0
+RETRY_500_BASE_SECONDS = 3.0
+
+# ============== v1internal 路径 ==============
+V1INTERNAL_PATH_TEMPLATE = "/v1internal:{action}"
+
+# ============== Signature 错误关键字（用于 400 错误检测） ==============
+SIGNATURE_ERROR_KEYWORDS = (
+    "Invalid `signature`",
+    "thinking.signature",
+    "thinking.thinking",
+    "Corrupted thought signature",
+)
+
+# ============== Antigravity System Instruction ==============
+ANTIGRAVITY_SYSTEM_INSTRUCTION = (
+    "You are Antigravity, a powerful agentic AI coding assistant designed by the "
+    "Google Deepmind team working on Advanced Agentic Coding.\n"
+    "You are pair programming with a USER to solve their coding task. The task may "
+    "require creating a new codebase, modifying or debugging an existing codebase, "
+    "or simply answering a question.\n"
+    "**Absolute paths only**\n"
+    "**Proactiveness**"
+)
+
+# ============== JSON Schema 禁止字段 ==============
+FORBIDDEN_SCHEMA_FIELDS = frozenset(
+    {
+        "multipleOf",
+        "exclusiveMinimum",
+        "exclusiveMaximum",
+        "contentEncoding",
+        "contentMediaType",
+    }
+)
+
+__all__ = [
+    "ANTIGRAVITY_SYSTEM_INSTRUCTION",
+    "DAILY_BASE_URL",
+    "DUMMY_THOUGHT_SIGNATURE",
+    "FORBIDDEN_SCHEMA_FIELDS",
+    "HTTP_USER_AGENT",
+    "MIN_SIGNATURE_LENGTH",
+    "PROD_BASE_URL",
+    "REQUEST_USER_AGENT",
+    "RETRY_429_BASE_SECONDS",
+    "RETRY_500_BASE_SECONDS",
+    "RETRY_503_BASE_SECONDS",
+    "RETRY_503_MAX_SECONDS",
+    "SANDBOX_BASE_URL",
+    "SIGNATURE_ERROR_KEYWORDS",
+    "THINKING_BUDGET_AUTO_CAP",
+    "THINKING_BUDGET_DEFAULT_INJECT",
+    "THINKING_MODELS_AUTO_INJECT_KEYWORDS",
+    "URL_UNAVAILABLE_TTL_SECONDS",
+    "V1INTERNAL_PATH_TEMPLATE",
+    "VERSION_FETCH_URL",
+    "get_http_user_agent",
+    "parse_version_string",
+    "update_user_agent_version",
+]

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

@@ -0,0 +1,560 @@
+"""Antigravity v1internal request/response envelope helpers.
+
+Antigravity reuses the `gemini:cli` endpoint signature but wraps the actual
+wire format:
+- Request: V1InternalRequest (top-level metadata + nested GeminiRequest)
+- Response: V1InternalResponse (top-level responseId + nested GeminiResponse)
+
+对齐 Antigravity-Manager wrapper.rs 的处理逻辑：
+- Claude model tool ID 注入（request + response）
+- Thinking budget capping (Auto cap 24576)
+- [undefined] 字符串深度清理
+- parametersJsonSchema → parameters 重命名
+- JSON Schema 禁止字段清洗
+- Antigravity System Instruction 注入
+- Signature 错误检测
+"""
+
+from __future__ import annotations
+
+import uuid
+from typing import Any
+
+from src.services.provider.adapters.antigravity.constants import (
+    ANTIGRAVITY_SYSTEM_INSTRUCTION,
+    FORBIDDEN_SCHEMA_FIELDS,
+)
+from src.services.provider.adapters.antigravity.constants import (
+    REQUEST_USER_AGENT as ANTIGRAVITY_REQUEST_USER_AGENT,
+)
+from src.services.provider.adapters.antigravity.constants import (
+    SIGNATURE_ERROR_KEYWORDS,
+    THINKING_BUDGET_AUTO_CAP,
+    THINKING_BUDGET_DEFAULT_INJECT,
+    THINKING_MODELS_AUTO_INJECT_KEYWORDS,
+    get_http_user_agent,
+)
+from src.services.provider.adapters.antigravity.url_availability import url_availability
+from src.services.provider.request_context import get_selected_base_url
+
+# ---------------------------------------------------------------------------
+# Request body 预处理工具函数（对齐 AM wrapper.rs / common_utils.rs）
+# ---------------------------------------------------------------------------
+
+
+def _deep_clean_undefined(obj: Any) -> None:
+    """In-place 递归清理 '[undefined]' 字符串值。
+
+    Cherry Studio 等客户端会在请求中注入 '[undefined]' 字符串，
+    可能导致上游 API 解析错误。
+    """
+    if isinstance(obj, dict):
+        keys_to_remove = [k for k, v in obj.items() if v == "[undefined]"]
+        for k in keys_to_remove:
+            del obj[k]
+        for v in obj.values():
+            if isinstance(v, (dict, list)):
+                _deep_clean_undefined(v)
+    elif isinstance(obj, list):
+        i = 0
+        while i < len(obj):
+            if obj[i] == "[undefined]":
+                obj.pop(i)
+            else:
+                if isinstance(obj[i], (dict, list)):
+                    _deep_clean_undefined(obj[i])
+                i += 1
+
+
+def _inject_claude_tool_ids_request(inner_request: dict[str, Any], model: str) -> None:
+    """为 Claude 模型注入 functionCall/functionResponse 的 id 字段。
+
+    Google v1internal 在目标模型为 Claude 时要求 functionCall 带有 id 字段，
+    但标准 Gemini 协议不包含此字段。对齐 AM wrapper.rs #1522。
+    """
+    if "claude" not in model.lower():
+        return
+
+    contents = inner_request.get("contents")
+    if not isinstance(contents, list):
+        return
+
+    for content in contents:
+        if not isinstance(content, dict):
+            continue
+        parts = content.get("parts")
+        if not isinstance(parts, list):
+            continue
+
+        # 每条消息维护独立的计数器（确保 Call 和 Response 生成匹配的 ID）
+        name_counters: dict[str, int] = {}
+
+        for part in parts:
+            if not isinstance(part, dict):
+                continue
+
+            # 1. functionCall（Assistant 请求调用工具）
+            fc = part.get("functionCall")
+            if isinstance(fc, dict) and fc.get("id") is None:
+                name = fc.get("name", "unknown")
+                if not isinstance(name, str):
+                    name = "unknown"
+                count = name_counters.get(name, 0)
+                fc["id"] = f"call_{name}_{count}"
+                name_counters[name] = count + 1
+
+            # 2. functionResponse（User 回复工具结果）
+            fr = part.get("functionResponse")
+            if isinstance(fr, dict) and fr.get("id") is None:
+                name = fr.get("name", "unknown")
+                if not isinstance(name, str):
+                    name = "unknown"
+                count = name_counters.get(name, 0)
+                fr["id"] = f"call_{name}_{count}"
+                name_counters[name] = count + 1
+
+
+def _process_thinking_budget(inner_request: dict[str, Any], model: str) -> None:
+    """处理 Thinking Budget：自动注入 + Auto Cap。
+
+    对齐 AM wrapper.rs：
+    - 对 flash/pro/thinking 模型处理 thinkingConfig
+    - 自动注入 thinkingConfig（对已知需要 thinking 的模型）
+    - Auto Cap：budget 超过 24576 时裁剪
+    """
+    lower_model = model.lower()
+    if not any(kw in lower_model for kw in ("flash", "pro", "thinking")):
+        return
+
+    # 确保 generationConfig 存在
+    gen_config = inner_request.setdefault("generationConfig", {})
+    if not isinstance(gen_config, dict):
+        return
+
+    # 自动注入 thinkingConfig（对已知需要 thinking 的模型）
+    if gen_config.get("thinkingConfig") is None:
+        should_inject = any(kw in lower_model for kw in THINKING_MODELS_AUTO_INJECT_KEYWORDS)
+        if should_inject:
+            gen_config["thinkingConfig"] = {
+                "includeThoughts": True,
+                "thinkingBudget": THINKING_BUDGET_DEFAULT_INJECT,
+            }
+
+    # Auto Cap
+    thinking_config = gen_config.get("thinkingConfig")
+    if not isinstance(thinking_config, dict):
+        return
+
+    budget = thinking_config.get("thinkingBudget")
+    if isinstance(budget, int) and budget > THINKING_BUDGET_AUTO_CAP:
+        thinking_config["thinkingBudget"] = THINKING_BUDGET_AUTO_CAP
+
+
+def _clean_tool_declarations(inner_request: dict[str, Any]) -> None:
+    """清洗工具声明：重命名字段 + 移除禁止的 Schema 字段 + 过滤搜索声明。
+
+    对齐 AM wrapper.rs：
+    - parametersJsonSchema → parameters（Gemini CLI 兼容）
+    - 移除 Gemini 不支持的 Schema 字段（multipleOf 等）
+    - 过滤 web_search / google_search 工具声明
+    """
+    tools = inner_request.get("tools")
+    if not isinstance(tools, list):
+        return
+
+    for tool in tools:
+        if not isinstance(tool, dict):
+            continue
+        decls = tool.get("functionDeclarations")
+        if not isinstance(decls, list):
+            continue
+
+        # 1. 过滤搜索关键字函数
+        decls[:] = [
+            d
+            for d in decls
+            if not (
+                isinstance(d, dict)
+                and isinstance(d.get("name"), str)
+                and d["name"] in ("web_search", "google_search")
+            )
+        ]
+
+        # 2. 重命名 + 清洗
+        for decl in decls:
+            if not isinstance(decl, dict):
+                continue
+
+            # parametersJsonSchema → parameters
+            if "parametersJsonSchema" in decl:
+                params = decl.pop("parametersJsonSchema")
+                if isinstance(params, dict):
+                    _clean_json_schema(params)
+                decl["parameters"] = params
+            elif "parameters" in decl:
+                params = decl["parameters"]
+                if isinstance(params, dict):
+                    _clean_json_schema(params)
+
+
+def _clean_json_schema(schema: dict[str, Any]) -> None:
+    """递归移除 Gemini 不支持的 JSON Schema 字段。"""
+    for field in FORBIDDEN_SCHEMA_FIELDS:
+        schema.pop(field, None)
+
+    # Recurse into properties
+    props = schema.get("properties")
+    if isinstance(props, dict):
+        for prop_schema in props.values():
+            if isinstance(prop_schema, dict):
+                _clean_json_schema(prop_schema)
+
+    # Recurse into items
+    items = schema.get("items")
+    if isinstance(items, dict):
+        _clean_json_schema(items)
+
+    # Recurse into additionalProperties
+    addl = schema.get("additionalProperties")
+    if isinstance(addl, dict):
+        _clean_json_schema(addl)
+
+    # Recurse into anyOf / oneOf / allOf
+    for combo_key in ("anyOf", "oneOf", "allOf"):
+        combo = schema.get(combo_key)
+        if isinstance(combo, list):
+            for sub_schema in combo:
+                if isinstance(sub_schema, dict):
+                    _clean_json_schema(sub_schema)
+
+
+def _inject_system_instruction(inner_request: dict[str, Any]) -> None:
+    """注入 Antigravity 身份系统指令。
+
+    对齐 AM wrapper.rs：
+    - 如果已有 systemInstruction：在前面插入（避免重复）
+    - 如果没有：创建新的
+    - 补全 role: user（Gemini API 要求）
+    """
+    system_instruction = inner_request.get("systemInstruction")
+
+    if isinstance(system_instruction, dict):
+        # 补全 role
+        if "role" not in system_instruction:
+            system_instruction["role"] = "user"
+
+        parts = system_instruction.get("parts")
+        if isinstance(parts, list):
+            # 检查是否已包含 Antigravity 身份（避免重复注入）
+            has_antigravity = False
+            if parts and isinstance(parts[0], dict):
+                text = parts[0].get("text", "")
+                if isinstance(text, str) and "You are Antigravity" in text:
+                    has_antigravity = True
+
+            if not has_antigravity:
+                parts.insert(0, {"text": ANTIGRAVITY_SYSTEM_INSTRUCTION})
+    else:
+        # 没有 systemInstruction，创建新的
+        inner_request["systemInstruction"] = {
+            "role": "user",
+            "parts": [{"text": ANTIGRAVITY_SYSTEM_INSTRUCTION}],
+        }
+
+
+# ---------------------------------------------------------------------------
+# Response 后处理工具函数
+# ---------------------------------------------------------------------------
+
+
+def _inject_claude_tool_ids_response(response: dict[str, Any], model: str) -> None:
+    """为 Claude 模型的响应注入 functionCall 的 id 字段。
+
+    对齐 AM wrapper.rs inject_ids_to_response：
+    让下游客户端（如 OpenCode/Vercel AI SDK）能感知 tool call ID，
+    并在下一轮对话中原样带回。
+    """
+    if "claude" not in model.lower():
+        return
+
+    candidates = response.get("candidates")
+    if not isinstance(candidates, list):
+        return
+
+    for candidate in candidates:
+        if not isinstance(candidate, dict):
+            continue
+        content = candidate.get("content")
+        if not isinstance(content, dict):
+            continue
+        parts = content.get("parts")
+        if not isinstance(parts, list):
+            continue
+
+        name_counters: dict[str, int] = {}
+        for part in parts:
+            if not isinstance(part, dict):
+                continue
+            fc = part.get("functionCall")
+            if isinstance(fc, dict) and fc.get("id") is None:
+                name = fc.get("name", "unknown")
+                if not isinstance(name, str):
+                    name = "unknown"
+                count = name_counters.get(name, 0)
+                fc["id"] = f"call_{name}_{count}"
+                name_counters[name] = count + 1
+
+
+# ---------------------------------------------------------------------------
+# Core wrap / unwrap 函数
+# ---------------------------------------------------------------------------
+
+
+def _generate_stable_session_id(inner_request: dict[str, Any]) -> str:
+    """生成稳定的 sessionId（对齐 CLIProxyAPI generateStableSessionID）。
+
+    基于请求内容的哈希生成确定性 sessionId，相同内容的请求会得到相同的 sessionId，
+    有助于上游维持会话上下文。
+    """
+    import hashlib
+    import json
+
+    try:
+        # 使用 contents 字段生成稳定哈希（与 CLIProxyAPI 对齐）
+        contents = inner_request.get("contents")
+        if contents:
+            raw = json.dumps(contents, sort_keys=True, separators=(",", ":"))
+        else:
+            raw = json.dumps(inner_request, sort_keys=True, separators=(",", ":"))
+        digest = hashlib.sha256(raw.encode()).hexdigest()[:32]
+        return f"session-{digest}"
+    except Exception:
+        return f"session-{uuid.uuid4().hex[:32]}"
+
+
+def wrap_v1internal_request(
+    gemini_request: dict[str, Any],
+    *,
+    project_id: str,
+    model: str,
+    request_type: str = "agent",
+) -> dict[str, Any]:
+    """Wrap a GeminiRequest into Antigravity V1InternalRequest.
+
+    处理流程（对齐 AM wrapper.rs）：
+    1. 移除 model（移到顶层）
+    2. 移除 safetySettings（v1internal 不支持）
+    3. 深度清理 [undefined] 字符串
+    4. Claude model tool ID 注入
+    5. Thinking budget 处理（自动注入 + Auto Cap）
+    6. 工具声明清洗（schema 清理 + 字段重命名）
+    7. System Instruction 注入
+    8. 注入 sessionId（对齐 CLIProxyAPI）
+    9. 构建 v1internal 信封
+    """
+    inner_request = dict(gemini_request)
+    inner_request.pop("model", None)
+    inner_request.pop("safetySettings", None)
+
+    # 1. 深度清理 [undefined]
+    _deep_clean_undefined(inner_request)
+
+    # 2. Claude tool ID 注入
+    _inject_claude_tool_ids_request(inner_request, model)
+
+    # 3. Thinking budget 处理
+    _process_thinking_budget(inner_request, model)
+
+    # 4. 工具声明清洗
+    _clean_tool_declarations(inner_request)
+
+    # 5. System Instruction 注入
+    _inject_system_instruction(inner_request)
+
+    # 6. 注入 sessionId（对齐 CLIProxyAPI/sub2api）
+    if "sessionId" not in inner_request:
+        inner_request["sessionId"] = _generate_stable_session_id(inner_request)
+
+    return {
+        "project": project_id,
+        "requestId": f"agent-{uuid.uuid4()}",
+        "request": inner_request,
+        "model": model,
+        "userAgent": ANTIGRAVITY_REQUEST_USER_AGENT,
+        "requestType": request_type,
+    }
+
+
+def unwrap_v1internal_response(response: dict[str, Any]) -> dict[str, Any]:
+    """Unwrap Antigravity V1InternalResponse into a GeminiResponse-like dict."""
+    inner = response.get("response")
+    if isinstance(inner, dict):
+        unwrapped = dict(inner)
+        resp_id = response.get("responseId")
+        if resp_id is not None:
+            unwrapped["_v1internal_response_id"] = resp_id
+        return unwrapped
+    return response
+
+
+def cache_thought_signatures(model: str, response: dict[str, Any]) -> None:
+    """Best-effort cache for Antigravity thought signatures.
+
+    同时缓存到 legacy (text) 层和 tool (Layer 1) 层。
+    """
+    try:
+        from src.services.provider.adapters.antigravity.signature_cache import signature_cache
+    except Exception:
+        return
+
+    try:
+        candidates = response.get("candidates")
+        if not isinstance(candidates, list):
+            return
+
+        for cand in candidates:
+            if not isinstance(cand, dict):
+                continue
+            content = cand.get("content")
+            if not isinstance(content, dict):
+                continue
+            parts = content.get("parts")
+            if not isinstance(parts, list):
+                continue
+
+            for part in parts:
+                if not isinstance(part, dict):
+                    continue
+
+                # 缓存 thinking signature（legacy text 层）
+                text = part.get("text")
+                sig = (
+                    part.get("thoughtSignature")
+                    or part.get("thought_signature")
+                    or part.get("signature")
+                )
+                if isinstance(text, str) and text and isinstance(sig, str) and sig:
+                    signature_cache.cache(model, text, sig)
+
+                # 缓存 tool call signature（Layer 1）
+                fc = part.get("functionCall")
+                if isinstance(fc, dict) and isinstance(sig, str) and sig:
+                    tool_id = fc.get("id")
+                    if isinstance(tool_id, str) and tool_id:
+                        signature_cache.cache_tool_signature(tool_id, sig)
+    except Exception:
+        # Never fail request path due to cache issues.
+        return
+
+
+def is_signature_error(status_code: int, error_body: str) -> bool:
+    """检测 400 错误是否为 thinking signature 相关错误。
+
+    对齐 AM handlers/common.rs：用于判断是否应该移除 thinking 配置后重试。
+    """
+    if status_code != 400:
+        return False
+    return any(kw in error_body for kw in SIGNATURE_ERROR_KEYWORDS)
+
+
+# ---------------------------------------------------------------------------
+# Envelope 类（Provider Hook 接口）
+# ---------------------------------------------------------------------------
+
+
+class AntigravityV1InternalEnvelope:
+    """Provider envelope hooks for Antigravity v1internal wrapper."""
+
+    name = "antigravity:v1internal"
+
+    def extra_headers(self) -> dict[str, str] | None:
+        return {"User-Agent": get_http_user_agent()}
+
+    def wrap_request(
+        self,
+        request_body: dict[str, Any],
+        *,
+        model: str,
+        url_model: str | None,
+        decrypted_auth_config: dict[str, Any] | None,
+    ) -> tuple[dict[str, Any], str | None]:
+        from src.core.logger import logger as _envelope_logger
+
+        project_id = (decrypted_auth_config or {}).get("project_id")
+        if not isinstance(project_id, str) or not project_id:
+            from src.core.exceptions import ProviderNotAvailableException
+
+            raise ProviderNotAvailableException(
+                "Antigravity OAuth 配置缺少 project_id，请重新授权",
+                provider_name="antigravity",
+                upstream_response="missing auth_config.project_id",
+            )
+
+        wrapped = wrap_v1internal_request(
+            request_body,
+            project_id=project_id,
+            model=model,
+        )
+
+        # Debug: 打印 v1internal 请求的关键字段（不打印完整 body 避免日志过大）
+        _envelope_logger.debug(
+            "[Antigravity Envelope] model={}, project_id={}, requestType={}, "
+            "userAgent={}, has_sessionId={}, has_systemInstruction={}, "
+            "has_contents={}, has_generationConfig={}",
+            wrapped.get("model"),
+            str(wrapped.get("project", ""))[:8] + "...",
+            wrapped.get("requestType"),
+            wrapped.get("userAgent"),
+            "sessionId" in (wrapped.get("request") or {}),
+            "systemInstruction" in (wrapped.get("request") or {}),
+            "contents" in (wrapped.get("request") or {}),
+            "generationConfig" in (wrapped.get("request") or {}),
+        )
+
+        # Antigravity's model lives in the request body, not the URL path.
+        return wrapped, None
+
+    def unwrap_response(self, data: Any) -> Any:
+        if isinstance(data, dict):
+            return unwrap_v1internal_response(data)
+        return data
+
+    def postprocess_unwrapped_response(self, *, model: str, data: Any) -> None:
+        if isinstance(data, dict):
+            # Claude model: 注入 tool call ID（对齐 AM wrapper.rs inject_ids_to_response）
+            _inject_claude_tool_ids_response(data, model)
+            cache_thought_signatures(model, data)
+
+    def capture_selected_base_url(self) -> str | None:
+        return get_selected_base_url()
+
+    def on_http_status(self, *, base_url: str | None, status_code: int) -> None:
+        if not base_url:
+            return
+        if status_code == 200:
+            url_availability.mark_success(base_url)
+        elif status_code in (429, 500, 502, 503, 504):
+            url_availability.mark_unavailable(base_url)
+
+    def on_connection_error(self, *, base_url: str | None, exc: Exception) -> None:  # noqa: ARG002
+        if not base_url:
+            return
+        url_availability.mark_unavailable(base_url)
+
+    def force_stream_rewrite(self) -> bool:
+        # Streaming must be rewritten even when endpoint signature matches, because
+        # Antigravity wraps chunks in v1internal envelope.
+        return True
+
+
+antigravity_v1internal_envelope = AntigravityV1InternalEnvelope()
+
+
+__all__ = [
+    "AntigravityV1InternalEnvelope",
+    "antigravity_v1internal_envelope",
+    "cache_thought_signatures",
+    "is_signature_error",
+    "unwrap_v1internal_response",
+    "wrap_v1internal_request",
+]

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

@@ -0,0 +1,330 @@
+"""Antigravity provider plugin — 统一注册入口。
+
+将 Antigravity 对各通用 registry 的注册集中在一个文件中：
+- Envelope (v1internal 信封)
+- Transport Hook (URL 构建)
+- Auth Enricher (OAuth enrichment)
+- Model Fetcher (模型获取)
+- Behavior Variants (格式变体)
+
+新增 provider 时参照此文件创建对应的 plugin.py 即可。
+"""
+
+from __future__ import annotations
+
+import time
+from typing import Any
+from urllib.parse import urlencode
+
+from src.core.logger import logger
+from src.services.provider.adapters.antigravity.constants import V1INTERNAL_PATH_TEMPLATE
+from src.services.provider.adapters.antigravity.url_availability import url_availability
+from src.services.provider.request_context import set_selected_base_url
+
+# ---------------------------------------------------------------------------
+# Transport Hook
+# ---------------------------------------------------------------------------
+
+
+def build_antigravity_url(
+    endpoint: Any,
+    *,
+    is_stream: bool,
+    effective_query_params: dict[str, Any],
+) -> str:
+    """构建 Antigravity v1internal URL。
+
+    使用 url_availability 选择最优端点，构建 v1internal:generateContent 或
+    v1internal:streamGenerateContent URL。
+    """
+    ordered_urls = url_availability.get_ordered_urls(prefer_daily=True)
+    base_url = ordered_urls[0] if ordered_urls else endpoint.base_url
+
+    # 存入 contextvars（供后续 Handler 层 envelope 获取）
+    set_selected_base_url(str(base_url) if base_url is not None else None)
+
+    action = "streamGenerateContent" if is_stream else "generateContent"
+    path = V1INTERNAL_PATH_TEMPLATE.format(action=action)
+
+    # v1internal 流式请求同样支持 ?alt=sse
+    if is_stream:
+        effective_query_params.setdefault("alt", "sse")
+
+    url = f"{str(base_url).rstrip('/')}{path}"
+    if effective_query_params:
+        query_string = urlencode(effective_query_params, doseq=True)
+        if query_string:
+            url = f"{url}?{query_string}"
+
+    return url
+
+
+# ---------------------------------------------------------------------------
+# Auth Enricher
+# ---------------------------------------------------------------------------
+
+
+async def enrich_antigravity(
+    auth_config: dict[str, Any],
+    token_response: dict[str, Any],
+    access_token: str,
+    proxy_config: dict[str, Any] | None,
+) -> dict[str, Any]:
+    """Antigravity auth_config enrichment.
+
+    1. 通过 Google userinfo API 获取 email
+    2. 通过 loadCodeAssist 获取 project_id / tier
+    3. 未激活账号尝试 onboardUser
+    4. fallback 到随机 project_id
+    """
+    from src.core.provider_oauth_utils import fetch_google_email
+    from src.services.provider.adapters.antigravity.client import (
+        extract_project_id,
+        extract_tier_id,
+        generate_fallback_project_id,
+        load_code_assist,
+        onboard_user,
+    )
+
+    # Email（仅在缺失时获取）
+    if not auth_config.get("email"):
+        email = await fetch_google_email(
+            access_token,
+            proxy_config=proxy_config,
+            timeout_seconds=10.0,
+        )
+        if email:
+            auth_config["email"] = email
+
+    # Project ID + Tier（需要 loadCodeAssist 时一起获取）
+    need_project = not auth_config.get("project_id")
+    # Tier 每次 enrich 都重新获取（确保归一化为 Free/Pro/Ultra）
+    need_tier = True
+
+    if need_project or need_tier:
+        project_id = auth_config.get("project_id", "")
+        try:
+            code_assist = await load_code_assist(access_token, proxy_config=proxy_config)
+
+            # 提取 tier 信息（对齐 sub2api：优先 paidTier，fallback currentTier）
+            tier_str = _extract_tier_from_code_assist(code_assist)
+            if tier_str:
+                auth_config["tier"] = tier_str
+                logger.info("[enrich] Antigravity tier: {}", tier_str)
+
+            if need_project:
+                project_id = extract_project_id(code_assist)
+
+                # 未激活：尝试 onboardUser
+                if not project_id and code_assist.get("allowedTiers"):
+                    tier_id = extract_tier_id(code_assist)
+                    logger.info("[enrich] Antigravity onboardUser tier={}", tier_id)
+                    project_id = await onboard_user(
+                        access_token,
+                        tier_id=tier_id,
+                        proxy_config=proxy_config,
+                    )
+        except Exception as e:
+            logger.warning("[enrich] Antigravity loadCodeAssist/onboardUser 失败: {}", e)
+
+        if need_project:
+            if project_id:
+                auth_config["project_id"] = project_id
+                logger.info("[enrich] Antigravity project_id: {}", project_id[:8] + "...")
+            else:
+                fallback = generate_fallback_project_id()
+                auth_config["project_id"] = fallback
+                logger.info("[enrich] Antigravity 随机 project_id fallback: {}", fallback)
+
+    return auth_config
+
+
+def _extract_tier_from_code_assist(code_assist: dict[str, Any]) -> str:
+    """从 loadCodeAssist 响应中提取用户层级，返回 Free/Pro/Ultra。
+
+    对齐 sub2api/CLIProxyAPI：优先 paidTier，fallback currentTier。
+    兼容 tier 为字符串或 {"id": "...", "tierType": "..."} 两种格式。
+    """
+    # 优先 paidTier（付费订阅级别）
+    paid_tier = code_assist.get("paidTier")
+    tier = _normalize_tier(_extract_tier_raw(paid_tier))
+    if tier:
+        return tier
+
+    # fallback currentTier
+    current_tier = code_assist.get("currentTier")
+    tier = _normalize_tier(_extract_tier_raw(current_tier))
+    if tier:
+        return tier
+
+    return "Free"
+
+
+def _extract_tier_raw(tier_obj: Any) -> str:
+    """从 tier 对象中提取原始标识，兼容字符串和字典两种格式。"""
+    if isinstance(tier_obj, str) and tier_obj.strip():
+        return tier_obj.strip()
+    if isinstance(tier_obj, dict):
+        for key in ("id", "tierType"):
+            val = tier_obj.get(key)
+            if isinstance(val, str) and val.strip():
+                return val.strip()
+    return ""
+
+
+def _normalize_tier(raw: str) -> str:
+    """将上游 tier 标识归一化为 Free/Pro/Ultra。
+
+    上游格式示例：
+    - id: "free-tier", "g1-pro-tier", "g1-ultra-tier"
+    - tierType: "FREE", "PAID"
+    """
+    if not raw:
+        return ""
+    lower = raw.lower()
+    if "ultra" in lower:
+        return "Ultra"
+    if "pro" in lower or "paid" in lower:
+        return "Pro"
+    if "free" in lower or "legacy" in lower:
+        return "Free"
+    return raw
+
+
+# ---------------------------------------------------------------------------
+# Model Fetcher
+# ---------------------------------------------------------------------------
+
+# Antigravity 内部测试/调试模型，不应暴露给用户
+BLOCKED_MODELS: frozenset[str] = frozenset({"chat_23310", "chat_20706"})
+
+
+async def fetch_models_antigravity(
+    ctx: Any,
+    timeout_seconds: float,
+) -> tuple[list[dict], list[str], bool, dict[str, Any] | None]:
+    """Antigravity 模型获取策略。
+
+    调用 v1internal:fetchAvailableModels 获取可用模型列表，
+    解析配额信息，过滤黑名单模型。
+    """
+    from src.services.provider.adapters.antigravity.client import fetch_available_models
+
+    auth_config = ctx.auth_config or {}
+    project_id = auth_config.get("project_id")
+    if not isinstance(project_id, str) or not project_id.strip():
+        return [], ["antigravity: missing auth_config.project_id (please re-auth)"], False, None
+
+    try:
+        data = await fetch_available_models(
+            ctx.api_key_value,
+            project_id=project_id.strip(),
+            proxy_config=ctx.proxy_config,
+            timeout_seconds=timeout_seconds,
+        )
+    except Exception as e:
+        return [], [f"antigravity: fetchAvailableModels error: {e}"], False, None
+
+    raw_models = data.get("models")
+    if not isinstance(raw_models, dict):
+        return [], ["antigravity: invalid response (missing models)"], False, None
+
+    models: list[dict] = []
+    quota_by_model: dict[str, dict[str, Any]] = {}
+
+    for model_id, model_data in raw_models.items():
+        if not isinstance(model_id, str) or not model_id.strip():
+            continue
+        if model_id.strip() in BLOCKED_MODELS:
+            continue
+        if not isinstance(model_data, dict):
+            model_data = {}
+
+        display_name = model_data.get("displayName")
+        if not isinstance(display_name, str) or not display_name:
+            display_name = model_id
+
+        models.append(
+            {
+                "id": model_id,
+                "owned_by": "antigravity",
+                "display_name": display_name,
+                "api_format": "gemini:cli",
+            }
+        )
+
+        quota_info = model_data.get("quotaInfo")
+        if not isinstance(quota_info, dict):
+            continue
+
+        remaining = quota_info.get("remainingFraction")
+        reset_time = quota_info.get("resetTime")
+
+        remaining_fraction: float | None = None
+        try:
+            if remaining is not None:
+                remaining_fraction = float(remaining)
+        except Exception:
+            remaining_fraction = None
+
+        if remaining_fraction is None:
+            continue
+
+        used_percent = (1.0 - remaining_fraction) * 100.0
+        if used_percent < 0:
+            used_percent = 0.0
+        if used_percent > 100:  # noqa: PLR2004
+            used_percent = 100.0
+
+        payload: dict[str, Any] = {
+            "remaining_fraction": remaining_fraction,
+            "used_percent": used_percent,
+        }
+        if isinstance(reset_time, str) and reset_time.strip():
+            payload["reset_time"] = reset_time.strip()
+        quota_by_model[model_id] = payload
+
+    upstream_metadata: dict[str, Any] | None = None
+    if quota_by_model:
+        upstream_metadata = {
+            "antigravity": {
+                "updated_at": int(time.time()),
+                "quota_by_model": quota_by_model,
+            }
+        }
+
+    return models, [], True, upstream_metadata
+
+
+# ---------------------------------------------------------------------------
+# Unified Registration
+# ---------------------------------------------------------------------------
+
+
+def register_all() -> None:
+    """一次性注册 Antigravity 的所有 hooks 到各通用 registry。"""
+    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.transport import register_transport_hook
+
+    # Envelope
+    register_envelope("antigravity", "gemini:cli", antigravity_v1internal_envelope)
+    register_envelope("antigravity", "", antigravity_v1internal_envelope)
+
+    # Transport
+    register_transport_hook("antigravity", "gemini:cli", build_antigravity_url)
+
+    # Auth
+    register_auth_enricher("antigravity", enrich_antigravity)
+
+    # Model Fetcher
+    UpstreamModelsFetcherRegistry.register(
+        provider_types=["antigravity"],
+        fetcher=fetch_models_antigravity,
+    )
+
+    # Behavior
+    register_behavior_variant("antigravity", cross_format=True)

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

@@ -0,0 +1,231 @@
+"""Antigravity thinking block signature cache (triple-layer).
+
+与 Antigravity-Manager 对齐的三层缓存设计：
+  Layer 1: tool_use_id → thoughtSignature  （工具调用签名恢复）
+  Layer 2: signature  → model_family       （跨模型兼容校验）
+  Layer 3: session_id → latest signature   （会话级签名追踪 + rewind 检测）
+
+同时保留原有的 model:text → signature 兼容层。
+"""
+
+from __future__ import annotations
+
+import hashlib
+import threading
+import time
+from typing import Any
+
+from src.services.provider.adapters.antigravity.constants import (
+    DUMMY_THOUGHT_SIGNATURE,
+    MIN_SIGNATURE_LENGTH,
+)
+
+# TTL: 2 小时（与 Antigravity-Manager 对齐）
+_SIGNATURE_TTL_SECONDS = 2 * 60 * 60
+
+# 各层缓存上限
+_TOOL_CACHE_LIMIT = 500
+_FAMILY_CACHE_LIMIT = 200
+_SESSION_CACHE_LIMIT = 1000
+_TEXT_CACHE_LIMIT = 1000
+
+
+class _CacheEntry:
+    """带时间戳的缓存条目，支持 TTL 过期。"""
+
+    __slots__ = ("data", "created_at")
+
+    def __init__(self, data: Any) -> None:
+        self.data = data
+        self.created_at: float = time.monotonic()
+
+    def is_expired(self, now: float | None = None) -> bool:
+        return ((now or time.monotonic()) - self.created_at) > _SIGNATURE_TTL_SECONDS
+
+
+class _SessionEntry:
+    """Session 层缓存数据，包含消息计数用于 rewind 检测。"""
+
+    __slots__ = ("signature", "message_count")
+
+    def __init__(self, signature: str, message_count: int) -> None:
+        self.signature = signature
+        self.message_count = message_count
+
+
+class ThinkingSignatureCache:
+    """Triple-layer thinking signature cache.
+
+    Layer 1 (tool): tool_use_id → thoughtSignature
+        当客户端(如 OpenCode) 在 tool_result 中丢弃了 signature 时用于恢复。
+
+    Layer 2 (family): signature → model_family
+        防止跨模型签名污染（Claude 签名不能用在 Gemini 上）。
+
+    Layer 3 (session): session_id → latest signature + message_count
+        会话级追踪，支持 rewind 检测（用户删除消息后不会注入来自"未来"的签名）。
+
+    Legacy (text): SHA256(model + text) → signature
+        向后兼容的 get_or_dummy() 接口。
+    """
+
+    def __init__(self) -> None:
+        self._tool_sigs: dict[str, _CacheEntry] = {}
+        self._families: dict[str, _CacheEntry] = {}
+        self._sessions: dict[str, _CacheEntry] = {}
+        self._text_sigs: dict[str, _CacheEntry] = {}
+        self._lock = threading.Lock()
+
+    # ===== Layer 1: Tool Use ID → Signature =====
+
+    def cache_tool_signature(self, tool_use_id: str, signature: str) -> None:
+        """缓存工具调用对应的 thinking signature。"""
+        if len(signature) < MIN_SIGNATURE_LENGTH:
+            return
+        with self._lock:
+            self._tool_sigs[tool_use_id] = _CacheEntry(signature)
+            if len(self._tool_sigs) > _TOOL_CACHE_LIMIT:
+                self._prune(self._tool_sigs)
+
+    def get_tool_signature(self, tool_use_id: str) -> str | None:
+        """查找工具调用对应的 signature。"""
+        with self._lock:
+            entry = self._tool_sigs.get(tool_use_id)
+            if entry is None:
+                return None
+            if entry.is_expired():
+                self._tool_sigs.pop(tool_use_id, None)
+                return None
+            return entry.data
+
+    # ===== Layer 2: Signature → Model Family =====
+
+    def cache_thinking_family(self, signature: str, family: str) -> None:
+        """记录 signature 所属的模型家族。"""
+        if len(signature) < MIN_SIGNATURE_LENGTH:
+            return
+        with self._lock:
+            self._families[signature] = _CacheEntry(family)
+            if len(self._families) > _FAMILY_CACHE_LIMIT:
+                self._prune(self._families)
+
+    def get_signature_family(self, signature: str) -> str | None:
+        """查找 signature 所属的模型家族。"""
+        with self._lock:
+            entry = self._families.get(signature)
+            if entry is None:
+                return None
+            if entry.is_expired():
+                self._families.pop(signature, None)
+                return None
+            return entry.data
+
+    # ===== Layer 3: Session ID → Latest Signature =====
+
+    def cache_session_signature(
+        self, session_id: str, signature: str, message_count: int = 0
+    ) -> None:
+        """存储会话的最新 thinking signature。
+
+        Rewind 检测：当 message_count 小于已缓存值时，说明用户删除了消息，
+        强制更新签名以避免注入来自"未来"的签名。
+        """
+        if len(signature) < MIN_SIGNATURE_LENGTH:
+            return
+
+        with self._lock:
+            existing = self._sessions.get(session_id)
+            should_store = True
+
+            if existing and not existing.is_expired():
+                entry: _SessionEntry = existing.data
+                if message_count < entry.message_count:
+                    # Rewind detected: 用户删除了消息，强制更新
+                    pass
+                elif message_count == entry.message_count:
+                    # 同一轮消息：仅当新签名更长（更完整）时才替换
+                    should_store = len(signature) > len(entry.signature)
+                # else: 正常递增，更新
+
+            if should_store:
+                self._sessions[session_id] = _CacheEntry(_SessionEntry(signature, message_count))
+                if len(self._sessions) > _SESSION_CACHE_LIMIT:
+                    self._prune(self._sessions)
+
+    def get_session_signature(self, session_id: str) -> str | None:
+        """获取会话的最新 thinking signature。"""
+        with self._lock:
+            entry = self._sessions.get(session_id)
+            if entry is None:
+                return None
+            if entry.is_expired():
+                self._sessions.pop(session_id, None)
+                return None
+            return entry.data.signature
+
+    # ===== Legacy: model:text → signature（向后兼容） =====
+
+    def get_or_dummy(self, model: str, thinking_text: str) -> str | None:
+        """Legacy: 根据 model + thinking_text 查找 signature。
+
+        Gemini 模型在未命中时返回 DUMMY_THOUGHT_SIGNATURE（跳过验证）。
+        """
+        key = self._text_key(model, thinking_text)
+        with self._lock:
+            entry = self._text_sigs.get(key)
+            if entry is not None:
+                if entry.is_expired():
+                    self._text_sigs.pop(key, None)
+                else:
+                    return entry.data
+        if str(model).startswith("gemini-"):
+            return DUMMY_THOUGHT_SIGNATURE
+        return None
+
+    def cache(self, model: str, thinking_text: str, signature: str) -> None:
+        """Legacy: 缓存 model + thinking_text → signature。"""
+        if len(signature) < MIN_SIGNATURE_LENGTH:
+            return
+
+        key = self._text_key(model, thinking_text)
+        with self._lock:
+            if key in self._text_sigs:
+                self._text_sigs[key] = _CacheEntry(signature)
+                return
+
+            if len(self._text_sigs) >= _TEXT_CACHE_LIMIT:
+                # FIFO 淘汰 1/4
+                evict_n = max(1, _TEXT_CACHE_LIMIT // 4)
+                for k in list(self._text_sigs.keys())[:evict_n]:
+                    self._text_sigs.pop(k, None)
+
+            self._text_sigs[key] = _CacheEntry(signature)
+
+    # ===== Utilities =====
+
+    @staticmethod
+    def _text_key(model: str, thinking_text: str) -> str:
+        # 使用 \x00 作为分隔符避免 model 中含 ':' 时的歧义
+        content = f"{model}\x00{thinking_text}"
+        return hashlib.sha256(content.encode("utf-8")).hexdigest()[:32]
+
+    @staticmethod
+    def _prune(d: dict[str, _CacheEntry]) -> None:
+        """清理已过期的缓存条目。"""
+        now = time.monotonic()
+        expired = [k for k, v in d.items() if v.is_expired(now)]
+        for k in expired:
+            d.pop(k, None)
+
+    def clear(self) -> None:
+        """清空所有缓存层（用于测试或手动重置）。"""
+        with self._lock:
+            self._tool_sigs.clear()
+            self._families.clear()
+            self._sessions.clear()
+            self._text_sigs.clear()
+
+
+signature_cache = ThinkingSignatureCache()
+
+__all__ = ["ThinkingSignatureCache", "signature_cache"]

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

@@ -5,9 +5,10 @@ from __future__ import annotations
 import threading
 import time
 
-from src.services.antigravity.constants import (
+from src.services.provider.adapters.antigravity.constants import (
     DAILY_BASE_URL,
     PROD_BASE_URL,
+    SANDBOX_BASE_URL,
     URL_UNAVAILABLE_TTL_SECONDS,
 )
 
@@ -50,8 +51,11 @@ class URLAvailability:
         with self._mu:
             self._prune()
 
+            # Antigravity-Manager 顺序：sandbox → daily → prod
             base_order = (
-                [DAILY_BASE_URL, PROD_BASE_URL] if prefer_daily else [PROD_BASE_URL, DAILY_BASE_URL]
+                [SANDBOX_BASE_URL, DAILY_BASE_URL, PROD_BASE_URL]
+                if prefer_daily
+                else [PROD_BASE_URL, SANDBOX_BASE_URL, DAILY_BASE_URL]
             )
 
             if self._last_success and self._last_success in base_order:

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

@@ -75,4 +75,5 @@ class CodexOAuthEnvelope:
 
 codex_oauth_envelope = CodexOAuthEnvelope()
 
+
 __all__ = ["CodexOAuthEnvelope", "codex_oauth_envelope"]

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

@@ -10,7 +10,9 @@ Codex Provider 元数据采集器
 
 from typing import Any, ClassVar
 
-from src.services.provider.metadata_collectors import MetadataCollector
+from src.services.provider.metadata_collectors import (  # noqa: E501 — core registry
+    MetadataCollector,
+)
 
 
 def _safe_float(value: str | None) -> float | None:

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

@@ -0,0 +1,106 @@
+"""Codex provider plugin — 统一注册入口。
+
+将 Codex 对各通用 registry 的注册集中在一个文件中：
+- Envelope (OAuth headers)
+- Transport Hook (URL 构建)
+- Auth Enricher (OAuth enrichment)
+- Behavior Variants (格式变体)
+
+新增 provider 时参照此文件创建对应的 plugin.py 即可。
+"""
+
+from __future__ import annotations
+
+from typing import Any
+from urllib.parse import urlencode
+
+from src.core.logger import logger
+
+# ---------------------------------------------------------------------------
+# Transport Hook
+# ---------------------------------------------------------------------------
+
+
+def build_codex_url(
+    endpoint: Any,
+    *,
+    is_stream: bool,
+    effective_query_params: dict[str, Any],
+) -> str:
+    """构建 Codex OAuth URL。
+
+    Codex upstream (chatgpt.com/backend-api/codex) 使用 /responses
+    而非标准 OpenAI 的 /v1/responses。
+    """
+    _ = is_stream  # Codex 不需要根据 stream 切换路径
+
+    base = str(endpoint.base_url).rstrip("/")
+    path = "/responses"
+    # 如果用户已在 base_url 中包含了最终路径，不要重复
+    url = base if base.endswith(path) else f"{base}{path}"
+    if effective_query_params:
+        query_string = urlencode(effective_query_params, doseq=True)
+        if query_string:
+            url = f"{url}?{query_string}"
+    return url
+
+
+# ---------------------------------------------------------------------------
+# Auth Enricher
+# ---------------------------------------------------------------------------
+
+
+async def enrich_codex(
+    auth_config: dict[str, Any],
+    token_response: dict[str, Any],
+    access_token: str,
+    proxy_config: dict[str, Any] | None,
+) -> dict[str, Any]:
+    """Codex auth_config enrichment: parse id_token -> email/account_id/plan_type/user_id."""
+    from src.core.provider_oauth_utils import parse_codex_id_token
+
+    id_token = token_response.get("id_token")
+    logger.debug(
+        "Codex enrich_auth_config: id_token_present={} token_keys={}",
+        bool(id_token),
+        list(token_response.keys()),
+    )
+    codex_info = parse_codex_id_token(str(id_token) if id_token else None)
+    if codex_info:
+        logger.debug("Codex parsed id_token fields: {}", list(codex_info.keys()))
+    if codex_info.get("email"):
+        auth_config["email"] = codex_info["email"]
+    if codex_info.get("account_id"):
+        auth_config["account_id"] = codex_info["account_id"]
+    if codex_info.get("plan_type"):
+        auth_config["plan_type"] = codex_info["plan_type"]
+    if codex_info.get("user_id"):
+        auth_config["user_id"] = codex_info["user_id"]
+    return auth_config
+
+
+# ---------------------------------------------------------------------------
+# Unified Registration
+# ---------------------------------------------------------------------------
+
+
+def register_all() -> None:
+    """一次性注册 Codex 的所有 hooks 到各通用 registry。"""
+    from src.core.provider_oauth_utils import register_auth_enricher
+    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
+
+    # Envelope
+    register_envelope("codex", "openai:cli", codex_oauth_envelope)
+    register_envelope("codex", "", codex_oauth_envelope)
+
+    # Transport
+    register_transport_hook("codex", "openai:cli", build_codex_url)
+
+    # Auth
+    register_auth_enricher("codex", enrich_codex)
+
+    # Behavior
+    register_behavior_variant("codex", same_format=True, cross_format=True)

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

@@ -1,20 +1,28 @@
 """
-Codex provider request patching helpers.
+Codex provider request patching helpers (standalone / passthrough path).
 
-Codex (OpenAI-compatible) gateways may reject or behave unexpectedly with some parameters in
-OpenAI CLI / Responses-style requests. These helpers apply a minimal, safe transformation:
+In the main request pipeline, Codex-specific transformations are handled by the
+``openai:cli`` normalizer with ``target_variant="codex"`` (triggered automatically
+via ``register_behavior_variant("codex", same_format=True)``).
 
-- Force `store=false` (avoid persistence features not supported by some gateways).
-- Ensure `instructions` exists (Codex expects it in some deployments).
-- Convert `role=system` messages to `role=developer` (Codex may not accept `system`).
+This module provides an **equivalent** standalone patcher for contexts where the full
+normalizer pipeline is not used (e.g. external tooling, one-off scripts, or future
+passthrough-only paths). It is intentionally kept in sync with the normalizer logic.
+
+Transformations applied:
+- Force ``store=false`` (avoid persistence features not supported by some gateways).
+- Ensure ``instructions`` exists (Codex expects it in some deployments).
+- Convert ``role=system`` messages to ``role=developer`` (Codex may not accept ``system``).
 - Drop request parameters known to be rejected by Codex gateways.
-- Ensure `include` contains "reasoning.encrypted_content" for parity with CLI behavior.
+- Ensure ``include`` contains ``"reasoning.encrypted_content"`` for parity with CLI behavior.
 """
 
 from __future__ import annotations
 
 from typing import Any
 
+from src.core.provider_types import ProviderType
+
 _REJECTED_PARAMS: frozenset[str] = frozenset(
     {
         "max_output_tokens",
@@ -96,7 +104,7 @@ def maybe_patch_request_for_codex(
     - Non OpenAI CLI / Responses-style endpoints
     - Non-dict request bodies
     """
-    if (provider_type or "").lower() != "codex":
+    if (provider_type or "").lower() != ProviderType.CODEX:
         return request_body
     if (provider_api_format or "").lower() != "openai:cli":
         return request_body

src/services/provider/behavior.py

@@ -12,8 +12,31 @@ from __future__ import annotations
 
 from dataclasses import dataclass
 
+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:
@@ -28,14 +51,11 @@ def get_provider_behavior(
     provider_type: str | None,
     endpoint_sig: str | None,
 ) -> ProviderBehavior:
-    pt = str(provider_type or "").strip().lower()
+    pt = normalize_provider_type(provider_type)
     envelope = get_provider_envelope(provider_type=pt, endpoint_sig=endpoint_sig)
 
-    # same-format variant: apply on top of passthrough (e.g. OpenAI Responses -> Codex quirks)
-    same_format_variant = pt if pt in {"codex"} else None
-
-    # cross-format variant: apply during format conversion (e.g. Claude thinking -> Gemini thought parts)
-    cross_format_variant = pt if pt in {"codex", "antigravity"} else None
+    same_format_variant = pt if pt in _same_format_variants else None
+    cross_format_variant = pt if pt in _cross_format_variants else None
 
     return ProviderBehavior(
         provider_type=pt,
@@ -45,4 +65,7 @@ def get_provider_behavior(
     )
 
 
-__all__ = ["ProviderBehavior", "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/envelope.py

@@ -10,6 +10,7 @@ provider-specific envelopes live in their own service modules.
 
 from __future__ import annotations
 
+import threading
 from typing import Any, Protocol
 
 
@@ -50,32 +51,83 @@ class ProviderEnvelope(Protocol):
         """Whether streaming should always go through the rewrite/conversion path."""
 
 
+# ---------------------------------------------------------------------------
+# Envelope Registry
+# ---------------------------------------------------------------------------
+# key: (provider_type, endpoint_sig) — endpoint_sig="" 表示通配
+_envelope_registry: dict[tuple[str, str], ProviderEnvelope] = {}
+
+
+def register_envelope(
+    provider_type: str,
+    endpoint_sig: str,
+    envelope: ProviderEnvelope,
+) -> None:
+    """注册 provider 特有的 envelope。
+
+    Args:
+        provider_type: 如 "antigravity"
+        endpoint_sig: 如 "gemini:cli"，传 "" 表示该 provider 的所有 endpoint
+        envelope: 实现了 ProviderEnvelope 协议的实例
+    """
+    from src.core.provider_types import normalize_provider_type
+
+    pt = normalize_provider_type(provider_type)
+    sig = str(endpoint_sig or "").strip().lower()
+    _envelope_registry[(pt, sig)] = envelope
+
+
 def get_provider_envelope(
     *,
     provider_type: str | None,
     endpoint_sig: str | None,
 ) -> ProviderEnvelope | None:
     """Return envelope hooks for the given provider_type + endpoint signature."""
+    ensure_providers_bootstrapped()
 
-    pt = str(provider_type or "").strip().lower()
+    from src.core.provider_types import normalize_provider_type
+
+    pt = normalize_provider_type(provider_type)
     sig = str(endpoint_sig or "").strip().lower()
 
     if not pt:
         return None
 
-    # Antigravity wraps Gemini CLI responses in a v1internal envelope.
-    if pt == "antigravity" and (sig == "gemini:cli" or not sig):
-        from src.services.antigravity.envelope import antigravity_v1internal_envelope
-
-        return antigravity_v1internal_envelope
-
-    # Codex OAuth upstream requires a few fixed headers (SSE, session id, etc.).
-    if pt == "codex" and (sig == "openai:cli" or not sig):
-        from src.services.codex.envelope import codex_oauth_envelope
-
-        return codex_oauth_envelope
-
-    return None
+    # 精确匹配优先，再尝试通配
+    return _envelope_registry.get((pt, sig)) or _envelope_registry.get((pt, ""))
 
 
-__all__ = ["ProviderEnvelope", "get_provider_envelope"]
+# ---------------------------------------------------------------------------
+# Provider Bootstrap（惰性 + 幂等）
+# ---------------------------------------------------------------------------
+# 所有 registry 共享同一个 bootstrap，首次访问任何 registry 时自动触发。
+# 不再依赖模块 import 顺序。
+_bootstrapped = False
+_bootstrap_lock = threading.Lock()
+
+
+def ensure_providers_bootstrapped() -> None:
+    """确保所有 provider plugin 已注册（幂等，只执行一次）。"""
+    global _bootstrapped  # noqa: PLW0603
+    if _bootstrapped:
+        return
+    with _bootstrap_lock:
+        if _bootstrapped:
+            return
+        _bootstrapped = True
+
+        from src.services.provider.adapters.antigravity.plugin import (
+            register_all as _reg_antigravity,
+        )
+        from src.services.provider.adapters.codex.plugin import register_all as _reg_codex
+
+        _reg_antigravity()
+        _reg_codex()
+
+
+__all__ = [
+    "ProviderEnvelope",
+    "ensure_providers_bootstrapped",
+    "get_provider_envelope",
+    "register_envelope",
+]

src/services/provider/metadata_collectors/__init__.py

@@ -47,7 +47,7 @@ class MetadataCollectorRegistry:
         cls._collectors.append(collector)
         for pt in collector.PROVIDER_TYPES:
             cls._type_index[pt.lower()] = collector
-        logger.info(
+        logger.debug(
             "[MetadataCollectorRegistry] 注册: {} -> {}",
             collector.__class__.__name__,
             collector.PROVIDER_TYPES,
@@ -79,7 +79,7 @@ def _ensure_collectors_registered() -> None:
     _initialized = True
 
     # 延迟导入，避免模块加载时的循环依赖
-    from src.services.provider.metadata_collectors.codex import CodexMetadataCollector
+    from src.services.provider.adapters.codex.metadata_collector import CodexMetadataCollector
 
     MetadataCollectorRegistry.register(CodexMetadataCollector())
 

src/services/provider/oauth_token.py

@@ -0,0 +1,118 @@
+"""Provider OAuth token helpers.
+
+These helpers are for *upstream Provider* OAuth keys (ProviderAPIKey.auth_type == "oauth"),
+not for user-login OAuth.
+
+Why:
+- Request path uses `get_provider_auth()` which may refresh the access_token lazily.
+- Some background/admin paths (model fetch/query, etc.) need the same behavior but must
+  avoid sharing a SQLAlchemy Session across concurrent async tasks.
+
+Strategy:
+- Run `get_provider_auth()` on a detached key-like object (no DB session held during HTTP).
+- If refresh updated encrypted fields, persist them back to DB in a short transaction.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from types import SimpleNamespace
+from typing import Any
+
+from src.core.logger import logger
+from src.database import create_session
+from src.models.database import ProviderAPIKey
+
+# ---------------------------------------------------------------------------
+# Account-level block 结构化标记
+# ---------------------------------------------------------------------------
+# oauth_invalid_reason 以此前缀开头的，属于"账号级别"异常（如 Google 要求验证账号）；
+# 刷新 token 无法修复，必须由用户手动解决后再由管理员手动清除。
+# 其余 reason 属于 token 级别异常，成功刷新 token 后自动清除。
+OAUTH_ACCOUNT_BLOCK_PREFIX = "[ACCOUNT_BLOCK] "
+
+
+def is_account_level_block(reason: str | None) -> bool:
+    """判断 oauth_invalid_reason 是否属于账号级别的 block（刷新 token 无法修复）。"""
+    if not reason:
+        return False
+    return str(reason).startswith(OAUTH_ACCOUNT_BLOCK_PREFIX)
+
+
+@dataclass(frozen=True, slots=True)
+class OAuthAccessTokenResult:
+    access_token: str
+    decrypted_auth_config: dict[str, Any] | None
+    refreshed: bool
+
+
+async def resolve_oauth_access_token(
+    *,
+    key_id: str,
+    encrypted_api_key: str,
+    encrypted_auth_config: str | None,
+    provider_proxy_config: dict[str, Any] | None = None,
+    endpoint_api_format: str | None = None,
+) -> OAuthAccessTokenResult:
+    """Resolve (and lazily refresh) OAuth access_token for a ProviderAPIKey.
+
+    This helper is safe to call from concurrent async tasks because it does not
+    rely on the caller's SQLAlchemy Session:
+    - It runs refresh logic without an ORM session.
+    - If refresh succeeded (encrypted fields changed), it persists the new encrypted
+      values to DB using a short, independent session.
+    """
+
+    # Local import to avoid circular imports during app startup.
+    from src.api.handlers.base.request_builder import get_provider_auth
+
+    # Build detached key-like objects for get_provider_auth().
+    provider_obj = (
+        SimpleNamespace(proxy=provider_proxy_config) if provider_proxy_config is not None else None
+    )
+    endpoint_obj = SimpleNamespace(api_format=str(endpoint_api_format or ""))
+    key_obj = SimpleNamespace(
+        id=str(key_id),
+        auth_type="oauth",
+        api_key=encrypted_api_key,
+        auth_config=encrypted_auth_config,
+        provider=provider_obj,
+    )
+
+    orig_api_key = key_obj.api_key
+    orig_auth_config = key_obj.auth_config
+
+    auth_info = await get_provider_auth(endpoint_obj, key_obj)  # type: ignore[arg-type]
+    if auth_info is None:
+        # Should not happen for auth_type="oauth", but keep defensive.
+        return OAuthAccessTokenResult(access_token="", decrypted_auth_config=None, refreshed=False)
+
+    access_token = str(auth_info.auth_value or "").removeprefix("Bearer ").strip()
+    refreshed = (key_obj.api_key != orig_api_key) or (key_obj.auth_config != orig_auth_config)
+
+    if refreshed:
+        # Persist refreshed token/config back to DB.
+        try:
+            with create_session() as db:
+                row = db.query(ProviderAPIKey).filter(ProviderAPIKey.id == str(key_id)).first()
+                if row is not None:
+                    row.api_key = key_obj.api_key
+                    row.auth_config = key_obj.auth_config
+                    # Refresh succeeded => clear token-level invalid markers.
+                    # Preserve account-level blocks (刷新 token 无法修复).
+                    if not is_account_level_block(getattr(row, "oauth_invalid_reason", None)):
+                        row.oauth_invalid_at = None
+                        row.oauth_invalid_reason = None
+                    db.commit()
+        except Exception as e:
+            # Don't fail caller path; token is still usable for this request.
+            logger.debug("[OAUTH_REFRESH] persist refreshed token failed for key {}: {}", key_id, e)
+
+    return OAuthAccessTokenResult(
+        access_token=access_token,
+        decrypted_auth_config=auth_info.decrypted_auth_config,
+        refreshed=refreshed,
+    )
+
+
+__all__ = ["OAuthAccessTokenResult", "resolve_oauth_access_token"]

src/services/provider/stream_policy.py

@@ -16,6 +16,7 @@ from enum import Enum
 from typing import Any
 
 from src.core.api_format.metadata import resolve_endpoint_definition
+from src.core.provider_types import ProviderType
 
 
 class UpstreamStreamPolicy(str, Enum):
@@ -72,7 +73,7 @@ def get_upstream_stream_policy(
         if parsed != UpstreamStreamPolicy.AUTO:
             # Codex upstream requires streaming; do not allow forcing non-stream.
             if (
-                pt == "codex"
+                pt == ProviderType.CODEX
                 and sig == "openai:cli"
                 and parsed == UpstreamStreamPolicy.FORCE_NON_STREAM
             ):
@@ -80,7 +81,7 @@ def get_upstream_stream_policy(
             return parsed
 
     # Safe-by-default: Codex Responses OAuth behaves like SSE-only.
-    if pt == "codex" and sig == "openai:cli":
+    if pt == ProviderType.CODEX and sig == "openai:cli":
         return UpstreamStreamPolicy.FORCE_STREAM
 
     return UpstreamStreamPolicy.AUTO

src/services/provider/transport.py

@@ -10,7 +10,7 @@
 from __future__ import annotations
 
 import re
-from typing import TYPE_CHECKING, Any
+from typing import TYPE_CHECKING, Any, Callable
 from urllib.parse import urlencode
 
 from src.core.api_format import (
@@ -19,11 +19,7 @@ from src.core.api_format import (
     make_signature_key,
 )
 from src.core.logger import logger
-from src.services.antigravity.constants import PROVIDER_TYPE as ANTIGRAVITY_PROVIDER_TYPE
-from src.services.antigravity.constants import (
-    V1INTERNAL_PATH_TEMPLATE,
-)
-from src.services.antigravity.url_availability import url_availability
+from src.core.provider_types import ProviderType, normalize_provider_type
 from src.services.provider.format import normalize_endpoint_signature
 from src.services.provider.request_context import (
     get_selected_base_url,
@@ -34,6 +30,25 @@ from src.utils.url_utils import is_codex_url
 if TYPE_CHECKING:
     from src.models.database import ProviderAPIKey, ProviderEndpoint
 
+# ---------------------------------------------------------------------------
+# Transport Hook Registry
+# ---------------------------------------------------------------------------
+# key: (provider_type, endpoint_sig)
+# value: Callable(endpoint, *, is_stream, effective_query_params) -> str
+_TransportHookFn = Callable[..., str]
+_transport_hooks: dict[tuple[str, str], _TransportHookFn] = {}
+
+
+def register_transport_hook(
+    provider_type: str,
+    endpoint_sig: str,
+    hook: _TransportHookFn,
+) -> None:
+    """注册 provider 特有的 URL 构建 hook。"""
+    pt = normalize_provider_type(provider_type)
+    sig = str(endpoint_sig or "").strip().lower()
+    _transport_hooks[(pt, sig)] = hook
+
 
 # URL 中需要脱敏的查询参数（正则模式）
 _SENSITIVE_QUERY_PARAMS_PATTERN = re.compile(
@@ -174,43 +189,21 @@ def build_provider_url(
     if endpoint_sig.startswith("gemini:"):
         effective_query_params.pop("key", None)
 
-    # Antigravity 特殊处理：复用 gemini:cli endpoint signature，但走 v1internal 端点
-    if provider_type == ANTIGRAVITY_PROVIDER_TYPE and endpoint_sig == "gemini:cli":
-        ordered_urls = url_availability.get_ordered_urls(prefer_daily=True)
-        base_url = ordered_urls[0] if ordered_urls else endpoint.base_url  # type: ignore[arg-type]
+    # Provider transport hook: 如果有注册的 hook 则委托处理
+    from src.services.provider.envelope import ensure_providers_bootstrapped
 
-        # 存入 contextvars（供后续 Handler 层获取）
-        set_selected_base_url(str(base_url) if base_url is not None else None)
+    ensure_providers_bootstrapped()
+    if provider_type and endpoint_sig:
+        hook = _transport_hooks.get((provider_type, endpoint_sig))
+        # Codex hook 仅在无 custom_path 时生效
+        if hook and not (provider_type == ProviderType.CODEX and endpoint.custom_path):
+            return hook(
+                endpoint,
+                is_stream=is_stream,
+                effective_query_params=effective_query_params,
+            )
 
-        action = "streamGenerateContent" if is_stream else "generateContent"
-        path = V1INTERNAL_PATH_TEMPLATE.format(action=action)
-
-        # v1internal 流式请求同样支持 `?alt=sse`
-        if is_stream:
-            effective_query_params.setdefault("alt", "sse")
-
-        url = f"{str(base_url).rstrip('/')}{path}"
-        if effective_query_params:
-            query_string = urlencode(effective_query_params, doseq=True)
-            if query_string:
-                url = f"{url}?{query_string}"
-
-        return url
-
-    # Codex OAuth upstream (chatgpt.com/backend-api/codex) uses `/responses` instead of `/v1/responses`.
-    # We special-case this at transport layer so fixed providers work without requiring custom_path.
-    if provider_type == "codex" and endpoint_sig == "openai:cli" and not endpoint.custom_path:
-        base = str(endpoint.base_url).rstrip("/")
-        path = "/responses"
-        # If user already included the final path in base_url, don't duplicate it.
-        url = base if base.endswith(path) else f"{base}{path}"
-        if effective_query_params:
-            query_string = urlencode(effective_query_params, doseq=True)
-            if query_string:
-                url = f"{url}?{query_string}"
-        return url
-
-    # 非 Antigravity：清除 contextvar，避免跨请求污染
+    # 非 hook 路径：清除 contextvar，避免跨请求污染
     set_selected_base_url(None)
 
     # 准备路径参数（Gemini chat/cli 需要 action）

src/services/provider_ops/registry.py

@@ -63,20 +63,23 @@ class ArchitectureRegistry:
         ]
 
         for arch_cls in builtin:
-            self.register(arch_cls())
+            self.register(arch_cls(), _quiet=True)
+        logger.debug(f"内置架构注册完成: {', '.join(self._architectures.keys())}")
 
-    def register(self, architecture: ProviderArchitecture) -> None:
+    def register(self, architecture: ProviderArchitecture, *, _quiet: bool = False) -> None:
         """
         注册架构
 
         Args:
             architecture: 架构实例
+            _quiet: 内部参数，批量注册时抑制逐条日志
         """
         if architecture.architecture_id in self._architectures:
             logger.warning(f"架构 {architecture.architecture_id} 已存在，将被覆盖")
 
         self._architectures[architecture.architecture_id] = architecture
-        logger.debug(f"注册架构: {architecture.architecture_id}")
+        if not _quiet:
+            logger.debug(f"注册架构: {architecture.architecture_id}")
 
     def unregister(self, architecture_id: str) -> bool:
         """

src/services/provider_ops/service.py

@@ -292,12 +292,6 @@ class ProviderOpsService:
             actual_credentials = credentials
         else:
             actual_credentials = self._decrypt_credentials(config.connector_credentials)
-            logger.debug(
-                f"解密凭据: provider_id={provider_id}, "
-                f"encrypted_keys={list(config.connector_credentials.keys())}, "
-                f"decrypted_keys={list(actual_credentials.keys())}, "
-                f"has_api_key={bool(actual_credentials.get('api_key'))}"
-            )
 
         if not actual_credentials:
             return False, "未提供凭据"
@@ -590,9 +584,6 @@ class ProviderOpsService:
         }
 
         await CacheService.set(cache_key, cache_data, BALANCE_CACHE_TTL)
-        logger.debug(
-            f"余额缓存已写入: provider_id={provider_id}, extra={data.get('extra') if data else None}"
-        )
 
     async def _cache_balance_from_verify(
         self,
@@ -845,7 +836,6 @@ class ProviderOpsService:
         # 使用信号量限制并发数，避免同时发起过多请求耗尽连接池
         concurrency = _get_batch_balance_concurrency()
         semaphore = asyncio.Semaphore(concurrency)
-        logger.debug(f"批量余额查询: providers={len(provider_ids)}, concurrency={concurrency}")
 
         async def _query_with_limit(provider_id: str) -> tuple[str, ActionResult]:
             async with semaphore:

src/services/system/audit.py

@@ -94,8 +94,7 @@ class AuditService:
                 logger.warning(log_message)
             elif event_type in [AuditEventType.LOGIN_FAILED, AuditEventType.REQUEST_FAILED]:
                 logger.info(log_message)
-            else:
-                logger.debug(log_message)
+            # request_success 已由 Pipeline 日志覆盖，不再重复输出到控制台
 
         return audit_log
 

src/services/system/maintenance_scheduler.py

@@ -9,7 +9,6 @@
 - 连接池监控：定期检查数据库连接池状态
 - Pending 状态清理：清理异常的 Pending 状态记录
 - Gemini 文件映射清理：清理过期的 Gemini 文件→Key 映射
-- OAuth Token 刷新：主动刷新即将过期的 OAuth token
 
 使用 APScheduler 进行任务调度，支持时区配置。
 """
@@ -39,8 +38,6 @@ class MaintenanceScheduler:
 
     # 签到任务的 job_id
     CHECKIN_JOB_ID = "provider_checkin"
-    # OAuth 刷新任务的 job_id
-    OAUTH_REFRESH_JOB_ID = "oauth_token_refresh"
     # 用户配额重置任务的 job_id
     USER_QUOTA_RESET_JOB_ID = "user_quota_reset"
 
@@ -49,18 +46,6 @@ class MaintenanceScheduler:
         self._interval_tasks = []
         self._stats_aggregation_lock = asyncio.Lock()
 
-    def trigger_oauth_refresh_check(self) -> None:
-        """
-        触发 OAuth Token 刷新检查
-
-        当新增 OAuth Key 时调用此方法，重新调度刷新任务。
-        会取消当前的调度，并立即重新计算下次执行时间。
-        """
-        if not self.running:
-            return
-
-        asyncio.create_task(self._schedule_next_oauth_refresh())
-
     def _get_checkin_time(self) -> tuple[int, int]:
         """获取签到任务的执行时间
 
@@ -276,11 +261,6 @@ class MaintenanceScheduler:
             name="Provider签到",
         )
 
-        # OAuth Token 刷新任务 - 动态调度
-        # 根据最近即将过期的 token 时间来调度，避免固定间隔频繁查询
-        # 启动时先执行一次，计算下次执行时间
-        asyncio.create_task(self._schedule_next_oauth_refresh())
-
         # 用户配额重置任务 - 根据配置时间执行（按周期配置决定是否执行）
         quota_reset_hour, quota_reset_minute = self._get_user_quota_reset_time()
         scheduler.add_cron_job(
@@ -362,157 +342,6 @@ class MaintenanceScheduler:
         """Provider 签到任务（定时调用）"""
         await self._perform_provider_checkin()
 
-    async def _scheduled_oauth_token_refresh(self) -> None:
-        """OAuth Token 刷新任务（定时调用）"""
-        await self._perform_oauth_token_refresh()
-        # 执行完成后，调度下次执行
-        await self._schedule_next_oauth_refresh()
-
-    async def _schedule_next_oauth_refresh(self) -> None:
-        """
-        动态调度下次 OAuth Token 刷新任务
-
-        策略：
-        - 查询所有 OAuth Key 的 expires_at
-        - 找到最近即将过期的 token（在 refresh_threshold 内）
-        - 设置下次执行时间为：最近过期时间 - 提前量（如提前 1 小时刷新）
-        - 如果没有即将过期的 token，设置默认间隔（如 6 小时后）
-        """
-        import json
-        import time
-
-        from src.core.crypto import crypto_service
-        from src.models.database import ProviderAPIKey
-
-        # 延迟启动，等待系统初始化
-        await asyncio.sleep(5)
-
-        scheduler = get_scheduler()
-        job_id = "oauth_token_refresh"
-
-        try:
-            db = create_session()
-            try:
-                # 检查配置开关
-                if not SystemConfigService.get_config(db, "enable_oauth_token_refresh", True):
-                    logger.info("OAuth Token 自动刷新已禁用，不调度任务")
-                    # 移除已调度的 job（如果存在）
-                    try:
-                        scheduler.remove_job(job_id)
-                    except Exception:
-                        pass
-                    return
-                # 查找所有活跃的 OAuth 类型 Key
-                oauth_keys = (
-                    db.query(ProviderAPIKey)
-                    .filter(
-                        ProviderAPIKey.auth_type == "oauth",
-                        ProviderAPIKey.is_active == True,  # noqa: E712
-                    )
-                    .all()
-                )
-
-                if not oauth_keys:
-                    # 没有 OAuth Key，6 小时后再检查
-                    next_run = datetime.now(timezone.utc) + timedelta(hours=6)
-                    scheduler.add_date_job(
-                        self._scheduled_oauth_token_refresh,
-                        run_date=next_run,
-                        job_id=job_id,
-                        name="OAuth Token刷新",
-                    )
-                    logger.info("没有 OAuth Key，下次检查时间: {}", next_run.isoformat())
-                    return
-
-                now = int(time.time())
-                # 24 小时内过期的都需要刷新（含提前量）
-                refresh_window = 24 * 3600
-                # 提前 1 小时执行刷新
-                refresh_advance = 1 * 3600
-                refresh_threshold_seconds = refresh_window + refresh_advance
-                refresh_threshold = now + refresh_threshold_seconds
-
-                earliest_expires_at: int | None = None
-
-                for key in oauth_keys:
-                    if not key.auth_config:
-                        continue
-
-                    try:
-                        decrypted_config = crypto_service.decrypt(key.auth_config)
-                        token_meta = json.loads(decrypted_config)
-                        expires_at = token_meta.get("expires_at")
-
-                        if expires_at is None:
-                            continue
-
-                        expires_at_int = int(expires_at)
-
-                        # 已经过期或在阈值内，需要立即刷新
-                        if expires_at_int <= refresh_threshold:
-                            # 立即执行
-                            next_run = datetime.now(timezone.utc) + timedelta(seconds=10)
-                            scheduler.add_date_job(
-                                self._scheduled_oauth_token_refresh,
-                                run_date=next_run,
-                                job_id=job_id,
-                                name="OAuth Token刷新",
-                            )
-                            logger.info(
-                                "发现即将过期的 OAuth Token，立即执行刷新: {}",
-                                next_run.isoformat(),
-                            )
-                            return
-
-                        # 记录最近的过期时间
-                        if earliest_expires_at is None or expires_at_int < earliest_expires_at:
-                            earliest_expires_at = expires_at_int
-
-                    except Exception:
-                        continue
-
-                # 计算下次执行时间
-                if earliest_expires_at is not None:
-                    # 在最近过期时间前 24 小时 + 提前量执行
-                    next_run_ts = earliest_expires_at - refresh_threshold_seconds
-                    # 确保不会是过去的时间
-                    if next_run_ts <= now:
-                        next_run_ts = now + 60  # 1 分钟后
-                    next_run = datetime.fromtimestamp(next_run_ts, tz=timezone.utc)
-                else:
-                    # 没有有效的过期时间，6 小时后再检查
-                    next_run = datetime.now(timezone.utc) + timedelta(hours=6)
-
-                # 限制最大间隔为 24 小时
-                max_next_run = datetime.now(timezone.utc) + timedelta(hours=24)
-                if next_run > max_next_run:
-                    next_run = max_next_run
-
-                scheduler.add_date_job(
-                    self._scheduled_oauth_token_refresh,
-                    run_date=next_run,
-                    job_id=job_id,
-                    name="OAuth Token刷新",
-                )
-                logger.info("OAuth Token 刷新任务已调度，下次执行时间: {}", next_run.isoformat())
-
-            finally:
-                db.close()
-
-        except Exception as e:
-            logger.exception("调度 OAuth Token 刷新任务失败: {}", e)
-            # 出错时 1 小时后重试
-            next_run = datetime.now(timezone.utc) + timedelta(hours=1)
-            try:
-                scheduler.add_date_job(
-                    self._scheduled_oauth_token_refresh,
-                    run_date=next_run,
-                    job_id=job_id,
-                    name="OAuth Token刷新",
-                )
-            except Exception:
-                pass
-
     async def _scheduled_user_quota_reset(self) -> None:
         """用户配额重置任务（定时调用）"""
         await self._perform_user_quota_reset()
@@ -1376,294 +1205,6 @@ class MaintenanceScheduler:
 
         return total_deleted
 
-    async def _perform_oauth_token_refresh(self) -> None:
-        """
-        主动刷新即将过期的 OAuth token
-
-        策略：
-        - 查找所有 auth_type='oauth' 且 is_active=True 的 Key
-        - 检查 auth_config 中的 expires_at，如果在 24 小时内过期则刷新
-        - 使用 refresh_token 换取新的 access_token
-        - 更新数据库中的 token 信息
-        """
-        import json
-        import time
-
-        from src.core.crypto import crypto_service
-        from src.core.provider_oauth_utils import enrich_auth_config, post_oauth_token
-        from src.core.provider_templates.fixed_providers import FIXED_PROVIDERS
-        from src.core.provider_templates.types import ProviderType
-        from src.models.database import ProviderAPIKey
-
-        # 检查配置开关
-        check_db = create_session()
-        try:
-            if not SystemConfigService.get_config(check_db, "enable_oauth_token_refresh", True):
-                logger.info("OAuth Token 自动刷新已禁用，跳过任务")
-                return
-        finally:
-            check_db.close()
-
-        logger.info("开始执行 OAuth Token 刷新任务...")
-
-        db = create_session()
-        refreshed_count = 0
-        failed_count = 0
-        skipped_count = 0
-
-        try:
-            # 查找所有活跃的 OAuth 类型 Key
-            oauth_keys = (
-                db.query(ProviderAPIKey)
-                .filter(
-                    ProviderAPIKey.auth_type == "oauth",
-                    ProviderAPIKey.is_active == True,  # noqa: E712
-                )
-                .all()
-            )
-
-            if not oauth_keys:
-                logger.info("没有找到需要刷新的 OAuth Key")
-                return
-
-            logger.info("找到 {} 个 OAuth Key，开始检查过期状态...", len(oauth_keys))
-
-            now = int(time.time())
-            # 24 小时内过期的都刷新（含提前量）
-            refresh_window = 24 * 3600
-            # 提前 1 小时执行刷新
-            refresh_advance = 1 * 3600
-            refresh_threshold = now + refresh_window + refresh_advance
-
-            for key in oauth_keys:
-                try:
-                    # 解密 auth_config
-                    if not key.auth_config:
-                        skipped_count += 1
-                        continue
-
-                    try:
-                        decrypted_config = crypto_service.decrypt(key.auth_config)
-                        token_meta = json.loads(decrypted_config)
-                    except Exception:
-                        logger.warning("Key {} auth_config 解密失败，跳过", key.id)
-                        skipped_count += 1
-                        continue
-
-                    expires_at = token_meta.get("expires_at")
-                    refresh_token = token_meta.get("refresh_token")
-                    provider_type = str(token_meta.get("provider_type") or "")
-
-                    # 检查是否需要刷新
-                    if expires_at is None:
-                        skipped_count += 1
-                        continue
-
-                    try:
-                        expires_at_int = int(expires_at)
-                    except (ValueError, TypeError):
-                        skipped_count += 1
-                        continue
-
-                    if expires_at_int > refresh_threshold:
-                        # 还没到刷新时间
-                        skipped_count += 1
-                        continue
-
-                    if not refresh_token or not provider_type:
-                        logger.warning(
-                            "Key {} 缺少 refresh_token 或 provider_type，无法刷新", key.id
-                        )
-                        skipped_count += 1
-                        continue
-
-                    # 获取 provider 模板
-                    try:
-                        provider_type_enum = ProviderType(provider_type)
-                    except ValueError:
-                        logger.warning("Key {} 未知的 provider_type: {}", key.id, provider_type)
-                        skipped_count += 1
-                        continue
-
-                    template = FIXED_PROVIDERS.get(provider_type_enum)
-                    if not template or not template.oauth:
-                        logger.warning("Key {} provider {} 不支持 OAuth", key.id, provider_type)
-                        skipped_count += 1
-                        continue
-
-                    # 获取代理配置
-                    proxy_config = None
-                    if key.provider and key.provider.endpoints:
-                        for endpoint in key.provider.endpoints:
-                            if endpoint.proxy:
-                                proxy_config = endpoint.proxy
-                                break
-
-                    # 执行刷新
-                    token_url = template.oauth.token_url
-                    is_json = "anthropic.com" in token_url
-
-                    if is_json:
-                        body = {
-                            "grant_type": "refresh_token",
-                            "client_id": template.oauth.client_id,
-                            "refresh_token": str(refresh_token),
-                        }
-                        headers = {
-                            "Content-Type": "application/json",
-                            "Accept": "application/json",
-                        }
-                        data = None
-                        json_body = body
-                    else:
-                        form = {
-                            "grant_type": "refresh_token",
-                            "client_id": template.oauth.client_id,
-                            "refresh_token": str(refresh_token),
-                        }
-                        if template.oauth.client_secret:
-                            form["client_secret"] = template.oauth.client_secret
-                        headers = {
-                            "Content-Type": "application/x-www-form-urlencoded",
-                            "Accept": "application/json",
-                        }
-                        data = form
-                        json_body = None
-
-                    logger.info(
-                        "刷新 Key {} ({}) 的 OAuth token，当前过期时间: {}",
-                        key.id,
-                        key.name,
-                        datetime.fromtimestamp(expires_at_int, tz=timezone.utc).isoformat(),
-                    )
-
-                    resp = await post_oauth_token(
-                        provider_type=provider_type,
-                        token_url=token_url,
-                        headers=headers,
-                        data=data,
-                        json_body=json_body,
-                        proxy_config=proxy_config,
-                        timeout_seconds=30.0,
-                    )
-
-                    if 200 <= resp.status_code < 300:
-                        token = resp.json()
-                        access_token = str(token.get("access_token") or "")
-                        new_refresh_token = str(token.get("refresh_token") or "")
-                        expires_in = token.get("expires_in")
-                        new_expires_at = None
-
-                        try:
-                            if expires_in is not None:
-                                new_expires_at = int(time.time()) + int(expires_in)
-                        except Exception:
-                            new_expires_at = None
-
-                        if access_token:
-                            # 更新 token_meta
-                            token_meta["token_type"] = token.get("token_type")
-                            if new_refresh_token:
-                                token_meta["refresh_token"] = new_refresh_token
-                            token_meta["expires_at"] = new_expires_at
-                            token_meta["scope"] = token.get("scope")
-                            token_meta["updated_at"] = int(time.time())
-
-                            # 提取额外信息
-                            token_meta = await enrich_auth_config(
-                                provider_type=provider_type,
-                                auth_config=token_meta,
-                                token_response=token,
-                                access_token=access_token,
-                                proxy_config=proxy_config,
-                            )
-
-                            # 更新数据库
-                            encrypted_token = crypto_service.encrypt(access_token)
-                            encrypted_config = crypto_service.encrypt(json.dumps(token_meta))
-
-                            key.api_key = encrypted_token
-                            key.auth_config = encrypted_config
-                            # 刷新成功，清除失效标记
-                            key.oauth_invalid_at = None
-                            key.oauth_invalid_reason = None
-                            db.commit()
-
-                            refreshed_count += 1
-                            new_expires_str = (
-                                datetime.fromtimestamp(new_expires_at, tz=timezone.utc).isoformat()
-                                if new_expires_at
-                                else "unknown"
-                            )
-                            logger.info(
-                                "Key {} ({}) OAuth token 刷新成功，新过期时间: {}",
-                                key.id,
-                                key.name,
-                                new_expires_str,
-                            )
-                        else:
-                            failed_count += 1
-                            logger.warning(
-                                "Key {} ({}) 刷新响应中没有 access_token", key.id, key.name
-                            )
-                    else:
-                        failed_count += 1
-                        # 解析错误原因
-                        error_reason = "HTTP {}".format(resp.status_code)
-                        try:
-                            error_body = resp.json()
-                            if "error" in error_body:
-                                error_reason = str(
-                                    error_body.get("error_description") or error_body.get("error")
-                                )
-                        except Exception:
-                            error_reason = (
-                                resp.text[:100] if resp.text else "HTTP {}".format(resp.status_code)
-                            )
-
-                        # 标记为失效（400/401/403 通常表示永久性错误）
-                        if resp.status_code in (400, 401, 403):
-                            key.oauth_invalid_at = datetime.now(timezone.utc)
-                            key.oauth_invalid_reason = error_reason
-                            db.commit()
-                            logger.warning(
-                                "Key {} ({}) OAuth token 刷新失败，已标记为失效: {}",
-                                key.id,
-                                key.name,
-                                error_reason,
-                            )
-                        else:
-                            logger.warning(
-                                "Key {} ({}) OAuth token 刷新失败，状态码: {}，响应: {}",
-                                key.id,
-                                key.name,
-                                resp.status_code,
-                                resp.text[:200],
-                            )
-
-                except Exception as e:
-                    failed_count += 1
-                    logger.exception("Key {} OAuth token 刷新出错: {}", key.id, e)
-                    try:
-                        db.rollback()
-                    except Exception:
-                        pass
-
-                # 避免请求过于频繁
-                await asyncio.sleep(1)
-
-        except Exception as e:
-            logger.exception("OAuth Token 刷新任务执行出错: {}", e)
-        finally:
-            db.close()
-
-        logger.info(
-            "OAuth Token 刷新任务完成: 刷新 {} 个，失败 {} 个，跳过 {} 个",
-            refreshed_count,
-            failed_count,
-            skipped_count,
-        )
-
 
 # 全局单例
 _maintenance_scheduler = None

src/services/system/stats_aggregator.py

@@ -277,9 +277,6 @@ class StatsAggregatorService:
         if commit:
             db.commit()
 
-        logger.info(
-            f"[StatsAggregator] 聚合日期 {date.date()} 完成: {computed['total_requests']} 请求"
-        )
         return stats
 
     @staticmethod
@@ -347,7 +344,6 @@ class StatsAggregatorService:
 
         if commit:
             db.commit()
-        logger.info(f"[StatsAggregator] 聚合日期 {date.date()} 模型统计完成: {len(results)} 个模型")
         return results
 
     @staticmethod
@@ -416,9 +412,6 @@ class StatsAggregatorService:
 
         if commit:
             db.commit()
-        logger.info(
-            f"[StatsAggregator] 聚合日期 {date.date()} 供应商统计完成: {len(results)} 个供应商"
-        )
         return results
 
     @staticmethod
@@ -483,9 +476,6 @@ class StatsAggregatorService:
 
         if commit:
             db.commit()
-        logger.info(
-            f"[StatsAggregator] 聚合日期 {date.date()} API Key 统计完成: {len(results)} 个密钥"
-        )
         return results
 
     @staticmethod
@@ -527,7 +517,6 @@ class StatsAggregatorService:
 
         if commit:
             db.commit()
-        logger.info(f"[StatsAggregator] 聚合日期 {date.date()} 错误统计完成: {len(results)} 条")
         return results
 
     @staticmethod

src/services/task/impl/video_poller.py

@@ -585,6 +585,7 @@ class VideoTaskPollerAdapter:
             endpoint_sig,
             upstream_key,
             endpoint_headers=extra_headers,
+            header_rules=getattr(endpoint, "header_rules", None),
         )
         if auth_info:
             headers.pop("x-goog-api-key", None)

src/services/task/service.py

@@ -8,6 +8,7 @@ from sqlalchemy.orm import Session
 
 from src.config.settings import config
 from src.core.logger import logger
+from src.core.provider_types import ProviderType
 from src.models.database import ApiKey
 from src.services.candidate.failover import FailoverEngine
 from src.services.candidate.policy import RetryPolicy, SkipPolicy
@@ -598,7 +599,7 @@ class TaskService:
         if stage <= 0 and request_body_ref.get("_rectified", False):
             stage = 1
 
-        if stage >= 2 or (stage >= 1 and provider_type_norm != "antigravity"):
+        if stage >= 2 or (stage >= 1 and provider_type_norm != ProviderType.ANTIGRAVITY):
             logger.warning("  [{}] Thinking 错误：已整流仍失败，终止重试", request_id)
             self._mark_thinking_error_failed(
                 candidate_record_id,
@@ -631,7 +632,7 @@ class TaskService:
             request_body_ref["_rectified_this_turn"] = True
             request_body_ref["_rectify_stage"] = next_stage
 
-            if provider_type_norm == "antigravity":
+            if provider_type_norm == ProviderType.ANTIGRAVITY:
                 try:
                     from src.core.metrics import antigravity_degradation_total
 
@@ -1501,6 +1502,7 @@ class TaskService:
             provider_format,
             upstream_key,
             endpoint_headers=extra_headers,
+            header_rules=getattr(endpoint, "header_rules", None),
         )
 
         client = await HTTPClientPool.get_default_client_async()

src/services/usage/service.py

@@ -1457,13 +1457,23 @@ class UsageService:
                 req_id = record.get("request_id")
                 if req_id and req_id in existing_usages:
                     existing_usage = existing_usages[req_id]
-                    # 只更新 pending/streaming 状态的记录
-                    # 已经是 completed/failed/cancelled 的记录跳过
-                    if existing_usage.status in ("pending", "streaming"):
+                    # 以 billing_status 为幂等闸门：
+                    # - pending: 允许更新（补全 tokens/cost/headers/body 等）
+                    # - settled/void: 跳过（避免重复记账/重复覆盖）
+                    #
+                    # 注意：stream_telemetry 在 usage_queue_enabled 时会“直接更新 status”
+                    # 来减少 UI 延迟，但不会同步更新 billing_status。
+                    # 这会导致出现 status=completed 但 billing_status=pending 的中间态，
+                    # 此时仍应允许 completed/failed/cancelled 事件落库补全详情。
+                    billing_status = getattr(existing_usage, "billing_status", None)
+                    if billing_status == "pending":
                         records_to_update.append(record)
                     else:
                         logger.debug(
-                            f"批量记录预过滤: 跳过已完成的 request_id={req_id} (status={existing_usage.status})"
+                            "批量记录预过滤: 跳过已结算的 request_id={} (status={}, billing_status={})",
+                            req_id,
+                            getattr(existing_usage, "status", None),
+                            billing_status,
                         )
                 else:
                     records_to_insert.append(record)
@@ -1472,7 +1482,7 @@ class UsageService:
 
         if records_to_update:
             logger.debug(
-                f"批量记录: 需要更新 {len(records_to_update)} 条已存在的 pending/streaming 记录"
+                f"批量记录: 需要更新 {len(records_to_update)} 条已存在的 billing_status=pending 记录"
             )
 
         usages: list[Usage] = []
@@ -1582,6 +1592,9 @@ class UsageService:
         update_results = prepared_results[: len(update_params_list)]
         insert_results = prepared_results[len(update_params_list) :]
 
+        finalized_at = datetime.now(timezone.utc)
+        terminal_statuses = {"completed", "failed", "cancelled"}
+
         # 1. 处理需要更新的记录
         for i, (record, request_id, params) in enumerate(update_params_list):
             try:
@@ -1596,6 +1609,14 @@ class UsageService:
 
                 # 更新已存在的 Usage 记录
                 cls._update_existing_usage(existing_usage, usage_params, record.get("target_model"))
+                # 结算标记：pending -> settled（幂等闸门由 prefilter 控制）
+                if (
+                    usage_params.get("status") in terminal_statuses
+                    and getattr(existing_usage, "billing_status", None) == "pending"
+                ):
+                    existing_usage.billing_status = "settled"
+                    if getattr(existing_usage, "finalized_at", None) is None:
+                        existing_usage.finalized_at = finalized_at
                 usages.append(existing_usage)
                 updated_count += 1
 
@@ -1634,6 +1655,12 @@ class UsageService:
 
                 # 创建 Usage 记录
                 usage = Usage(**usage_params)
+                # 新建记录默认 billing_status=settled，但补齐 finalized_at，便于审计与幂等判断
+                if usage_params.get("status") in terminal_statuses:
+                    if getattr(usage, "billing_status", None) in (None, "pending"):
+                        usage.billing_status = "settled"
+                    if getattr(usage, "finalized_at", None) is None:
+                        usage.finalized_at = finalized_at
                 db.add(usage)
                 usages.append(usage)