fix(realtime): bound offers and preserve duration billing

litellm/cost_calculator.py

@@ -2601,8 +2601,6 @@ def handle_live_session_duration_cost(
     custom_llm_provider: str,
     litellm_model_name: str,
 ) -> float:
-    if any(event.get("type") == "response.done" for event in results):
-        return 0.0
     terminal: Final = next((event for event in reversed(results) if event.get("type") == "session.closed"), None)
     if terminal is None:
         return 0.0

litellm/proxy/realtime_endpoints/call_sessions.py

@@ -223,7 +223,31 @@ def decode_call(token: str, authorization: str) -> CodexRealtimeCall:
     return call
 
 
+MAX_REALTIME_OFFER_BYTES: Final = 8 * 1024 * 1024
+
+
+async def _cache_bounded_offer_body(request: Request) -> None:
+    try:
+        if int(request.headers.get("content-length", "")) > MAX_REALTIME_OFFER_BYTES:
+            raise HTTPException(413, "Realtime offer exceeds the 8 MiB limit")
+    except ValueError:
+        pass
+    if hasattr(request, "_body"):
+        if len(request._body) > MAX_REALTIME_OFFER_BYTES:  # pyright: ignore[reportPrivateUsage]  # validate Starlette's cached body without consuming it again
+            raise HTTPException(413, "Realtime offer exceeds the 8 MiB limit")
+        return
+    if request._form is not None and request._stream_consumed:  # pyright: ignore[reportPrivateUsage]  # a mixed-case empty form cache may leave the stream unread
+        return
+    body: Final = bytearray()
+    async for chunk in request.stream():
+        if len(body) + len(chunk) > MAX_REALTIME_OFFER_BYTES:
+            raise HTTPException(413, "Realtime offer exceeds the 8 MiB limit")
+        body.extend(chunk)
+    request._body = bytes(body)  # pyright: ignore[reportPrivateUsage]  # Starlette has no public setter for its shared body cache
+
+
 async def read_codex_offer(request: Request) -> CodexRealtimeOffer:
+    await _cache_bounded_offer_body(request)
     content_type: Final = request.headers.get("content-type", "")
     if _normalize_media_type(content_type) == "multipart/form-data":
         if content_type.split(";", 1)[0] != "multipart/form-data" and not await request.form():

tests/test_litellm/proxy/realtime_endpoints/test_call_sessions.py

@@ -11,6 +11,110 @@ from litellm.llms.chatgpt.codex import CodexRealtimeCall
 from litellm.proxy.realtime_endpoints.call_sessions import decode_call, encode_call
 
 
