fix(deepgram): refuse /listen sessions that have no streaming price

A caller could pick a model with only a pre-recorded registry row, or no row at all, and the session would be billed at the pre-recorded rate or logged at zero cost, so budgets did not apply. The route now closes the WebSocket with 1008 before dialing Deepgram unless deepgram/streaming/<model> (or the -multilingual row for language=multi) is an exact registry hit, and the logging handler applies the same check so a registry change under a live session records the duration with no cost instead of a substitute rate

Regression tests cover the route refusal, an operator-supplied streaming row for another model being accepted, and the handler never substituting the pre-recorded rate

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

litellm/llms/deepgram/common_utils.py

@@ -6,8 +6,10 @@ from urllib.parse import parse_qs, urlparse
 
 import httpx
 
+import litellm
 from litellm.constants import DEEPGRAM_DEFAULT_API_BASE, DEEPGRAM_LISTEN_DEFAULT_MODEL
 from litellm.llms.base_llm.chat.transformation import BaseLLMException
+from litellm.types.utils import LlmProviders
 
 _WEBSOCKET_SCHEMES: Final = MappingProxyType({"https": "wss", "http": "ws", "wss": "wss", "ws": "ws"})
 DEEPGRAM_LISTEN_CALLBACK_PARAMS: Final = frozenset({"callback", "callback_method"})
@@ -57,19 +59,30 @@ def _param_enabled(values: Sequence[str]) -> bool:
     return any(value.strip().lower() not in _DISABLED_PARAM_VALUES for value in values)
 
 
-def deepgram_listen_base_pricing_models(upstream_url: str) -> tuple[str, ...]:
-    """Registry keys to try, in order, for the per-second base rate of a streaming session: the streaming entry for
-    the language mode Deepgram bills (multilingual when ``language=multi``), then the plain streaming entry, then
-    the pre-recorded entry for models that have no streaming price of their own."""
-    model: Final = deepgram_listen_model(upstream_url)
-    params: Final = parse_qs(urlparse(upstream_url).query)
-    streaming: Final = f"{DEEPGRAM_LISTEN_STREAMING_PRICING_PREFIX}{model}"
-    multilingual: Final = params.get("language", ("",))[-1].strip().lower() == DEEPGRAM_LISTEN_MULTILINGUAL_LANGUAGE
-    return (
-        (f"{streaming}{DEEPGRAM_LISTEN_MULTILINGUAL_PRICING_SUFFIX}", streaming, model)
-        if multilingual
-        else (streaming, model)
-    )
+def deepgram_listen_pricing_model(upstream_url: str) -> str:
+    """Registry key, without the provider prefix, for the per-second base rate Deepgram bills a streaming session at:
+    the multilingual streaming entry when ``language=multi``, otherwise the model's own streaming entry. Pre-recorded
+    entries are never a substitute: Deepgram prices the two products differently."""
+    streaming: Final = f"{DEEPGRAM_LISTEN_STREAMING_PRICING_PREFIX}{deepgram_listen_model(upstream_url)}"
+    language: Final = parse_qs(urlparse(upstream_url).query).get("language", ("",))[-1]
+    if language.strip().lower() == DEEPGRAM_LISTEN_MULTILINGUAL_LANGUAGE:
+        return f"{streaming}{DEEPGRAM_LISTEN_MULTILINGUAL_PRICING_SUFFIX}"
+    return streaming
+
+
+def deepgram_listen_registry_key(upstream_url: str) -> str:
+    return f"{LlmProviders.DEEPGRAM.value}/{deepgram_listen_pricing_model(upstream_url)}"
+
+
+def deepgram_listen_is_priced(upstream_url: str) -> bool:
+    """Only an exact registry hit counts: the cost calculator resolves a missing ``streaming/<model>`` row to the
+    pre-recorded ``<model>`` row, which is not the rate Deepgram bills a WebSocket session at."""
+    registry_key: Final = deepgram_listen_registry_key(upstream_url)
+    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
 
 
 def deepgram_listen_addon_pricing_models(upstream_url: str) -> tuple[str, ...]:

