feat: Antigravity 和 Codex 服务支持

- 新增 Antigravity 服务：签名缓存、URL 可用性检测、信封处理
- 新增 Codex 服务：信封处理、元数据收集器
- 重构 provider transport 支持新的服务架构
- 新增 stream_bridge 和 upstream_stream_bridge 处理流式响应
- 优化 OAuth 工具函数
- 添加相关测试用例

src/services/antigravity/__init__.py

@@ -0,0 +1 @@
+"""Antigravity integration package."""

src/services/antigravity/client.py

@@ -0,0 +1,75 @@
+"""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

@@ -0,0 +1,40 @@
+"""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

@@ -0,0 +1,177 @@
+"""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

@@ -0,0 +1,58 @@
+"""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/antigravity/url_availability.py

@@ -0,0 +1,78 @@
+"""Antigravity URL 可用性管理（带 TTL 自动恢复）。"""
+
+from __future__ import annotations
+
+import threading
+import time
+
+from src.services.antigravity.constants import (
+    DAILY_BASE_URL,
+    PROD_BASE_URL,
+    URL_UNAVAILABLE_TTL_SECONDS,
+)
+
+
+class URLAvailability:
+    """管理 Antigravity API 端点可用性（进程内）。"""
+
+    _instance: "URLAvailability | None" = None
+    _lock = threading.Lock()
+
+    def __new__(cls) -> "URLAvailability":
+        if cls._instance is None:
+            with cls._lock:
+                if cls._instance is None:
+                    cls._instance = super().__new__(cls)
+                    cls._instance._init()
+        return cls._instance
+
+    def _init(self) -> None:
+        self._unavailable: dict[str, float] = {}  # url -> recover_at(ts)
+        self._last_success: str | None = None
+        self._mu = threading.RLock()
+
+    def _prune(self, now: float | None = None) -> None:
+        now_ts = time.time() if now is None else now
+        self._unavailable = {u: t for u, t in self._unavailable.items() if t > now_ts}
+
+    def is_available(self, url: str) -> bool:
+        with self._mu:
+            self._prune()
+            return url not in self._unavailable
+
+    def get_ordered_urls(self, *, prefer_daily: bool = True) -> list[str]:
+        """返回优先级排序的可用 URL 列表。
+
+        - 默认 daily 优先（通常限流更宽松）
+        - 最近成功的 URL 会被提升到最前
+        - 若全部被标记不可用，则返回 base_order（允许继续尝试，等待 TTL 自动恢复）
+        """
+        with self._mu:
+            self._prune()
+
+            base_order = (
+                [DAILY_BASE_URL, PROD_BASE_URL] if prefer_daily else [PROD_BASE_URL, DAILY_BASE_URL]
+            )
+
+            if self._last_success and self._last_success in base_order:
+                base_order.remove(self._last_success)
+                base_order.insert(0, self._last_success)
+
+            available = [u for u in base_order if u not in self._unavailable]
+            return available if available else base_order
+
+    def mark_success(self, url: str) -> None:
+        with self._mu:
+            self._last_success = url
+            self._unavailable.pop(url, None)
+
+    def mark_unavailable(self, url: str) -> None:
+        with self._mu:
+            self._unavailable[url] = time.time() + URL_UNAVAILABLE_TTL_SECONDS
+            if self._last_success == url:
+                self._last_success = None
+
+
+url_availability = URLAvailability()
+
+__all__ = ["URLAvailability", "url_availability"]

src/services/codex/__init__.py

@@ -0,0 +1,3 @@
+"""Codex provider integration package."""
+
+__all__ = []

src/services/codex/envelope.py

