fix(passthrough): bill every Deepgram channel, not just wall-clock duration

Deepgram charges for the total processed audio across channels, so a stereo /listen session with multichannel=true costs twice its duration. The handler now multiplies the session duration by a validated channel count taken from Metadata.channels, then the widest Results channel_index, then the channels query parameter, defaulting to one. Booleans, floats, strings, zero and negative values are ignored

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

litellm/llms/deepgram/common_utils.py

@@ -36,6 +36,46 @@ def deepgram_listen_model(upstream_url: str) -> str:
     return models[0] if models else DEEPGRAM_LISTEN_DEFAULT_MODEL
 
 
+def _channel_count(value: object) -> int | None:
+    if isinstance(value, bool) or not isinstance(value, int):
+        return None
+    return value if value >= 1 else 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])
+
+
+def _declared_channel_count(upstream_url: str) -> int | None:
+    declared: Final = parse_qs(urlparse(upstream_url).query).get("channels")
+    if not declared or not declared[0].isdigit():
+        return None
+    return _channel_count(int(declared[0]))
+
+
+def deepgram_listen_channel_count(websocket_messages: Sequence[Mapping[str, object]], upstream_url: str) -> int:
+    metadata_channels: Final = tuple(
+        channels
+        for frame in websocket_messages
+        if frame.get("type") == "Metadata"
+        if (channels := _channel_count(frame.get("channels"))) is not None
+    )
+    if metadata_channels:
+        return metadata_channels[-1]
+    results_channels: Final = tuple(
+        channels
+        for frame in websocket_messages
+        if frame.get("type") == "Results"
+        if (channels := _results_channel_count(frame)) is not None
+    )
+    if results_channels:
+        return max(results_channels)
+    return _declared_channel_count(upstream_url) or 1
+
+
 def _seconds(value: object) -> float | None:
     if isinstance(value, bool) or not isinstance(value, (int, float)):
         return None

litellm/proxy/pass_through_endpoints/llm_provider_handlers/deepgram_listen_passthrough_logging_handler.py

@@ -8,6 +8,7 @@ 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_listen_audio_seconds,
+    deepgram_listen_channel_count,
     deepgram_listen_model,
     deepgram_listen_transcript,
 )
@@ -45,8 +46,10 @@ class DeepgramListenPassthroughLoggingHandler:
     ) -> PassThroughEndpointLoggingTypedDict:
         model: Final = deepgram_listen_model(upstream_url)
         audio_seconds: Final = deepgram_listen_audio_seconds(websocket_messages)
+        channels: Final = deepgram_listen_channel_count(websocket_messages, upstream_url)
+        billed_seconds: Final = audio_seconds * channels
         response: Final = TranscriptionResponse(text=deepgram_listen_transcript(websocket_messages))
-        response._hidden_params["audio_transcription_duration"] = audio_seconds  # pyright: ignore[reportPrivateUsage]  # the cost calculator reads the billed duration off the response's hidden params
+        response._hidden_params["audio_transcription_duration"] = billed_seconds  # pyright: ignore[reportPrivateUsage]  # the cost calculator reads the billed duration off the response's hidden params
         response_cost: Final = _audio_cost(response, model)
         response._hidden_params["response_cost"] = response_cost  # pyright: ignore[reportPrivateUsage]  # the logger reads a precomputed cost off the response's hidden params
 
@@ -56,9 +59,10 @@ class DeepgramListenPassthroughLoggingHandler:
         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 listen passthrough cost tracking: model %s, audio seconds %s, cost %s",
+            "Deepgram listen passthrough cost tracking: model %s, audio seconds %s, channels %s, cost %s",
             model,
             audio_seconds,