litellm/proxy/pass_through_endpoints/llm_passthrough_endpoints.py

@@ -38,6 +38,8 @@ from litellm.llms.azure.passthrough.transformation import foreign_azure_deployme
 from litellm.llms.custom_httpx.http_handler import get_async_httpx_client
 from litellm.llms.deepgram.common_utils import (
     deepgram_listen_callback_params,
+    deepgram_listen_is_priced,
+    deepgram_listen_registry_key,
     deepgram_listen_requested_model,
     deepgram_listen_websocket_target,
 )
@@ -2897,6 +2899,9 @@ _DEEPGRAM_WS_MISSING_KEY_REASON: Final = (
     "Required 'DEEPGRAM_API_KEY' in environment to make pass-through calls to Deepgram."
 )
 _DEEPGRAM_WS_CALLBACK_REASON: Final = "Deepgram callback delivery is not supported through the proxy: remove {params}"
+_DEEPGRAM_WS_UNPRICED_REASON: Final = (
+    "No streaming price for '{registry_key}': add it to the model cost map to enable it"
+)
 
 
 async def deepgram_listen_user_api_key_auth(websocket: WebSocket) -> UserAPIKeyAuth:
@@ -2929,12 +2934,20 @@ async def deepgram_listen_websocket_route(
         )
         return
 