@@ -0,0 +1,78 @@
+"""Codex upstream envelope hooks.
+
+Codex OAuth upstreams (e.g. `chatgpt.com/backend-api/codex`) behave like the OpenAI
+Responses API (`openai:cli`) but may require additional transport-level headers
+to avoid upstream blocks (Cloudflare, etc.).
+
+Request/response shape quirks should live in the conversion layer as a same-format
+variant (`target_variant="codex"` in the `openai:cli` normalizer). This envelope
+only adds headers and keeps the rest as a no-op wrapper.
+"""
+
+from __future__ import annotations
+
+import uuid
+from typing import Any
+
+from src.config.settings import config
+from src.services.provider.request_context import get_selected_base_url
+
+
+class CodexOAuthEnvelope:
+    """Provider envelope hooks for Codex OAuth upstream."""
+
+    name = "codex:oauth"
+
+    def extra_headers(self) -> dict[str, str] | None:
+        # These headers are best-effort: Codex upstream is stricter than public OpenAI API.
+        # Keep them provider-scoped (via ProviderEnvelope) to avoid leaking to other upstreams.
+        headers: dict[str, str] = {
+            "x-oai-web-search-eligible": "true",
+            "session_id": str(uuid.uuid4()),
+            "originator": "codex_cli_rs",
+            # Ensure SSE is returned when upstream is forced to streaming mode.
+            "Accept": "text/event-stream",
+        }
+
+        ua = str(getattr(config, "internal_user_agent_openai_cli", "") or "").strip()
+        if ua:
+            headers["User-Agent"] = ua
+
+        return headers
+
+    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]:
+        # No wire envelope for Codex; keep request body as-is.
+        _ = model, decrypted_auth_config
+        return request_body, url_model
+
+    def unwrap_response(self, data: Any) -> Any:
+        # No response envelope for Codex.
+        return data
+
+    def postprocess_unwrapped_response(self, *, model: str, data: Any) -> None:  # noqa: ARG002
+        return
+
+    def capture_selected_base_url(self) -> str | None:
+        # Keep interface consistent with Antigravity. Transport currently doesn't set this for Codex.
+        return get_selected_base_url()
+
+    def on_http_status(self, *, base_url: str | None, status_code: int) -> None:  # noqa: ARG002
+        return
+
+    def on_connection_error(self, *, base_url: str | None, exc: Exception) -> None:  # noqa: ARG002
+        return
+
+    def force_stream_rewrite(self) -> bool:
+        return False
+
+
+codex_oauth_envelope = CodexOAuthEnvelope()
+
+__all__ = ["CodexOAuthEnvelope", "codex_oauth_envelope"]

src/services/message/thinking_rectifier.py

