fix(e2e): recover realtime suite; fail fast on error events, defer-setup flag, pipecat pin

collect_until now fails immediately with the payload when the server sends an
error event instead of waiting out the timeout and reporting the opaque
"no 'session.created' within 20s; got ['error']" from the stage logs (LIT-4482).

The compose stack sets LITELLM_GEMINI_LIVE_DEFER_SETUP=true: the suite configures
sessions through the client's first session.update, which only reaches Gemini Live
in deferred-setup mode; without it every gemini/vertex raw-ws cell times out
waiting for session.updated.

test_pipecat_tool_smoke seeds the same tool-use instruction the raw-ws tool test
uses so the smoke no longer depends on the model spontaneously calling the tool.

pipecat is pinned to <1.5 with an import-time guard: 1.5.0 breaks the
azure/gemini/vertex realtime paths through the proxy (LIT-4511 tracks proxy-side
compat so the pin can be dropped).

tests/e2e/docker-compose.yml

@@ -135,6 +135,7 @@ services:
     environment:
       LITELLM_MASTER_KEY: sk-1234
       STORE_MODEL_IN_DB: "True"
+      LITELLM_GEMINI_LIVE_DEFER_SETUP: "true"
       DD_API_KEY: local-sink-noauth
       DD_SITE: datadoghq.com
       DD_BASE_URL: http://dd-sink:8080

tests/e2e/llm_translation/realtime/REALTIME_COVERAGE_MATRIX.md

@@ -22,7 +22,9 @@ appears). That raw-websocket tool path is the source of truth for tool calling.
 `test_pipecat_tool_smoke` is a realism layer through pipecat for openai, azure,
 and gemini only (not vertex_ai: native-audio live is flaky under pipecat tool
 calling while raw-ws tools pass; see pipecat-ai/pipecat#2544). Assertions are
-coarse; raw-ws remains authoritative. Requires `pipecat-ai`.
+coarse; raw-ws remains authoritative. Requires `pipecat-ai[openai]<1.5`:
+1.5.0 broke the azure/gemini/vertex realtime paths through the proxy
+(`pipecat_service.py` fails loudly on >=1.5).
 
 Pipecat audio coverage lives in `test_realtime_pipecat_audio_e2e.py` (VAD / audio
 I/O).
@@ -57,7 +59,12 @@ to turn its tests green.
 ## Running
 
 Start a proxy with the provider keys set in its environment (the suite registers
-the deployments itself), then
+the deployments itself). The proxy must also run with
+`LITELLM_GEMINI_LIVE_DEFER_SETUP=true` (the `tests/e2e` compose stack sets it):
+the suite configures sessions through the client's first `session.update`, and
+without deferred setup the Gemini Live bridge sends its own `setup` at connect,
+ignores that `session.update`, and never echoes `session.updated`, so every
+gemini and vertex_ai case times out. Then
 
 ```
 uv run pytest tests/e2e/llm_translation/realtime/ -v

tests/e2e/llm_translation/realtime/pipecat_service.py

@@ -9,17 +9,29 @@ three overrides from bot.py needed to talk to the proxy, the keepalive-disabling
 
 Importing this module skips the collecting test when pipecat is not installed:
 
-    uv pip install "pipecat-ai[openai]"
+    uv pip install "pipecat-ai[openai]<1.5"
+
+pipecat 1.5.0 broke the azure/gemini/vertex realtime paths through the proxy
+(openai still passes; raw-ws passes for every provider), so imports fail loudly
+on >=1.5 instead of letting the suites fail as opaque no-response timeouts.
 """
 
 # pipecat is an optional, dynamically typed dependency loaded behind importorskip,
 # so its symbols are Unknown to the type checker; relax those rules for this file.
 # pyright: reportUnknownMemberType=false, reportUnknownVariableType=false, reportUnknownArgumentType=false, reportAttributeAccessIssue=false, reportUntypedBaseClass=false, reportUnknownParameterType=false, reportMissingParameterType=false
 
+from importlib.metadata import version as _distribution_version
+
 import pytest
 
 pytest.importorskip("pipecat", reason="pipecat-ai not installed")
 
+_PIPECAT_VERSION = _distribution_version("pipecat-ai")
+assert tuple(int(part) for part in _PIPECAT_VERSION.split(".")[:2]) < (1, 5), (
+    f"pipecat-ai {_PIPECAT_VERSION} is installed, but >=1.5 breaks the "
+    'azure/gemini/vertex realtime paths; install "pipecat-ai[openai]<1.5"'
+)
+
 from pipecat.services.openai.realtime.llm import (  # noqa: E402
     OpenAIRealtimeLLMService,
 )

tests/e2e/llm_translation/realtime/realtime_client.py

@@ -307,6 +307,11 @@ class RealtimeSession:
     def collect_until(
         self, stop_type: str, *, timeout: float
     ) -> tuple[ReceivedEvent, ...]:
+        """Collect events until `stop_type` arrives. A server `error` event fails
+        immediately with the error payload instead of burning the rest of the
+        timeout: upstream failures (bad key, exhausted quota) arrive as an `error`
+        event on an otherwise-open socket, and waiting out the timeout used to
+        bury the cause in a bare "no session.created; got ['error']" (LIT-4482)."""
         deadline = time.monotonic() + timeout
         collected: list[ReceivedEvent] = []
         while time.monotonic() < deadline:
@@ -322,6 +327,10 @@ class RealtimeSession:
             collected.append(event)
             if event.type == stop_type:
                 return tuple(collected)
+            if event.type == "error":
+                raise AssertionError(
+                    f"server sent 'error' while waiting for {stop_type!r}: {event.payload}"
+                )
         raise TimeoutError(
             f"no {stop_type!r} within {timeout}s; got {[e.type for e in collected]}"
         )

tests/e2e/llm_translation/realtime/test_realtime_pipecat_e2e.py

@@ -10,7 +10,7 @@ the raw-websocket suite is the source of truth.
 The harness is synchronous, so each test stays a normal sync function and drives
 the async pipecat pipeline with asyncio.run. Skips unless pipecat is installed:
 
-    uv pip install "pipecat-ai[openai]"
+    uv pip install "pipecat-ai[openai]<1.5"
 
 Known caveat: pipecat tool calling over the realtime service has been flaky
 upstream (pipecat-ai/pipecat#2544). A failure here with the matching raw-ws tool
@@ -103,7 +103,18 @@ async def _run_pipeline(key: str, model: str) -> tuple[bool, bool]:
     )
     llm.register_function("get_weather", get_weather)
 
-    context = LLMContext(tools=WEATHER_TOOL)
+    context = LLMContext(
+        messages=[
+            {
+                "role": "system",
+                "content": (
+                    "Use the get_weather tool whenever the user asks about weather. "
+                    "After receiving the result, state the temperature."
+                ),
+            }
+        ],
+        tools=WEATHER_TOOL,
+    )
     aggregator = LLMContextAggregatorPair(context)
     capture = _CaptureText()
     task = PipelineTask(