+@pytest.mark.asyncio
+@pytest.mark.parametrize("multipart", [False, True])
+@pytest.mark.parametrize("content_length", [None, "1", "999999999"])
+async def test_oversized_offer_stops_before_auth_or_multipart_files(monkeypatch, multipart, content_length):
+    import json
+    from unittest.mock import AsyncMock, Mock
+
+    from fastapi import Request
+
+    monkeypatch.setattr(codex, "MAX_REALTIME_OFFER_BYTES", 1024)
+    if multipart:
+        body = (
+            b'--Boundary\r\nContent-Disposition: form-data; name="extra"; filename="large.bin"\r\n\r\n'
+            + b"x" * 2048
+            + b"\r\n--Boundary--\r\n"
+        )
+        media_type = b"Multipart/Form-Data; boundary=Boundary"
+    else:
+        body = json.dumps({"sdp": "x" * 2048, "session": {"model": "voice"}}).encode()
+        media_type = b"application/json"
+    chunks = [body[offset : offset + 256] for offset in range(0, len(body), 256)]
+    received = []
+
+    async def receive():
+        chunk = chunks.pop(0)
+        received.append(len(chunk))
+        return {"type": "http.request", "body": chunk, "more_body": bool(chunks)}
+
+    headers = [(b"content-type", media_type)]
+    if content_length is not None:
+        headers.append((b"content-length", content_length.encode()))
+    request = Request({"type": "http", "headers": headers}, receive)
+    if multipart:
+        from litellm.proxy.common_utils.http_parsing_utils import _read_request_body
+
+        assert await _read_request_body(request) == {}
+    authenticate = AsyncMock()
+    create_file = Mock(side_effect=AssertionError("Oversized offers must not create temporary files"))
+    monkeypatch.setattr(codex, "user_api_key_auth", authenticate)
+    monkeypatch.setattr("starlette.formparsers.SpooledTemporaryFile", create_file)
+    with pytest.raises(HTTPException) as rejected:
+        await codex.create_codex_realtime_call(request)
+    assert rejected.value.status_code == 413
+    assert sum(received) <= 1280
+    assert chunks
+    authenticate.assert_not_awaited()
+    create_file.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_offer_at_size_limit_keeps_body_available_for_custom_auth(monkeypatch):
+    import json
+
+    from fastapi import Request
+
+    monkeypatch.setattr(codex, "MAX_REALTIME_OFFER_BYTES", 1024)
+    empty = {"sdp": "", "session": {"model": "voice"}}
+    sdp = "x" * (1024 - len(json.dumps(empty).encode()))
+    body = json.dumps({"sdp": sdp, "session": {"model": "voice"}}).encode()
+    chunks = [body[:512], body[512:]]
+
+    async def receive():
+        return {"type": "http.request", "body": chunks.pop(0), "more_body": bool(chunks)}
+
+    request = Request({"type": "http", "headers": [(b"content-type", b"application/json")]}, receive)
+    offer = await codex.read_codex_offer(request)
+    assert offer.sdp == sdp
+    assert offer.session.model == "voice"
+    assert await request.body() == body
+    assert not chunks
+
+
+@pytest.mark.asyncio
+async def test_empty_pre_read_multipart_offer_returns_invalid_offer():
+    from fastapi import Request
+
+    async def receive():
+        return {"type": "http.request", "body": b"--Boundary--\r\n", "more_body": False}
+
+    request = Request(
+        {"type": "http", "headers": [(b"content-type", b"multipart/form-data; boundary=Boundary")]}, receive
+    )
+    assert not await request.form()
+    with pytest.raises(HTTPException) as rejected:
+        await codex.create_codex_realtime_call(request)
+    assert rejected.value.status_code == 400
+
+
+@pytest.mark.asyncio
+async def test_oversized_pre_read_offer_is_rejected_before_decoding(monkeypatch):
+    from fastapi import Request
+
+    monkeypatch.setattr(codex, "MAX_REALTIME_OFFER_BYTES", 1024)
+
+    async def receive():
+        return {"type": "http.request", "body": b"x" * 2048, "more_body": False}
+
+    request = Request({"type": "http", "headers": [(b"content-type", b"application/json")]}, receive)
+    await request.body()
+    with pytest.raises(HTTPException) as rejected:
+        await codex.read_codex_offer(request)
+    assert rejected.value.status_code == 413
+
+
 @pytest.mark.asyncio
 @pytest.mark.parametrize("malformed", [False, True])
 async def test_mixed_case_offer_preserves_boundary_metadata_and_closes_extra_files(monkeypatch, malformed):

tests/test_litellm/test_cost_calculator.py

@@ -4827,12 +4827,43 @@ def test_live_terminal_is_not_counted_twice(monkeypatch):
     assert handle_realtime_stream_cost_calculation(
         [_live_terminal_event(), _live_terminal_event()], Usage(), "chatgpt", "live-priced-test"
     ) == pytest.approx(0.1)
-    assert (
-        handle_realtime_stream_cost_calculation(
-            [{"type": "response.done", "response": {"usage": {}}}, _live_terminal_event()],
-            Usage(),
-            "chatgpt",
-            "live-priced-test",
-        )
-        == 0
+
+
+@pytest.mark.parametrize("with_tokens", [False, True])
+@pytest.mark.parametrize("terminal_count", [1, 2])
+@pytest.mark.parametrize("duration_priced", [False, True])
+def test_live_terminal_with_response_done_preserves_configured_billing(
+    monkeypatch, with_tokens, terminal_count, duration_priced
+):
+    monkeypatch.setitem(
+        litellm.model_cost,
+        "realtime-deployment-test",
+        {
+            "litellm_provider": "chatgpt",
+            "mode": "realtime",
+            **(
+                {"input_cost_per_second": 0.025}
+                if duration_priced
+                else {"input_cost_per_token": 0.001, "output_cost_per_token": 0.002}
+            ),
+        },
     )
+    events = [
+        {
+            "type": "response.done",
+            "response": {
+                "usage": ({"input_tokens": 10, "output_tokens": 5, "total_tokens": 15} if with_tokens else {})
+            },
+        },
+        *(_live_terminal_event() for _ in range(terminal_count)),
+    ]
+    usage = RealtimeAPITokenUsageProcessor.collect_and_combine_usage_from_realtime_stream_results(events)
+    result = RealtimeAPITokenUsageProcessor.create_logging_realtime_object(usage, events)
+    assert completion_cost(
+        completion_response=result,
+        model="gpt-live-1-codex" if duration_priced else "gpt-realtime-1.5",
+        custom_llm_provider="chatgpt",
+        call_type="_arealtime",
+        custom_pricing=True,
+        router_model_id="realtime-deployment-test",
+    ) == pytest.approx(0.1 if duration_priced else (0.02 if with_tokens else 0))