@@ -13,6 +13,7 @@ Thinking 整流器（Rectifier）
 """
 
 import copy
+import json
 from typing import Any
 
 from src.core.logger import logger
@@ -67,6 +68,103 @@ class ThinkingRectifier:
 
         return rectified_body, modified
 
+    @staticmethod
+    def rectify_signature_sensitive_blocks(
+        request_body: dict[str, Any],
+    ) -> tuple[dict[str, Any], bool]:
+        """Second-stage rectification for signature-related failures.
+
+        This is a more aggressive fallback than `rectify()`:
+        - Removes all thinking/redacted_thinking blocks
+        - Removes signature fields on remaining blocks
+        - Degrades tool_use/tool_result blocks into plain text blocks
+        - Disables top-level `thinking` when enabled
+        """
+        if not request_body:
+            return request_body, False
+
+        rectified_body = copy.deepcopy(request_body)
+        modified = False
+
+        messages = rectified_body.get("messages", [])
+        if isinstance(messages, list) and messages:
+            new_messages: list[Any] = []
+            for message in messages:
+                if not isinstance(message, dict):
+                    new_messages.append(message)
+                    continue
+
+                new_message = dict(message)
+                content = message.get("content")
+                if isinstance(content, list):
+                    new_content: list[Any] = []
+                    for block in content:
+                        if not isinstance(block, dict):
+                            new_content.append(block)
+                            continue
+
+                        block_type = block.get("type")
+
+                        if block_type in ("thinking", "redacted_thinking"):
+                            modified = True
+                            continue
+
+                        if block_type == "tool_use":
+                            # Degrade into text to avoid strict structure/signature validation.
+                            name = block.get("name")
+                            inp = block.get("input")
+                            try:
+                                inp_text = json.dumps(inp, ensure_ascii=False)
+                            except Exception:
+                                inp_text = str(inp)
+                            new_content.append(
+                                {
+                                    "type": "text",
+                                    "text": f"[tool_use] name={name} input={inp_text}",
+                                }
+                            )
+                            modified = True
+                            continue
+
+                        if block_type == "tool_result":
+                            raw = block.get("content")
+                            try:
+                                raw_text = json.dumps(raw, ensure_ascii=False)
+                            except Exception:
+                                raw_text = str(raw)
+                            new_content.append(
+                                {
+                                    "type": "text",
+                                    "text": f"[tool_result] {raw_text}",
+                                }
+                            )
+                            modified = True
+                            continue
+
+                        # Remove signature field (for any non-thinking block).
+                        if "signature" in block:
+                            new_block = {k: v for k, v in block.items() if k != "signature"}
+                            new_content.append(new_block)
+                            modified = True
+                            continue
+
+                        new_content.append(block)
+
+                    new_message["content"] = new_content
+
+                new_messages.append(new_message)
+
+            rectified_body["messages"] = new_messages
+
+        # Stage-2: disable top-level thinking unconditionally when enabled.
+        thinking_param = rectified_body.get("thinking")
+        if isinstance(thinking_param, dict) and thinking_param.get("type") == "enabled":
+            del rectified_body["thinking"]
+            modified = True
+            logger.info("ThinkingRectifier(stage2): 已移除顶层 thinking 参数")
+
+        return rectified_body, modified
+
     @staticmethod
     def _rectify_messages(messages: list[dict[str, Any]]) -> tuple[list[dict[str, Any]], bool]:
         """

src/services/orchestration/error_classifier.py

@@ -176,6 +176,9 @@ class ErrorClassifier:
         "expected `thinking`, found",  # 带反引号变体
         "expected redacted_thinking, found",
         "expected `redacted_thinking`, found",
+        # Antigravity / Gemini-internal: thought signature validation
+        "thoughtsignature",
+        "thought_signature",
     )
 
     def _parse_error_response(self, error_text: str | None) -> dict[str, Any]:

src/services/provider/behavior.py