+            channels,
             response_cost,
         )
         logging_result: Final[PassThroughEndpointLoggingTypedDict] = {

tests/test_litellm/llms/deepgram/test_deepgram_common_utils.py

@@ -8,6 +8,7 @@ import litellm
 from litellm.llms.deepgram.common_utils import (
     deepgram_listen_audio_seconds,
     deepgram_listen_callback_params,
+    deepgram_listen_channel_count,
     deepgram_listen_model,
     deepgram_listen_transcript,
     deepgram_listen_websocket_target,
@@ -16,18 +17,25 @@ from litellm.llms.deepgram.common_utils import (
 NOVA_3_URL: Final = "wss://api.deepgram.com/v1/listen?model=nova-3&encoding=linear16&sample_rate=16000"
 
 
-def _results(start: object, duration: object, transcript: str = "", is_final: object = True) -> dict[str, object]:
+def _results(
+    start: object,
+    duration: object,
+    transcript: str = "",
+    is_final: object = True,
+    channel_index: object = (0, 1),
+) -> dict[str, object]:
     return {
         "type": "Results",
         "start": start,
         "duration": duration,
         "is_final": is_final,
+        "channel_index": list(channel_index) if isinstance(channel_index, tuple) else channel_index,
         "channel": {"alternatives": [{"transcript": transcript, "confidence": 0.9}]},
     }
 
 
-def _metadata(duration: object) -> dict[str, object]:
-    return {"type": "Metadata", "request_id": "req-1", "duration": duration, "channels": 1}
+def _metadata(duration: object, channels: object = 1) -> dict[str, object]:
+    return {"type": "Metadata", "request_id": "req-1", "duration": duration, "channels": channels}
 
 
 @pytest.mark.parametrize(
@@ -107,6 +115,62 @@ def test_deepgram_listen_audio_seconds(frames: Sequence[Mapping[str, object]], e
     assert deepgram_listen_audio_seconds(frames) == expected_seconds
 
 
+@pytest.mark.parametrize(
+    ("frames", "upstream_url", "expected_channels"),
+    [
+        pytest.param((_results(0.0, 2.0), _metadata(6.25)), NOVA_3_URL, 1, id="mono"),
+        pytest.param((_results(0.0, 2.0, channel_index=(0, 2)), _metadata(6.25, 2)), NOVA_3_URL, 2, id="stereo"),
+        pytest.param((_metadata(1.0, 3), _metadata(1.0, 5)), NOVA_3_URL, 5, id="last metadata wins"),
+        pytest.param(
+            (_metadata(1.0, 20), _results(0.0, 1.0, channel_index=(1, 2))),
+            NOVA_3_URL,
+            20,
+            id="metadata beats channel_index",
+        ),
+        pytest.param(
+            (_results(0.0, 1.0, channel_index=(0, 2)), _results(0.0, 1.0, channel_index=(3, 4))),
+            NOVA_3_URL,
+            4,
+            id="widest channel_index without metadata",
+        ),
+        pytest.param(
+            (_results(0.0, 1.0, channel_index=(0, 2)),),
+            f"{NOVA_3_URL}&channels=7&multichannel=true",
+            2,
+            id="frames beat the declared query",
+        ),
+        pytest.param((), f"{NOVA_3_URL}&channels=7&multichannel=true", 7, id="declared query when no frames"),
+        pytest.param((), f"{NOVA_3_URL}&channels=0", 1, id="zero declared channels"),
+        pytest.param((), f"{NOVA_3_URL}&channels=-2", 1, id="negative declared channels"),
+        pytest.param((), f"{NOVA_3_URL}&channels=two", 1, id="non numeric declared channels"),
+        pytest.param((), NOVA_3_URL, 1, id="nothing declared"),
+        pytest.param((_metadata(1.0, "2"), _metadata(1.0, True), _metadata(1.0, 0)), NOVA_3_URL, 1, id="bad metadata"),
+        pytest.param((_metadata(1.0, 3), _metadata(1.0, True)), NOVA_3_URL, 3, id="boolean does not shadow a count"),
+        pytest.param((_metadata(1.0, 2.0), _metadata(1.0, -1)), NOVA_3_URL, 1, id="float and negative metadata"),
+        pytest.param(
+            (_metadata(1.0, 2), {**_results(0.0, 1.0), "channels": 9}, {"type": "UtteranceEnd", "channels": 11}),
+            NOVA_3_URL,
+            2,
+            id="channels on non metadata frames ignored",
+        ),
+        pytest.param(
+            (
+                _results(0.0, 1.0, channel_index=[0]),
+                _results(0.0, 1.0, channel_index=(0, "2")),
+                _results(0.0, 1.0, channel_index=(0, 0)),
+            ),
+            NOVA_3_URL,
+            1,
+            id="bad channel_index",
+        ),
+    ],
+)
+def test_deepgram_listen_channel_count(
+    frames: Sequence[Mapping[str, object]], upstream_url: str, expected_channels: int
+):
+    assert deepgram_listen_channel_count(frames, upstream_url) == expected_channels
+
+
 def test_deepgram_listen_transcript_joins_final_results_only():
     frames = (
         _results(0.0, 1.0, "hello wor", is_final=False),

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

@@ -30,8 +30,8 @@ def _results(start: object, duration: object, transcript: str = "", is_final: ob
     }
 
 
-def _metadata(duration: object) -> dict[str, object]:
-    return {"type": "Metadata", "request_id": "req-1", "duration": duration, "channels": 1}
+def _metadata(duration: object, channels: int = 1) -> dict[str, object]:
+    return {"type": "Metadata", "request_id": "req-1", "duration": duration, "channels": channels}
 
 
 @pytest.mark.parametrize(
@@ -119,6 +119,34 @@ def test_handler_charges_more_for_more_audio_on_the_same_model():
     assert short["kwargs"]["response_cost"] > 0
 
 
+def test_handler_bills_every_channel_of_a_multichannel_session():
+    """Deepgram bills processed audio per channel (deepgram.com/pricing FAQ, 2026-09-17), so a stereo session must be
+    charged for twice its wall-clock duration or budgets can be bypassed by requesting more channels."""
+    mono = DeepgramListenPassthroughLoggingHandler().deepgram_listen_passthrough_handler(
+        websocket_messages=(_metadata(30.0),), logging_obj=_logging_obj(), upstream_url=NOVA_3_URL
+    )
+    stereo = DeepgramListenPassthroughLoggingHandler().deepgram_listen_passthrough_handler(
+        websocket_messages=(_metadata(30.0, channels=2),),
+        logging_obj=_logging_obj(),
+        upstream_url=f"{NOVA_3_URL}&multichannel=true&channels=2",
+    )
+
+    assert stereo["result"]._hidden_params["audio_transcription_duration"] == 60.0
+    assert stereo["kwargs"]["response_cost"] == pytest.approx(2 * mono["kwargs"]["response_cost"])
+    assert stereo["kwargs"]["response_cost"] == pytest.approx(_registry_cost("nova-3", 60.0))
+
+
+def test_handler_bills_the_declared_channels_when_the_stream_dies_before_any_frame_reports_them():
+    handler_result = DeepgramListenPassthroughLoggingHandler().deepgram_listen_passthrough_handler(
+        websocket_messages=(_results(0.0, 10.0, "a"),),
+        logging_obj=_logging_obj(),
+        upstream_url=f"{NOVA_3_URL}&multichannel=true&channels=3",
+    )
+
+    assert handler_result["result"]._hidden_params["audio_transcription_duration"] == 30.0
+    assert handler_result["kwargs"]["response_cost"] == pytest.approx(_registry_cost("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),),