test(translation): two-sided google differential gates over the characterization corpus

vertex_gemini / gemini / vertex_anthropic requests, responses, and streams:
snapshot == v1-at-HEAD == v2 (canonical JSON), plus a quirk corpus pinning
the 3-way structured-output fork, gemini-3 defaults and id forwarding,
multi-system parts, top_k passthrough on both routes, and the fail-closed
shapes (large cache markers, AI-Studio https media, xhigh effort, blocked
responses, mid-stream error objects).

litellm/translation/providers/google_genai/params.py

@@ -216,18 +216,17 @@ _TEMPERATURE_KEYS = (
 def sampling_entries(
     request: ChatRequest, deps: TranslationDeps, target: GoogleTarget
 ) -> dict[str, PlainJson] | TranslationError:
+    """top_k rides for BOTH targets: it is not an OpenAI param, so v1's
+    get_optional_params forwards it as a provider kwarg even on AI Studio
+    (verified in-process; the drift list's supported-params delta only gates
+    OpenAI-named params)."""
+    del deps, target
     params = request.params
     entries: dict[str, PlainJson] = {}
     for attr, key in _TEMPERATURE_KEYS:
         value = getattr(params, attr).default_value(None)
         if value is None:
             continue
-        if attr == "top_k" and target == "gemini":
-            if deps.drop_params:
-                continue
-            return TranslationError.of_unsupported(
-                "top_k on google ai studio; v1 raises UnsupportedParamsError without drop_params"
-            )
         entries = {**entries, key: value}
     max_tokens = params.max_tokens.default_value(None)
     if max_tokens is not None:

litellm/translation/providers/google_genai/serialize.py

@@ -214,11 +214,8 @@ def _json_schema_entries(
         return Ok(({**entries, "response_schema": built}, None))
     # v1 response_schema_prompt consults litellm.custom_prompt_dict; the seam
     # only routes here when that ambient dict is empty, so the default prompt
-    # applies (str(dict) formatting included).
-    prompt = """Use this JSON schema:
-    ```json
-    {}
-    ```""".format(built)
+    # applies (str(dict) formatting AND the trailing spaces included).
+    prompt = "Use this JSON schema: \n    ```json \n    {}\n    ```".format(built)
     return Ok((entries, prompt))
 
 

litellm/translation_seam_google.py

@@ -0,0 +1,302 @@
+"""Google-route adapters for translation v2 (vertex gemini, AI Studio gemini,
+vertex claude). Lives OUTSIDE litellm/translation like translation_seam.py:
+ambient litellm state (model-map capability lookups keyed per provider,
+vertex OAuth tokens, uuid/time) enters here as values; the translation
+package stays pure. Route decisions call v1's own helpers
+(``get_vertex_ai_model_route``) — never re-derived string matching.
+"""
+
+from __future__ import annotations
+
+import json
+from typing import Any, Dict, List, Optional, cast
+
+import litellm
+from litellm.llms.anthropic.common_utils import AnthropicModelInfo
+
+from litellm.translation import TranslationDeps
+from litellm.translation.ir import Body
+from litellm.translation_seam import enabled_providers
+
+GOOGLE_PROVIDER_KEYS = ("vertex_ai", "gemini", "vertex_anthropic")
+
+_THOUGHT_SIGNATURE_SEPARATOR = "__thought__"
+
+_VERTEX_RESPONSE_METADATA_FIELDS = (
+    "vertex_ai_grounding_metadata",
+    "vertex_ai_url_context_metadata",
+    "vertex_ai_safety_results",
+    "vertex_ai_citation_metadata",
+)
+
+
+def _supports_google(model: str, key: str, provider: str) -> bool:
+    if key == "supports_response_schema":
+        from litellm.utils import supports_response_schema
+
+        return supports_response_schema(model, provider)
+    from litellm.utils import _supports_factory
+
+    try:
+        return _supports_factory(model=model, custom_llm_provider=provider, key=key)
+    except Exception:
+        return False
+
+
+def _flag_google(model: str, key: str, provider: str) -> Optional[bool]:
+    candidates = (model, f"{provider}/{model}")
+    for candidate in candidates:
+        value = litellm.model_cost.get(candidate, {}).get(key)
+        if isinstance(value, bool):
+            return value
+    return None
+
+
+def _vertex_claude_candidates(model: str) -> tuple:
+    return (model, f"vertex_ai/{model}")
+
+
+def _supports_vertex_claude(model: str, key: str) -> bool:
+    return any(
+        AnthropicModelInfo._supports_model_capability(candidate, key)
+        for candidate in _vertex_claude_candidates(model)
+    )
+
+
+def _flag_vertex_claude(model: str, key: str) -> Optional[bool]:
+    for candidate in _vertex_claude_candidates(model):
+        value = AnthropicModelInfo._get_model_capability(candidate, key)
+        if value is not None:
+            return value
+    return None
+
+
+def _max_tokens_vertex_claude(model: str) -> Optional[int]:
+    for candidate in _vertex_claude_candidates(model):
+        try:
+            value = litellm.utils.get_max_tokens(candidate)
+        except Exception:
+            value = None
+        if value is not None:
+            return value
+    return None
+
+
+def _count_response_tokens(text: str) -> int:
+    from litellm.utils import token_counter
+
+    return token_counter(text=text, count_response_tokens=True)
+
+
+def build_google_deps(
+    provider_key: str, request_drop_params: Optional[bool] = None
+) -> TranslationDeps:
+    """Capability lookups resolve against the PROVIDER's model-map rows (the
+    dossier's drift item 5: supports_reasoning can disagree between the
+    vertex and gemini rows of the same model)."""
+    drop_params_global = litellm.drop_params is True
+    if provider_key == "vertex_anthropic":
+        supports = _supports_vertex_claude
+        flag = _flag_vertex_claude
+        max_tokens = _max_tokens_vertex_claude
+    else:
+
+        def supports(model: str, key: str) -> bool:
+            return _supports_google(model, key, provider_key)
+
+        def flag(model: str, key: str) -> Optional[bool]:
+            return _flag_google(model, key, provider_key)
+
+        def max_tokens(model: str) -> Optional[int]:
+            try:
+                return litellm.utils.get_max_tokens(model)
+            except Exception:
+                return None
+
+    return TranslationDeps(
+        max_tokens_for_model=max_tokens,
+        supports_capability=supports,
+        capability_flag=flag,
+        count_response_tokens=_count_response_tokens,
+        drop_params=drop_params_global or request_drop_params is True,
+        drop_params_global=drop_params_global,
+        modify_params=litellm.modify_params is True,
+    )
+
+
+def _mint_tool_call_id(raw_id: object) -> object:
+    """v1 mints ``call_<uuid4.hex[:28]>`` per functionCall without a native
+    id; the IR carries an empty prefix (optionally followed by the
+    thought-signature suffix) as the sentinel."""
+    if not isinstance(raw_id, str):
+        return raw_id
+    if raw_id == "" or raw_id.startswith(_THOUGHT_SIGNATURE_SEPARATOR):
+        import uuid
+
+        return f"call_{uuid.uuid4().hex[:28]}{raw_id}"
+    return raw_id
+
+
+def _minted_message(message: Dict[str, Any]) -> Dict[str, Any]:
+    tool_calls = message.get("tool_calls")
+    if not isinstance(tool_calls, list):
+        return message
+    minted = [
+        (
+            {**entry, "id": _mint_tool_call_id(entry.get("id"))}
+            if isinstance(entry, dict)
+            else entry
+        )
+        for entry in tool_calls
+    ]
+    return {**message, "tool_calls": minted}
+
+
+def _build_usage_gemini(payload: Dict[str, Any]):
+    """Construct ``Usage`` with v1 ``_calculate_usage``'s exact kwarg set:
+    a five-field PromptTokensDetailsWrapper and a CompletionTokensDetails
+    wrapper whose fields are only assigned when the wire reported them."""
+    from litellm.types.utils import (
+        CompletionTokensDetailsWrapper,
+        PromptTokensDetailsWrapper,
+        Usage,
+    )
+
+    prompt_details = payload.get("prompt_tokens_details") or {}
+    completion_payload = payload.get("completion_tokens_details")
+    completion_details = None
+    if isinstance(completion_payload, dict) and completion_payload:
+        completion_details = CompletionTokensDetailsWrapper()
+        for key, value in completion_payload.items():
+            setattr(completion_details, key, value)
+    return Usage(
+        prompt_tokens=payload.get("prompt_tokens"),
+        completion_tokens=payload.get("completion_tokens"),
+        total_tokens=payload.get("total_tokens"),
+        prompt_tokens_details=PromptTokensDetailsWrapper(
+            cached_tokens=prompt_details.get("cached_tokens"),
+            audio_tokens=prompt_details.get("audio_tokens"),
+            text_tokens=prompt_details.get("text_tokens"),
+            image_tokens=prompt_details.get("image_tokens"),
+            video_tokens=prompt_details.get("video_tokens"),
+        ),
+        cache_read_input_tokens=payload.get("cache_read_input_tokens"),
+        reasoning_tokens=payload.get("reasoning_tokens"),
+        completion_tokens_details=completion_details,
+    )
+
+
+def to_model_response_google(body: Body, model_response=None):
+    """Adapt a v2 gemini-dialect response body onto ModelResponse the way
+    v1's ``_transform_google_generate_content_to_openai_model_response``
+    assembles it (fresh Choices list, vertex metadata attrs, responseId)."""
+    import time
+
+    from litellm.types.utils import Choices, Message, ModelResponse
+
+    response = model_response if model_response is not None else ModelResponse()
+    choices = body.get("choices")
+    first = choices[0] if isinstance(choices, list) and choices else {}
+    message_payload = first.get("message") if isinstance(first, dict) else {}
+    finish = first.get("finish_reason") if isinstance(first, dict) else None
+    message = Message(
+        **cast(Dict[str, Any], _minted_message(cast(Dict[str, Any], message_payload)))
+    )
+    response.choices = [
+        Choices(
+            finish_reason=finish if isinstance(finish, str) else "stop",
+            index=0,
+            message=message,
+            logprobs=None,
+            enhancements=None,
+        )
+    ]
+    usage_payload = body.get("usage")
+    if isinstance(usage_payload, dict):
+        setattr(response, "usage", _build_usage_gemini(usage_payload))
+    response.created = int(time.time())
+    model = body.get("model")
+    if isinstance(model, str):
+        response.model = model
+    response_id = body.get("id")
+    if isinstance(response_id, str) and response_id:
+        response.id = response_id
+    for field in _VERTEX_RESPONSE_METADATA_FIELDS:
+        setattr(response, field, [])
+        response._hidden_params[field] = []
+    return response
+
+
+def to_model_response_stream_google(body: Body):
+    """One v2 gemini chunk body -> ModelResponseStream, mirroring the two
+    construction sites in v1 (the iterator's content chunks and the
+    wrapper-synthesized finish chunk)."""
+    from litellm.types.utils import (
+        Delta,
+        ModelResponseStream,
+        StreamingChoices,
+    )
+
+    choices_payload = cast(List[Dict[str, Any]], body.get("choices") or [{}])
+    first = choices_payload[0]
+    delta_payload = cast(Dict[str, Any], first.get("delta") or {})
+    finish = first.get("finish_reason")
+    if finish is not None:
+        chunk = ModelResponseStream(
+            id=cast(Optional[str], body.get("id")),
+            model=cast(Optional[str], body.get("model")),
+            choices=[
+                StreamingChoices(
+                    finish_reason=finish,
+                    index=0,
+                    delta=Delta(),
+                    logprobs=None,
+                    enhancements=None,
+                )
+            ],
+        )
+        return chunk
+    tool_calls = delta_payload.get("tool_calls")
+    if isinstance(tool_calls, list):
+        tool_calls = [
+            (
+                {**entry, "id": _mint_tool_call_id(entry.get("id"))}
+                if isinstance(entry, dict)
+                else entry
+            )
+            for entry in tool_calls
+        ]
+    delta = Delta(
+        content=delta_payload.get("content"),
+        reasoning_content=delta_payload.get("reasoning_content"),
+        tool_calls=tool_calls,
+        images=None,
+        function_call=None,
+        annotations=None,
+        provider_specific_fields=delta_payload.get("provider_specific_fields"),
+        role=delta_payload.get("role"),
+    )
+    chunk = ModelResponseStream(
+        id=cast(Optional[str], body.get("id")),
+        model=cast(Optional[str], body.get("model")),
+        choices=[
+            StreamingChoices(
+                finish_reason=None,
+                index=0,
+                delta=delta,
+                logprobs=None,
+                enhancements=None,
+            )
+        ],
+        system_fingerprint=None,
+    )
+    setattr(chunk, "citations", None)
+    for field in (
+        "vertex_ai_grounding_metadata",
+        "vertex_ai_url_context_metadata",
+        "vertex_ai_safety_ratings",
+        "vertex_ai_safety_results",
+        "vertex_ai_citation_metadata",
+    ):
+        setattr(chunk, field, body.get(field, []))
+    return chunk

tests/test_litellm/translation/_google_corpus.py

@@ -0,0 +1,323 @@
+"""Shared plumbing for the google differential gates (vertex gemini, AI
+Studio gemini, vertex claude).
+
+The reference corpus under ``characterization_google/`` is a verbatim copy of
+the translation characterization corpus (mateo/translation-characterization-
+providers branch). The v1 invokers reproduce that corpus's ``_seams.py``
+exactly: the gemini body builder is ``sync_transform_request_body`` (v1's
+``transform_request`` raises NotImplementedError; the wrapper is hermetic
+below the 1024-token cache minimum with the vertex token fetch stubbed), and
+the vertex claude body goes through ``VertexAIAnthropicConfig`` with the
+partner route's ``anthropic_version``/``is_vertex_request`` injection plus
+the beta-filtering step. Each differential row proves
+snapshot == v1-at-HEAD == v2.
+"""
+
+import copy
+import json
+import pathlib
+import time
+from typing import Any, Dict, List, Tuple
+
+import httpx
+
+import litellm
+from litellm.litellm_core_utils.get_litellm_params import get_litellm_params
+from litellm.litellm_core_utils.litellm_logging import Logging
+from litellm.litellm_core_utils.streaming_handler import CustomStreamWrapper
+from litellm.types.utils import LlmProviders
+from litellm.utils import ProviderConfigManager, get_optional_params
+
+CORPUS_DIR = pathlib.Path(__file__).parent / "characterization_google"
+CASES_DIR = CORPUS_DIR / "cases"
+FIXTURES_DIR = CORPUS_DIR / "fixtures"
+SNAPSHOTS_DIR = CORPUS_DIR / "snapshots"
+
+PROVIDERS: Dict[str, str] = {
+    "vertex_gemini": "vertex_ai/gemini-2.5-pro",
+    "gemini": "gemini/gemini-2.5-flash",
+    "vertex_anthropic": "vertex_ai/claude-sonnet-4@20250514",
+}
+
+# differential provider key -> translation v2 provider key
+V2_PROVIDERS: Dict[str, str] = {
+    "vertex_gemini": "vertex_ai",
+    "gemini": "gemini",
+    "vertex_anthropic": "vertex_anthropic",
+}
+
+GEMINI_API_KEY = "char-gemini-test-key"
+VERTEX_TOKEN = "char-vertex-token"
+VERTEX_PROJECT = "char-test-project"
+VERTEX_LOCATION_GEMINI = "us-central1"
+
+FROZEN_TIME = 1718064000.0
+
+
+def load_json(path: pathlib.Path) -> Any:
+    with open(path) as f:
+        return json.load(f)
+
+
+def cases() -> Dict[str, Dict[str, Any]]:
+    return {path.stem: load_json(path) for path in sorted(CASES_DIR.glob("*.json"))}
+
+
+def jsonable(obj: Any) -> Any:
+    if hasattr(obj, "model_dump"):
+        return jsonable(obj.model_dump())
+    if isinstance(obj, dict):
+        return {str(k): jsonable(v) for k, v in obj.items()}
+    if isinstance(obj, (list, tuple)):
+        return [jsonable(v) for v in obj]
+    if isinstance(obj, (str, int, float, bool)) or obj is None:
+        return obj
+    return repr(obj)
+
+
+def canonical_json(obj: Any) -> str:
+    return json.dumps(jsonable(obj), indent=2, sort_keys=True) + "\n"
+
+
+def resolve_model(model_alias: str) -> Tuple[str, str, Any]:
+    model, custom_llm_provider, _, _ = litellm.get_llm_provider(model=model_alias)
+    config = ProviderConfigManager.get_provider_chat_config(
+        model=model, provider=LlmProviders(custom_llm_provider)
+    )
+    assert config is not None
+    return model, custom_llm_provider, config
+
+
+def resolve(provider_key: str) -> Tuple[str, str, Any]:
+    return resolve_model(PROVIDERS[provider_key])
+
+
+def make_logging(model: str, messages: List[dict], stream: bool = False) -> Logging:
+    logging_obj = Logging(
+        model=model,
+        messages=messages,
+        stream=stream,
+        call_type="completion",
+        start_time=time.time(),
+        litellm_call_id="diff-litellm-call-id",
+        function_id="diff-function-id",
+    )
+    logging_obj.update_environment_variables(
+        model=model, user=None, optional_params={}, litellm_params={}
+    )
+    return logging_obj
+
+
+def _gemini_request_body(
+    model: str,
+    custom_llm_provider: str,
+    messages: List[dict],
+    optional_params: Dict[str, Any],
+    litellm_params: Dict[str, Any],
+) -> Dict[str, Any]:
+    from litellm.llms.vertex_ai.gemini.transformation import (
+        sync_transform_request_body,
+    )
+    from litellm.llms.vertex_ai.gemini.vertex_and_google_ai_studio_gemini import (
+        VertexGeminiConfig,
+        VertexLLM,
+    )
+
+    is_vertex = custom_llm_provider != "gemini"
+    gemini_api_key = None if is_vertex else GEMINI_API_KEY
+    vertex_project = VERTEX_PROJECT if is_vertex else None
+    vertex_location = VERTEX_LOCATION_GEMINI if is_vertex else None
+
+    vertex_llm = VertexLLM()
+    _auth_header, project = vertex_llm._ensure_access_token(
+        credentials=None,
+        project_id=vertex_project,
+        custom_llm_provider=custom_llm_provider,  # type: ignore[arg-type]
+    )
+    auth_header, _url = vertex_llm._get_token_and_url(
+        model=model,
+        gemini_api_key=gemini_api_key,
+        auth_header=_auth_header,
+        vertex_project=project or None,
+        vertex_location=vertex_location,
+        vertex_credentials=None,
+        stream=None,
+        custom_llm_provider=custom_llm_provider,  # type: ignore[arg-type]
+        api_base=None,
+        should_use_v1beta1_features=False,
+    )
+    VertexGeminiConfig().validate_environment(
+        api_key=auth_header,
+        headers=None,
+        model=model,
+        messages=messages,
+        optional_params=optional_params,
+        litellm_params=litellm_params,
+    )
+    return dict(
+        sync_transform_request_body(
+            gemini_api_key=gemini_api_key,
+            messages=messages,
+            api_base=None,
+            model=model,
+            client=None,
+            timeout=None,
+            extra_headers=None,
+            optional_params=optional_params,
+            logging_obj=make_logging(model, messages),
+            custom_llm_provider=custom_llm_provider,  # type: ignore[arg-type]
+            litellm_params=litellm_params,
+            vertex_project=project or None,
+            vertex_location=vertex_location,
+            vertex_auth_header=auth_header,
+        )
+    )
+
+
+def _vertex_anthropic_request_body(
+    model: str,
+    config: Any,
+    messages: List[dict],
+    optional_params: Dict[str, Any],
+    litellm_params: Dict[str, Any],
+) -> Dict[str, Any]:
+    from litellm.anthropic_beta_headers_manager import (
+        update_request_with_filtered_beta,
+    )
+    from litellm.llms.anthropic.chat.transformation import AnthropicConfig
+
+    optional_params.update(
+        {"anthropic_version": "vertex-2023-10-16", "is_vertex_request": True}
+    )
+    optional_params.pop("stream", None)
+    optional_params.pop("json_mode", None)
+    is_vertex_request = optional_params.pop("is_vertex_request", False)
+    merged_params = {**optional_params, "is_vertex_request": is_vertex_request}
+    headers = AnthropicConfig().validate_environment(
+        api_key=VERTEX_TOKEN,
+        headers={"Authorization": f"Bearer {VERTEX_TOKEN}"},
+        model=model,
+        messages=messages,
+        optional_params=merged_params,
+        litellm_params=litellm_params,
+    )
+    data = config.transform_request(
+        model=model,
+        messages=messages,
+        optional_params=merged_params,
+        litellm_params=litellm_params,
+        headers=headers,
+    )
+    _headers, data = update_request_with_filtered_beta(
+        headers=headers, request_data=data, provider="vertex_ai"
+    )
+    return data
+
+
+def run_v1_request_transform(
+    provider_key: str, case: Dict[str, Any], drop_params: bool = False
+) -> Dict[str, Any]:
+    return run_v1_request_transform_for_model(
+        PROVIDERS[provider_key], case, drop_params=drop_params
+    )
+
+
+def run_v1_request_transform_for_model(
+    model_alias: str, case: Dict[str, Any], drop_params: bool = False
+) -> Dict[str, Any]:
+    model, custom_llm_provider, config = resolve_model(model_alias)
+    messages = copy.deepcopy(case["messages"])
+    optional_params = get_optional_params(
+        model=model,
+        custom_llm_provider=custom_llm_provider,
+        messages=messages,
+        drop_params=drop_params or None,
+        **copy.deepcopy(case["params"]),
+    )
+    litellm_params = get_litellm_params(custom_llm_provider=custom_llm_provider)
+    if custom_llm_provider in ("vertex_ai", "gemini") and "claude" not in model:
+        return _gemini_request_body(
+            model, custom_llm_provider, messages, optional_params, litellm_params
+        )
+    return _vertex_anthropic_request_body(
+        model, config, messages, optional_params, litellm_params
+    )
+
+
+def run_v1_response_transform(
+    provider_key: str,
+    provider_response: Dict[str, Any],
+    messages: List[dict],
+) -> litellm.ModelResponse:
+    model, _, config = resolve(provider_key)
+    raw_response = httpx.Response(
+        status_code=200,
+        json=provider_response,
+        request=httpx.Request("POST", "https://differential.invalid/generateContent"),
+    )
+    return config.transform_response(
+        model=model,
+        raw_response=raw_response,
+        model_response=litellm.ModelResponse(),
+        logging_obj=make_logging(model, messages),
+        request_data={},
+        messages=messages,
+        optional_params={},
+        litellm_params={},
+        encoding=litellm.encoding,
+        api_key=None,
+        json_mode=None,
+    )
+
+
+def _wrap_stream(
+    model: str, custom_llm_provider: str, completion_stream: Any
+) -> List[dict]:
+    wrapper = CustomStreamWrapper(
+        completion_stream=iter(completion_stream),
+        model=model,
+        custom_llm_provider=custom_llm_provider,
+        logging_obj=make_logging(
+            model, [{"role": "user", "content": "stream"}], stream=True
+        ),
+    )
+    return [chunk.model_dump() for chunk in wrapper]
+
+
+def replay_v1_gemini_sse(provider_key: str, sse_lines: List[str]) -> List[dict]:
+    """Raw ``alt=sse`` lines through the REAL vertex ``ModelResponseIterator``
+    inside ``CustomStreamWrapper`` (which v1 tags ``vertex_ai_beta`` for both
+    google routes)."""
+    from litellm.llms.vertex_ai.gemini.vertex_and_google_ai_studio_gemini import (
+        ModelResponseIterator,
+    )
+
+    model, _, _ = resolve(provider_key)
+    iterator = ModelResponseIterator(
+        streaming_response=iter(sse_lines),
+        sync_stream=True,
+        logging_obj=make_logging(
+            model, [{"role": "user", "content": "stream"}], stream=True
+        ),
+    )
+    return _wrap_stream(model, "vertex_ai_beta", iterator)
+
+
+def replay_v1_vertex_anthropic_sse(sse_lines: List[str]) -> List[dict]:
+    from litellm.llms.anthropic.chat.handler import ModelResponseIterator
+
+    model, _, _ = resolve("vertex_anthropic")
+    iterator = ModelResponseIterator(
+        streaming_response=iter(sse_lines), sync_stream=True
+    )
+    return _wrap_stream(model, "anthropic", iterator)
+
+
+def sse_events(sse_lines: List[str]) -> List[dict]:
+    """The parsed-event seam for gemini streams: strip the SSE framing that
+    is transport plumbing in front of ``chunk_parser``."""
+    return [
+        json.loads(line[len("data: ") :])
+        for line in sse_lines
+        if line.startswith("data: ")
+    ]

tests/test_litellm/translation/conftest.py

@@ -17,6 +17,7 @@ os.environ.setdefault("LITELLM_LOCAL_MODEL_COST_MAP", "True")
 os.environ.setdefault("AWS_ACCESS_KEY_ID", "AKIADIFFTESTKEY00000")
 os.environ.setdefault("AWS_SECRET_ACCESS_KEY", "diff-test-secret")
 os.environ.setdefault("AWS_REGION_NAME", "us-east-1")
+os.environ.setdefault("GEMINI_API_KEY", "char-gemini-test-key")
 
 from litellm.llms.anthropic.common_utils import AnthropicModelInfo  # noqa: E402
 from litellm.utils import get_max_tokens, token_counter  # noqa: E402
@@ -76,3 +77,21 @@ def frozen_ambient(monkeypatch):
     monkeypatch.setattr(litellm._uuid, "uuid4", fake_uuid4)
     monkeypatch.setattr(time, "time", lambda: 1718064000.0)
     yield
+
+
+@pytest.fixture()
+def vertex_token_stub(monkeypatch):
+    """Stub the vertex credential fetch at its narrowest point
+    (``VertexBase.get_access_token``), mirroring the characterization corpus:
+    everything downstream runs real v1 code with this fixed token/project."""
+    from litellm.llms.vertex_ai.vertex_llm_base import VertexBase
+
+    monkeypatch.setattr(
+        VertexBase,
+        "get_access_token",
+        lambda self, credentials, project_id: (
+            "char-vertex-token",
+            project_id or "char-test-project",
+        ),
+    )
+    yield

tests/test_litellm/translation/test_differential_google_request.py

@@ -0,0 +1,350 @@
+"""Differential parity for the google request transforms.
+
+Two-sided gate over the characterization corpus (cases copied verbatim from
+mateo/translation-characterization-providers):
+
+1. v1-at-HEAD must still equal the committed snapshot (drift guard);
+2. v2 must equal the snapshot BYTE-FOR-BYTE (canonical JSON).
+
+A quirk corpus pins the drift list and the 3-way structured-output fork
+against v1 in-process: responseJsonSchema (2.x regex) vs responseSchema +
+propertyOrdering (model-map capability) vs schema-as-user-message, AI-Studio
+top_k dropping, gemini-3 default temperature / thinkingLevel / function-call
+id forwarding, multi-message systems, and the function_response name
+recovery. Shapes outside the proven surface must return a typed error (the
+seam falls back to v1), never a divergent body.
+"""
+
+import copy
+import json
+
+import pytest
+
+from litellm.translation import translate_chat_request
+from litellm.translation_seam_google import build_google_deps
+
+from . import _google_corpus as corpus
+
+CASES = corpus.cases()
+
+EXPECTED_FALLBACKS = {
+    "pdf_base64": "file/document parts are outside the v2 inbound surface",
+}
+
+
+def _v2_raw(provider_key: str, case: dict) -> dict:
+    model, _, _ = corpus.resolve(provider_key)
+    return {
+        "model": model,
+        "messages": copy.deepcopy(case["messages"]),
+        **copy.deepcopy(case["params"]),
+    }
+
+
+def _v2_translate(provider_key: str, raw: dict, drop_params: bool = False):
+    v2_provider = corpus.V2_PROVIDERS[provider_key]
+    deps = build_google_deps(v2_provider, request_drop_params=drop_params)
+    return translate_chat_request(raw, v2_provider, deps)
+
+
+@pytest.mark.parametrize("case_id", sorted(CASES))
+@pytest.mark.parametrize("provider_key", sorted(corpus.PROVIDERS))
+def test_v1_still_matches_snapshot(
+    provider_key: str, case_id: str, vertex_token_stub
+) -> None:
+    case = CASES[case_id]
+    if provider_key in case["skip"]:
+        pytest.skip(case["skip"][provider_key])
+    snapshot = corpus.SNAPSHOTS_DIR / "requests" / provider_key / f"{case_id}.json"
+    body = corpus.run_v1_request_transform(provider_key, case)
+    assert corpus.canonical_json(body) == snapshot.read_text(), (
+        f"v1 drifted from the characterization snapshot for {case_id}; "
+        "regenerate the corpus and ship the diff as its own PR"
+    )
+
+
+@pytest.mark.parametrize("case_id", sorted(CASES))
+@pytest.mark.parametrize("provider_key", sorted(corpus.PROVIDERS))
+def test_v2_matches_snapshot_or_falls_back(provider_key: str, case_id: str) -> None:
+    case = CASES[case_id]
+    if provider_key in case["skip"]:
+        pytest.skip(case["skip"][provider_key])
+    result = _v2_translate(provider_key, _v2_raw(provider_key, case))
+    if case_id in EXPECTED_FALLBACKS:
+        assert result.is_error(), EXPECTED_FALLBACKS[case_id]
+        return
+    assert result.is_ok(), result.error.summary
+    snapshot = corpus.SNAPSHOTS_DIR / "requests" / provider_key / f"{case_id}.json"
+    assert corpus.canonical_json(result.ok) == snapshot.read_text()
+
+
+# ---------------------------------------------------------------------------
+# google-only quirk corpus: reference is v1 in-process (the same invocation
+# as the characterization seam), asserted JSON-equal.
+# ---------------------------------------------------------------------------
+
+_USER = {"role": "user", "content": "What is the capital of France?"}
+
+_JSON_SCHEMA_RF = {
+    "type": "json_schema",
+    "json_schema": {
+        "name": "capital",
+        "strict": True,
+        "schema": {
+            "type": "object",
+            "properties": {"capital": {"type": "string"}},
+            "required": ["capital"],
+            "additionalProperties": False,
+        },
+    },
+}
+
+_TOOL_HISTORY = [
+    {"role": "user", "content": "Weather in Paris?"},
+    {
+        "role": "assistant",
+        "content": None,
+        "tool_calls": [
+            {
+                "id": "call_g3_001",
+                "type": "function",
+                "function": {
+                    "name": "get_weather",
+                    "arguments": json.dumps({"city": "Paris"}),
+                },
+            }
+        ],
+    },
+    {"role": "tool", "tool_call_id": "call_g3_001", "content": "18C"},
+]
+
+_WEATHER_TOOL = {
+    "type": "function",
+    "function": {
+        "name": "get_weather",
+        "parameters": {
+            "type": "object",
+            "properties": {"city": {"type": "string"}},
+            "required": ["city"],
+        },
+    },
+}
+
+# (alias, case, drop_params) quirks; every row references v1 in-process.
+QUIRKS = {
+    "studio_response_schema_property_ordering": (
+        # supports_response_schema=True but fails the 2.x regex -> the
+        # responseSchema + propertyOrdering tier of the 3-way fork.
+        "gemini/gemini-exp-1206",
+        {"messages": [_USER], "params": {"response_format": _JSON_SCHEMA_RF}},
+        False,
+    ),
+    "vertex_schema_prompt_for_unsupported_capability": (
+        "vertex_ai/gemini-pro-latest",
+        {"messages": [_USER], "params": {"response_format": _JSON_SCHEMA_RF}},
+        False,
+    ),
+    "studio_schema_prompt_for_unsupported_model": (
+        "gemini/gemini-1.5-flash",
+        {"messages": [_USER], "params": {"response_format": _JSON_SCHEMA_RF}},
+        False,
+    ),
+    "studio_top_k_passthrough": (
+        # top_k is not an OpenAI param; v1 forwards it on BOTH google routes.
+        "gemini/gemini-2.5-flash",
+        {"messages": [_USER], "params": {"max_tokens": 64, "top_k": 5}},
+        False,
+    ),
+    "vertex_top_k_passthrough": (
+        "vertex_ai/gemini-2.5-pro",
+        {"messages": [_USER], "params": {"max_tokens": 64, "top_k": 5}},
+        False,
+    ),
+    "multi_system_messages_two_parts": (
+        "vertex_ai/gemini-2.5-pro",
+        {
+            "messages": [
+                {"role": "system", "content": "You are terse."},
+                {"role": "system", "content": "Answer in French."},
+                _USER,
+            ],
+            "params": {"max_tokens": 64},
+        },
+        False,
+    ),
+    "system_only_blank_user_message": (
+        "vertex_ai/gemini-2.5-pro",
+        {
+            "messages": [{"role": "system", "content": "You are terse."}],
+            "params": {},
+        },
+        False,
+    ),
+    "stop_as_string": (
+        "vertex_ai/gemini-2.5-pro",
+        {"messages": [_USER], "params": {"stop": "END", "max_tokens": 32}},
+        False,
+    ),
+    "tool_choice_none_mode": (
+        "vertex_ai/gemini-2.5-pro",
+        {
+            "messages": [_USER],
+            "params": {"tools": [copy.deepcopy(_WEATHER_TOOL)], "tool_choice": "none"},
+        },
+        False,
+    ),
+    "tool_without_parameters": (
+        "vertex_ai/gemini-2.5-pro",
+        {
+            "messages": [_USER],
+            "params": {
+                "tools": [{"type": "function", "function": {"name": "ping"}}]
+            },
+        },
+        False,
+    ),
+    "parallel_tool_calls_never_reaches_wire": (
+        "vertex_ai/gemini-2.5-pro",
+        {
+            "messages": [_USER],
+            "params": {
+                "tools": [copy.deepcopy(_WEATHER_TOOL)],
+                "parallel_tool_calls": False,
+            },
+        },
+        False,
+    ),
+    "reasoning_effort_minimal_model_budget": (
+        "vertex_ai/gemini-2.5-pro",
+        {"messages": [_USER], "params": {"reasoning_effort": "minimal"}},
+        False,
+    ),
+    "thinking_budget_zero": (
+        "vertex_ai/gemini-2.5-pro",
+        {
+            "messages": [_USER],
+            "params": {"thinking": {"type": "enabled", "budget_tokens": 0}},
+        },
+        False,
+    ),
+    "image_url_format_override": (
+        "vertex_ai/gemini-2.5-pro",
+        {
+            "messages": [
+                {
+                    "role": "user",
+                    "content": [
+                        {"type": "text", "text": "Look."},
+                        {
+                            "type": "image_url",
+                            "image_url": {
+                                "url": "https://example.com/cat.png",
+                                "format": "image/webp",
+                            },
+                        },
+                    ],
+                }
+            ],
+            "params": {"max_tokens": 32},
+        },
+        False,
+    ),
+    "gemini3_default_temperature_and_level": (
+        "vertex_ai/gemini-3-pro-preview",
+        {"messages": [_USER], "params": {"reasoning_effort": "low"}},
+        False,
+    ),
+    "gemini3_studio_forwards_function_call_ids": (
+        "gemini/gemini-3-pro-preview",
+        {"messages": copy.deepcopy(_TOOL_HISTORY), "params": {"max_tokens": 64}},
+        False,
+    ),
+}
+
+
+@pytest.mark.parametrize("name", sorted(QUIRKS))
+def test_quirks_match_v1(name: str, vertex_token_stub) -> None:
+    alias, case, drop_params = QUIRKS[name]
+    v1 = corpus.run_v1_request_transform_for_model(
+        alias, copy.deepcopy(case), drop_params=drop_params
+    )
+    model, custom_llm_provider, _ = corpus.resolve_model(alias)
+    raw = {
+        "model": model,
+        "messages": copy.deepcopy(case["messages"]),
+        **copy.deepcopy(case["params"]),
+    }
+    provider_key = {"vertex_ai": "vertex_gemini", "gemini": "gemini"}[
+        custom_llm_provider
+    ]
+    result = _v2_translate(provider_key, raw, drop_params=drop_params)
+    assert result.is_ok(), result.error.summary
+    assert corpus.canonical_json(result.ok) == corpus.canonical_json(v1)
+
+
+def test_studio_https_image_falls_back() -> None:
+    raw = {
+        "model": "gemini-2.5-flash",
+        "max_tokens": 64,
+        "messages": [
+            {
+                "role": "user",
+                "content": [
+                    {"type": "text", "text": "Look."},
+                    {
+                        "type": "image_url",
+                        "image_url": {"url": "https://example.com/cat.png"},
+                    },
+                ],
+            }
+        ],
+    }
+    result = _v2_translate("gemini", raw)
+    assert result.is_error()
+    assert result.error.tag == "unsupported"
+
+
+def test_large_cache_marker_falls_back() -> None:
+    raw = {
+        "model": "gemini-2.5-pro",
+        "max_tokens": 64,
+        "messages": [
+            {
+                "role": "user",
+                "content": [
+                    {
+                        "type": "text",
+                        "text": "x" * 5000,
+                        "cache_control": {"type": "ephemeral"},
+                    }
+                ],
+            }
+        ],
+    }
+    result = _v2_translate("vertex_gemini", raw)
+    assert result.is_error()
+    assert result.error.tag == "unsupported"
+
+
+def test_reasoning_effort_xhigh_falls_back() -> None:
+    raw = {
+        "model": "gemini-2.5-pro",
+        "reasoning_effort": "xhigh",
+        "messages": [_USER],
+    }
+    result = _v2_translate("vertex_gemini", raw)
+    assert result.is_error()
+    assert result.error.tag == "unsupported"
+
+
+def test_vertex_anthropic_response_format_with_thinking_falls_back() -> None:
+    raw = {
+        "model": "claude-sonnet-4@20250514",
+        "max_tokens": 2048,
+        "thinking": {"type": "enabled", "budget_tokens": 1024},
+        "response_format": {"type": "json_object"},
+        "messages": [_USER],
+    }
+    result = _v2_translate("vertex_anthropic", raw)
+    assert result.is_error()
+    assert result.error.tag == "unsupported"

tests/test_litellm/translation/test_differential_google_response.py

@@ -0,0 +1,119 @@
+"""Differential parity for the google response transforms.
+
+Each recorded provider payload goes through v1's ``transform_response``
+(``VertexGeminiConfig`` for the gemini routes, ``VertexAIAnthropicConfig``
+for vertex claude) and v2's ``parse_response`` -> ``serialize_response`` ->
+the seam adapter; the ``ModelResponse`` dumps must be identical AND match
+the committed corpus snapshot. uuid/time are frozen (gemini mints
+``call_<uuid>`` tool ids); the gemini response id comes from the wire
+``responseId`` while the anthropic-family chatcmpl id is ambient and
+normalized.
+"""
+
+import copy
+import json
+
+import pytest
+
+from litellm.translation.engine.pipeline import (
+    _RESPONSE_PARSERS,
+    response_dialect,
+)
+from litellm.translation.inbound.openai_chat import parse_request
+from litellm.translation.inbound.openai_chat.response import serialize_response
+from litellm.translation_seam import to_model_response
+from litellm.translation_seam_google import (
+    build_google_deps,
+    to_model_response_google,
+)
+
+from . import _google_corpus as corpus
+
+_MESSAGES = [{"role": "user", "content": "What is the capital of France?"}]
+
+
+def _fixture_ids(provider_key: str) -> list:
+    return sorted(
+        path.stem
+        for path in (corpus.FIXTURES_DIR / "responses" / provider_key).glob("*.json")
+    )
+
+
+def _norm(payload: dict) -> str:
+    return json.dumps({**payload, "id": "chatcmpl-X"}, sort_keys=True, default=str)
+
+
+def _v2_model_response(provider_key: str, payload: dict) -> dict:
+    model, _, _ = corpus.resolve(provider_key)
+    v2_provider = corpus.V2_PROVIDERS[provider_key]
+    parsed = parse_request(
+        {"model": model, "max_tokens": 256, "messages": copy.deepcopy(_MESSAGES)}
+    )
+    assert parsed.is_ok(), parsed.error.summary
+    result = _RESPONSE_PARSERS[v2_provider](copy.deepcopy(payload), parsed.ok)
+    assert result.is_ok(), result.error.summary
+    dialect = response_dialect(v2_provider)
+    deps = build_google_deps(v2_provider)
+    body = serialize_response(result.ok, deps, dialect)
+    if dialect == "gemini":
+        return to_model_response_google(body).model_dump()
+    return to_model_response(body, usage_style=dialect).model_dump()
+
+
+@pytest.mark.parametrize(
+    "provider_key,fixture_id",
+    [(p, f) for p in sorted(corpus.PROVIDERS) for f in _fixture_ids(p)],
+)
+def test_v2_response_matches_v1_and_snapshot(
+    provider_key: str, fixture_id: str, frozen_ambient
+) -> None:
+    payload = corpus.load_json(
+        corpus.FIXTURES_DIR / "responses" / provider_key / f"{fixture_id}.json"
+    )
+    v1 = corpus.run_v1_response_transform(
+        provider_key, copy.deepcopy(payload), copy.deepcopy(_MESSAGES)
+    ).model_dump()
+    v2 = _v2_model_response(provider_key, payload)
+    assert _norm(v2) == _norm(v1)
+    snapshot = corpus.load_json(
+        corpus.SNAPSHOTS_DIR / "responses" / provider_key / f"{fixture_id}.json"
+    )
+    assert _norm(v2) == _norm(snapshot), (
+        f"v2/v1 drifted from the characterization snapshot for {fixture_id}; "
+        "regenerate the corpus and ship the diff as its own PR"
+    )
+
+
+def test_blocked_prompt_feedback_fails_closed() -> None:
+    model, _, _ = corpus.resolve("vertex_gemini")
+    parsed = parse_request(
+        {"model": model, "max_tokens": 64, "messages": copy.deepcopy(_MESSAGES)}
+    )
+    assert parsed.is_ok()
+    result = _RESPONSE_PARSERS["vertex_ai"](
+        {"promptFeedback": {"blockReason": "SAFETY"}, "candidates": []}, parsed.ok
+    )
+    assert result.is_error()
+    assert result.error.tag == "unsupported"
+
+
+def test_flagged_finish_reason_fails_closed() -> None:
+    model, _, _ = corpus.resolve("vertex_gemini")
+    parsed = parse_request(
+        {"model": model, "max_tokens": 64, "messages": copy.deepcopy(_MESSAGES)}
+    )
+    assert parsed.is_ok()
+    result = _RESPONSE_PARSERS["vertex_ai"](
+        {
+            "candidates": [
+                {
+                    "content": {"role": "model", "parts": [{"text": "x"}]},
+                    "finishReason": "SAFETY",
+                }
+            ],
+            "usageMetadata": {"promptTokenCount": 1, "totalTokenCount": 1},
+        },
+        parsed.ok,
+    )
+    assert result.is_error()
+    assert result.error.tag == "unsupported"

tests/test_litellm/translation/test_differential_google_stream.py

@@ -0,0 +1,156 @@
+"""Differential parity for google streaming.
+
+Gemini routes pin at the parsed-event seam: the recorded ``alt=sse`` lines
+replay through the REAL ``ModelResponseIterator`` inside
+``CustomStreamWrapper`` on the v1 side, while v2 folds the decoded
+``GenerateContentResponse`` events (SSE framing is transport plumbing). The
+fold reproduces v1's stateful bits: cumulative tool index across chunks, the
+``has_seen_tool_calls`` stop->tool_calls rewrite, the wrapper-synthesized
+trailing finish chunk, withheld usage, and thought signatures riding inside
+tool-call ids. Vertex claude streams are anthropic SSE through the anthropic
+parser (the bedrock_invoke precedent), id-normalized like the other
+anthropic-family gates; gemini chunk ids are the wire ``responseId`` and
+compare verbatim.
+"""
+
+import copy
+import json
+
+import pytest
+
+from litellm.translation.engine.stream import fold_events, fold_lines
+from litellm.translation.inbound.openai_chat import parse_request
+from litellm.translation.inbound.openai_chat.stream import initial_state
+from litellm.translation.providers.anthropic.stream import (
+    parse_sse_line,
+    reverse_names,
+)
+from litellm.translation.providers.google_genai.stream import parse_event
+from litellm.translation_seam import to_model_response_stream
+from litellm.translation_seam_google import to_model_response_stream_google
+
+from . import _google_corpus as corpus
+
+_GEMINI_PROVIDERS = ("gemini", "vertex_gemini")
+
+
+def _fixture_ids(provider_key: str) -> list:
+    return sorted(
+        path.stem
+        for path in (corpus.FIXTURES_DIR / "streams" / provider_key).glob("*.txt")
+    )
+
+
+def _read_lines(provider_key: str, fixture_id: str) -> list:
+    path = corpus.FIXTURES_DIR / "streams" / provider_key / f"{fixture_id}.txt"
+    return path.read_text().splitlines()
+
+
+def _norm(chunks: list, normalize_id: bool) -> str:
+    if normalize_id:
+        chunks = [{**chunk, "id": "chatcmpl-X"} for chunk in chunks]
+    return json.dumps(chunks, sort_keys=True, default=str)
+
+
+def _v2_gemini_chunks(provider_key: str, lines: list) -> list:
+    model, _, _ = corpus.resolve(provider_key)
+    events = corpus.sse_events(lines)
+    folded = fold_events(
+        events, parse_event, initial_state(model=model, dialect="gemini")
+    )
+    assert folded.is_ok(), folded.error.summary
+    return [to_model_response_stream_google(body).model_dump() for body in folded.ok]
+
+
+def _v2_vertex_anthropic_chunks(lines: list) -> list:
+    model, _, _ = corpus.resolve("vertex_anthropic")
+    parsed = parse_request(
+        {
+            "model": model,
+            "max_tokens": 64,
+            "messages": [{"role": "user", "content": "stream"}],
+        }
+    )
+    assert parsed.is_ok(), parsed.error.summary
+    reverse = reverse_names(parsed.ok)
+    folded = fold_lines(
+        lines,
+        lambda line: parse_sse_line(line, reverse),
+        initial_state(model=model, dialect="anthropic"),
+    )
+    assert folded.is_ok(), folded.error.summary
+    return [
+        to_model_response_stream(chunk, "chatcmpl-X").model_dump()
+        for chunk in folded.ok
+    ]
+
+
+@pytest.mark.parametrize(
+    "provider_key,fixture_id",
+    [(p, f) for p in _GEMINI_PROVIDERS for f in _fixture_ids(p)],
+)
+def test_v2_gemini_stream_matches_v1_and_snapshot(
+    provider_key: str, fixture_id: str, frozen_ambient
+) -> None:
+    lines = _read_lines(provider_key, fixture_id)
+    v1 = corpus.replay_v1_gemini_sse(provider_key, copy.deepcopy(lines))
+    v2 = _v2_gemini_chunks(provider_key, lines)
+    assert _norm(v2, False) == _norm(v1, False)
+    snapshot = corpus.load_json(
+        corpus.SNAPSHOTS_DIR / "streams" / provider_key / f"{fixture_id}.json"
+    )
+    assert _norm(v2, False) == _norm(snapshot, False), (
+        f"v2/v1 drifted from the characterization snapshot for {fixture_id}; "
+        "regenerate the corpus and ship the diff as its own PR"
+    )
+
+
+@pytest.mark.parametrize("fixture_id", _fixture_ids("vertex_anthropic"))
+def test_v2_vertex_anthropic_stream_matches_v1_and_snapshot(
+    fixture_id: str, frozen_ambient
+) -> None:
+    lines = _read_lines("vertex_anthropic", fixture_id)
+    v1 = corpus.replay_v1_vertex_anthropic_sse(copy.deepcopy(lines))
+    v2 = _v2_vertex_anthropic_chunks(lines)
+    assert _norm(v2, True) == _norm(v1, True)
+    snapshot = corpus.load_json(
+        corpus.SNAPSHOTS_DIR / "streams" / "vertex_anthropic" / f"{fixture_id}.json"
+    )
+    assert _norm(v2, True) == _norm(snapshot, True), (
+        f"v2/v1 drifted from the characterization snapshot for {fixture_id}; "
+        "regenerate the corpus and ship the diff as its own PR"
+    )
+
+
+def test_mid_stream_error_object_is_loud() -> None:
+    folded = fold_events(
+        [{"error": {"code": 429, "message": "RESOURCE_EXHAUSTED"}}],
+        parse_event,
+        initial_state(model="gemini-2.5-pro", dialect="gemini"),
+    )
+    assert folded.is_error()
+
+
+def test_finish_only_chunk_rewrites_stop_to_tool_calls() -> None:
+    events = [
+        {
+            "candidates": [
+                {
+                    "content": {
+                        "role": "model",
+                        "parts": [
+                            {"functionCall": {"name": "get_weather", "args": {}}}
+                        ],
+                    }
+                }
+            ],
+            "responseId": "r1",
+        },
+        {"candidates": [{"finishReason": "STOP"}], "responseId": "r1"},
+    ]
+    folded = fold_events(
+        events, parse_event, initial_state(model="gemini-2.5-pro", dialect="gemini")
+    )
+    assert folded.is_ok(), folded.error.summary
+    chunks = list(folded.ok)
+    assert chunks[-1]["choices"][0]["finish_reason"] == "tool_calls"