@@ -0,0 +1,48 @@
+"""Provider behavior resolver.
+
+Keep provider-specific quirks centralized so handler code stays generic.
+
+Concepts:
+- envelope: wire-level request/response wrappers and transport side-effects
+- same_format_variant: subtle same-format differences (e.g. Codex)
+- cross_format_variant: cross-format conversion tweaks (e.g. Antigravity thinking blocks)
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+from src.services.provider.envelope import ProviderEnvelope, get_provider_envelope
+
+
+@dataclass(frozen=True, slots=True)
+class ProviderBehavior:
+    provider_type: str
+    envelope: ProviderEnvelope | None
+    same_format_variant: str | None
+    cross_format_variant: str | None
+
+
+def get_provider_behavior(
+    *,
+    provider_type: str | None,
+    endpoint_sig: str | None,
+) -> ProviderBehavior:
+    pt = str(provider_type or "").strip().lower()
+    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
+
+    return ProviderBehavior(
+        provider_type=pt,
+        envelope=envelope,
+        same_format_variant=same_format_variant,
+        cross_format_variant=cross_format_variant,
+    )
+
+
+__all__ = ["ProviderBehavior", "get_provider_behavior"]

src/services/provider/envelope.py

@@ -0,0 +1,81 @@
+"""Provider request/response envelope hooks.
+
+Some upstreams expose an API that is *almost* compatible with an existing
+endpoint signature (family:kind), but wrap the wire format in an extra envelope
+or require small transport-level behaviors.
+
+This module provides a small hook mechanism so handlers can stay generic while
+provider-specific envelopes live in their own service modules.
+"""
+
+from __future__ import annotations
+
+from typing import Any, Protocol
+
+
+class ProviderEnvelope(Protocol):
+    """Provider-specific envelope transformation and side-effects."""
+
+    name: str
+
+    def extra_headers(self) -> dict[str, str] | None:
+        """Extra upstream request headers to merge into the RequestBuilder."""
+
+    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]:
+        """Wrap request payload and optionally override url_model (e.g. move model into body)."""
+
+    def unwrap_response(self, data: Any) -> Any:
+        """Unwrap upstream response payload (streaming chunk or full JSON)."""
+
+    def postprocess_unwrapped_response(self, *, model: str, data: Any) -> None:
+        """Best-effort post processing after unwrap (e.g. cache signatures)."""
+
+    def capture_selected_base_url(self) -> str | None:
+        """Capture the base_url selected by transport layer (if any)."""
+
+    def on_http_status(self, *, base_url: str | None, status_code: int) -> None:
+        """Called after receiving upstream HTTP status code."""
+
+    def on_connection_error(self, *, base_url: str | None, exc: Exception) -> None:
+        """Called when a connection-type exception happens."""
+
+    def force_stream_rewrite(self) -> bool:
+        """Whether streaming should always go through the rewrite/conversion path."""
+
+
+def get_provider_envelope(
+    *,
+    provider_type: str | None,
+    endpoint_sig: str | None,
+) -> ProviderEnvelope | None:
+    """Return envelope hooks for the given provider_type + endpoint signature."""
+
+    pt = str(provider_type or "").strip().lower()
+    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
+
+
+__all__ = ["ProviderEnvelope", "get_provider_envelope"]

src/services/provider/request_context.py

@@ -0,0 +1,28 @@
+"""Per-request context shared across layers.
+
+We use `contextvars` so the transport layer (URL builder) can pass small bits of
+state to the handler layer without changing existing return types.
+
+This is intentionally minimal; only add fields that are safe and cheap to carry
+per request.
+"""
+
+from __future__ import annotations
+
+import contextvars
+
+_selected_base_url: contextvars.ContextVar[str | None] = contextvars.ContextVar(
+    "provider_selected_base_url",
+    default=None,
+)
+
+
+def set_selected_base_url(url: str | None) -> None:
+    _selected_base_url.set(url)
+
+
+def get_selected_base_url() -> str | None:
+    return _selected_base_url.get()
+
+
+__all__ = ["get_selected_base_url", "set_selected_base_url"]

src/services/provider/stream_policy.py

@@ -0,0 +1,139 @@
+"""Upstream streaming execution policy (per endpoint).
+
+This is about how we talk to the upstream provider, not what the client asked for.
+
+Motivation:
+- Some upstreams require streaming only (e.g. Codex Responses OAuth endpoint).
+- Some upstreams do not support streaming (or are flaky with SSE).
+
+We allow forcing upstream request mode per ProviderEndpoint, while the gateway still
+returns what the client requested by doing internal sync<->stream bridging.
+"""
+
+from __future__ import annotations
+
+from enum import Enum
+from typing import Any
+
+from src.core.api_format.metadata import resolve_endpoint_definition
+
+
+class UpstreamStreamPolicy(str, Enum):
+    AUTO = "auto"  # follow client request
+    FORCE_STREAM = "force_stream"
+    FORCE_NON_STREAM = "force_non_stream"
+
+
+def parse_upstream_stream_policy(value: Any) -> UpstreamStreamPolicy:
+    if value is None:
+        return UpstreamStreamPolicy.AUTO
+
+    if isinstance(value, bool):
+        return UpstreamStreamPolicy.FORCE_STREAM if value else UpstreamStreamPolicy.FORCE_NON_STREAM
+
+    raw = str(value).strip().lower()
+    if raw in {"", "auto", "follow", "client", "default"}:
+        return UpstreamStreamPolicy.AUTO
+    if raw in {"force_stream", "stream", "sse", "true", "1", "yes"}:
+        return UpstreamStreamPolicy.FORCE_STREAM
+    if raw in {"force_non_stream", "force_sync", "non_stream", "sync", "false", "0", "no"}:
+        return UpstreamStreamPolicy.FORCE_NON_STREAM
+
+    return UpstreamStreamPolicy.AUTO
+
+
+def get_upstream_stream_policy(
+    endpoint: Any,
+    *,
+    provider_type: str | None = None,
+    endpoint_sig: str | None = None,
+) -> UpstreamStreamPolicy:
+    """Resolve policy for an endpoint.
+
+    Config source: endpoint.config["upstream_stream_policy"] (preferred).
+
+    Defaults:
+    - Codex + openai:cli: FORCE_STREAM (Codex upstream requires stream=true).
+    """
+
+    provider_obj = getattr(endpoint, "provider", None)
+    pt = str(provider_type or getattr(provider_obj, "provider_type", "") or "").strip().lower()
+    sig = str(endpoint_sig or getattr(endpoint, "api_format", "") or "").strip().lower()
+
+    # Explicit config wins (unless upstream has a hard constraint).
+    cfg = getattr(endpoint, "config", None)
+    if isinstance(cfg, dict):
+        val = (
+            cfg.get("upstream_stream_policy")
+            or cfg.get("upstreamStreamPolicy")
+            or cfg.get("upstream_stream")
+        )
+        parsed = parse_upstream_stream_policy(val)
+        if parsed != UpstreamStreamPolicy.AUTO:
+            # Codex upstream requires streaming; do not allow forcing non-stream.
+            if (
+                pt == "codex"
+                and sig == "openai:cli"
+                and parsed == UpstreamStreamPolicy.FORCE_NON_STREAM
+            ):
+                return UpstreamStreamPolicy.FORCE_STREAM
+            return parsed
+
+    # Safe-by-default: Codex Responses OAuth behaves like SSE-only.
+    if pt == "codex" and sig == "openai:cli":
+        return UpstreamStreamPolicy.FORCE_STREAM
+
+    return UpstreamStreamPolicy.AUTO
+
+
+def resolve_upstream_is_stream(
+    *,
+    client_is_stream: bool,
+    policy: UpstreamStreamPolicy,
+) -> bool:
+    if policy == UpstreamStreamPolicy.FORCE_STREAM:
+        return True
+    if policy == UpstreamStreamPolicy.FORCE_NON_STREAM:
+        return False
+    return bool(client_is_stream)
+
+
+def enforce_stream_mode_for_upstream(
+    request_body: dict[str, Any],
+    *,
+    provider_api_format: str,
+    upstream_is_stream: bool,
+) -> dict[str, Any]:
+    """Force upstream stream/sync mode in request body (best-effort).
+
+    Note: Some formats (Gemini) do not use a `stream` field in body; for those we
+    remove it to avoid leaking client intent.
+    """
+
+    meta = resolve_endpoint_definition(provider_api_format)
+    provider_uses_stream = meta.stream_in_body if meta is not None else True
+
+    if provider_uses_stream:
+        request_body["stream"] = bool(upstream_is_stream)
+    else:
+        request_body.pop("stream", None)
+
+    # OpenAI Chat Completions: request usage in streaming mode.
+    provider_fmt = str(provider_api_format or "").strip().lower()
+    if upstream_is_stream and provider_fmt == "openai:chat":
+        stream_options = request_body.get("stream_options")
+        if not isinstance(stream_options, dict):
+            stream_options = {}
+        stream_options["include_usage"] = True
+        request_body["stream_options"] = stream_options
+
+    return request_body
+
+
+__all__ = [
+    "UpstreamStreamPolicy",
+    "enforce_stream_mode_for_upstream",
+    "get_upstream_stream_policy",
+    "parse_upstream_stream_policy",
+    "resolve_upstream_is_stream",
+]

src/services/provider/transport.py

@@ -19,7 +19,16 @@ 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.services.provider.format import normalize_endpoint_signature
+from src.services.provider.request_context import (
+    get_selected_base_url,
+    set_selected_base_url,
+)
 from src.utils.url_utils import is_codex_url
 
 if TYPE_CHECKING:
@@ -72,6 +81,35 @@ def _normalize_base_url(base_url: str, path: str) -> str:
     return base
 
 
+def get_antigravity_base_url() -> str | None:
+    """Backward-compat alias for `get_selected_base_url()`."""
+    return get_selected_base_url()
+
+
+def _get_provider_type(endpoint: Any, key: "ProviderAPIKey" | None = None) -> str | None:
+    """尽力获取 Provider.provider_type（用于 Antigravity 等 Provider 特判）。"""
+    try:
+        provider = getattr(endpoint, "provider", None)
+        if provider is not None:
+            pt = getattr(provider, "provider_type", None)
+            if pt:
+                return str(pt).lower()
+    except Exception:
+        pass
+
+    try:
+        if key is not None:
+            provider = getattr(key, "provider", None)
+            if provider is not None:
+                pt = getattr(provider, "provider_type", None)
+                if pt:
+                    return str(pt).lower()
+    except Exception:
+        pass
+
+    return None
+
+
 def build_provider_url(
     endpoint: ProviderEndpoint,
     *,
@@ -97,6 +135,9 @@ def build_provider_url(
         key: Provider API Key（用于 Vertex AI 等需要从密钥配置读取信息的场景）
         decrypted_auth_config: 已解密的认证配置（避免重复解密，由 get_provider_auth 提供）
     """
