test(e2e): cover google-native generateContent framing and prometheus queue time (#34650)

* test(e2e): cover google-native generateContent framing and prometheus queue time

Adds live coverage for three shipped regressions that had none, all reached
through surfaces a customer drives from Google SDKs and operator dashboards.

The managed google-native route (`/v1beta/models/{model}:generateContent`) had
no harness support at all, so EndpointsClient gains generate_content and
stream_generate_content plus the request body models, and a new suite asserts
the two contracts that broke there: the response carries
x-litellm-response-cost so SDK traffic reconciles against spend (LIT-4076), and
the stream relays single-prefixed SSE frames with no OpenAI [DONE] terminator.
A doubled `data:` prefix, a leaked bytes literal, or the [DONE] sentinel each
fail the stream test; [DONE] absence is only asserted once real content has
arrived, because a first-chunk upstream error legitimately falls back to the
OpenAI error shape and does emit it.

The prometheus test pins litellm_request_queue_time_seconds to an actual
observation on our own key's series rather than to the family merely existing,
which is the distinction the original regression turned on: the histogram stayed
registered while nothing was ever written to it (LIT-2034).

Each assertion was mutation-checked against the live proxy; inverting the
[DONE] expectation, the cost-header expectation, or the metric name fails the
corresponding test.

* refactor(e2e): simplify google native coverage

tests/e2e/coverage_registry/llm_nonconversational.yaml

@@ -37,6 +37,8 @@
 - {id: llm.rerank.cohere.basic.nonstream.works, module: llm, tier: P1, subject_endpoint: rerank, route: cohere, capability: basic, streaming: nonstream, assertions: [works], source: "test_rerank_e2e.py:29", rationale: "Cohere rerank, top_n + relevance_score"}
 - {id: llm.files.openai.content.nonstream.works, module: llm, tier: P0, subject_endpoint: files, route: openai, capability: basic, streaming: nonstream, assertions: [works], source: "test_batches_e2e.py", rationale: "GET /v1/files/{id}/content returns uploaded batch JSONL bytes"}
 - {id: llm.realtime.bedrock_converse.basic.stream.works, module: llm, tier: P0, subject_endpoint: realtime, route: bedrock_converse, capability: basic, streaming: stream, assertions: [works], source: "test_realtime_bedrock_e2e.py", rationale: "Nova Sonic realtime session emits response.done (LIT-2239)"}
+- {id: llm.google_native.gemini.basic.nonstream.cost_logged, module: llm, tier: P0, subject_endpoint: google_native, route: gemini, capability: basic, streaming: nonstream, assertions: [cost_logged], source: "LIT-4076 / proxy/google_endpoints/endpoints.py", fail_before_fix: proven, rationale: "google-native generateContent must stamp x-litellm-response-cost so SDK traffic reconciles against spend"}
+- {id: llm.google_native.gemini.basic.stream.works, module: llm, tier: P0, subject_endpoint: google_native, route: gemini, capability: basic, streaming: stream, assertions: [works], source: "PR #28213 / proxy/proxy_server.py async_data_generator", fail_before_fix: proven, rationale: "streamGenerateContent must relay single-prefixed SSE frames with no [DONE] sentinel; doubled data: prefixes and the OpenAI terminator both break the Vertex Java SDK"}
 - {id: llm.realtime.openai.basic.nonstream.works, module: llm, tier: P1, subject_endpoint: realtime, route: openai, capability: basic, streaming: nonstream, assertions: [works], source: "vendor strategy §9.19 / LIT-4778", rationale: "HTTP /v1/realtime/client_secrets returns an ephemeral credential"}
 - {id: llm.vector_stores.openai.basic.nonstream.works, module: llm, tier: P1, subject_endpoint: vector_stores, route: openai, capability: basic, streaming: nonstream, assertions: [works], source: "vendor strategy §9.17 / LIT-4778", rationale: "Vector store create/list/retrieve/delete lifecycle"}
 - {id: llm.vector_stores.openai.input_validation.nonstream.works, module: llm, tier: P1, subject_endpoint: vector_stores, route: openai, capability: input_validation, streaming: nonstream, assertions: [works], source: "vendor strategy §9.17 / LIT-4778", rationale: "Vector store search and invalid id errors"}

tests/e2e/coverage_registry/logging.yaml

@@ -6,6 +6,7 @@
 - {id: logging.datadog.stream.exports_metric, module: logging, tier: P0, event: stream, assertions: [exports_metric], exercised_on: [chat_completions, messages, responses], source: "integrations/datadog/datadog.py", rationale: "Streaming aggregates usage after the last chunk; delivery and cost must survive that path"}
 - {id: logging.datadog.failure.exports_metric, module: logging, tier: P0, event: failure, assertions: [exports_metric], exercised_on: [chat_completions], source: "integrations/datadog/datadog.py", rationale: "Failure metrics for alerting/SLO"}
 - {id: logging.prometheus.success.exports_metric, module: logging, tier: P0, event: success, assertions: [exports_metric], exercised_on: [chat_completions, messages, embeddings], source: "integrations/prometheus.py", rationale: "Standard OSS metrics; per-key cardinality (existing e2e)"}
+- {id: logging.prometheus.success.records_queue_time, module: logging, tier: P1, event: success, assertions: [records_queue_time], exercised_on: [chat_completions], source: "integrations/prometheus.py / LIT-2034", fail_before_fix: proven, rationale: "Queue time feeds saturation alerting; the family stayed registered while no observation was ever recorded, so presence alone is not the contract"}
 - {id: logging.otel.success.exports_metric, module: logging, tier: P0, event: success, assertions: [exports_metric], exercised_on: [chat_completions, messages, responses, embeddings], source: "integrations/otel/logger.py", rationale: "OTEL spans on every call path"}
 - {id: logging.otel.stream.exports_metric, module: logging, tier: P0, event: stream, assertions: [exports_metric], exercised_on: [chat_completions, messages, responses], source: "integrations/otel/logger.py", rationale: "Streaming closes the LLM span from the stream path; historically prone to duplicate/orphaned spans"}
 - {id: logging.otel.stream.records_ttft, module: logging, tier: P1, event: stream, assertions: [records_ttft], exercised_on: [chat_completions, messages, responses], source: "integrations/otel/mappers/genai.py", rationale: "TTFT is the streaming latency SLI; a zero or span-length value silently corrupts dashboards"}

tests/e2e/coverage_registry/schema.py

@@ -40,6 +40,7 @@ LlmEndpoint = Literal[
     "audio_transcriptions",
     "moderations",
     "realtime",
+    "google_native",
     "vector_stores",
     "ocr",
     "bedrock_native",

tests/e2e/llm_translation/endpoints_client.py

@@ -136,6 +136,19 @@ class ModerationRequest(BaseModel):
     input: str
 
 
+class GenerateContentPart(BaseModel):
+    text: str
+
+
+class GenerateContentContent(BaseModel):
+    role: Literal["user"] = "user"
+    parts: tuple[GenerateContentPart, ...]
+
+
+class GenerateContentBody(BaseModel):
+    contents: tuple[GenerateContentContent, ...]
+
+
 class ResponsesOutputContent(BaseModel):
     type: str | None = None
     text: str | None = None
@@ -434,6 +447,19 @@ class EndpointsClient:
             response_type=ImagesResult,
         )
 
+    def generate_content(
+        self, key: str, model: str, text: str, *, stream: bool = False
+    ) -> StreamingResponse:
+        operation = "streamGenerateContent" if stream else "generateContent"
+        return self._send(
+            f"/v1beta/models/{model}:{operation}",
+            key,
+            GenerateContentBody(
+                contents=(GenerateContentContent(parts=(GenerateContentPart(text=text),)),)
+            ),
+            stream=stream,
+        )
+
 
 def build_endpoints_client(proxy: ProxyClient) -> EndpointsClient:
     return EndpointsClient(proxy=proxy)

tests/e2e/llm_translation/test_google_native_e2e.py

@@ -0,0 +1,107 @@
+from __future__ import annotations
+
+import pytest
+from pydantic import BaseModel
+
+from e2e_config import unique_marker
+from e2e_http import StreamingResponse, require_successful_call
+from endpoints_client import EndpointsClient
+from lifecycle import ResourceManager
+from models import LiteLLMParamsBody
+
+pytestmark = pytest.mark.e2e
+
+UPSTREAM_MODEL = "gemini/gemini-2.5-flash"
+
+
+class _StreamPart(BaseModel):
+    text: str | None = None
+
+
+class _StreamContent(BaseModel):
+    parts: tuple[_StreamPart, ...] = ()
+
+
+class _StreamCandidate(BaseModel):
+    content: _StreamContent | None = None
+
+
+class _StreamEvent(BaseModel):
+    candidates: tuple[_StreamCandidate, ...] = ()
+
+
+def _managed_deployment(client: EndpointsClient, resources: ResourceManager) -> str:
+    model = f"e2e-google-native-{unique_marker()}"
+    model_id = client.create_model(
+        model,
+        LiteLLMParamsBody(model=UPSTREAM_MODEL, api_key="os.environ/GEMINI_API_KEY"),
+    )
+    resources.defer(lambda: client.delete_model(model_id))
+    return model
+
+
+def _streamed_text(result: StreamingResponse) -> str:
+    return "".join(
+        part.text
+        for event in result.stream_events
+        for candidate in _StreamEvent.model_validate_json(event).candidates
+        for part in (candidate.content.parts if candidate.content else ())
+        if part.text
+    )
+
+
+class TestGoogleNativeGenerateContent:
+    @pytest.mark.covers("llm.google_native.gemini.basic.nonstream.cost_logged")
+    def test_generate_content_returns_response_cost_header(
+        self,
+        endpoints_client: EndpointsClient,
+        resources: ResourceManager,
+        scoped_key: str,
+    ) -> None:
+        model = _managed_deployment(endpoints_client, resources)
+
+        result = endpoints_client.generate_content(
+            scoped_key, model, f"Reply with the single word ok. {unique_marker()}"
+        )
+
+        require_successful_call(result)
+        assert result.call_id, "generateContent must stamp x-litellm-call-id"
+        assert result.response_cost is not None, (
+            "generateContent returned no x-litellm-response-cost header; "
+            "google-native traffic cannot be reconciled against spend without it"
+        )
+        assert result.response_cost > 0, f"x-litellm-response-cost must be a real cost, got {result.response_cost}"
+
+    @pytest.mark.covers("llm.google_native.gemini.basic.stream.works")
+    def test_stream_generate_content_frames_sse_the_way_google_sdks_expect(
+        self,
+        endpoints_client: EndpointsClient,
+        resources: ResourceManager,
+        scoped_key: str,
+    ) -> None:
+        model = _managed_deployment(endpoints_client, resources)
+
+        result = endpoints_client.generate_content(
+            scoped_key,
+            model,
+            f"Count from one to five, one number per line. {unique_marker()}",
+            stream=True,
+        )
+
+        require_successful_call(result)
+        assert result.is_streaming, f"expected text/event-stream, got content-type {result.content_type!r}"
+        assert result.stream_error is None, f"stream carried an error: {result.stream_error}"
+        assert result.stream_events, f"stream delivered no data events (chunks={result.chunks})"
+
+        doubled = tuple(event for event in result.stream_events if event.lstrip().startswith("data:"))
+        assert not doubled, (
+            f"{len(doubled)} event(s) carry a second data: prefix, so the proxy re-wrapped "
+            f"already-framed SSE; first offender: {doubled[0][:120]!r}"
+        )
+        leaked = tuple(event for event in result.stream_events if event.startswith("b'"))
+        assert not leaked, f"event serialized as a Python bytes literal instead of text: {leaked[0][:120]!r}"
+        assert _streamed_text(result).strip(), "stream delivered events but no candidate text"
+        assert not result.stream_done, (
+            "google-native stream emitted the OpenAI [DONE] sentinel; Google never sends it "
+            "and the Vertex Java SDK rejects the stream when it appears"
+        )

tests/e2e/logging/test_prometheus_queue_time_e2e.py

@@ -0,0 +1,58 @@
+from __future__ import annotations
+
+import time
+
+import pytest
+from prometheus_client.parser import text_string_to_metric_families
+
+from e2e_config import unique_marker
+from lifecycle import ResourceManager
+from logging_client import LoggingClient
+
+pytestmark = pytest.mark.e2e
+
+DRIVER_MODEL = "gemini-2.5-flash"
+QUEUE_TIME_METRIC = "litellm_request_queue_time_seconds"
+ALIAS_LABEL = "api_key_alias"
+
+
+def _observation_count(exposition: str, alias: str) -> float | None:
+    return next(
+        (
+            sample.value
+            for family in text_string_to_metric_families(exposition)
+            for sample in family.samples
+            if sample.name == f"{QUEUE_TIME_METRIC}_count" and sample.labels.get(ALIAS_LABEL) == alias
+        ),
+        None,
+    )
+
+
+class TestPrometheusRequestQueueTime:
+    @pytest.mark.covers("logging.prometheus.success.records_queue_time")
+    def test_queue_time_histogram_records_an_observation(
+        self, client: LoggingClient, resources: ResourceManager
+    ) -> None:
+        alias = f"e2e-queue-time-{unique_marker()}"
+        key = client.key_with_alias(alias, models=[DRIVER_MODEL])
+        resources.defer(lambda: client.delete_key(key))
+
+        response = client.chat(key, DRIVER_MODEL, f"reply with one word {alias}")
+        assert response.model, f"driver call returned no model: {response}"
+
+        deadline = time.monotonic() + client.proxy.poll_timeout
+        count: float | None = None
+        while time.monotonic() < deadline:
+            count = _observation_count(client.scrape_metrics(), alias)
+            if count is not None and count > 0:
+                break
+            time.sleep(client.proxy.poll_interval)
+
+        assert count is not None, (
+            f"{QUEUE_TIME_METRIC} has no series for {ALIAS_LABEL}={alias}; the histogram was "
+            f"never observed for a request that succeeded"
+        )
+        assert count > 0, (
+            f"{QUEUE_TIME_METRIC} series for {alias} exists but recorded {count} observations; "
+            f"the metric is registered yet never written"
+        )