test(speech): make the speech callback tests order-independent and cover the Vertex AI await

The sync callback test now starts from a fresh callback registration, so it passes after other speech tests in the same run. The internal sub-call cache test records real callback events instead of mocking the logging object, and a Vertex AI test fails when that branch passes aspeech=False

tests/unit/caching/test_caching_handler.py

@@ -3095,9 +3095,29 @@ async def test_async_get_cache_answers_400_for_a_single_object_embedding_input()
 @pytest.mark.parametrize("internal", [False, True])
 async def test_a_cache_hit_inside_an_internal_sub_call_fires_no_callbacks_of_its_own(
     monkeypatch: pytest.MonkeyPatch, internal: bool
-):
-    cache: Final = Cache(type="local")
+) -> None:
+    from litellm.integrations.custom_logger import CustomLogger
+
+    logged: Final[list[tuple[str, object]]] = []
+
+    class Recorder(CustomLogger):
+        def log_success_event(
+            self, kwargs: dict[str, object], response_obj: object, start_time: object, end_time: object
+        ) -> None:
+            logged.append(("sync", kwargs.get("litellm_call_id")))
+
+        async def async_log_success_event(
+            self, kwargs: dict[str, object], response_obj: object, start_time: object, end_time: object
+        ) -> None:
+            logged.append(("async", kwargs.get("litellm_call_id")))
+
+    recorder: Final = Recorder()
+    cache: Final = Cache(type=LiteLLMCacheType.LOCAL)
     monkeypatch.setattr(litellm, "cache", cache)
+    monkeypatch.setattr(litellm, "callbacks", [])
+    monkeypatch.setattr(litellm, "input_callback", [])
+    monkeypatch.setattr(litellm, "success_callback", [recorder])
+    monkeypatch.setattr(litellm, "_async_success_callback", [recorder])
     kwargs: Final = {"model": "gpt-5.4", "messages": [{"role": "user", "content": "hello"}], "caching": True}
     cache.add_cache(
         result=litellm.ModelResponse(
@@ -3108,10 +3128,7 @@ async def test_a_cache_hit_inside_an_internal_sub_call_fires_no_callbacks_of_its
         **kwargs,
     )
     async_logging: Final = _build_logging_obj(CallTypes.acompletion.value, stream=False)
-    async_logging.async_success_handler = AsyncMock()
-    async_logging.handle_sync_success_callbacks_for_async_calls = MagicMock()
     sync_logging: Final = _build_logging_obj(CallTypes.completion.value, stream=False)
-    sync_logging.handle_sync_cache_hit_success_callbacks = MagicMock()
 
     with internal_sub_call() if internal else nullcontext():
         async_hit: Final = await LLMCachingHandler(
@@ -3136,8 +3153,17 @@ async def test_a_cache_hit_inside_an_internal_sub_call_fires_no_callbacks_of_its
             kwargs=kwargs,
         )
 
+    expected: Final = (
+        Counter()
+        if internal
+        else Counter({("async", async_logging.litellm_call_id): 1, ("sync", sync_logging.litellm_call_id): 1})
+    )
+    for _ in range(100):
+        if len(logged) >= expected.total():
+            break
+        await asyncio.sleep(0.05)
+    await asyncio.sleep(0.5)
+
     assert async_hit is not None and async_hit.cached_result is not None
     assert sync_hit.cached_result is not None
-    assert async_logging.async_success_handler.called is not internal
-    assert async_logging.handle_sync_success_callbacks_for_async_calls.called is not internal
-    assert sync_logging.handle_sync_cache_hit_success_callbacks.called is not internal
+    assert Counter(logged) == expected, logged

tests/unit/llms/vertex_ai/text_to_speech/test_transformation.py

@@ -1,7 +1,9 @@
+import asyncio
 import base64
 from typing import Final
 import json
 from collections.abc import Iterator, Mapping
+from concurrent.futures import ThreadPoolExecutor
 from unittest.mock import MagicMock, Mock, patch
 
 import httpx
@@ -1169,3 +1171,41 @@ async def test_aspeech_vertex_ai_gemini_tts_no_cache_request_reaches_vertex(
     await litellm.aspeech(**speech_kwargs, cache={"no-cache": True})
 
     assert vertex_route.call_count == 2
+
+
+@pytest.mark.asyncio
+async def test_aspeech_vertex_ai_gemini_tts_waits_for_a_stalled_upstream_without_holding_a_worker_thread(
+    respx_mock: respx.MockRouter, google_token_endpoint: responses.RequestsMock
+) -> None:
+    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=_GENERATE_CONTENT_RESPONSE)
+
+    respx_mock.post(
+        url__regex=r"https://aiplatform\.googleapis\.com/.*/models/gemini-3\.8-flash-tts:generateContent$"
+    ).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="vertex_ai/gemini-3.8-flash-tts",
+                input="Say hello",
+                voice="Kore",
+                vertex_credentials=_AUTHORIZED_USER,
+                vertex_location="global",
+            )
+        )
+        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 _PCM_AUDIO in response.content

tests/unit/test_main.py

@@ -4547,7 +4547,6 @@ async def test_aspeech_gemini_bridge_logs_only_the_speech_call_under_its_call_id
     _assert_logged_once_as_speech(events)
 
 
-
 @pytest.mark.asyncio
 async def test_aspeech_gemini_bridge_runs_sync_function_callbacks_once_with_the_speech_response(
     respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch
@@ -4560,6 +4559,7 @@ async def test_aspeech_gemini_bridge_runs_sync_function_callbacks_once_with_the_
     def sync_callback(kwargs: dict[str, object], response_obj: object, start_time: object, end_time: object) -> None:
         logged.append((type(response_obj), kwargs.get("litellm_call_id")))
 
+    monkeypatch.setattr(litellm.utils, "callback_list", [])
     monkeypatch.setattr(litellm, "success_callback", [sync_callback])
     respx_mock.post(
         url__regex=r"https://generativelanguage\.googleapis\.com/v1beta/models/gemini-2\.5-flash-preview-tts:generateContent.*"
@@ -5097,7 +5097,7 @@ def test_speech_gemini_tts_keeps_cached_audio_only_for_the_request_ttl(
 
 async def test_aspeech_gemini_tts_waits_for_a_stalled_upstream_without_holding_a_worker_thread(
     respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch
-):
+) -> None:
     monkeypatch.setattr(litellm, "disable_aiohttp_transport", True)
     upstream_reached: Final = asyncio.Event()
     release_upstream: Final = asyncio.Event()