+    # 默认清理，避免上一次请求的 selected_base_url 泄漏到其他请求
+    set_selected_base_url(None)
+
     # 检查是否为 Vertex AI 认证类型
     auth_type = getattr(key, "auth_type", "api_key") if key else "api_key"
     if auth_type == "vertex_ai":
@@ -123,6 +164,55 @@ def build_provider_url(
     # endpoint_sig 为空时保持为空（更安全：默认路径回退到 "/"，避免误判为 claude:chat）
     endpoint_sig = normalize_endpoint_signature(endpoint_sig) if endpoint_sig else ""
 
+    provider_type = _get_provider_type(endpoint, key)
+
+    # 合并查询参数（部分逻辑需要先拿到 query_params）
+    effective_query_params = dict(query_params) if query_params else {}
+
+    # Gemini family 下清除可能存在的 key 参数（避免客户端传入的认证信息泄露到上游）
+    # 上游认证始终使用 header 方式，不使用 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]
+
+        # 存入 contextvars（供后续 Handler 层获取）
+        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
+
+    # 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，避免跨请求污染
+    set_selected_base_url(None)
+
     # 准备路径参数（Gemini chat/cli 需要 action）
     effective_path_params = dict(path_params) if path_params else {}
     if endpoint_sig.startswith("gemini:"):
@@ -166,17 +256,10 @@ def build_provider_url(
     base = _normalize_base_url(endpoint.base_url, path)  # type: ignore[arg-type]
     url = f"{base}{path}"
 
-    # 合并查询参数
-    effective_query_params = dict(query_params) if query_params else {}
-
-    # Gemini family 下清除可能存在的 key 参数（避免客户端传入的认证信息泄露到上游）
-    # 上游认证始终使用 header 方式，不使用 URL 参数
-    if endpoint_sig.startswith("gemini:"):
-        effective_query_params.pop("key", None)
-        # Gemini streamGenerateContent 官方支持 `?alt=sse` 返回 SSE（data: {...}）。
-        # 网关侧统一使用 SSE 输出，优先向上游请求 SSE 以减少解析分支；同时保留 JSON-array 兜底解析。
-        if is_stream:
-            effective_query_params.setdefault("alt", "sse")
+    # Gemini streamGenerateContent 官方支持 `?alt=sse` 返回 SSE（data: {...}）。
+    # 网关侧统一使用 SSE 输出，优先向上游请求 SSE 以减少解析分支；同时保留 JSON-array 兜底解析。
+    if endpoint_sig.startswith("gemini:") and is_stream:
+        effective_query_params.setdefault("alt", "sse")
 
     # 添加查询参数
     if effective_query_params:

src/services/task/service.py

@@ -549,6 +549,8 @@ class TaskService:
         self,
         *,
         converted_error: Any,
+        provider_type: str | None,
+        model_name: str | None,
         request_id: str | None,
         candidate_record_id: str,
         elapsed_ms: int,
@@ -585,32 +587,77 @@ class TaskService:
             )
             raise converted_error
 
-        if request_body_ref.get("_rectified", False):
+        provider_type_norm = str(provider_type or "").lower()
+
+        # Rectification may have multiple stages (Antigravity only).
+        stage_raw = request_body_ref.get("_rectify_stage", 0)
+        try:
+            stage = int(stage_raw or 0)
+        except Exception:
+            stage = 0
+        if stage <= 0 and request_body_ref.get("_rectified", False):
+            stage = 1
+
+        if stage >= 2 or (stage >= 1 and provider_type_norm != "antigravity"):
             logger.warning("  [{}] Thinking 错误：已整流仍失败，终止重试", request_id)
             self._mark_thinking_error_failed(
                 candidate_record_id,
                 converted_error,
                 elapsed_ms,
                 captured_key_concurrent,
-                {**serializable_extra_data, "rectified": True},
+                {**serializable_extra_data, "rectified": True, "rectify_stage": stage},
             )
             raise converted_error
 
         request_body = request_body_ref.get("body", {})
-        rectified_body, modified = ThinkingRectifier.rectify(request_body)
+
+        stage_label = "thinking_only"
+        next_stage = 1
+        if stage == 0:
+            rectified_body, modified = ThinkingRectifier.rectify(request_body)
+            stage_label = "thinking_only"
+            next_stage = 1
+        else:
+            # Stage 2 only applies to Antigravity.
+            rectified_body, modified = ThinkingRectifier.rectify_signature_sensitive_blocks(
+                request_body
+            )
+            stage_label = "thinking_and_tools"
+            next_stage = 2
 
         if modified:
             request_body_ref["body"] = rectified_body
             request_body_ref["_rectified"] = True
             request_body_ref["_rectified_this_turn"] = True
+            request_body_ref["_rectify_stage"] = next_stage
 
-            logger.info("  [{}] 请求已整流，在当前候选上重试", request_id)
+            if provider_type_norm == "antigravity":
+                try:
+                    from src.core.metrics import antigravity_degradation_total
+
+                    antigravity_degradation_total.labels(
+                        stage=stage_label,
+                        model=str(model_name or "unknown"),
+                    ).inc()
+                except Exception:
+                    pass
+
+            logger.info(
+                "  [{}] 请求已整流(stage={})，在当前候选上重试",
+                request_id,
+                next_stage,
+            )
             self._mark_thinking_error_failed(
                 candidate_record_id,
                 converted_error,
                 elapsed_ms,
                 captured_key_concurrent,
-                {**serializable_extra_data, "rectified": True},
+                {
+                    **serializable_extra_data,
+                    "rectified": True,
+                    "rectify_stage": next_stage,
+                    "rectify_stage_label": stage_label,
+                },
             )
             return "continue"
 
@@ -770,6 +817,8 @@ class TaskService:
             if isinstance(converted_error, ThinkingSignatureException):
                 action = self._handle_thinking_signature_error(
                     converted_error=converted_error,
+                    provider_type=str(getattr(provider, "provider_type", "") or "").lower(),
+                    model_name=str(global_model_id or ""),
                     request_id=request_id,
                     candidate_record_id=candidate_record_id,
                     elapsed_ms=elapsed_ms,