fix(speech): await the TTS bridge under aspeech and forward ttl and caching like chat

Under aspeech the Gemini and Vertex AI speech bridge now awaits acompletion
instead of running completion in a worker thread, so a slow gateway no longer
holds an executor thread for the whole wait. The request's top-level ttl and
the SDK's caching flag now reach the inner chat call, and the deployment's
extra_headers are passed on as is, so a non-string header fails like chat

litellm/endpoints/speech/speech_to_completion_bridge/handler.py

@@ -2,7 +2,7 @@
 Handler for transforming /chat/completions api requests to litellm.responses requests
 """
 
-from collections.abc import Mapping
+from collections.abc import Coroutine, Mapping
 from typing import TYPE_CHECKING, Final
 
 from typing_extensions import TypedDict
@@ -92,13 +92,13 @@ class SpeechToCompletionBridgeHandler:
         logging_obj: "LiteLLMLoggingObj",
         custom_llm_provider: str,
         api_base: str | None,
-        extra_headers: Mapping[str, str] | None,
+        extra_headers: Mapping[str, object] | None,
         timeout: "float | httpx.Timeout | None",
         cache_controls: "CacheControlsHint",
-    ) -> "HttpxBinaryResponseContent":
+        aspeech: bool,
+    ) -> "HttpxBinaryResponseContent | Coroutine[object, object, HttpxBinaryResponseContent]":
         received_args: Final = locals()
-        from litellm import completion
+        from litellm import acompletion, completion
-        from litellm.types.utils import ModelResponse
 
         validated_kwargs: Final = self.validate_input_kwargs(received_args)
         model = validated_kwargs["model"]
@@ -127,20 +127,40 @@ class SpeechToCompletionBridgeHandler:
             voice=voice,
         )
 
+        requested_response_format: Final = optional_params.get("response_format")
+        response_format: Final = requested_response_format if isinstance(requested_response_format, str) else None
+        if aspeech:
+            return self._aspeech(acompletion(**request_data), model, response_format, logging_obj.litellm_call_id)
+
         with internal_sub_call(logging_obj.litellm_call_id):
             result: Final = completion(
                 **request_data,
             )
+        return self._to_speech_response(result, model, response_format)
 
-        requested_response_format: Final = optional_params.get("response_format")
+    async def _aspeech(
-        if isinstance(result, ModelResponse):
+        self,
-            return self.transformation_handler.transform_response(
+        pending_completion: Coroutine[object, object, object],
-                model_response=result,
+        model: str,
-                model=model,
+        response_format: str | None,
-                response_format=requested_response_format if isinstance(requested_response_format, str) else None,
+        litellm_call_id: str | None,
-            )
+    ) -> "HttpxBinaryResponseContent":
-        else:
+        with internal_sub_call(litellm_call_id):
+            result: Final = await pending_completion
+        return self._to_speech_response(result, model, response_format)
+
+    def _to_speech_response(
+        self, result: object, model: str, response_format: str | None
+    ) -> "HttpxBinaryResponseContent":
+        from litellm.types.utils import ModelResponse
+
+        if not isinstance(result, ModelResponse):
             raise Exception(f"Unmapped response type. Got type: {type(result)}")
+        return self.transformation_handler.transform_response(
+            model_response=result,
+            model=model,
+            response_format=response_format,
+        )
 
 
 speech_to_completion_bridge_handler = SpeechToCompletionBridgeHandler()

litellm/endpoints/speech/speech_to_completion_bridge/transformation.py

@@ -135,7 +135,7 @@ class SpeechToCompletionBridgeTransformationHandler:
         headers: dict,
         litellm_logging_obj: "LiteLLMLoggingObj",
         custom_llm_provider: str,
-    ) -> dict:
+    ) -> Mapping[str, object]:
         self._validate_response_format(model, custom_llm_provider, optional_params)
         user_message: Final[ChatCompletionUserMessage] = {"role": "user", "content": input}
         return_kwargs: Final = {

litellm/main.py

@@ -9242,6 +9242,7 @@ def speech(
                 extra_headers=_DEPLOYMENT_HEADERS_HINT.validate_python(kwargs).get("extra_headers"),
                 timeout=timeout,
                 cache_controls=_CACHE_CONTROLS_HINT.validate_python(kwargs),
+                aspeech=aspeech or False,
             )
 
         # Vertex AI Text-to-Speech (Google Cloud TTS)
@@ -9297,6 +9298,7 @@ def speech(
             extra_headers=_DEPLOYMENT_HEADERS_HINT.validate_python(kwargs).get("extra_headers"),
             timeout=timeout,
             cache_controls=_CACHE_CONTROLS_HINT.validate_python(kwargs),
+            aspeech=aspeech or False,
         )
     elif custom_llm_provider == "runwayml":
         from litellm.llms.runwayml.text_to_speech.transformation import (

litellm/types/utils.py

@@ -5402,7 +5402,8 @@ class BaseUrlRoutingHint(TypedDict, total=False):
 class CacheControlsHint(TypedDict, total=False):
     cache: ReadOnly[OnErrorOmit[Mapping[str, object] | None]]
     caching: ReadOnly[OnErrorOmit[JsonValue]]
+    ttl: ReadOnly[OnErrorOmit[JsonValue]]
 
 
 class DeploymentHeadersHint(TypedDict, total=False):
-    extra_headers: ReadOnly[OnErrorOmit[Mapping[str, str] | None]]
+    extra_headers: ReadOnly[OnErrorOmit[Mapping[str, object] | None]]

tests/unit/test_main.py

@@ -37,6 +37,7 @@ from litellm import (
 from litellm import main as litellm_main
 from litellm.caching.base_cache import BaseCache
 from litellm.caching.caching import Cache
+from litellm.caching.in_memory_cache import InMemoryCache
 from litellm.constants import CONTROL_OPTIONS_KEY
 from litellm.integrations.custom_logger import CustomLogger
 from litellm.integrations.custom_prompt_management import CustomPromptManagement
@@ -6328,7 +6329,7 @@ def test_speech_gemini_tts_honors_the_request_cache_controls(
 
 @pytest.mark.parametrize(
     "malformed_controls",
-    [{"caching": "yes"}, {"caching": 1}, {"cache": {1: True}}, {"extra_headers": {"x-gateway-key": 5}}],
+    [{"caching": "yes"}, {"caching": 1}, {"cache": {1: True}}],
 )
 def test_speech_gemini_tts_without_a_cache_ignores_malformed_controls(
     respx_mock: respx.MockRouter,
@@ -6352,6 +6353,91 @@ def test_speech_gemini_tts_without_a_cache_ignores_malformed_controls(
     assert response.content.startswith(b"RIFF")
 
 
+@pytest.mark.parametrize("ttl", [60, "60"])
+def test_speech_gemini_tts_keeps_cached_audio_only_for_the_request_ttl(
+    respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch, ttl: int | str
+):
+    now: Final = [0.0]
+    cache: Final = Cache(type="local")
+    cache.cache = InMemoryCache(clock=lambda: now[0])
+    monkeypatch.setattr(litellm, "cache", cache)
+    route: Final = respx_mock.post(url__regex=GEMINI_DEFAULT_GENERATE_CONTENT_URL).mock(
+        return_value=httpx.Response(200, json=_gemini_tts_generate_content_response())
+    )
+
+    def speak() -> None:
+        litellm.speech(
+            model=f"gemini/{GEMINI_TTS_MODEL}", input="hello", voice="Kore", api_key="fake-gemini-key", ttl=ttl
+        )
+
+    speak()
+    now[0] = 30.0
+    speak()
+    assert route.call_count == 1
+    now[0] = 61.0
+    speak()
+    assert route.call_count == 2
+
+
+def test_speech_gemini_tts_fails_like_chat_on_a_non_string_deployment_header(respx_mock: respx.MockRouter):
+    route: Final = respx_mock.post(url__regex=GEMINI_DEFAULT_GENERATE_CONTENT_URL).mock(
+        return_value=httpx.Response(200, json=_gemini_tts_generate_content_response())
+    )
+    deployment_headers: Final = {"x-gateway-key": "gateway-secret", "x-org-id": 12345}
+
+    with pytest.raises(litellm.APIConnectionError) as chat_error:
+        litellm.completion(
+            model=f"gemini/{GEMINI_TTS_MODEL}",
+            messages=[{"role": "user", "content": "hello"}],
+            modalities=["audio"],
+            audio={"voice": "Kore", "format": "pcm16"},
+            api_key="fake-gemini-key",
+            extra_headers=deployment_headers,
+        )
+    with pytest.raises(litellm.APIConnectionError) as speech_error:
+        litellm.speech(
+            model=f"gemini/{GEMINI_TTS_MODEL}",
+            input="hello",
+            voice="Kore",
+            api_key="fake-gemini-key",
+            extra_headers=deployment_headers,
+        )
+
+    assert "Header value must be str or bytes" in str(chat_error.value)
+    assert "Header value must be str or bytes" in str(speech_error.value)
+    assert route.call_count == 0
+
+
+async def test_aspeech_gemini_tts_waits_for_a_stalled_upstream_without_holding_a_worker_thread(
+    respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch
+):
+    monkeypatch.setattr(litellm, "disable_aiohttp_transport", True)
+    upstream_reached: Final = asyncio.Event()
+    release_upstream: Final = asyncio.Event()
+
+    async def stalled_upstream(request: httpx.Request) -> httpx.Response:
+        upstream_reached.set()
+        await release_upstream.wait()
+        return httpx.Response(200, json=_gemini_tts_generate_content_response())
+
+    respx_mock.post(url__regex=GEMINI_DEFAULT_GENERATE_CONTENT_URL).mock(side_effect=stalled_upstream)
+    loop: Final = asyncio.get_running_loop()
+    with ThreadPoolExecutor(max_workers=1) as single_worker:
+        loop.set_default_executor(single_worker)
+        speech_task: Final = asyncio.create_task(
+            litellm.aspeech(model=f"gemini/{GEMINI_TTS_MODEL}", input="hello", voice="Kore", api_key="fake-gemini-key")
+        )
+        await asyncio.wait_for(upstream_reached.wait(), timeout=10)
+
+        worker_answer: Final = await asyncio.wait_for(loop.run_in_executor(None, lambda: "free"), timeout=10)
+
+        release_upstream.set()
+        response: Final = await asyncio.wait_for(speech_task, timeout=10)
+
+    assert worker_answer == "free"
+    assert b"pcm-audio-bytes" in response.content
+
+
 FOUNDRY_HOST: Final = "https://my-project.services.ai.azure.com"