+    target: Final = deepgram_listen_websocket_target(
+        api_base=get_secret_str("DEEPGRAM_API_BASE"),
+        query_string=websocket.url.query,
+    )
+    if not deepgram_listen_is_priced(target):
+        await websocket.close(
+            code=1008,
+            reason=_DEEPGRAM_WS_UNPRICED_REASON.format(registry_key=deepgram_listen_registry_key(target)),
+        )
+        return
+
     await relay(
         websocket=websocket,
-        target=deepgram_listen_websocket_target(
-            api_base=get_secret_str("DEEPGRAM_API_BASE"),
-            query_string=websocket.url.query,
-        ),
+        target=target,
         custom_headers={  # mutable-ok: websocket_passthrough_request requires a plain dict of upstream headers
             "Authorization": f"Token {deepgram_api_key}"
         },

litellm/proxy/pass_through_endpoints/llm_provider_handlers/deepgram_listen_passthrough_logging_handler.py

@@ -9,9 +9,11 @@ from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLogging
 from litellm.llms.deepgram.common_utils import (
     deepgram_listen_addon_pricing_models,
     deepgram_listen_audio_seconds,
-    deepgram_listen_base_pricing_models,
     deepgram_listen_channel_count,
+    deepgram_listen_is_priced,
     deepgram_listen_model,
+    deepgram_listen_pricing_model,
+    deepgram_listen_registry_key,
     deepgram_listen_transcript,
 )
 from litellm.proxy._types import PassThroughEndpointLoggingTypedDict
@@ -34,19 +36,14 @@ def _registry_cost(response: TranscriptionResponse, pricing_model: str) -> float
 
 
 def _audio_cost(response: TranscriptionResponse, upstream_url: str) -> float | None:
-    base_cost: Final = next(
-        (
-            cost
-            for pricing_model in deepgram_listen_base_pricing_models(upstream_url)
-            if (cost := _registry_cost(response, pricing_model)) is not None
-        ),
-        None,
-    )
-    if base_cost is None:
+    if not deepgram_listen_is_priced(upstream_url):
         verbose_proxy_logger.warning(
-            "Deepgram listen passthrough: no pricing for model '%s'", deepgram_listen_model(upstream_url)
+            "Deepgram listen passthrough: no registry entry '%s'", deepgram_listen_registry_key(upstream_url)
         )
         return None
+    base_cost: Final = _registry_cost(response, deepgram_listen_pricing_model(upstream_url))
+    if base_cost is None:
+        return None
     addon_costs: Final = tuple(
         _registry_cost(response, pricing_model) for pricing_model in deepgram_listen_addon_pricing_models(upstream_url)
     )

tests/test_litellm/llms/deepgram/test_deepgram_common_utils.py

@@ -8,10 +8,12 @@ import litellm
 from litellm.llms.deepgram.common_utils import (
     deepgram_listen_addon_pricing_models,
     deepgram_listen_audio_seconds,
-    deepgram_listen_base_pricing_models,
     deepgram_listen_callback_params,
     deepgram_listen_channel_count,
+    deepgram_listen_is_priced,
     deepgram_listen_model,
+    deepgram_listen_pricing_model,
+    deepgram_listen_registry_key,
     deepgram_listen_requested_model,
     deepgram_listen_transcript,
     deepgram_listen_websocket_target,
@@ -220,27 +222,51 @@ def test_requested_model_is_the_model_the_upstream_target_will_carry(query_strin
 @pytest.mark.parametrize(
     ("upstream_url", "expected"),
     [
-        pytest.param(NOVA_3_URL, ("streaming/nova-3", "nova-3"), id="monolingual"),
-        pytest.param(f"{NOVA_3_URL}&language=en", ("streaming/nova-3", "nova-3"), id="explicit language"),
-        pytest.param(
-            f"{NOVA_3_URL}&language=multi",
-            ("streaming/nova-3-multilingual", "streaming/nova-3", "nova-3"),
-            id="multilingual",
-        ),
-        pytest.param(
-            f"{NOVA_3_URL}&language=MULTI",
-            ("streaming/nova-3-multilingual", "streaming/nova-3", "nova-3"),
-            id="multilingual any case",
-        ),
+        pytest.param(NOVA_3_URL, "streaming/nova-3", id="monolingual"),
+        pytest.param(f"{NOVA_3_URL}&language=en", "streaming/nova-3", id="explicit language"),
+        pytest.param(f"{NOVA_3_URL}&language=multi", "streaming/nova-3-multilingual", id="multilingual"),
+        pytest.param(f"{NOVA_3_URL}&language=MULTI", "streaming/nova-3-multilingual", id="multilingual any case"),
         pytest.param(
             "wss://api.deepgram.com/v1/listen?model=nova-2&language=multi",
-            ("streaming/nova-2-multilingual", "streaming/nova-2", "nova-2"),
+            "streaming/nova-2-multilingual",
             id="other model",
         ),
+        pytest.param("wss://api.deepgram.com/v1/listen?encoding=linear16", "streaming/nova-3", id="default model"),
     ],
 )
-def test_deepgram_listen_base_pricing_models(upstream_url: str, expected: tuple[str, ...]):
-    assert deepgram_listen_base_pricing_models(upstream_url) == expected
+def test_deepgram_listen_pricing_model_is_the_streaming_entry_never_the_prerecorded_one(
+    upstream_url: str, expected: str
+):
+    assert deepgram_listen_pricing_model(upstream_url) == expected
+    assert deepgram_listen_registry_key(upstream_url) == f"deepgram/{expected}"
+
+
+NOVA_2_URL: Final = "wss://api.deepgram.com/v1/listen?model=nova-2"
+
+
+@pytest.mark.usefixtures("local_model_cost_map")
+@pytest.mark.parametrize(
+    ("upstream_url", "extra_rows", "expected"),
+    [
+        pytest.param(NOVA_3_URL, (), True, id="streaming entry present"),
+        pytest.param(f"{NOVA_3_URL}&language=multi", (), True, id="multilingual entry present"),
+        pytest.param(NOVA_2_URL, (), False, id="only the pre-recorded entry"),
+        pytest.param(f"{NOVA_2_URL}&language=multi", ("deepgram/streaming/nova-2",), False, id="needs multilingual"),
+        pytest.param("wss://api.deepgram.com/v1/listen?model=nova-99-unmapped", (), False, id="nothing priced"),
+        pytest.param(NOVA_2_URL, ("deepgram/streaming/nova-2",), True, id="operator-supplied streaming entry"),
+        pytest.param(NOVA_2_URL, ("streaming/nova-2",), False, id="a row under another key is not the entry"),
+    ],
+)
+def test_deepgram_listen_is_priced(
+    monkeypatch: pytest.MonkeyPatch, upstream_url: str, extra_rows: tuple[str, ...], expected: bool
+):
+    """The bundled map prices only nova-3 for streaming; nova-2 has a pre-recorded row, which must never count."""
+    monkeypatch.delitem(litellm.model_cost, "deepgram/streaming/nova-2", raising=False)
+    assert "deepgram/nova-2" in litellm.model_cost
+    for row in extra_rows:
+        monkeypatch.setitem(litellm.model_cost, row, dict(litellm.model_cost["deepgram/streaming/nova-3"]))
+
+    assert deepgram_listen_is_priced(upstream_url) is expected
 
 
 @pytest.mark.parametrize(

tests/test_litellm/proxy/pass_through_endpoints/llm_provider_handlers/test_deepgram_listen_passthrough_logging_handler.py

@@ -158,17 +158,25 @@ def test_handler_add_ons_scale_with_channels_like_the_base_rate():
     assert stereo_redacted - stereo_plain == pytest.approx(_registry_cost("streaming/redact", 120.0))
 
 
-def test_handler_falls_back_to_the_prerecorded_rate_for_a_model_without_a_streaming_entry():
-    assert "deepgram/streaming/nova-2" not in litellm.model_cost
+@pytest.mark.parametrize(
+    "upstream_url",
+    [
+        pytest.param("wss://api.deepgram.com/v1/listen?model=nova-2", id="only a pre-recorded entry"),
+        pytest.param("wss://api.deepgram.com/v1/listen?model=nova-99-not-in-registry", id="no entry at all"),
+    ],
+)
+def test_handler_never_substitutes_another_rate_for_a_missing_streaming_entry(monkeypatch, upstream_url):
+    """The route refuses these sessions up front; should the registry change under a live one, the spend row
+    keeps the duration and carries no cost, rather than the pre-recorded rate or any other stand-in."""
+    monkeypatch.delitem(litellm.model_cost, "deepgram/streaming/nova-2", raising=False)
+    assert "deepgram/nova-2" in litellm.model_cost
 
     handler_result = DeepgramListenPassthroughLoggingHandler().deepgram_listen_passthrough_handler(
-        websocket_messages=(_metadata(60.0),),
-        logging_obj=_logging_obj(),
-        upstream_url="wss://api.deepgram.com/v1/listen?model=nova-2",
+        websocket_messages=(_metadata(60.0),), logging_obj=_logging_obj(), upstream_url=upstream_url
     )
 
-    assert handler_result["kwargs"]["model"] == "nova-2"
-    assert handler_result["kwargs"]["response_cost"] == pytest.approx(_registry_cost("nova-2", 60.0))
+    assert handler_result["kwargs"]["response_cost"] is None
+    assert handler_result["result"]._hidden_params["audio_transcription_duration"] == 60.0
 
 
 def test_handler_falls_back_to_results_frames_when_the_stream_ends_without_metadata():
@@ -222,18 +230,6 @@ def test_handler_bills_the_declared_channels_when_the_stream_dies_before_any_fra
     assert handler_result["kwargs"]["response_cost"] == pytest.approx(_registry_cost("streaming/nova-3", 30.0))
 
 
-def test_handler_keeps_the_spend_row_but_no_cost_for_an_unpriced_model():
-    handler_result = DeepgramListenPassthroughLoggingHandler().deepgram_listen_passthrough_handler(
-        websocket_messages=(_metadata(12.5),),
-        logging_obj=_logging_obj(),
-        upstream_url="wss://api.deepgram.com/v1/listen?model=nova-99-not-in-registry",
-    )
-
-    assert handler_result["kwargs"]["model"] == "nova-99-not-in-registry"
-    assert handler_result["kwargs"]["response_cost"] is None
-    assert handler_result["result"]._hidden_params["audio_transcription_duration"] == 12.5
-
-
 class _CapturingLogger(CustomLogger):
     def __init__(self) -> None:
         super().__init__()

tests/test_litellm/proxy/pass_through_endpoints/test_deepgram_ws_passthrough_routes.py

@@ -30,6 +30,14 @@ 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")
+NOVA_2_STREAMING_KEY: Final = "deepgram/streaming/nova-2"
+
+pytestmark: Final = pytest.mark.usefixtures("local_model_cost_map")
+
+
+def _price_nova_2_streaming(monkeypatch: pytest.MonkeyPatch) -> None:
+    """An operator-supplied streaming row: the bundled map prices only nova-3 for streaming."""
+    monkeypatch.setitem(litellm.model_cost, NOVA_2_STREAMING_KEY, dict(litellm.model_cost["deepgram/streaming/nova-3"]))
 
 
 class _FakeWebSocket:
@@ -138,6 +146,7 @@ async def test_deepgram_listen_forwards_query_and_injects_only_provider_auth(pat
 @pytest.mark.asyncio
 async def test_deepgram_listen_keeps_caller_chosen_model(monkeypatch):
     monkeypatch.delenv("DEEPGRAM_API_BASE", raising=False)
+    _price_nova_2_streaming(monkeypatch)
     websocket = _FakeWebSocket("/deepgram/v1/listen", "model=nova-2&language=en")
 
     with patch(GET_CREDENTIALS, return_value="dg-provider-key"):
@@ -236,6 +245,53 @@ async def test_deepgram_listen_rejects_callback_delivery_that_would_go_unbilled(
     assert "dg-provider-key" not in websocket.closed[1]
 
 
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    ("query", "missing_key"),
+    [
+        pytest.param("model=nova-2", "deepgram/streaming/nova-2", id="model with only a pre-recorded price"),
+        pytest.param("model=nova-99", "deepgram/streaming/nova-99", id="model unknown to the registry"),
+        pytest.param(
+            "model=nova-3&language=multi",
+            "deepgram/streaming/nova-3-multilingual",
+            id="multilingual session without its own price",
+        ),
+    ],
+)
+async def test_deepgram_listen_refuses_sessions_it_cannot_price(query, missing_key, monkeypatch):
+    """A session with no streaming price would be logged at zero (or at the pre-recorded rate), letting a caller run
+    up unmetered spend, so the proxy closes it before Deepgram is contacted and names the registry row to add."""
+    monkeypatch.delenv("DEEPGRAM_API_BASE", raising=False)
+    monkeypatch.delitem(litellm.model_cost, missing_key, raising=False)
+    assert "deepgram/nova-2" in litellm.model_cost
+    websocket = _FakeWebSocket("/deepgram/v1/listen", query)
+
+    with patch(GET_CREDENTIALS, return_value="dg-provider-key"):
+        relay = await _serve(websocket)
+
+    assert relay.calls == []
+    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]
+
+
+@pytest.mark.asyncio
+async def test_deepgram_listen_relays_once_the_operator_prices_the_model(monkeypatch):
+    monkeypatch.delenv("DEEPGRAM_API_BASE", raising=False)
+    websocket = _FakeWebSocket("/deepgram/v1/listen", "model=nova-2")
+    with patch(GET_CREDENTIALS, return_value="dg-provider-key"):
+        assert (await _serve(websocket)).calls == []
+
+    _price_nova_2_streaming(monkeypatch)
+    priced_websocket = _FakeWebSocket("/deepgram/v1/listen", "model=nova-2")
+    with patch(GET_CREDENTIALS, return_value="dg-provider-key"):
+        relay = await _serve(priced_websocket)
+
+    assert [call.target for call in relay.calls] == ["wss://api.deepgram.com/v1/listen?model=nova-2"]
+    assert priced_websocket.closed is None
+
+
 def _app_with_relay(relay: _FakeRelay) -> FastAPI:
     app = FastAPI()
     app.include_router(router)
@@ -332,6 +388,7 @@ def test_deepgram_listen_authorizes_the_model_it_will_actually_send_upstream(que
     in its default: the real key auth path must see the same model the upstream target will carry."""
     monkeypatch.delenv("DEEPGRAM_API_BASE", raising=False)
     monkeypatch.setattr(litellm, "max_budget", 0.0)
+    _price_nova_2_streaming(monkeypatch)
     cache = asyncio.run(_cache_restricted_key("sk-only-nova-2", ["nova-2"]))
     relay = _FakeRelay()
     client = TestClient(_app_with_relay(relay))