From 930f4417246429c5ecbb64bf4b6913f0bd464958 Mon Sep 17 00:00:00 2001 From: shrey kharbanda Date: Mon, 26 Oct 2026 14:55:35 -0700 Subject: [PATCH] feat(passthrough): add Deepgram streaming text-to-speech WebSocket route Deepgram streams Aura text-to-speech over a WebSocket at /v1/speak. The proxy now relays it at /deepgram/v1/speak and /deepgram/speak with the same key auth (Bearer or a Deepgram SDK's Token scheme), provider key injection and server-configured DEEPGRAM_API_BASE as the listen route. A caller-supplied api_base is never honored The route names exactly one voice upstream, the first model in the query or aura-asteria-en when none is given, and authorizes the key against that same voice. A missing DEEPGRAM_API_KEY is closed with 1011 before the socket is accepted A session is billed at one cost-map row. An aura-2-- voice uses deepgram/aura-2, an aura-- voice uses deepgram/aura, and any other voice uses a row named after itself, deepgram/. When that row has no input_cost_per_character the socket is closed with 1008 before Deepgram is contacted, naming the row to add, so a new generation such as aura-3-x-en is refused rather than billed at another generation's rate, and adding its row is enough to enable it Spend is the relay's client text count times that rate. Only the length of each Speak message's text is counted, whether it arrives in a text or a binary frame, so Flush, Clear, Close and malformed frames add nothing, and the text never reaches the logging payload. The success handler gets one speak branch next to the listen branch, which takes its dispatcher one branch past the C901 limit, so the def carries a noqa with the reason That branch passes the dispatcher's untyped request_body and kwargs, as every branch does, so the listen channel_index check becomes a match statement to keep the type-check gate's unknown-argument count flat Refs LIT-10070 --- litellm/constants.py | 1 + litellm/llms/deepgram/common_utils.py | 77 ++++++- .../llm_passthrough_endpoints.py | 38 ++++ ...pgram_speak_passthrough_logging_handler.py | 79 +++++++ .../pass_through_endpoints/success_handler.py | 16 +- .../deepgram/test_deepgram_common_utils.py | 146 ++++++++++++ ...pgram_speak_passthrough_logging_handler.py | 174 ++++++++++++++ .../test_deepgram_ws_passthrough_routes.py | 214 +++++++++++++++++- .../test_pass_through_endpoints.py | 61 +++++ 9 files changed, 788 insertions(+), 18 deletions(-) create mode 100644 litellm/proxy/pass_through_endpoints/llm_provider_handlers/deepgram_speak_passthrough_logging_handler.py create mode 100644 tests/unit/proxy/pass_through_endpoints/llm_provider_handlers/test_deepgram_speak_passthrough_logging_handler.py diff --git a/litellm/constants.py b/litellm/constants.py index 1d5c6404bfd..796e4456bbb 100644 --- a/litellm/constants.py +++ b/litellm/constants.py @@ -423,6 +423,7 @@ WEBSOCKET_CLOSE_REASON_MAX_BYTES: Final = 123 DEEPGRAM_DEFAULT_API_BASE: Final = "https://api.deepgram.com/v1" NADIR_DEFAULT_API_BASE: Final = "https://api.getnadir.com/v1" DEEPGRAM_LISTEN_DEFAULT_MODEL: Final = "nova-3" +DEEPGRAM_SPEAK_DEFAULT_MODEL: Final = "aura-asteria-en" BEDROCK_REALTIME_PENDING_OPENING_MESSAGE_SCOPE_KEY: Final = "litellm.bedrock_realtime.pending_opening_message" BEDROCK_REALTIME_SESSION_COMMITTED_SCOPE_KEY: Final = "litellm.bedrock_realtime.session_committed" diff --git a/litellm/llms/deepgram/common_utils.py b/litellm/llms/deepgram/common_utils.py index 5638f2e0e4b..a4a413370f7 100644 --- a/litellm/llms/deepgram/common_utils.py +++ b/litellm/llms/deepgram/common_utils.py @@ -1,16 +1,18 @@ import math +import re from collections.abc import Mapping, Sequence from types import MappingProxyType -from typing import Final +from typing import Final, Literal from urllib.parse import parse_qs, urlparse import httpx +from pydantic import BaseModel, ConfigDict, ValidationError import litellm -from litellm.constants import DEEPGRAM_DEFAULT_API_BASE, DEEPGRAM_LISTEN_DEFAULT_MODEL +from litellm.constants import DEEPGRAM_DEFAULT_API_BASE, DEEPGRAM_LISTEN_DEFAULT_MODEL, DEEPGRAM_SPEAK_DEFAULT_MODEL from litellm.litellm_core_utils.url_utils import join_api_base from litellm.llms.base_llm.chat.transformation import BaseLLMException -from litellm.types.utils import LlmProviders +from litellm.types.utils import LlmProviders, ModelInfo _WEBSOCKET_SCHEMES: Final = MappingProxyType({"https": "wss", "http": "ws", "wss": "wss", "ws": "ws"}) DEEPGRAM_LISTEN_CALLBACK_PARAMS: Final = frozenset({"callback", "callback_method"}) @@ -26,6 +28,11 @@ DEEPGRAM_LISTEN_ADDON_PRICING_PARAMS: Final = MappingProxyType( "diarize_model": "diarize", } ) +# Deepgram bills every voice of an Aura generation at one rate, so each generation has one cost-map row +DEEPGRAM_SPEAK_PRICING_FAMILIES: Final = ( + (re.compile(r"aura-2-[a-z]+-[a-z]{2}"), "aura-2"), + (re.compile(r"aura-[a-z]+-[a-z]{2}"), "aura"), +) _DISABLED_PARAM_VALUES: Final = frozenset({"", "false"}) _SINGLE_VALUED_PARAMS: Final = frozenset({"model", "language"}) @@ -101,15 +108,15 @@ def deepgram_listen_registry_key(upstream_url: str) -> str: def deepgram_listen_is_priced(upstream_url: str) -> bool: """Only an exact registry hit counts: the cost calculator resolves a missing ``streaming/`` row to the pre-recorded ```` row, which is not the rate Deepgram bills a WebSocket session at.""" - return _has_exact_registry_entry(deepgram_listen_registry_key(upstream_url)) + return _exact_registry_entry(deepgram_listen_registry_key(upstream_url)) is not None -def _has_exact_registry_entry(registry_key: str) -> bool: +def _exact_registry_entry(registry_key: str) -> ModelInfo | None: try: model_info: Final = litellm.get_model_info(model=registry_key, custom_llm_provider=LlmProviders.DEEPGRAM.value) except Exception: - return False - return model_info["key"] == registry_key + return None + return model_info if model_info["key"] == registry_key else None def deepgram_listen_addon_pricing_models(upstream_url: str) -> tuple[str, ...]: @@ -132,10 +139,11 @@ def _channel_count(value: object) -> int | None: def _results_channel_count(frame: Mapping[str, object]) -> int | None: - channel_index: Final = frame.get("channel_index") - if not isinstance(channel_index, list) or len(channel_index) != 2: - return None - return _channel_count(channel_index[1]) + match frame.get("channel_index"): + case [object(), int() as channels]: + return _channel_count(channels) + case _: + return None def _declared_channel_count(upstream_url: str) -> int | None: @@ -214,3 +222,50 @@ def deepgram_listen_transcript(websocket_messages: Sequence[Mapping[str, object] if frame.get("type") == "Results" if (transcript := _final_transcript(frame)) is not None ) + + +def deepgram_speak_requested_model(query_string: str) -> str: + return _requested_model(query_string, DEEPGRAM_SPEAK_DEFAULT_MODEL) + + +def deepgram_speak_websocket_target(api_base: str | None, query_string: str) -> str: + return _deepgram_websocket_target(api_base, "speak", query_string, DEEPGRAM_SPEAK_DEFAULT_MODEL) + + +def deepgram_speak_model(upstream_url: str) -> str: + return _upstream_model(upstream_url, DEEPGRAM_SPEAK_DEFAULT_MODEL) + + +def deepgram_speak_registry_key(upstream_url: str) -> str: + """The row a speak session is billed at: its Aura generation's row, else a row named after the voice itself""" + model: Final = deepgram_speak_model(upstream_url) + pricing_model: Final = next( + (family for pattern, family in DEEPGRAM_SPEAK_PRICING_FAMILIES if pattern.fullmatch(model)), model + ) + return f"{LlmProviders.DEEPGRAM.value}/{pricing_model}" + + +def deepgram_speak_character_rate(upstream_url: str) -> float | None: + model_info: Final = _exact_registry_entry(deepgram_speak_registry_key(upstream_url)) + return None if model_info is None else model_info.get("input_cost_per_character") + + +def deepgram_speak_is_priced(upstream_url: str) -> bool: + return deepgram_speak_character_rate(upstream_url) is not None + + +class _DeepgramSpeakMessage(BaseModel): + model_config = ConfigDict(frozen=True) + + type: Literal["Speak"] + text: str + + +def deepgram_speak_characters(frame: str | bytes) -> int: + """Characters Deepgram bills for one client frame: the length of a Speak message's text, else 0. It never raises, + because the relay runs it on every frame and a traceback from a malformed frame would carry text that must not be + logged""" + try: + return len(_DeepgramSpeakMessage.model_validate_json(frame).text) + except ValidationError: + return 0 diff --git a/litellm/proxy/pass_through_endpoints/llm_passthrough_endpoints.py b/litellm/proxy/pass_through_endpoints/llm_passthrough_endpoints.py index dfbe87bd496..4e4b5de90f8 100644 --- a/litellm/proxy/pass_through_endpoints/llm_passthrough_endpoints.py +++ b/litellm/proxy/pass_through_endpoints/llm_passthrough_endpoints.py @@ -63,6 +63,11 @@ from litellm.llms.deepgram.common_utils import ( deepgram_listen_registry_key, deepgram_listen_requested_model, deepgram_listen_websocket_target, + deepgram_speak_characters, + deepgram_speak_is_priced, + deepgram_speak_registry_key, + deepgram_speak_requested_model, + deepgram_speak_websocket_target, ) from litellm.llms.fal_ai.cost_calculator import fal_ai_passthrough_cost, fal_ai_queue_base from litellm.llms.nvidia_nim.passthrough.transformation import nvidia_nim_model_group_in_path @@ -3354,6 +3359,7 @@ class _WebsocketRelay(Protocol): forward_headers: bool, endpoint: str, accept_websocket: bool, + count_client_text: Callable[[str | bytes], int] | None = None, ) -> None: ... @@ -3484,6 +3490,7 @@ async def _relay_deepgram_websocket( relay: _WebsocketRelay, target: str, refusal: str | None, + count_client_text: Callable[[str | bytes], int] | None = None, ) -> None: deepgram_api_key: Final = passthrough_endpoint_router.get_credentials( custom_llm_provider=litellm.LlmProviders.DEEPGRAM.value, @@ -3506,6 +3513,7 @@ async def _relay_deepgram_websocket( forward_headers=False, endpoint=websocket.url.path, accept_websocket=False, + count_client_text=count_client_text, ) @@ -3520,6 +3528,36 @@ async def deepgram_listen_websocket_route( await _relay_deepgram_websocket(websocket, user_api_key_dict, relay, target, _deepgram_listen_refusal(target)) +async def deepgram_speak_user_api_key_auth(websocket: WebSocket) -> UserAPIKeyAuth: + return await user_api_key_auth_websocket_for_model( + websocket, model=deepgram_speak_requested_model(websocket.url.query) + ) + + +def _deepgram_speak_refusal(target: str) -> str | None: + if deepgram_speak_is_priced(target): + return None + return _DEEPGRAM_WS_UNPRICED_REASON.format(registry_key=deepgram_speak_registry_key(target)) + + +@router.websocket("/deepgram/v1/speak") +@router.websocket("/deepgram/speak") +async def deepgram_speak_websocket_route( + websocket: WebSocket, + user_api_key_dict: Annotated[UserAPIKeyAuth, Depends(deepgram_speak_user_api_key_auth)], + relay: Annotated[_WebsocketRelay, Depends(_websocket_relay)], +) -> None: + target: Final = deepgram_speak_websocket_target(get_secret_str("DEEPGRAM_API_BASE"), websocket.url.query) + await _relay_deepgram_websocket( + websocket, + user_api_key_dict, + relay, + target, + _deepgram_speak_refusal(target), + count_client_text=deepgram_speak_characters, + ) + + class BaseOpenAIPassThroughHandler: @staticmethod async def base_openai_pass_through_handler( diff --git a/litellm/proxy/pass_through_endpoints/llm_provider_handlers/deepgram_speak_passthrough_logging_handler.py b/litellm/proxy/pass_through_endpoints/llm_provider_handlers/deepgram_speak_passthrough_logging_handler.py new file mode 100644 index 00000000000..572d183ecc2 --- /dev/null +++ b/litellm/proxy/pass_through_endpoints/llm_provider_handlers/deepgram_speak_passthrough_logging_handler.py @@ -0,0 +1,79 @@ +import json +from collections.abc import Mapping +from types import MappingProxyType +from typing import Final +from urllib.parse import urlparse + +import litellm +from litellm._logging import verbose_proxy_logger +from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj +from litellm.llms.deepgram.common_utils import ( + deepgram_speak_character_rate, + deepgram_speak_model, + deepgram_speak_registry_key, +) +from litellm.proxy._types import PassThroughEndpointLoggingTypedDict +from litellm.types.passthrough_endpoints.pass_through_endpoints import WEBSOCKET_CLIENT_TEXT_COUNT_KEY +from litellm.types.utils import StandardPassThroughResponseObject + +DEEPGRAM_SPEAK_ROUTE_SUFFIX: Final = "/speak" + + +def _billed_characters(request_body: Mapping[str, object]) -> int: + characters: Final = request_body.get(WEBSOCKET_CLIENT_TEXT_COUNT_KEY) + return characters if isinstance(characters, int) else 0 + + +def _speech_cost(upstream_url: str, characters: int) -> float | None: + rate: Final = deepgram_speak_character_rate(upstream_url) + if rate is None: + verbose_proxy_logger.warning( + "Deepgram speak passthrough: no per-character registry entry '%s'", + deepgram_speak_registry_key(upstream_url), + ) + return None + return characters * rate + + +class DeepgramSpeakPassthroughLoggingHandler: + """Bills a Deepgram speak session by the characters of Speak text the client sent. The relay hands over only the + count, never the text, so the spend row and every callback see a model, a count and a cost and nothing the caller + asked to have spoken""" + + @staticmethod + def is_deepgram_speak_route(url_route: str, request_method: str) -> bool: + path: Final = urlparse(url_route).path + return request_method == "WEBSOCKET" and "/deepgram/" in path and path.endswith(DEEPGRAM_SPEAK_ROUTE_SUFFIX) + + @staticmethod + def deepgram_speak_passthrough_handler( + request_body: Mapping[str, object], + logging_obj: LiteLLMLoggingObj, + upstream_url: str, + kwargs: Mapping[str, object] = MappingProxyType({}), + ) -> PassThroughEndpointLoggingTypedDict: + model: Final = deepgram_speak_model(upstream_url) + characters: Final = _billed_characters(request_body) + response_cost: Final = _speech_cost(upstream_url, characters) + + provider: Final = litellm.LlmProviders.DEEPGRAM.value + logging_obj.model = model # rebind-ok: the spend logger reads model and cost off the shared logging object + logging_obj.model_call_details["model"] = model # rebind-ok: same shared logging object + logging_obj.model_call_details["custom_llm_provider"] = provider # rebind-ok: same shared logging object + logging_obj.model_call_details["response_cost"] = response_cost # rebind-ok: same shared logging object + verbose_proxy_logger.debug( + "Deepgram speak passthrough cost tracking: model %s, characters %s, cost %s", + model, + characters, + response_cost, + ) + logging_result: Final[PassThroughEndpointLoggingTypedDict] = { + "result": StandardPassThroughResponseObject(response=json.dumps({"characters": characters})), + "kwargs": { + **kwargs, + "model": model, + "custom_llm_provider": provider, + "response_cost": response_cost, + }, + } + return logging_result diff --git a/litellm/proxy/pass_through_endpoints/success_handler.py b/litellm/proxy/pass_through_endpoints/success_handler.py index 9c6c3111209..8ad1a221f52 100644 --- a/litellm/proxy/pass_through_endpoints/success_handler.py +++ b/litellm/proxy/pass_through_endpoints/success_handler.py @@ -29,6 +29,9 @@ from .llm_provider_handlers.cursor_passthrough_logging_handler import ( from .llm_provider_handlers.deepgram_listen_passthrough_logging_handler import ( DeepgramListenPassthroughLoggingHandler, ) +from .llm_provider_handlers.deepgram_speak_passthrough_logging_handler import ( + DeepgramSpeakPassthroughLoggingHandler, +) from .llm_provider_handlers.fal_ai_passthrough_logging_handler import ( FalAIPassthroughLoggingHandler, ) @@ -150,7 +153,7 @@ class PassThroughEndpointLogging: _handle_logging = handle_logging - def normalize_llm_passthrough_logging_payload( + def normalize_llm_passthrough_logging_payload( # noqa: C901 # provider dispatch, one branch per provider self, httpx_response: httpx.Response, response_body: dict | list[dict[str, object]] | None, @@ -400,6 +403,17 @@ class PassThroughEndpointLogging: ) standard_logging_response_object = deepgram_handler_result["result"] # rebind-ok: elif-chain kwargs = deepgram_handler_result["kwargs"] # rebind-ok: elif-chain contract + elif DeepgramSpeakPassthroughLoggingHandler.is_deepgram_speak_route(url_route, httpx_response.request.method): + deepgram_speak_handler_result: Final = ( + DeepgramSpeakPassthroughLoggingHandler.deepgram_speak_passthrough_handler( + request_body=request_body, + logging_obj=logging_obj, + upstream_url=str(httpx_response.request.url), + kwargs=kwargs, + ) + ) + standard_logging_response_object = deepgram_speak_handler_result["result"] # rebind-ok: elif-chain + kwargs = deepgram_speak_handler_result["kwargs"] # rebind-ok: elif-chain contract elif FalAIPassthroughLoggingHandler.is_fal_ai_route(url_route, custom_llm_provider): fal_ai_handler_result: Final = FalAIPassthroughLoggingHandler().fal_ai_passthrough_handler( response_body=response_body if isinstance(response_body, dict) else MappingProxyType({}), diff --git a/tests/unit/llms/deepgram/test_deepgram_common_utils.py b/tests/unit/llms/deepgram/test_deepgram_common_utils.py index 7f480d3459f..c0ea885aebf 100644 --- a/tests/unit/llms/deepgram/test_deepgram_common_utils.py +++ b/tests/unit/llms/deepgram/test_deepgram_common_utils.py @@ -1,3 +1,4 @@ +import json import math from collections.abc import Mapping, Sequence from typing import Final @@ -18,6 +19,13 @@ from litellm.llms.deepgram.common_utils import ( deepgram_listen_requested_model, deepgram_listen_transcript, deepgram_listen_websocket_target, + deepgram_speak_character_rate, + deepgram_speak_characters, + deepgram_speak_is_priced, + deepgram_speak_model, + deepgram_speak_registry_key, + deepgram_speak_requested_model, + deepgram_speak_websocket_target, ) NOVA_3_URL: Final = "wss://api.deepgram.com/v1/listen?model=nova-3&encoding=linear16&sample_rate=16000" @@ -344,3 +352,141 @@ def test_aura_rows_bill_speech_per_character_at_the_published_rate( assert prompt_cost == pytest.approx(usd_per_thousand_characters) assert completion_cost == 0.0 + + +SPEAK_BASE: Final = "wss://api.deepgram.com/v1/speak" + + +@pytest.mark.parametrize( + ("api_base", "query_string", "expected"), + [ + pytest.param(None, "", f"{SPEAK_BASE}?model=aura-asteria-en", id="default model added"), + pytest.param(None, "model=", f"{SPEAK_BASE}?model=aura-asteria-en", id="empty model replaced"), + pytest.param( + None, + "model=aura-2-thalia-en&encoding=linear16&sample_rate=24000", + f"{SPEAK_BASE}?model=aura-2-thalia-en&encoding=linear16&sample_rate=24000", + id="query kept in order", + ), + pytest.param( + None, + "model=aura-2-thalia-en&model=aura-2-zeus-en", + f"{SPEAK_BASE}?model=aura-2-thalia-en", + id="only the first model", + ), + pytest.param( + "http://127.0.0.1:8080/v1", "model=aura-2-thalia-en", "ws://127.0.0.1:8080/v1/speak?model=aura-2-thalia-en" + ), + pytest.param( + "https://dg.internal/v1/", + "encoding=mulaw", + "wss://dg.internal/v1/speak?encoding=mulaw&model=aura-asteria-en", + ), + ], +) +def test_deepgram_speak_websocket_target(api_base: str | None, query_string: str, expected: str): + assert deepgram_speak_websocket_target(api_base=api_base, query_string=query_string) == expected + + +@pytest.mark.parametrize( + "query_string", + ["", "model=aura-2-thalia-en", "model=aura-2-thalia-en&model=aura-2-zeus-en", "encoding=mulaw&model=aura-luna-en"], +) +def test_the_authorized_speak_model_is_the_only_model_the_upstream_target_carries(query_string: str): + """The key is checked against deepgram_speak_requested_model, so the upstream must use exactly that voice""" + target: Final = deepgram_speak_websocket_target(api_base=None, query_string=query_string) + + assert parse_qs(urlparse(target).query)["model"] == [deepgram_speak_requested_model(query_string)] + assert deepgram_speak_model(target) == deepgram_speak_requested_model(query_string) + + +@pytest.mark.parametrize( + ("model", "expected_family"), + [ + # Voice names from https://developers.deepgram.com/docs/tts-models, checked 2026-10-26 + pytest.param("aura-2-thalia-en", "aura-2", id="aura-2 english"), + pytest.param("aura-2-celeste-es", "aura-2", id="aura-2 spanish"), + pytest.param("aura-2-rhea-nl", "aura-2", id="aura-2 dutch"), + pytest.param("aura-2-agathe-fr", "aura-2", id="aura-2 french"), + pytest.param("aura-2-viktoria-de", "aura-2", id="aura-2 german"), + pytest.param("aura-2-livia-it", "aura-2", id="aura-2 italian"), + pytest.param("aura-2-izanami-ja", "aura-2", id="aura-2 japanese"), + pytest.param("aura-asteria-en", "aura", id="aura-1"), + pytest.param("aura-zeus-en", "aura", id="another aura-1"), + pytest.param("aura-3-foo-en", "aura-3-foo-en", id="unknown aura generation"), + pytest.param("aura-2", "aura-2", id="bare aura-2"), + pytest.param("aura-x", "aura-x", id="aura without a language"), + pytest.param("aura-2-thalia-en-extra", "aura-2-thalia-en-extra", id="trailing part"), + pytest.param("flux-tts-en", "flux-tts-en", id="other family"), + pytest.param("nova-3", "nova-3", id="listen model"), + ], +) +def test_deepgram_speak_registry_key_names_the_aura_generation_row_else_the_voice(model: str, expected_family: str): + target: Final = deepgram_speak_websocket_target(api_base=None, query_string=f"model={model}") + + assert deepgram_speak_registry_key(target) == f"deepgram/{expected_family}" + + +@pytest.mark.usefixtures("local_model_cost_map") +@pytest.mark.parametrize( + ("model", "expected"), + [ + pytest.param("aura-2-thalia-en", True, id="aura-2"), + pytest.param("aura-asteria-en", True, id="aura-1"), + pytest.param("aura-3-foo-en", False, id="unknown aura generation"), + pytest.param("flux-tts-en", False, id="unpriced family"), + pytest.param("nova-3", False, id="a listen model has no speech row"), + ], +) +def test_deepgram_speak_is_priced(model: str, expected: bool): + assert deepgram_speak_is_priced(f"{SPEAK_BASE}?model={model}") is expected + + +@pytest.mark.usefixtures("local_model_cost_map") +def test_deepgram_speak_is_not_priced_once_its_family_row_is_gone(monkeypatch: pytest.MonkeyPatch): + monkeypatch.delitem(litellm.model_cost, "deepgram/aura-2") + + assert deepgram_speak_is_priced(f"{SPEAK_BASE}?model=aura-2-thalia-en") is False + + +@pytest.mark.usefixtures("local_model_cost_map") +def test_a_cost_map_row_named_after_a_new_voice_prices_it_at_that_rate(monkeypatch: pytest.MonkeyPatch): + """The 1008 refusal tells the operator to add the voice's row, so adding it must be enough""" + url: Final = f"{SPEAK_BASE}?model=aura-3-foo-en" + monkeypatch.setitem( + litellm.model_cost, + "deepgram/aura-3-foo-en", + {"input_cost_per_character": 4e-05, "litellm_provider": "deepgram", "mode": "audio_speech"}, + ) + + assert deepgram_speak_character_rate(url) == 4e-05 + assert deepgram_speak_is_priced(url) is True + + +@pytest.mark.parametrize( + ("frame", "expected"), + [ + pytest.param( + json.dumps({"type": "Speak", "text": "The quick brown fox jumps over the lazy dog."}), 44, id="speak" + ), + pytest.param(json.dumps({"type": "Speak", "text": "Gr\u00fc\u00dfe \U0001f44b"}), 7, id="code points"), + pytest.param( + json.dumps({"type": "Speak", "text": "Sent as a binary frame."}).encode(), 23, id="speak in a binary frame" + ), + pytest.param(b"\xff\xfe\x00", 0, id="binary frame that is not utf-8"), + pytest.param(json.dumps({"type": "Speak", "text": ""}), 0, id="empty text"), + pytest.param(json.dumps({"type": "Flush"}), 0, id="flush"), + pytest.param(json.dumps({"type": "Clear"}), 0, id="clear"), + pytest.param(json.dumps({"type": "Close"}), 0, id="close"), + pytest.param(json.dumps({"type": "Speak"}), 0, id="speak without text"), + pytest.param(json.dumps({"type": "Speak", "text": 12345}), 0, id="text not a string"), + pytest.param(json.dumps({"type": "Speak", "text": ["a", "b"]}), 0, id="text a list"), + pytest.param(json.dumps([{"type": "Speak", "text": "hello"}]), 0, id="not an object"), + pytest.param("hello there", 0, id="not json"), + pytest.param("", 0, id="empty frame"), + pytest.param("[" * 100_000, 0, id="nesting deep enough to recurse"), + pytest.param("1" * 5_000, 0, id="integer longer than the parser allows"), + ], +) +def test_deepgram_speak_characters_counts_speak_text_and_never_raises(frame: str | bytes, expected: int): + assert deepgram_speak_characters(frame) == expected diff --git a/tests/unit/proxy/pass_through_endpoints/llm_provider_handlers/test_deepgram_speak_passthrough_logging_handler.py b/tests/unit/proxy/pass_through_endpoints/llm_provider_handlers/test_deepgram_speak_passthrough_logging_handler.py new file mode 100644 index 00000000000..96247df28f1 --- /dev/null +++ b/tests/unit/proxy/pass_through_endpoints/llm_provider_handlers/test_deepgram_speak_passthrough_logging_handler.py @@ -0,0 +1,174 @@ +"""Deepgram ``/v1/speak`` WebSocket passthrough: per-character cost tracking from the relay's client text count.""" + +import json +from datetime import datetime +from types import SimpleNamespace +from typing import Final + +import pytest + +import litellm +from litellm.integrations.custom_logger import CustomLogger +from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj +from litellm.proxy.pass_through_endpoints.llm_provider_handlers.deepgram_speak_passthrough_logging_handler import ( + DeepgramSpeakPassthroughLoggingHandler, +) +from litellm.proxy.pass_through_endpoints.success_handler import PassThroughEndpointLogging +from litellm.types.passthrough_endpoints.pass_through_endpoints import ( + WEBSOCKET_CLIENT_TEXT_COUNT_KEY, + PassthroughStandardLoggingPayload, +) +from litellm.types.utils import StandardLoggingPayload + +AURA_2_URL: Final = "wss://api.deepgram.com/v1/speak?model=aura-2-thalia-en&encoding=linear16&sample_rate=24000" +AURA_1_URL: Final = "wss://api.deepgram.com/v1/speak?model=aura-asteria-en" +SPOKEN_CHARACTERS: Final = len("The quick brown fox jumps over the lazy dog.") +# Pay as you go rates from https://deepgram.com/pricing, checked 2026-10-26 +AURA_2_USD_PER_CHARACTER: Final = 0.030 / 1000 +AURA_1_USD_PER_CHARACTER: Final = 0.0150 / 1000 + +pytestmark: Final = pytest.mark.usefixtures("local_model_cost_map") + + +@pytest.mark.parametrize( + ("url_route", "request_method", "expected"), + [ + ("/deepgram/v1/speak", "WEBSOCKET", True), + ("/deepgram/speak", "WEBSOCKET", True), + ("/deepgram/v1/speak?model=aura-2-thalia-en", "WEBSOCKET", True), + ("/deepgram/v1/listen", "WEBSOCKET", False), + ("/deepgram/v1/speak/extra", "WEBSOCKET", False), + ("/openai/v1/audio/speech", "WEBSOCKET", False), + ("", "WEBSOCKET", False), + pytest.param("https://gateway.internal/deepgram/v1/speak", "POST", False, id="http passthrough to a speak url"), + ], +) +def test_is_deepgram_speak_route(url_route: str, request_method: str, expected: bool): + assert DeepgramSpeakPassthroughLoggingHandler.is_deepgram_speak_route(url_route, request_method) is expected + + +def _logging_obj(call_id: str = "call-dg-speak") -> LiteLLMLoggingObj: + return LiteLLMLoggingObj( + model="unknown", + messages=[{"role": "user", "content": "WebSocket connection"}], + stream=True, + call_type="pass_through_endpoint", + start_time=datetime.now(), + litellm_call_id=call_id, + function_id="websocket_passthrough", + ) + + +def _handler_cost(upstream_url: str, characters: object) -> float | None: + handler_result: Final = DeepgramSpeakPassthroughLoggingHandler.deepgram_speak_passthrough_handler( + request_body={WEBSOCKET_CLIENT_TEXT_COUNT_KEY: characters}, + logging_obj=_logging_obj(), + upstream_url=upstream_url, + ) + response_cost: Final = handler_result["kwargs"]["response_cost"] + assert response_cost is None or isinstance(response_cost, float) + return response_cost + + +def test_handler_bills_spoken_characters_at_the_aura_2_rate_and_names_the_voice(): + logging_obj: Final = _logging_obj() + + handler_result: Final = DeepgramSpeakPassthroughLoggingHandler.deepgram_speak_passthrough_handler( + request_body={WEBSOCKET_CLIENT_TEXT_COUNT_KEY: SPOKEN_CHARACTERS}, + logging_obj=logging_obj, + upstream_url=AURA_2_URL, + kwargs={"litellm_params": {"metadata": {}}}, + ) + + assert handler_result["kwargs"]["response_cost"] == pytest.approx(SPOKEN_CHARACTERS * AURA_2_USD_PER_CHARACTER) + assert handler_result["kwargs"]["model"] == "aura-2-thalia-en" + assert handler_result["kwargs"]["custom_llm_provider"] == "deepgram" + assert handler_result["kwargs"]["litellm_params"] == {"metadata": {}} + assert handler_result["result"] == {"response": json.dumps({"characters": SPOKEN_CHARACTERS})} + assert logging_obj.model == "aura-2-thalia-en" + assert logging_obj.model_call_details["response_cost"] == pytest.approx( + SPOKEN_CHARACTERS * AURA_2_USD_PER_CHARACTER + ) + + +def test_handler_bills_an_aura_1_voice_at_the_aura_1_rate(): + assert _handler_cost(AURA_1_URL, 1000) == pytest.approx(1000 * AURA_1_USD_PER_CHARACTER) + + +@pytest.mark.parametrize( + "characters", + [pytest.param(0, id="nothing spoken"), pytest.param(None, id="no count")], +) +def test_handler_bills_zero_for_a_session_without_speak_text(characters: object): + assert _handler_cost(AURA_2_URL, characters) == 0.0 + + +@pytest.mark.parametrize("model", ["flux-tts-en", "aura-3-x-en", "aura-x", "nova-3"]) +def test_handler_reports_no_cost_rather_than_a_wrong_one_for_an_unpriced_voice(model: str): + assert _handler_cost(f"wss://api.deepgram.com/v1/speak?model={model}", 1000) is None + + +class _CapturingLogger(CustomLogger): + def __init__(self) -> None: + super().__init__() + self.payloads: list[StandardLoggingPayload] = [] # mutable-ok: the logger appends each payload it receives + + async def async_log_success_event(self, kwargs, response_obj, start_time, end_time): + self.payloads.append(kwargs["standard_logging_object"]) + + +@pytest.mark.parametrize( + ("characters", "expected_cost"), + [ + pytest.param(SPOKEN_CHARACTERS, SPOKEN_CHARACTERS * AURA_2_USD_PER_CHARACTER, id="one speak"), + pytest.param(0, 0.0, id="zero stays zero and is not recomputed"), + ], +) +async def test_success_handler_dispatches_deepgram_speak_and_logs_character_based_spend( + monkeypatch: pytest.MonkeyPatch, characters: int, expected_cost: float +): + """Drives the shared passthrough success handler the way the relay does at socket close and reads what a spend + logger receives""" + capturing_logger: Final = _CapturingLogger() + monkeypatch.setattr(litellm, "_async_success_callback", [capturing_logger]) + monkeypatch.setattr(litellm, "success_callback", []) + monkeypatch.setattr(litellm, "callbacks", []) + logging_obj: Final = _logging_obj("call-dg-speak-e2e") + litellm_params: Final = {"metadata": {"user_api_key": "hashed-key", "user_api_key_team_id": "team-tts"}} + logging_obj.update_environment_variables( + model="unknown", + user="unknown", + optional_params={}, + litellm_params=litellm_params, + call_type="pass_through_endpoint", + ) + + await PassThroughEndpointLogging().pass_through_async_success_handler( + httpx_response=SimpleNamespace( + status_code=200, + text="WebSocket connection successful", + headers={}, + request=SimpleNamespace(method="WEBSOCKET", url=AURA_2_URL), + ), + response_body=[{"type": "Metadata", "request_id": "req-1"}, {"type": "Flushed", "sequence_id": 0}], + logging_obj=logging_obj, + url_route="/deepgram/v1/speak", + result="websocket_connection_successful", + start_time=datetime.now(), + end_time=datetime.now(), + cache_hit=False, + request_body={WEBSOCKET_CLIENT_TEXT_COUNT_KEY: characters}, + passthrough_logging_payload=PassthroughStandardLoggingPayload( + url=AURA_2_URL, request_body={}, request_method="WEBSOCKET", cost_per_request=None + ), + litellm_params=litellm_params, + ) + + assert len(capturing_logger.payloads) == 1 + payload: Final = capturing_logger.payloads[0] + assert payload["model"] == "aura-2-thalia-en" + assert payload["custom_llm_provider"] == "deepgram" + assert payload["response_cost"] == pytest.approx(expected_cost) + assert payload["response_cost"] is not None + assert payload["metadata"]["user_api_key_team_id"] == "team-tts" + assert payload["id"] == "call-dg-speak-e2e" diff --git a/tests/unit/proxy/pass_through_endpoints/test_deepgram_ws_passthrough_routes.py b/tests/unit/proxy/pass_through_endpoints/test_deepgram_ws_passthrough_routes.py index 404ed0f5800..69106acb7e1 100644 --- a/tests/unit/proxy/pass_through_endpoints/test_deepgram_ws_passthrough_routes.py +++ b/tests/unit/proxy/pass_through_endpoints/test_deepgram_ws_passthrough_routes.py @@ -1,7 +1,8 @@ -"""Deepgram ``/v1/listen`` passthrough WebSocket route: registration, auth, credential injection, target URL.""" +"""Deepgram ``/v1/listen`` and ``/v1/speak`` passthrough WebSocket routes: registration, auth, credential injection, +target URL.""" import asyncio -from collections.abc import Mapping +from collections.abc import Callable, Mapping from dataclasses import dataclass from types import MappingProxyType, SimpleNamespace from typing import Final @@ -15,12 +16,14 @@ from starlette.websockets import WebSocketDisconnect import litellm from litellm.caching.dual_cache import DualCache +from litellm.llms.deepgram.common_utils import deepgram_speak_characters from litellm.proxy._lazy_features import LAZY_FEATURES from litellm.proxy._types import LiteLLMRoutes, UserAPIKeyAuth from litellm.proxy.auth.auth_checks import cache_key_object from litellm.proxy.pass_through_endpoints.llm_passthrough_endpoints import ( _websocket_relay, deepgram_listen_websocket_route, + deepgram_speak_websocket_route, router, ) from litellm.proxy.utils import hash_token @@ -30,6 +33,7 @@ GET_CREDENTIALS: Final = ( ) USER_API_KEY_AUTH: Final = "litellm.proxy.auth.user_api_key_auth.user_api_key_auth" LISTEN_PATHS: Final = ("/deepgram/v1/listen", "/deepgram/listen") +SPEAK_PATHS: Final = ("/deepgram/v1/speak", "/deepgram/speak") NOVA_2_STREAMING_KEY: Final = "deepgram/streaming/nova-2" pytestmark: Final = pytest.mark.usefixtures("local_model_cost_map") @@ -62,6 +66,7 @@ class _RelayCall: forward_headers: bool endpoint: str accept_websocket: bool + count_client_text: Callable[[str | bytes], int] | None = None class _FakeRelay: @@ -78,6 +83,7 @@ class _FakeRelay: forward_headers: bool, endpoint: str, accept_websocket: bool, + count_client_text: Callable[[str | bytes], int] | None = None, ) -> None: self.calls.append( _RelayCall( @@ -87,6 +93,7 @@ class _FakeRelay: forward_headers=forward_headers, endpoint=endpoint, accept_websocket=accept_websocket, + count_client_text=count_client_text, ) ) @@ -101,13 +108,13 @@ async def _serve(websocket: _FakeWebSocket, user_api_key_dict: UserAPIKeyAuth | return relay -def test_deepgram_listen_websocket_routes_registered(): +def test_deepgram_websocket_routes_registered(): ws_paths = {route.path for route in router.routes if isinstance(route, WebSocketRoute)} - assert set(LISTEN_PATHS) <= ws_paths + assert {*LISTEN_PATHS, *SPEAK_PATHS} <= ws_paths -@pytest.mark.parametrize("path", LISTEN_PATHS) -def test_deepgram_listen_is_a_lazily_loaded_mapped_pass_through_route(path): +@pytest.mark.parametrize("path", (*LISTEN_PATHS, *SPEAK_PATHS)) +def test_deepgram_websocket_routes_are_lazily_loaded_mapped_pass_through_routes(path): """The route must be reachable before the passthrough module is imported and must be authed and billed as a mapped pass-through route like the other provider prefixes.""" feature = next(feature for feature in LAZY_FEATURES if feature.name == "llm_passthrough") @@ -472,3 +479,198 @@ def test_deepgram_listen_echoes_the_browser_subprotocol_that_carries_the_litellm MappingProxyType({"Authorization": "Token dg-provider-key"}) ] assert [call.forward_headers for call in relay.calls] == [False] + + +async def _serve_speak(websocket: _FakeWebSocket, user_api_key_dict: UserAPIKeyAuth | None = None) -> _FakeRelay: + relay: Final = _FakeRelay() + await deepgram_speak_websocket_route( + websocket=websocket, + user_api_key_dict=user_api_key_dict or UserAPIKeyAuth(), + relay=relay, + ) + return relay + + +@pytest.mark.parametrize("path", SPEAK_PATHS) +async def test_deepgram_speak_forwards_query_injects_only_provider_auth_and_counts_characters(path, monkeypatch): + monkeypatch.delenv("DEEPGRAM_API_BASE", raising=False) + websocket: Final = _FakeWebSocket(path, "model=aura-2-thalia-en&encoding=linear16&sample_rate=24000") + caller: Final = UserAPIKeyAuth(api_key="sk-litellm-virtual", team_id="team-tts") + + with patch(GET_CREDENTIALS, return_value="dg-provider-key") as get_credentials: + relay: Final = await _serve_speak(websocket, caller) + + assert get_credentials.call_args.kwargs == {"custom_llm_provider": "deepgram", "region_name": None} + assert relay.calls == [ + _RelayCall( + target="wss://api.deepgram.com/v1/speak?model=aura-2-thalia-en&encoding=linear16&sample_rate=24000", + custom_headers=MappingProxyType({"Authorization": "Token dg-provider-key"}), + user_api_key_dict=caller, + forward_headers=False, + endpoint=path, + accept_websocket=False, + count_client_text=deepgram_speak_characters, + ) + ] + assert websocket.accepts == [None] + assert websocket.closed is None + + +async def test_deepgram_listen_does_not_count_client_text(monkeypatch): + """Listen bills audio seconds from upstream frames, so it must keep handing the relay no counter""" + monkeypatch.delenv("DEEPGRAM_API_BASE", raising=False) + + with patch(GET_CREDENTIALS, return_value="dg-provider-key"): + relay: Final = await _serve(_FakeWebSocket("/deepgram/v1/listen", "model=nova-3")) + + assert [call.count_client_text for call in relay.calls] == [None] + + +@pytest.mark.parametrize( + ("query", "expected_target"), + [ + pytest.param("", "wss://api.deepgram.com/v1/speak?model=aura-asteria-en", id="no query"), + pytest.param("model=", "wss://api.deepgram.com/v1/speak?model=aura-asteria-en", id="blank model"), + pytest.param( + "encoding=mulaw&sample_rate=8000", + "wss://api.deepgram.com/v1/speak?encoding=mulaw&sample_rate=8000&model=aura-asteria-en", + id="other params kept", + ), + pytest.param( + "model=aura-2-thalia-en&model=aura-asteria-en", + "wss://api.deepgram.com/v1/speak?model=aura-2-thalia-en", + id="only the first model", + ), + ], +) +async def test_deepgram_speak_names_exactly_one_model_upstream(query, expected_target, monkeypatch): + monkeypatch.delenv("DEEPGRAM_API_BASE", raising=False) + + with patch(GET_CREDENTIALS, return_value="dg-provider-key"): + relay: Final = await _serve_speak(_FakeWebSocket("/deepgram/speak", query)) + + assert [call.target for call in relay.calls] == [expected_target] + + +async def test_deepgram_speak_honours_server_configured_api_base_and_ignores_the_callers(monkeypatch): + """The server's Deepgram key must only ever go to the server-configured host""" + monkeypatch.setenv("DEEPGRAM_API_BASE", "https://api.eu.deepgram.com/v1/") + websocket: Final = _FakeWebSocket( + "/deepgram/v1/speak", "api_base=wss%3A%2F%2Fattacker.example%2Fv1&model=aura-2-thalia-en" + ) + + with patch(GET_CREDENTIALS, return_value="dg-provider-key"): + relay: Final = await _serve_speak(websocket) + + assert [call.target for call in relay.calls] == [ + "wss://api.eu.deepgram.com/v1/speak?api_base=wss%3A%2F%2Fattacker.example%2Fv1&model=aura-2-thalia-en" + ] + + +async def test_deepgram_speak_closes_before_accepting_when_provider_credentials_missing(): + websocket: Final = _FakeWebSocket("/deepgram/v1/speak", "model=aura-2-thalia-en") + + with patch(GET_CREDENTIALS, return_value=None): + relay: Final = await _serve_speak(websocket) + + assert websocket.closed is not None + assert websocket.closed[0] == 1011 + assert "DEEPGRAM_API_KEY" in websocket.closed[1] + assert websocket.accepts == [] + assert relay.calls == [] + + +@pytest.mark.parametrize( + ("query", "missing_key", "removed_row"), + [ + pytest.param("model=flux-tts-en", "deepgram/flux-tts-en", None, id="voice outside the priced families"), + pytest.param("model=aura-3-x-en", "deepgram/aura-3-x-en", None, id="unknown aura generation"), + pytest.param("model=aura-x", "deepgram/aura-x", None, id="aura voice without a language"), + pytest.param("model=nova-3", "deepgram/nova-3", None, id="transcription model with a per-second row"), + pytest.param("model=aura-2-thalia-en", "deepgram/aura-2", "deepgram/aura-2", id="family row removed"), + ], +) +async def test_deepgram_speak_refuses_sessions_it_cannot_price(query, missing_key, removed_row, monkeypatch): + """A session without a per-character price would be logged at no cost, so the proxy closes it before Deepgram is + contacted and names the registry row to add""" + monkeypatch.delenv("DEEPGRAM_API_BASE", raising=False) + if removed_row is not None: + monkeypatch.delitem(litellm.model_cost, removed_row) + websocket: Final = _FakeWebSocket("/deepgram/v1/speak", query) + + with patch(GET_CREDENTIALS, return_value="dg-provider-key"): + relay: Final = await _serve_speak(websocket) + + assert relay.calls == [] + assert websocket.accepts == [None] + assert websocket.closed is not None + assert websocket.closed[0] == 1008 + assert missing_key in websocket.closed[1] + assert "dg-provider-key" not in websocket.closed[1] + + +def test_deepgram_speak_rejects_connections_without_a_litellm_key(): + relay: Final = _FakeRelay() + client: Final = TestClient(_app_with_relay(relay)) + + with patch(GET_CREDENTIALS, return_value="dg-provider-key") as get_credentials: + with pytest.raises(WebSocketDisconnect) as disconnect: + with client.websocket_connect("/deepgram/v1/speak?model=aura-2-thalia-en"): + pass + + assert disconnect.value.code == 1008 + assert relay.calls == [] + get_credentials.assert_not_called() + + +@pytest.mark.parametrize( + ("query", "expected_target"), + [ + pytest.param( + "model=aura-2-thalia-en", + "wss://api.deepgram.com/v1/speak?model=aura-2-thalia-en", + id="allowed voice named", + ), + pytest.param( + "model=aura-2-thalia-en&model=aura-2-zeus-en", + "wss://api.deepgram.com/v1/speak?model=aura-2-thalia-en", + id="second voice stripped", + ), + pytest.param("model=aura-2-zeus-en", None, id="denied voice named"), + pytest.param("", None, id="voice omitted, default denied"), + pytest.param("model=&encoding=mulaw", None, id="voice blank, default denied"), + ], +) +def test_deepgram_speak_authorizes_the_voice_it_will_actually_send_upstream(query, expected_target, monkeypatch): + """A key allowed only one voice must not reach another by leaving ``model`` out or naming a second one, because + the real key auth path must see the same model the upstream target carries""" + monkeypatch.delenv("DEEPGRAM_API_BASE", raising=False) + monkeypatch.setattr(litellm, "max_budget", 0.0) + cache: Final = asyncio.run(_cache_restricted_key("sk-only-thalia", ["aura-2-thalia-en"])) + relay: Final = _FakeRelay() + client: Final = TestClient(_app_with_relay(relay)) + url: Final = f"/deepgram/v1/speak?{query}" + headers: Final = {"Authorization": "Bearer sk-only-thalia"} + + with ( + patch(GET_CREDENTIALS, return_value="dg-provider-key"), + patch.multiple( # test-quality-ok: the real key auth path reads these proxy_server globals and has no injection seam + "litellm.proxy.proxy_server", + master_key="sk-master", + prisma_client=MagicMock(), + user_api_key_cache=cache, + llm_model_list=None, + llm_router=None, + ), + ): + if expected_target is not None: + with client.websocket_connect(url, headers=headers): + pass + assert [call.target for call in relay.calls] == [expected_target] + return + with pytest.raises(WebSocketDisconnect) as disconnect: + with client.websocket_connect(url, headers=headers): + pass + + assert disconnect.value.code == 1008 + assert relay.calls == [] diff --git a/tests/unit/proxy/pass_through_endpoints/test_pass_through_endpoints.py b/tests/unit/proxy/pass_through_endpoints/test_pass_through_endpoints.py index a797f8084b9..abb59bc3b69 100644 --- a/tests/unit/proxy/pass_through_endpoints/test_pass_through_endpoints.py +++ b/tests/unit/proxy/pass_through_endpoints/test_pass_through_endpoints.py @@ -34,6 +34,7 @@ from litellm.constants import DEFAULT_REQUEST_TIMEOUT_SECONDS from litellm.integrations.custom_guardrail import CustomGuardrail from litellm.integrations.custom_logger import CustomLogger from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj +from litellm.llms.deepgram.common_utils import deepgram_speak_characters from litellm.proxy._lazy_features import LazyFeature, attach_lazy_features from litellm.proxy._types import ProxyException, UserAPIKeyAuth from litellm.proxy.litellm_pre_call_utils import LiteLLMProxyRequestSetup @@ -7954,6 +7955,66 @@ async def test_websocket_passthrough_counts_client_text_when_the_upstream_ends_w } +class _CapturingSpendLogger(CustomLogger): + def __init__(self) -> None: + super().__init__() + self.payloads: list[StandardLoggingPayload] = [] # mutable-ok: the logger appends each payload it receives + + async def async_log_success_event(self, kwargs, response_obj, start_time, end_time): + self.payloads.append(kwargs["standard_logging_object"]) + + +@pytest.mark.asyncio +@pytest.mark.usefixtures("local_model_cost_map") +async def test_deepgram_speak_session_logs_character_spend_once_and_never_the_spoken_text(monkeypatch): + """Runs the relay with the speak counter and the real success handler, so the spend logger sees what production + writes for one Speak message""" + capturing_logger = _CapturingSpendLogger() + monkeypatch.setattr(litellm, "_async_success_callback", [capturing_logger]) + monkeypatch.setattr(litellm, "success_callback", []) + monkeypatch.setattr(litellm, "callbacks", []) + queued: list[Coroutine[object, object, None]] = [] # mutable-ok: the fake logging worker keeps each coroutine + upstream_ws = SpeakingUpstreamWebSocket() + + with ( + patch("litellm.proxy.proxy_server.proxy_logging_obj") as proxy_logging, + patch( + "litellm.proxy.pass_through_endpoints.pass_through_endpoints.connect", + return_value=FakeUpstreamConnect(upstream_ws), + ), + patch("litellm.proxy.pass_through_endpoints.pass_through_endpoints.GLOBAL_LOGGING_WORKER") as worker, + ): + proxy_logging.pre_call_hook = AsyncMock(return_value={}) + proxy_logging.post_call_success_hook = AsyncMock() + proxy_logging.post_call_failure_hook = AsyncMock() + worker.ensure_initialized_and_enqueue = MagicMock( + side_effect=lambda async_coroutine: queued.append(async_coroutine) + ) + await asyncio.wait_for( + websocket_passthrough_request( + websocket=_speak_client(), + target=DEEPGRAM_SPEAK_TARGET, + custom_headers={"Authorization": "Token dg-provider-key"}, + user_api_key_dict=UserAPIKeyAuth(api_key="hashed-key", team_id="team-tts"), + forward_headers=False, + endpoint="/deepgram/v1/speak", + accept_websocket=False, + count_client_text=deepgram_speak_characters, + ), + timeout=RELAY_HANG_GUARD_SECONDS, + ) + for coroutine in queued: + await coroutine + + assert len(capturing_logger.payloads) == 1 + payload = capturing_logger.payloads[0] + assert payload["model"] == "aura-2-thalia-en" + assert payload["custom_llm_provider"] == "deepgram" + # Aura-2 pay as you go rate, $0.030 per 1,000 characters, from https://deepgram.com/pricing, checked 2026-10-26 + assert payload["response_cost"] == pytest.approx(len(SPOKEN_TEXT) * 0.030 / 1000) + assert SPOKEN_TEXT not in json.dumps(payload, default=str) + + @pytest.mark.asyncio async def test_passthrough_success_reconciles_budget_reservation(): """