feat(chatgpt): add public Live routes and Codex setup guide

docs/my-website/docs/providers/chatgpt.md

@@ -0,0 +1,121 @@
+# ChatGPT, Codex, and GPT-Live
+
+The proxy exposes the public GPT-Live session routes and retains the Codex-compatible `POST /live` route. Clients authenticate to LiteLLM with a LiteLLM virtual key. A `chatgpt` deployment uses one proxy-wide ChatGPT OAuth record, while an `openai` deployment uses its configured OpenAI API key. Do not send an upstream OAuth token as the proxy key
+
+## Configure LiteLLM deployments
+
+Codex text, image, and voice requests need separate LiteLLM aliases because their backends have different capabilities. The canonical aliases below keep the primary Qwen model separate from the ChatGPT OAuth models
+
+```yaml
+model_list:
+  - model_name: qwen3.8-flash-next-codex
+    litellm_params:
+      model: openai/qwen3.8-flash-next
+      api_base: https://qwen.example.com/v1
+      api_key: os.environ/QWEN_API_KEY
+
+  - model_name: gpt-image-2
+    litellm_params:
+      model: chatgpt/gpt-image-2
+
+  - model_name: gpt-image-2.5-flare
+    litellm_params:
+      model: chatgpt/gpt-image-2.5-flare
+
+  - model_name: gpt-image-2.5-sunburst
+    litellm_params:
+      model: chatgpt/gpt-image-2.5-sunburst
+
+  - model_name: gpt-realtime-1.5
+    litellm_params:
+      model: chatgpt/gpt-realtime-1.5
+
+  - model_name: gpt-live-1-codex
+    litellm_params:
+      model: chatgpt/gpt-live-1-codex
+```
+
+If clients use shorter local aliases, publish separate aliases such as `qwen-codex`, `images`, and `voice` that point to the corresponding deployments. Authorize the exact alias sent by the client in the key or team policy, and for a restricted team member set `allowed_models` to contain the aliases used for text, image, or voice requests. Selecting the Qwen alias does not give it image-generation or voice capabilities. The Qwen endpoint in this example uses the OpenAI-compatible adapter and must expose `/v1/responses`; a native vLLM deployment can use `hosted_vllm/qwen3.8-flash-next` when that endpoint is available. LiteLLM does not promise provider-specific tool, reasoning, or stream behavior parity. For a deployment using the public OpenAI API instead, configure `model: openai/gpt-live-1` and `api_key: os.environ/OPENAI_API_KEY` under its own alias. The public API documentation uses `gpt-live-1`; the Codex alias and its backend capabilities are separate
+
+### Configure global ChatGPT OAuth
+
+The ChatGPT provider reads one auth file for the LiteLLM process. Set these environment variables before starting the proxy when the default location is not suitable
+
+```bash
+export CHATGPT_TOKEN_DIR=/var/lib/litellm/chatgpt
+export CHATGPT_AUTH_FILE=auth.json
+```
+
+The defaults are `~/.config/litellm/chatgpt` and `auth.json`. Persist the directory and complete the provider's device-code OAuth flow. The provider refreshes the stored record when it expires. All `chatgpt/...` deployments in that process share this record. Live rejects per-deployment `chatgpt_auth_profile`, `chatgpt_token_dir`, and `chatgpt_auth_file` overrides
+
+### Image generation and editing
+
+Use the `gpt-image-2` alias for both `/v1/images/generations` and `/v1/images/edits`. Keep the exact image aliases requested by your Codex version; a shorter `images` alias only works for clients configured to request it. Codex sends image requests to its active provider's `base_url`, with no separate image URL override. LiteLLM then selects the image deployment independently of the primary text model. ChatGPT image generation and editing require a valid global ChatGPT OAuth login, but a successful login does not establish that the account or backend supports every image operation. Image editing accepts JSON reference images and multipart files, but not masks. Image 2.5 aliases preserve the requested model name; an accepted name does not prove which backend model executed
+
+## Configure Codex through LiteLLM
+
+Codex sends its Responses requests to LiteLLM's `/v1/responses` endpoint. Point the Codex provider at the proxy and select the primary Qwen alias (or the shorter alias you published)
+
+```toml
+model = "qwen3.8-flash-next-codex"
+model_provider = "litellm"
+experimental_realtime_ws_base_url = "https://litellm.example.com/v1"
+experimental_realtime_webrtc_call_base_url = "https://litellm.example.com/v1"
+
+[model_providers.litellm]
+name = "LiteLLM"
+base_url = "https://litellm.example.com/v1"
+wire_api = "responses"
+requires_openai_auth = true
+experimental_bearer_token = "<LITELLM_VIRTUAL_KEY>"
+```
+
+Keep Codex signed in with ChatGPT for its client-side capability checks. `experimental_bearer_token` is the LiteLLM virtual key issued by the proxy. It must never contain the upstream ChatGPT OAuth access or refresh token. `requires_openai_auth = true` enables the Codex OpenAI-auth capability path while LiteLLM remains responsible for the upstream provider credentials
+
+## Configure voice routing
+
+The two realtime settings in the TOML example are root-level Codex settings, not fields inside `[model_providers.litellm]`. `experimental_realtime_ws_base_url` routes the Realtime WebSocket and its sideband through LiteLLM. `experimental_realtime_webrtc_call_base_url` is optional and separately routes HTTP WebRTC call creation. The optional root setting `experimental_realtime_ws_model` overrides the voice model; leave it unset to retain your client's default. An override must match the active protocol: `gpt-realtime-1.5` for legacy Realtime v1/v2 or `gpt-live-1-codex` for frameless Live v3. The base URLs end at `/v1`; LiteLLM adds the protocol-specific path
+
+| Voice operation | LiteLLM path |
+| --- | --- |
+| Legacy Realtime WebSocket | `/v1/realtime` |
+| HTTP WebRTC call creation | `/v1/realtime/calls` |
+| Frameless Live v3 signaling | `/v1/live` and `/v1/live/{call_id}` |
+| Public Live session APIs | `/v1/live/sessions...` |
+
+WebSocket authentication uses `Authorization: Bearer <LITELLM_VIRTUAL_KEY>` by default. If the proxy sets `litellm_key_header_name`, send the virtual key in that configured header instead. Voice is experimental and pending retest: an observed mobile `POST /live` returned 201, but its sideband used the default `api.openai.com` and returned 404. The WebSocket base URL above addresses that routing gap; full bidirectional voice is not verified
+
+## Public Live routes
+
+Use the proxy host in place of `api.openai.com`. Send the configured LiteLLM alias in `session.model` when creating a session, or in the first `session.start` event for a primary WebSocket. Keep the returned session ID unchanged for subsequent operations
+
+| Method | Path | Request and response |
+| --- | --- | --- |
+| POST | `/v1/live/sessions` | JSON `session` and `transport: {type: "webrtc", sdp: "<offer>"}`; returns 201 JSON with `session.id` and `transport.sdp` |
+| POST | `/v1/live/sessions/{session_id}/fork` | JSON WebRTC `transport` and optional `session` overrides; returns 200 JSON with the new session ID and SDP answer |
+| GET | `/v1/live/sessions/{session_id}/content` | Downloads stored recording content without converting it to JSON |
+| POST | `/v1/live/sessions/{session_id}/accept` | JSON `session` with `type: "live"` and model; successful SIP acceptance returns an empty body |
+| POST | `/v1/live/sessions/{session_id}/reject` | JSON with required integer `status_code` from 300 through 699 |
+| POST | `/v1/live/sessions/{session_id}/refer` | JSON with `target_uri` for the SIP destination |
+| POST | `/v1/live/sessions/{session_id}/hangup` | No request body |
+| WebSocket | `/v1/live/sessions` | Start with `session.start`, then wait for `session.started` before sending audio or commands |
+| WebSocket | `/v1/live/sessions/{session_id}/attach` | Attach to an existing session; do not send `session.start` or input audio |
+| WebSocket | `/v1/live/sessions/{session_id}/fork` | Start with `session.start` and a required `session` overrides object, which may be empty |
+
+Public WebRTC creation uses JSON, not the multipart or raw SDP formats used by the Codex compatibility route. `POST /live` and its existing aliases remain available for Codex clients using that format. WebRTC audio travels on media tracks; its data channel carries Live JSON events. Primary WebSocket audio uses base64 chunks in `session.input_audio.append` and `session.output_audio.delta`
+
+The proxy preserves Live event payloads, including nested Responses events inside `response.event`, rather than translating them into Realtime events. Session routing rewrites the configured model alias to the selected upstream model. Audio, transcript, delegation and usage events retain their upstream format. Send `session.close` and wait for `session.closed` to obtain final usage; a disconnected socket alone does not confirm successful finalization
+
+## Availability and verification
+
+Route support does not establish that every configured backend or account supports every operation. The official API describes project API-key authentication; it does not guarantee equivalent capabilities for ChatGPT OAuth. An OAuth request reaching SDP validation proves only that the request reached that validation step. It does not prove a working audio session, recording, fork or SIP call. The routes listed here have not all been tested against a real upstream service
+
+Session controls require a session known to the proxy and owned by the authenticated caller. Incoming SIP calls originate upstream. A proxy administrator can accept or reject the raw ID from a verified incoming-call webhook by supplying `x-litellm-live-model` with an alias that resolves to exactly one deployment. Successful acceptance returns the proxy-owned handle in `x-litellm-live-session-id`, preserving the API's empty response body. Use that handle for subsequent controls. Ordinary virtual keys cannot enroll arbitrary upstream session IDs; a trusted webhook-to-owner enrollment flow is still required for those keys
+
+Live duration uses cumulative `usage.seconds`; legacy Codex milliseconds remain supported. WebRTC initialization has a 15-second minimum credited against running duration, not added to it. Nested terminal Responses usage is charged separately using its backend model and deduplicated by response ID. A failed observation connection cannot establish complete usage. Managed delegation also depends on receiving its backend usage events; the upstream sideband does not replay events emitted before attachment
+
+Managed Responses delegation authorizes its backend model as well as the voice model. Use client delegation when a key's per-model budgets or token/request limits require admission checks for each backend invocation. Restricted-model WebRTC keys must explicitly exclude `session.update` from frontend client events when using managed delegation, because that data channel connects directly to OpenAI and could otherwise change the backend model outside the proxy's checks
+
+For both HTTP and WebSocket forks, restricted-model keys must explicitly set `session.delegation` to `{ "type": "client" }` or `{ "type": "responses", "responses": { "model": "authorized-backend" } }`. Empty overrides cannot safely authorize an inherited backend: the session handle records startup configuration, while later updates may have changed the upstream model. Upstream rules still determine which delegation overrides a source session permits
+
+See the official [Live overview](https://developers.openai.com/api/docs/guides/live), [Live API reference](https://developers.openai.com/api/reference/resources/live), [session management](https://developers.openai.com/api/docs/guides/live-conversations), [WebRTC guide](https://developers.openai.com/api/docs/guides/voice-webrtc?api=live), [WebSocket guide](https://developers.openai.com/api/docs/guides/voice-websockets?api=live), [server controls](https://developers.openai.com/api/docs/guides/voice-server-controls?api=live) and [SIP guide](https://developers.openai.com/api/docs/guides/voice-sip?api=live) for the upstream contract. The voice guides also contain Realtime tabs with different routes and formats

litellm/cost_calculator.py

@@ -8,7 +8,7 @@ from types import MappingProxyType
 from typing import TYPE_CHECKING, Any, Final, Literal, cast
 
 from httpx import Response
-from pydantic import BaseModel, Field, ValidationError
+from pydantic import BaseModel, ValidationError
 from typing_extensions import ReadOnly, TypedDict
 
 import litellm
@@ -2471,6 +2471,13 @@ class RealtimeAPITokenUsageProcessor(BaseTokenUsageProcessor):
                 result["response"].get("usage", {})
             )
             usage_objects.append(usage_object)
+        usage_objects.extend(
+            ResponseAPILoggingUtils._transform_response_api_usage_to_chat_usage(  # pyright: ignore[reportPrivateUsage]  # reuse the existing Responses usage conversion for nested Live events
+                response.usage.model_dump()
+            )
+            for response in _live_backend_responses(results)
+            if response.usage is not None
+        )
         return usage_objects
 
     @staticmethod
@@ -2621,7 +2628,13 @@ def handle_realtime_stream_cost_calculation(
     potential_model_names.append(litellm_model_name)
     input_cost_per_token, output_cost_per_token = _first_priced_realtime_token_costs(
         potential_model_names=potential_model_names,
-        combined_usage_object=combined_usage_object,
+        combined_usage_object=(
+            RealtimeAPITokenUsageProcessor.collect_and_combine_usage_from_realtime_stream_results(
+                [event for event in results if event.get("type") != "response.event"]
+            )
+            if any(event.get("type") == "response.event" for event in results)
+            else combined_usage_object
+        ),
         custom_llm_provider=custom_llm_provider,
         data_residency=data_residency,
     )
@@ -2639,7 +2652,13 @@ def handle_realtime_stream_cost_calculation(
         custom_llm_provider=custom_llm_provider,
         litellm_model_name=litellm_model_name,
     )
-    total_cost: Final = input_cost_per_token + output_cost_per_token + transcription_cost + live_audio_cost
+    backend_cost: Final = sum(
+        _live_backend_response_cost(response, litellm_logging_obj)
+        for response in _live_backend_responses(results, litellm_logging_obj)
+    )
+    total_cost: Final = (
+        input_cost_per_token + output_cost_per_token + transcription_cost + live_audio_cost + backend_cost
+    )
 
     _store_cost_breakdown_in_logging_obj(
         litellm_logging_obj=litellm_logging_obj,
@@ -2649,7 +2668,11 @@ def handle_realtime_stream_cost_calculation(
         total_cost_usd_dollar=total_cost,
         additional_costs={  # mutable-ok: logging cost breakdown requires a concrete dict
             name: cost
-            for name, cost in (("transcription_cost", transcription_cost), ("live_audio_cost", live_audio_cost))
+            for name, cost in (
+                ("transcription_cost", transcription_cost),
+                ("live_audio_cost", live_audio_cost),
+                ("live_backend_cost", backend_cost),
+            )
             if cost > 0
         }
         or None,
@@ -2659,12 +2682,61 @@ def handle_realtime_stream_cost_calculation(
     return total_cost
 
 
-class _LiveSessionDurationUsage(BaseModel):
-    audio_duration_ms: float = Field(strict=True, ge=0, allow_inf_nan=False)
+class _LiveBackendEvent(BaseModel):
+    type: str
+    response: object = None
 
 
-class _LiveSessionClosedEvent(BaseModel):
-    usage: _LiveSessionDurationUsage
+class _LiveBackendEnvelope(BaseModel):
+    event: _LiveBackendEvent
+
+
+def _live_backend_responses(
+    results: OpenAIRealtimeStreamList, logging_obj: LitellmLoggingObject | None = None
+) -> tuple[ResponsesAPIResponse, ...]:
+    responses: Final = {
+        response.id: response
+        for result in results
+        if result.get("type") == "response.event"
+        and (response := _live_backend_response(result, logging_obj)) is not None
+    }
+    return tuple(responses.values())
+
+
+def _mark_live_backend_accounting_incomplete(logging_obj: LitellmLoggingObject | None) -> None:
+    verbose_logger.warning("Live backend accounting incomplete: missing valid terminal usage or model pricing")
+    if logging_obj is not None:
+        logging_obj.model_call_details["realtime_backend_accounting_incomplete"] = True
+
+
+def _live_backend_response(
+    result: Mapping[str, object], logging_obj: LitellmLoggingObject | None
+) -> ResponsesAPIResponse | None:
+    try:
+        event: Final = _LiveBackendEnvelope.model_validate(result).event
+    except ValidationError:
+        return None
+    if event.type not in ("response.completed", "response.incomplete", "response.failed"):
+        return None
+    try:
+        response: Final = ResponsesAPIResponse.model_validate(event.response)
+    except ValidationError:
+        _mark_live_backend_accounting_incomplete(logging_obj)
+        return None
+    if response.usage is None:
+        _mark_live_backend_accounting_incomplete(logging_obj)
+        return None
+    return response
+
+
+def _live_backend_response_cost(response: ResponsesAPIResponse, logging_obj: LitellmLoggingObject | None) -> float:
+    try:
+        return completion_cost(
+            completion_response=response, model=response.model, custom_llm_provider="openai", call_type="aresponses"
+        )
+    except Exception:  # noqa: BLE001  # preserve measured voice cost when backend pricing cannot be resolved
+        _mark_live_backend_accounting_incomplete(logging_obj)
+        return 0.0
 
 
 def handle_live_session_duration_cost(
@@ -2672,18 +2744,40 @@ def handle_live_session_duration_cost(
     custom_llm_provider: str,
     litellm_model_name: str,
 ) -> float:
-    terminal: Final = next((event for event in reversed(results) if event.get("type") == "session.closed"), None)
-    if terminal is None:
-        return 0.0
-    try:
-        usage: Final = _LiveSessionClosedEvent.model_validate(terminal).usage
-    except ValidationError:
-        return 0.0
+    seconds: Final = max(
+        (
+            duration
+            for event in results
+            if event.get("type") in ("session.closed", "session.usage.updated")
+            and (duration := _live_duration_seconds(event)) is not None
+        ),
+        default=0.0,
+    )
+    initialization_seconds: Final = max(
+        (
+            duration
+            for event in results
+            if event.get("type") == "litellm.live.initialization"
+            and (duration := _live_duration_seconds(event)) is not None
+        ),
+        default=0.0,
+    )
     try:
         model_info: Final = litellm.get_model_info(model=litellm_model_name, custom_llm_provider=custom_llm_provider)
     except Exception:
         return 0.0
-    return usage.audio_duration_ms / 1000 * (model_info.get("input_cost_per_second") or 0.0)
+    return max(seconds, initialization_seconds) * (model_info.get("input_cost_per_second") or 0.0)
+
+
+def _live_duration_seconds(event: Mapping[str, object]) -> float | None:
+    from litellm.types.realtime import LiveSessionUsageEvent
+
+    try:
+        usage: Final = LiveSessionUsageEvent.model_validate(event).usage
+    except ValidationError:
+        return None
+    raw_usage: Final = cast(Mapping[str, object], event.get("usage"))
+    return usage.duration / (1 if "seconds" in raw_usage else 1000)
 
 
 def handle_realtime_transcription_cost_calculation(

litellm/litellm_core_utils/realtime_streaming.py

@@ -15,6 +15,7 @@ from litellm._logging import redact_internal_details_from_client_message, verbos
 from litellm.litellm_core_utils.logging_worker import GLOBAL_LOGGING_WORKER
 from litellm.llms.base_llm.realtime.transformation import BaseRealtimeConfig
 from litellm.types.llms.openai import (
+    OpenAILiveResponseEvent,
     OpenAIRealtimeEvents,
     OpenAIRealtimeOutputItemDone,
     OpenAIRealtimeResponseDelta,
@@ -148,6 +149,7 @@ class RealTimeStreaming:
         logging_worker: _LoggingWorker = GLOBAL_LOGGING_WORKER,
         *,
         account_usage: bool = True,
+        live_initialization_seconds: float = 0,
     ):
         self.websocket: _ClientWebSocket = websocket
         self.backend_ws = backend_ws
@@ -155,6 +157,10 @@ class RealTimeStreaming:
         self._logging_worker = logging_worker
         self._account_usage = account_usage
         self.messages: list[OpenAIRealtimeEvents] = []
+        if account_usage and live_initialization_seconds > 0:
+            self.messages.append(
+                {"type": "litellm.live.initialization", "usage": {"seconds": live_initialization_seconds}}
+            )
         self._backend_sent_frames: bool = False
         self.input_message: dict = {}
         self.input_messages: list[dict[str, str]] = []
@@ -266,9 +272,16 @@ class RealTimeStreaming:
         else:
             message_obj = cast(dict[str, Any], json.loads(cast(str, message)))
         self._collect_tool_calls_from_response_done(cast(dict, message_obj))
-        if message_obj.get("type") == "session.closed" and isinstance(message_obj.get("usage"), dict):
+        if message_obj.get("type") in ("session.closed", "session.usage.updated") and isinstance(
+            message_obj.get("usage"), dict
+        ):
             self.messages.append(TypeAdapter(OpenAIRealtimeSessionClosed).validate_python(message_obj))
             return
+        if message_obj.get("type") == "response.event" and isinstance(message_obj.get("event"), dict):
+            nested: Final = message_obj["event"]
+            if nested.get("type") in ("response.completed", "response.incomplete", "response.failed"):
+                self.messages.append(TypeAdapter(OpenAILiveResponseEvent).validate_python(message_obj))
+            return
         if not self._should_store_message(message_obj):
             return
         try:

litellm/llms/chatgpt/live.py

@@ -0,0 +1,193 @@
+import re
+from collections.abc import Mapping
+from dataclasses import dataclass, field
+from types import MappingProxyType
+from typing import TYPE_CHECKING, Final, Literal, TypeAlias
+from unicodedata import category
+from urllib.parse import quote, unquote
+
+import httpx
+from pydantic import JsonValue
+
+from litellm.constants import REALTIME_WEBSOCKET_MAX_MESSAGE_SIZE_BYTES
+from litellm.llms.chatgpt.realtime import (
+    ChatGPTRealtime,
+    configured_realtime_headers,
+    configured_realtime_query,
+    realtime_headers,
+)
+from litellm.llms.custom_httpx.http_handler import (
+    get_async_httpx_client,  # pyright: ignore[reportUnknownVariableType]  # shared client has legacy untyped optional params
+    get_shared_realtime_ssl_context,
+)
+from litellm.types.router import GenericLiteLLMParams
+from litellm.types.utils import LlmProviders
+
+if TYPE_CHECKING:
+    from websockets.asyncio.client import ClientConnection
+
+LiveQuery: TypeAlias = Mapping[str, str | int | float | bool | None | tuple[str | int | float | bool | None, ...]]
+LiveBody: TypeAlias = Mapping[str, JsonValue]
+LiveOperation: TypeAlias = Literal["fork", "accept", "reject", "refer", "hangup", "content", "attach"]
+_PATH: Final = re.compile(r"live/sessions(?:/([^/]+)/(fork|accept|reject|refer|hangup|content|attach))?\Z")
+_ROUTING_QUERY: Final = frozenset(("model", "session_id", "call_id", "api_key", "api_base", "authorization"))
+
+
+@dataclass(frozen=True, slots=True)
+class LiveDeployment:
+    model: str
+    model_id: str | None = None
+    provider: Literal["chatgpt", "openai"] = "chatgpt"
+    api_base: str | None = None
+    api_key: str | None = field(default=None, repr=False)
+    extra_headers: Mapping[str, str] = field(default_factory=lambda: MappingProxyType({}), repr=False)
+    extra_query: LiveQuery = field(default_factory=lambda: MappingProxyType({}), repr=False)
+
+
+def _validate_session_id(session_id: str) -> None:
+    candidate: str = session_id  # rebind-ok: inspect every decoding layer without recursive stack exhaustion
+    while True:
+        if (
+            not candidate
+            or candidate in (".", "..")
+            or any(char in ("/", "\\") or category(char) in ("Cc", "Cs") for char in candidate)
+        ):
+            raise ValueError("Invalid Live session ID")
+        decoded: str = unquote(candidate, errors="strict")  # rebind-ok: validate successive decoding layers iteratively
+        if decoded == candidate:
+            return
+        candidate = decoded  # rebind-ok: each percent-decoding pass reduces the input length
+
+
+def _validate_path(path: str) -> None:
+    match: Final = _PATH.fullmatch(path)
+    if match is None:
+        raise ValueError("Invalid Live endpoint")
+    if match.group(1) is None:
+        return
+    session_id: Final = unquote(match.group(1), errors="strict")
+    if quote(session_id, safe="") != match.group(1):
+        raise ValueError("Noncanonical Live session path")
+    _validate_session_id(session_id)
+
+
+def live_session_path(session_id: str, operation: LiveOperation) -> str:
+    _validate_session_id(session_id)
+    path: Final = f"live/sessions/{quote(session_id, safe='')}/{operation}"
+    _validate_path(path)
+    return path
+
+
+class LiveTransport:
+    def __init__(
+        self,
+        deployment: LiveDeployment,
+        inbound_headers: Mapping[str, str],
+        *,
+        http_client: httpx.AsyncClient | None = None,
+    ) -> None:
+        self.deployment = deployment
+        self._http_client = http_client
+        params: Final = GenericLiteLLMParams.model_validate(
+            MappingProxyType(
+                {
+                    "api_base": deployment.api_base,
+                    "extra_query": deployment.extra_query,
+                }
+            )
+        )
+        self._query = configured_realtime_query(params)
+        self._headers = (
+            realtime_headers(params, inbound_headers, deployment.extra_headers)
+            if deployment.provider == "chatgpt"
+            else MappingProxyType(
+                {
+                    **MappingProxyType(
+                        {
+                            key.lower(): value
+                            for key, value in inbound_headers.items()
+                            if key.lower() in ("openai-alpha", "openai-beta", "x-session-id", "x-oai-attestation")
+                        }
+                    ),
+                    **configured_realtime_headers(deployment.extra_headers),
+                    "authorization": f"Bearer {deployment.api_key or ''}",
+                }
+            )
+        )
+
+    def _url(self, path: str, query: LiveQuery | None, *, websocket: bool) -> str:
+        _validate_path(path)
+        base: Final = httpx.URL(
+            ChatGPTRealtime.get_api_base(self.deployment.api_base)
+            if self.deployment.provider == "chatgpt"
+            else self.deployment.api_base or "https://api.openai.com/v1"
+        )
+        if base.scheme not in ("https", "http", "wss", "ws") or not base.host or base.userinfo or base.fragment:
+            raise ValueError("Invalid Live API base")
+        merged: Final = base.params.merge(query or MappingProxyType({})).merge(self._query)
+        safe_query: Final = tuple(
+            (key, value) for key, value in merged.multi_items() if key.lower() not in _ROUTING_QUERY
+        )
+        return str(
+            base.copy_with(
+                scheme=("wss" if base.scheme in ("https", "wss") else "ws")
+                if websocket
+                else ("https" if base.scheme in ("https", "wss") else "http"),
+                path=f"{base.path.rstrip('/')}/{path}",
+                params=safe_query,
+            )
+        )
+
+    async def request(
+        self,
+        method: str,
+        path: str,
+        body: LiveBody | None = None,
+        query: LiveQuery | None = None,
+    ) -> httpx.Response:
+        if (method, path.rsplit("/", 1)[-1]) not in (
+            ("POST", "sessions"),
+            ("POST", "fork"),
+            ("POST", "accept"),
+            ("POST", "reject"),
+            ("POST", "refer"),
+            ("POST", "hangup"),
+            ("GET", "content"),
+        ):
+            raise ValueError("Invalid Live HTTP operation")
+        url: Final = self._url(path, query, websocket=False)
+        client: Final = (
+            self._http_client
+            or get_async_httpx_client(
+                llm_provider=LlmProviders.CHATGPT if self.deployment.provider == "chatgpt" else LlmProviders.OPENAI
+            ).client
+        )
+        return await client.request(
+            method,
+            url,
+            headers=MappingProxyType({**self._headers, "content-type": "application/json"}),
+            json=dict(body) if body is not None else None,  # mutable-ok: JSON encoder requires a concrete dict
+            timeout=60,
+            follow_redirects=False,
+        )
+
+    async def connect(self, path: str, query: LiveQuery | None = None) -> "ClientConnection":
+        import websockets
+
+        class DirectConnect(websockets.connect):
+            def process_redirect(self, exc: Exception) -> Exception:
+                return exc
+
+        if path != "live/sessions" and path.rsplit("/", 1)[-1] not in ("attach", "fork"):
+            raise ValueError("Invalid Live WebSocket operation")
+        url: Final = self._url(path, query, websocket=True)
+        ssl_context: Final = get_shared_realtime_ssl_context() if url.startswith("wss://") else None
+        return await DirectConnect(
+            url,
+            additional_headers=self._headers,
+            max_size=REALTIME_WEBSOCKET_MAX_MESSAGE_SIZE_BYTES,
+            max_queue=16,
+            ssl=True if ssl_context is False else ssl_context,
+            open_timeout=20,
+            close_timeout=10,
+        )

litellm/proxy/_lazy_features.py

@@ -213,10 +213,15 @@ LAZY_FEATURES: Final[tuple[LazyFeature, ...]] = (
             "/watsonx/",
         ),
     ),
+    LazyFeature(
+        name="live",
+        module_path="litellm.proxy.realtime_endpoints.live",
+        path_prefixes=("/openai/v1/live/sessions", "/v1/live/sessions", "/live/sessions"),
+    ),
     LazyFeature(
         name="realtime",
         module_path="litellm.proxy.realtime_endpoints.endpoints",
-        path_prefixes=("/openai/v1/realtime", "/v1/realtime", "/realtime"),
+        path_prefixes=("/openai/v1/realtime", "/v1/realtime", "/realtime", "/openai/v1/live", "/v1/live", "/live"),
     ),
     LazyFeature(
         name="anthropic_passthrough",

litellm/proxy/_types.py

@@ -412,6 +412,33 @@ class LiteLLMRoutes(enum.Enum):
         "/live",
         "/v1/live",
         "/v1/live/{call_id}",
+        "/openai/v1/live",
+        "/live/{call_id}",
+        "/openai/v1/live/{call_id}",
+        "/live/sessions",
+        "/live/sessions/{session_id}/attach",
+        "/live/sessions/{session_id}/fork",
+        "/live/sessions/{session_id}/content",
+        "/live/sessions/{session_id}/accept",
+        "/live/sessions/{session_id}/reject",
+        "/live/sessions/{session_id}/refer",
+        "/live/sessions/{session_id}/hangup",
+        "/v1/live/sessions",
+        "/v1/live/sessions/{session_id}/attach",
+        "/v1/live/sessions/{session_id}/fork",
+        "/v1/live/sessions/{session_id}/content",
+        "/v1/live/sessions/{session_id}/accept",
+        "/v1/live/sessions/{session_id}/reject",
+        "/v1/live/sessions/{session_id}/refer",
+        "/v1/live/sessions/{session_id}/hangup",
+        "/openai/v1/live/sessions",
+        "/openai/v1/live/sessions/{session_id}/attach",
+        "/openai/v1/live/sessions/{session_id}/fork",
+        "/openai/v1/live/sessions/{session_id}/content",
+        "/openai/v1/live/sessions/{session_id}/accept",
+        "/openai/v1/live/sessions/{session_id}/reject",
+        "/openai/v1/live/sessions/{session_id}/refer",
+        "/openai/v1/live/sessions/{session_id}/hangup",
         # realtime (GA WebRTC HTTP routes)
         "/realtime/client_secrets",
         "/v1/realtime/client_secrets",

litellm/proxy/auth/auth_utils.py

@@ -1705,6 +1705,7 @@ _MODEL_ROUTING_COMPLETION_MODEL_ROUTE_MARKERS: Final = ("/evals",)
 _MODEL_ROUTING_SESSION_MODEL_ROUTE_MARKERS: Final = (
     "/realtime/client_secrets",
     "/realtime/calls",
+    "/live/sessions",
 )
 _MODEL_ROUTING_ID_FIELDS: Final = (
     "file_id",
@@ -1866,15 +1867,15 @@ def _extract_model_candidates_from_request(
     uses_completion_model_sources: Final = _route_matches_any_marker(
         route=route, markers=_MODEL_ROUTING_COMPLETION_MODEL_ROUTE_MARKERS
     )
-    session: Final[object] = (
-        request_data.get("session")
-        if _route_matches_any_marker(route=route, markers=_MODEL_ROUTING_SESSION_MODEL_ROUTE_MARKERS)
-        else None
-    )
+    uses_session_model: Final = _route_matches_any_marker(
+        route=route, markers=_MODEL_ROUTING_SESSION_MODEL_ROUTE_MARKERS
+    ) or route.rstrip("/") in ("/live", "/v1/live", "/openai/v1/live")
+    session: Final[object] = request_data.get("session") if uses_session_model else None
     parsed_session: Final[object] = safe_json_loads(session) if isinstance(session, str) else session
     session_model: Final[object] = parsed_session.get("model") if isinstance(parsed_session, dict) else None
     if (
-        _route_matches_any_marker(route=route, markers=("/realtime/calls",))
+        uses_session_model
+        and not _route_matches_any_marker(route=route, markers=("/realtime/client_secrets",))
         and isinstance(session_model, str)
         and session_model
     ):
@@ -1885,7 +1886,7 @@ def _extract_model_candidates_from_request(
     _append_model_candidates(candidates, body_model)
     if uses_body_target_model_sources or not body_model:
         _append_model_candidates(candidates, request_data.get("target_model_names"))
-    if _route_matches_any_marker(route=route, markers=_MODEL_ROUTING_SESSION_MODEL_ROUTE_MARKERS):
+    if uses_session_model:
         _append_model_candidates(candidates, session_model)
     if uses_completion_model_sources and isinstance(request_data.get("completion"), dict):
         _append_model_candidates(candidates, request_data["completion"].get("model"))

litellm/proxy/proxy_server.py

@@ -11920,7 +11920,12 @@ async def _reject_realtime_session(
 
 _CODEX_LIVE_AUTH_DEPENDENCY: Final = Depends(user_api_key_auth_websocket)
 
+reserve_lazy_slot(app, "live")
+reserve_lazy_slot(app, "realtime")
 
+
+@app.websocket("/openai/v1/live/{call_id}")
+@app.websocket("/live/{call_id}")
 @app.websocket("/v1/live/{call_id}")
 async def codex_live_sideband_endpoint(
     websocket: WebSocket,
@@ -11934,6 +11939,7 @@ async def codex_live_sideband_endpoint(
 
 @app.websocket("/v1/live")
 @app.websocket("/live")
+@app.websocket("/openai/v1/live")
 @app.websocket("/openai/v1/realtime")
 @app.websocket("/v1/realtime")
 @app.websocket("/realtime")

litellm/proxy/realtime_endpoints/call_sessions.py

@@ -356,7 +356,13 @@ async def _create_codex_realtime_call(request: Request) -> Response:
             valid_token=auth,
             llm_router=server.llm_router,
         )
-        data: Final = build_call_request(offer, request.query_params, request.headers)
+        live_signaling: Final = request.url.path.rstrip("/") in ("/live", "/v1/live", "/openai/v1/live")
+        query: Final = (
+            MappingProxyType({"intent": "quicksilver", "architecture": "avas", **request.query_params})
+            if live_signaling
+            else request.query_params
+        )
+        data: Final = build_call_request(offer, query, request.headers)
         signaling_auth: Final = auth.model_copy(update=MappingProxyType({"budget_reservation": None}))
         if isinstance(limiter, _PROXY_MaxParallelRequestsHandler) and (
             auth.max_parallel_requests is not None
@@ -409,7 +415,9 @@ async def _create_codex_realtime_call(request: Request) -> Response:
             response.content,
             status_code=response.status_code,
             media_type="application/sdp",
-            headers=MappingProxyType({"Location": f"/v1/realtime/calls/{token}"}),
+            headers=MappingProxyType(
+                {"Location": f"/v1/live/{token}" if live_signaling else f"/v1/realtime/calls/{token}"}
+            ),
         )
     finally:
         try:

litellm/proxy/realtime_endpoints/call_supervision.py

@@ -3,7 +3,7 @@ from collections.abc import AsyncIterator, Awaitable, Callable
 from contextlib import suppress
 from typing import Final, Protocol
 
-from pydantic import BaseModel, Field, ValidationError
+from pydantic import BaseModel, ValidationError
 from websockets.exceptions import ConnectionClosedOK
 
 from litellm._logging import verbose_proxy_logger
@@ -16,6 +16,7 @@ from litellm.proxy.spend_tracking.budget_reservation import (
     invalidate_budget_reservation_counters,
     release_or_invalidate_budget_reservation,
 )
+from litellm.types.realtime import LiveSessionUsageEvent
 
 
 class ObserverSocket(Protocol):
@@ -34,14 +35,6 @@ class _ObserverEvent(BaseModel):
     type: str
 
 
-class _LiveDurationUsage(BaseModel):
-    audio_duration_ms: float = Field(strict=True, ge=0, allow_inf_nan=False)
-
-
-class _LiveTerminalEvent(BaseModel):
-    usage: _LiveDurationUsage
-
-
 class CallSupervisor:
     def __init__(
         self,
@@ -57,6 +50,7 @@ class CallSupervisor:
         termination_timeout: float = 60,
         logging_timeout: float = LOGGING_WORKER_MAX_TIME_PER_COROUTINE,
         terminal_usage_required: bool = True,
+        connected_ready: bool = False,
         force_close_call: Callable[[], Awaitable[None]] | None = None,
         lease: RealtimeCallLease | None = None,
     ) -> None:
@@ -75,7 +69,9 @@ class CallSupervisor:
         self._terminal_usage_required = terminal_usage_required
         self._ready = asyncio.Event()
         self._stop = asyncio.Event()
-        self._started = False
+        self._started = connected_ready
+        if connected_ready:
+            self._ready.set()
         self._terminal = False
         self._terminal_usage_valid = False
         self._close_confirmed = False
@@ -128,7 +124,7 @@ class CallSupervisor:
             if event.type == "session.closed":
                 self._terminal = True
                 try:
-                    _LiveTerminalEvent.model_validate_json(message)
+                    LiveSessionUsageEvent.model_validate_json(message)
                 except ValidationError:
                     self._terminal_usage_valid = False
                 else:
@@ -206,7 +202,9 @@ class CallSupervisor:
                         await asyncio.wait_for(
                             self._stream.log_messages(wait_for_dispatch=True), timeout=self._logging_timeout
                         )
-                        self._accounting_complete = True
+                        self._accounting_complete = not bool(
+                            self._logging.model_call_details.get("realtime_backend_accounting_incomplete")
+                        )
                     except asyncio.TimeoutError:
                         verbose_proxy_logger.error("Realtime observer timed out dispatching usage accounting")
                     finally:

litellm/proxy/realtime_endpoints/endpoints.py

@@ -362,6 +362,15 @@ async def create_realtime_client_secret(
     return RealtimeClientSecretResponse(**upstream_json)
 
 
+@router.post("/v1/live", tags=["realtime"])
+@router.post("/live", tags=["realtime"])
+@router.post("/openai/v1/live", tags=["realtime"])
+async def proxy_live_calls(request: Request) -> Response:
+    from litellm.proxy.realtime_endpoints.call_sessions import create_codex_realtime_call
+
+    return await create_codex_realtime_call(request)
+
+
 @router.post(
     "/v1/realtime/calls",
     tags=["realtime"],

litellm/types/llms/openai.py

@@ -2037,10 +2037,15 @@ class OpenAIRealtimeStreamResponseBaseObject(TypedDict):
 
 
 class OpenAIRealtimeSessionClosed(TypedDict):
-    type: ReadOnly[Literal["session.closed"]]
+    type: ReadOnly[Literal["session.closed", "session.usage.updated", "litellm.live.initialization"]]
     usage: ReadOnly[Mapping[str, object]]
 
 
+class OpenAILiveResponseEvent(TypedDict):
+    type: ReadOnly[Literal["response.event"]]
+    event: ReadOnly[Mapping[str, object]]
+
+
 class OpenAIRealtimeConversationObject(TypedDict, total=False):
     id: str
     object: Required[Literal["realtime.conversation"]]
@@ -2295,6 +2300,7 @@ class OpenAIRealtimeEventTypes(Enum):
 OpenAIRealtimeEvents = (
     OpenAIRealtimeStreamResponseBaseObject
     | OpenAIRealtimeSessionClosed
+    | OpenAILiveResponseEvent
     | OpenAIRealtimeStreamSessionEvents
     | OpenAIRealtimeStreamResponseOutputItemAdded
     | OpenAIRealtimeResponseContentPartAdded

litellm/types/realtime.py

@@ -1,6 +1,6 @@
 from typing import Any, Literal
 
-from pydantic import BaseModel
+from pydantic import AliasChoices, BaseModel, Field
 from typing_extensions import ReadOnly, TypedDict
 
 from .llms.openai import (
@@ -12,6 +12,16 @@ from .llms.openai import (
 ALL_DELTA_TYPES = Literal["text", "audio"]
 
 
+class LiveSessionDurationUsage(BaseModel):
+    duration: float = Field(
+        strict=True, ge=0, allow_inf_nan=False, validation_alias=AliasChoices("seconds", "audio_duration_ms")
+    )
+
+
+class LiveSessionUsageEvent(BaseModel):
+    usage: LiveSessionDurationUsage
+
+
 class RealtimeResponseTransformInput(TypedDict):
     session_configuration_request: str | None
     current_output_item_id: (

tests/test_litellm/litellm_core_utils/test_realtime_streaming.py

@@ -3590,3 +3590,43 @@ async def test_provider_bytes_are_sent_raw_after_pacing():
 
     assert [call.args[0] for call in backend_ws.send.await_args_list] == [b"\x00\x01", '{"type":"endStream"}']
     provider_config.pace_backend_send.assert_awaited_once_with(b"\x00\x01")
+
+
+def test_public_live_accounting_survives_filtered_logging(monkeypatch):
+    monkeypatch.setattr(litellm, "logged_real_time_event_types", [])
+    stream = RealTimeStreaming(MagicMock(), MagicMock(), MagicMock())
+    events = [
+        {"type": "session.usage.updated", "usage": {"seconds": 15}},
+        {
+            "type": "response.event",
+            "event": {
+                "type": "response.completed",
+                "response": {
+                    "id": "resp_one",
+                    "model": "gpt-backend",
+                    "usage": {"total_tokens": 12},
+                },
+            },
+        },
+        {"type": "session.closed", "usage": {"seconds": 30}},
+    ]
+    for event in events:
+        stream.store_message({**event, "private_transcript": "do not retain"})
+    stream.store_message(
+        {"type": "response.event", "event": {"type": "response.output_text.delta", "delta": "private"}}
+    )
+    assert stream.messages == events
+
+
+@pytest.mark.parametrize("account_usage,expected", [(True, 1), (False, 0)])
+def test_live_initialization_is_retained_only_by_accounting_owner(account_usage, expected):
+    stream = RealTimeStreaming(
+        MagicMock(),
+        MagicMock(),
+        MagicMock(),
+        account_usage=account_usage,
+        live_initialization_seconds=15,
+    )
+    assert len(stream.messages) == expected
+    if account_usage:
+        assert stream.messages == [{"type": "litellm.live.initialization", "usage": {"seconds": 15}}]

tests/test_litellm/llms/chatgpt/test_live.py

@@ -0,0 +1,194 @@
+import json
+
+import httpx
+import pytest
+
+from litellm.llms.chatgpt.live import LiveDeployment, LiveTransport, live_session_path
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("provider", ["chatgpt", "openai"])
+@pytest.mark.parametrize("status", [201, 403, 429, 503])
+async def test_live_request_preserves_payload_status_and_selected_credentials(provider, status, chatgpt_tokens):
+    payload = {
+        "session": {"model": "deployment-model", "tools": [{"type": "function", "name": "lookup"}]},
+        "transport": {"type": "webrtc", "sdp": "v=0\r\n"},
+        "future_option": {"nested": [True, None, 3]},
+    }
+
+    def respond(request):
+        assert request.url.path == "/custom/v1/live/sessions"
+        assert request.headers["authorization"] == (
+            "Bearer test-token-default" if provider == "chatgpt" else "Bearer deployment-key"
+        )
+        assert request.headers.get("chatgpt-account-id") == ("test-account-default" if provider == "chatgpt" else None)
+        assert request.headers["x-gateway"] == "configured"
+        assert request.headers["openai-beta"] == "feature=v1"
+        assert "cookie" not in request.headers
+        assert json.loads(request.content) == payload
+        assert request.url.params.get_list("tag") == ["a +/&", "b"]
+        assert request.url.params["gateway"] == "trusted"
+        assert request.url.params["cursor"] == "opaque +/&"
+        assert not {"model", "call_id", "session_id", "api_key"}.intersection(request.url.params)
+        return httpx.Response(status, json={"result": "upstream"}, headers={"x-request-id": "provider-id"})
+
+    async with httpx.AsyncClient(transport=httpx.MockTransport(respond)) as client:
+        transport = LiveTransport(
+            LiveDeployment(
+                model="deployment-model",
+                provider=provider,
+                api_key="deployment-key",
+                api_base="https://gateway.example/custom/v1/?gateway=base",
+                extra_headers={"x-gateway": "configured", "Authorization": "bad", "ChatGPT-Account-Id": "bad"},
+                extra_query={"gateway": "trusted", "tag": ("a +/&", "b"), "model": "bad", "session_id": "bad"},
+            ),
+            {"Authorization": "Bearer proxy-key", "Cookie": "private", "OpenAI-Beta": "feature=v1"},
+            http_client=client,
+        )
+        response = await transport.request(
+            "POST",
+            "live/sessions",
+            payload,
+            {"gateway": "untrusted", "cursor": "opaque +/&", "call_id": "bad", "api_key": "bad"},
+        )
+        assert response.status_code == status
+        assert response.json() == {"result": "upstream"}
+        assert response.headers["x-request-id"] == "provider-id"
+        assert not client.is_closed
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("operation", ["fork", "accept", "reject", "refer", "hangup", "content"])
+async def test_live_all_http_operations(operation):
+    def respond(request):
+        assert request.url.path == f"/v1/live/sessions/sess_new-ID/{operation}"
+        assert request.method == ("GET" if operation == "content" else "POST")
+        assert request.url.params["output_format"] == "json"
+        return httpx.Response(204)
+
+    async with httpx.AsyncClient(transport=httpx.MockTransport(respond)) as client:
+        transport = LiveTransport(LiveDeployment("model", provider="openai", api_key="key"), {}, http_client=client)
+        response = await transport.request(
+            "GET" if operation == "content" else "POST",
+            live_session_path("sess_new-ID", operation),
+            query={"output_format": "json"},
+        )
+        assert response.status_code == 204
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("path", ["live/sessions", "live/sessions/sess_1/attach", "live/sessions/sess_1/fork"])
+async def test_live_websocket_paths_bounds_and_auth(path, chatgpt_tokens):
+    from websockets.asyncio.server import serve
+
+    async def observe(connection):
+        assert connection.request.headers["Authorization"] == "Bearer test-token-default"
+        await connection.send(connection.request.path)
+
+    async with serve(observe, "127.0.0.1", 0) as server:
+        port = server.sockets[0].getsockname()[1]
+        transport = LiveTransport(
+            LiveDeployment("model", api_base=f"http://127.0.0.1:{port}/v1", extra_query={"route": "a+&b"}),
+            {"Authorization": "Bearer proxy-key"},
+        )
+        connection = await transport.connect(path, {"checkpoint": "opaque+value"})
+        try:
+            received = await connection.recv()
+            url = httpx.URL(f"http://127.0.0.1{received}")
+            assert url.path == f"/v1/{path}"
+            assert url.params["route"] == "a+&b"
+            assert url.params["checkpoint"] == "opaque+value"
+        finally:
+            await connection.close()
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "path",
+    [
+        "https://evil.example/live/sessions",
+        "//evil.example/live/sessions",
+        "live/sessions/../accept",
+        "live/sessions/sess%2Fbad/accept",
+        "live/sessions/sess%5Cbad/accept",
+        "live/sessions/sess%252Fbad/accept",
+        "live/sessions/%252e%252e/accept",
+        "live/sessions/%2E%2E/accept",
+        "live/sessions/sess%00bad/accept",
+        "live/sessions/sess%0Abad/accept",
+        "live/sessions/sess.foo?query/accept",
+        "live/sessions/sess_1/accept?url=https://evil.example",
+        "live/sessions/sess_1/accept#fragment",
+        "live/sessions/sess_1/accept\n",
+    ],
+)
+async def test_live_rejects_noncanonical_paths_before_network(path):
+    transport = LiveTransport(LiveDeployment("model", provider="openai", api_key="key"), {})
+    with pytest.raises(ValueError, match=r"(?:Invalid|Noncanonical) Live"):
+        await transport.request("POST", path)
+    with pytest.raises(ValueError, match=r"(?:Invalid|Noncanonical) Live"):
+        await transport.connect(path)
+
+
+@pytest.mark.parametrize(
+    "session_id",
+    ["../x", "sess/x", "sess%2Fx", "sess\\x", "sess%255cx", ".", "..", "%252e%252e", "", "sess\n", "sess\x00"],
+)
+def test_live_session_ids_cannot_inject_path_or_query(session_id):
+    with pytest.raises(ValueError, match="Invalid Live session ID"):
+        live_session_path(session_id, "content")
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("session_id", ["sess.foo", "sess-\u00f1\u4e2d", "sess?x#y", "sess 50%", "x" * 1024])
+async def test_live_preserves_opaque_session_ids(session_id):
+    from urllib.parse import quote
+
+    def respond(request):
+        assert request.url.raw_path == f"/v1/live/sessions/{quote(session_id, safe='')}/content".encode()
+        assert request.url.params == httpx.QueryParams()
+        return httpx.Response(200, json={"session_id": session_id})
+
+    async with httpx.AsyncClient(transport=httpx.MockTransport(respond)) as client:
+        transport = LiveTransport(LiveDeployment("model", provider="openai", api_key="key"), {}, http_client=client)
+        response = await transport.request("GET", live_session_path(session_id, "content"))
+        assert response.json()["session_id"] == session_id
+
+
+@pytest.mark.asyncio
+async def test_live_does_not_redirect_credentials():
+    def respond(request):
+        assert request.url.host == "api.openai.com"
+        return httpx.Response(307, headers={"location": "https://elsewhere.example/collect"})
+
+    async with httpx.AsyncClient(transport=httpx.MockTransport(respond), follow_redirects=True) as client:
+        transport = LiveTransport(LiveDeployment("model", provider="openai", api_key="key"), {}, http_client=client)
+        response = await transport.request("POST", "live/sessions", {})
+        assert response.status_code == 307
+
+
+@pytest.mark.asyncio
+async def test_live_websocket_does_not_redirect_credentials():
+    from websockets.asyncio.server import serve
+    from websockets.datastructures import Headers
+    from websockets.exceptions import InvalidStatus
+    from websockets.http11 import Response
+
+    async def unused(connection):
+        pytest.fail("Redirected WebSocket must never open")
+
+    def redirect(connection, request):
+        assert request.headers["authorization"] == "Bearer deployment-key"
+        return Response(307, "Temporary Redirect", Headers({"Location": "/elsewhere"}))
+
+    async with serve(unused, "127.0.0.1", 0, process_request=redirect) as server:
+        port = server.sockets[0].getsockname()[1]
+        transport = LiveTransport(
+            LiveDeployment(
+                "model", provider="openai", api_key="deployment-key", api_base=f"http://127.0.0.1:{port}/v1"
+            ),
+            {},
+        )
+        with pytest.raises(InvalidStatus) as failure:
+            await transport.connect("live/sessions")
+        assert failure.value.response.status_code == 307

tests/test_litellm/proxy/auth/test_auth_utils.py

@@ -572,7 +572,11 @@ def _azure_relay_router():
         model_list=[
             {
                 "model_name": "gpt",
-                "litellm_params": {"model": "azure_ai/gpt-5.4-mini", "api_base": "https://a.services.ai.azure.com", "api_key": "k"},
+                "litellm_params": {
+                    "model": "azure_ai/gpt-5.4-mini",
+                    "api_base": "https://a.services.ai.azure.com",
+                    "api_key": "k",
+                },
             },
             {
                 "model_name": "other-group",
@@ -966,11 +970,24 @@ def test_get_model_from_request_extracts_realtime_session_model(route, encoded):
 
 
 @pytest.mark.parametrize("session", ['{"model":"actual-voice"}', {"model": "actual-voice"}])
-def test_realtime_calls_auth_uses_executed_session_model_despite_decoys(session):
+@pytest.mark.parametrize(
+    "route",
+    [
+        "/v1/realtime/calls",
+        "/v1/live",
+        "/live",
+        "/openai/v1/live",
+        "/v1/live/sessions",
+        "/live/sessions",
+        "/openai/v1/live/sessions",
+        "/v1/live/sessions/incoming/accept",
+    ],
+)
+def test_realtime_calls_auth_uses_executed_session_model_despite_decoys(session, route):
     assert (
         get_model_from_request(
             request_data={"model": "body-decoy", "session": session},
-            route="/v1/realtime/calls",
+            route=route,
             request_query_params={"model": "query-decoy"},
             request_headers={"x-litellm-model": "header-decoy"},
         )

tests/test_litellm/proxy/realtime_endpoints/test_call_sessions.py

@@ -178,7 +178,9 @@ async def test_mixed_case_offer_preserves_boundary_metadata_and_closes_extra_fil
 @pytest.mark.parametrize("policy", ["budget", "personal_models"])
 @pytest.mark.parametrize("mixed_case", [False, True])
 @pytest.mark.parametrize("pre_read", [False, True])
-async def test_offer_auth_enforces_session_model_policy_before_upstream(monkeypatch, multipart, policy, mixed_case, pre_read):
+async def test_offer_auth_enforces_session_model_policy_before_upstream(
+    monkeypatch, multipart, policy, mixed_case, pre_read
+):
     import json
     from unittest.mock import AsyncMock, MagicMock
 
@@ -557,8 +559,9 @@ async def test_realtime_endpoint_rejects_untrusted_call_ids(monkeypatch, call_id
 @pytest.mark.parametrize("multipart", [False, True])
 @pytest.mark.parametrize("credential", ["authorization", "api-key", "subprotocol", "x-litellm-api-key", "custom"])
 @pytest.mark.parametrize("signaling_credential", ["authorization", "api-key", "x-litellm-api-key", "mixed"])
+@pytest.mark.parametrize("signaling_path", ["/v1/realtime/calls", "/v1/live", "/live", "/openai/v1/live"])
 async def test_offer_exchange_wraps_call_and_filters_client_headers(
-    monkeypatch, multipart, credential, signaling_credential
+    monkeypatch, multipart, credential, signaling_credential, signaling_path
 ):
     import json
     from unittest.mock import AsyncMock
@@ -596,7 +599,7 @@ async def test_offer_exchange_wraps_call_and_filters_client_headers(
         {
             "type": "http",
             "method": "POST",
-            "path": "/v1/realtime/calls",
+            "path": signaling_path,
             "scheme": "http",
             "server": ("localhost", 80),
             "query_string": b"intent=quicksilver&architecture=avas&untrusted=bad",
@@ -672,6 +675,8 @@ async def test_offer_exchange_wraps_call_and_filters_client_headers(
     monkeypatch.setattr(codex, "supervise_codex_call", supervise)
     response = await codex.create_codex_realtime_call(request)
     assert response.status_code == 201
+    expected_prefix = "/v1/realtime/calls/" if signaling_path == "/v1/realtime/calls" else "/v1/live/"
+    assert response.headers["location"].startswith(expected_prefix)
     assert response.body == b"v=0\r\nanswer"
     token = response.headers["location"].rsplit("/", 1)[-1]
     call = codex.decode_call(token, "Bearer owner")
@@ -1147,7 +1152,9 @@ async def test_signaling_rejection_after_admission_refunds_parallel_slot(monkeyp
 
     class Reject(CustomLogger):
         async def async_pre_call_hook(self, user_api_key_dict, cache, data, call_type):
-            current = await proxy.internal_usage_cache.async_get_cache(key, litellm_parent_otel_span=None, local_only=True)
+            current = await proxy.internal_usage_cache.async_get_cache(
+                key, litellm_parent_otel_span=None, local_only=True
+            )
             assert limiter._gauge_in_flight_from_cache_value(current) == 1
             raise RuntimeError("Policy rejected after admission")
 

tests/test_litellm/proxy/realtime_endpoints/test_call_supervision.py

@@ -187,7 +187,8 @@ class Sink:
 @pytest.mark.parametrize(
     "duration,valid", [(0, True), (1000, True), (None, False), (-1, False), (True, False), ("1000", False)]
 )
-async def test_live_terminal_requires_valid_duration_for_accounting(monkeypatch, duration, valid):
+@pytest.mark.parametrize("duration_field", ["audio_duration_ms", "seconds"])
+async def test_live_terminal_requires_valid_duration_for_accounting(monkeypatch, duration, valid, duration_field):
     from litellm.proxy.realtime_endpoints import call_supervision
 
     socket = Socket()
@@ -201,7 +202,7 @@ async def test_live_terminal_requires_valid_duration_for_accounting(monkeypatch,
     await socket.messages.put({"type": "session.started"})
     await supervisor.start()
     await socket.messages.put(
-        {"type": "session.closed", **({"usage": {"audio_duration_ms": duration}} if duration is not None else {})}
+        {"type": "session.closed", **({"usage": {duration_field: duration}} if duration is not None else {})}
     )
     await supervisor.wait()
     close.assert_not_awaited()
@@ -695,3 +696,58 @@ async def test_ga_observer_disconnect_requires_confirmed_hangup(closure, hangup_
     assert bool(logger.model_call_details.get("realtime_usage_incomplete")) == (not hangup_succeeds)
     assert sink.logs == 1
     assert socket.closed
+
+
+@pytest.mark.asyncio
+async def test_live_connected_attach_is_ready_without_session_started_and_bills_once():
+    socket = Socket()
+    logger = MagicMock(spec=Logging)
+    logger.model_call_details = {}
+    sink = Sink(logger)
+
+    async def close():
+        await socket.messages.put({"type": "session.closed", "usage": {"seconds": 30}})
+
+    supervisor = CallSupervisor(
+        socket,
+        sink,
+        logger,
+        UserAPIKeyAuth(),
+        close,
+        connected_ready=True,
+        ready_timeout=0.1,
+    )
+    await supervisor.start()
+    await socket.messages.put({"type": "session.usage.updated", "usage": {"seconds": 15}})
+    await supervisor.close()
+    await supervisor.close()
+    assert sink.logs == 1
+    assert sink.events[-1] == {"type": "session.closed", "usage": {"seconds": 30}}
+    assert "realtime_usage_incomplete" not in logger.model_call_details
+    assert socket.closed
+
+
+@pytest.mark.asyncio
+async def test_live_missing_backend_accounting_invalidates_budget_after_dispatch(monkeypatch):
+    from litellm.proxy.realtime_endpoints import call_supervision
+
+    socket = Socket()
+    logger = MagicMock(spec=Logging)
+    logger.model_call_details = {}
+    invalidate = AsyncMock()
+    monkeypatch.setattr(call_supervision, "invalidate_budget_reservation_counters", invalidate)
+
+    class IncompleteSink(Sink):
+        async def log_messages(self, *, wait_for_dispatch=False):
+            await super().log_messages(wait_for_dispatch=wait_for_dispatch)
+            logger.model_call_details["realtime_backend_accounting_incomplete"] = True
+
+    sink = IncompleteSink(logger)
+    supervisor = CallSupervisor(socket, sink, logger, UserAPIKeyAuth(), AsyncMock(), connected_ready=True)
+    await supervisor.start()
+    await socket.messages.put({"type": "session.closed", "usage": {"seconds": 30}})
+    await supervisor.wait()
+    assert sink.logs == 1
+    assert logger.model_call_details["realtime_accounting_incomplete"] is True
+    assert "realtime_usage_incomplete" not in logger.model_call_details
+    invalidate.assert_awaited_once()

tests/test_litellm/proxy/realtime_endpoints/test_live.py

@@ -0,0 +1,930 @@
+import json
+import time
+from contextlib import asynccontextmanager
+from types import MappingProxyType, SimpleNamespace
+from unittest.mock import AsyncMock, Mock
+
+import httpx
+import pytest
+from fastapi import FastAPI, HTTPException, Request
+from fastapi.testclient import TestClient
+
+from litellm.llms.chatgpt.live import LiveDeployment
+from litellm.proxy._types import UserAPIKeyAuth
+from litellm.proxy.realtime_endpoints import live
+
+
+@pytest.fixture(autouse=True)
+def encryption_key(monkeypatch):
+    monkeypatch.setenv("LITELLM_SALT_KEY", "test-live-encryption-key")
+
+
+def handle(owner="owner"):
+    return live._new_handle(
+        "sess_upstream",
+        "voice",
+        LiveDeployment(model="gpt-live"),
+        UserAPIKeyAuth(api_key=owner),
+        None,
+    )
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("field", ["chatgpt_auth_profile", "chatgpt_token_dir", "chatgpt_auth_file"])
+async def test_live_rejects_unsupported_deployment_credentials(monkeypatch, field):
+    from litellm.proxy import proxy_server
+
+    router = SimpleNamespace(
+        async_get_available_deployment=AsyncMock(
+            return_value={
+                "litellm_params": {"model": "chatgpt/gpt-live-1", field: "other-account"},
+                "model_info": {"id": "voice"},
+            }
+        ),
+        async_routing_strategy_pre_call_checks=AsyncMock(),
+    )
+    monkeypatch.setattr(proxy_server, "llm_router", router)
+    with pytest.raises(HTTPException) as rejected:
+        await live._deployment("voice", {})
+    assert rejected.value.status_code == 400
+    assert "deployment auth overrides are unsupported" in rejected.value.detail
+
+
+def test_session_tokens_hide_credentials_and_enforce_owner_expiry_and_integrity():
+    original = handle()
+    token = live.encode_session(original)
+    assert "deployment-a" not in token and "sess_upstream" not in token
+    assert live.decode_session(token, live._owner(UserAPIKeyAuth(api_key="owner"))) == original
+    for candidate, owner in (
+        (token, "other-owner"),
+        ("sess_upstream", original.owner),
+        (token[:-8] + "aaaaaaaa", original.owner),
+        (live.encode_session(original.model_copy(update={"expires_at": time.time() - 1})), original.owner),
+    ):
+        with pytest.raises(HTTPException) as rejected:
+            live.decode_session(candidate, owner)
+        assert rejected.value.status_code == 403
+
+
+def test_handle_serializes_mappingproxy_without_losing_pinned_deployment():
+    deployment = LiveDeployment(
+        model="gpt-live",
+        model_id="deployment-a",
+        api_base="https://upstream.test/v1",
+        extra_headers=MappingProxyType({"openai-beta": "test"}),
+        extra_query=MappingProxyType({"architecture": "test"}),
+    )
+    original = live._new_handle("sess_upstream", "voice", deployment, UserAPIKeyAuth(api_key="owner"), None)
+    assert live._pinned(live.decode_session(live.encode_session(original), original.owner)) == deployment
+
+
+def test_only_protocol_session_ids_are_rewritten_and_application_values_survive():
+    event = {
+        "type": "session.started",
+        "session": {"id": "raw", "instructions": "raw"},
+        "session_id": "raw",
+        "delta": "raw",
+        "event": {"type": "response.output_text.delta", "delta": "raw", "session_id": "raw"},
+    }
+    rewritten = live.rewrite_session_ids(event, "raw", "public")
+    assert rewritten["session"]["id"] == "public"
+    assert rewritten["session_id"] == "public"
+    assert rewritten["session"]["instructions"] == "raw"
+    assert rewritten["delta"] == "raw"
+    assert rewritten["event"] == event["event"]
+    assert event["session"]["id"] == "raw"
+
+
+@pytest.fixture
+def route_client(monkeypatch):
+    auth = UserAPIKeyAuth(api_key="owner")
+    deployment = LiveDeployment(model="gpt-live", provider="openai", api_key="upstream-key", model_id="deployment-a")
+    transport = SimpleNamespace(
+        request=AsyncMock(
+            return_value=httpx.Response(
+                201, json={"session": {"id": "sess_upstream"}, "transport": {"type": "webrtc", "sdp": "answer"}}
+            )
+        )
+    )
+    selected = AsyncMock(return_value=deployment)
+    supervised = AsyncMock()
+    authenticated_bodies = []
+
+    async def authenticate(request):
+        authenticated_bodies.append(await request.json())
+        return auth
+
+    @asynccontextmanager
+    async def precall(request, auth, model, **kwargs):
+        yield live._Prepared({"model": model}, Mock(), None, kwargs.get("ownership"))
+
+    monkeypatch.setattr(live, "_auth", authenticate)
+    monkeypatch.setattr(live, "_precall", precall)
+    monkeypatch.setattr(live, "_deployment", selected)
+    monkeypatch.setattr(live, "_supervise", supervised)
+    factory = Mock(return_value=transport)
+    monkeypatch.setattr(live, "LiveTransport", factory)
+    app = FastAPI()
+    app.include_router(live.router)
+    return SimpleNamespace(
+        client=TestClient(app),
+        transport=transport,
+        selected=selected,
+        supervised=supervised,
+        auth=auth,
+        factory=factory,
+        bodies=authenticated_bodies,
+    )
+
+
+@pytest.mark.parametrize("prefix", ["/v1/live", "/live", "/openai/v1/live"])
+def test_create_preserves_configuration_and_returns_owned_json_session(route_client, prefix):
+    body = {
+        "session": {
+            "model": "voice",
+            "instructions": "hello",
+            "input": [{"role": "user", "content": "hi"}],
+            "audio": {"output": {"voice": "marin"}},
+            "delegation": {"type": "client"},
+            "future_option": {"enabled": True},
+        },
+        "transport": {"type": "webrtc", "sdp": "offer"},
+        "api_base": "https://untrusted.test",
+    }
+    result = route_client.client.post(prefix + "/sessions", json=body)
+    assert result.status_code == 201
+    output = result.json()
+    assert output["transport"] == {"type": "webrtc", "sdp": "answer"}
+    original = live.decode_session(output["session"]["id"], live._owner(route_client.auth))
+    assert original.session_id == "sess_upstream"
+    assert original.deployment["api_key"] == "upstream-key"
+    assert original.initialization_seconds == 15
+    forwarded = route_client.transport.request.await_args.kwargs["body"]
+    assert forwarded["session"] == {**body["session"], "model": "gpt-live"}
+    assert route_client.bodies[0] == {**body, "model": "voice"}
+    route_client.supervised.assert_awaited_once()
+    assert route_client.factory.call_args.args[0].api_base is None
+
+
+def test_fork_preserves_empty_overrides_and_pins_source_deployment(route_client):
+    source = handle()
+    source = source.model_copy(update={"deployment": {**source.deployment, "model_id": "deployment-a"}})
+    token = live.encode_session(source)
+    body = {"session": {}, "transport": {"type": "webrtc", "sdp": "offer"}}
+    result = route_client.client.post(f"/v1/live/sessions/{token}/fork", json=body)
+    assert result.status_code == 201
+    assert route_client.transport.request.await_args.args == ("POST", "live/sessions/sess_upstream/fork")
+    assert route_client.transport.request.await_args.kwargs["body"] == body
+    assert route_client.factory.call_args.args[0].model_id == "deployment-a"
+    route_client.selected.assert_not_awaited()
+
+
+def test_fork_cannot_change_model_even_to_same_alias(route_client):
+    token = live.encode_session(handle())
+    response = route_client.client.post(f"/v1/live/sessions/{token}/fork", json={"session": {"model": "voice"}})
+    assert response.status_code == 400
+    route_client.transport.request.assert_not_awaited()
+
+
+def test_cross_key_followup_and_raw_incoming_ids_never_contact_upstream(route_client):
+    token = live.encode_session(handle("different-key"))
+    for path in (f"{token}/hangup", "sess_other/accept", "sess_other/reject"):
+        response = route_client.client.post(f"/v1/live/sessions/{path}", json={})
+        assert response.status_code == 403
+    route_client.transport.request.assert_not_awaited()
+
+
+def test_recording_preserves_binary_body_status_and_content_headers(route_client):
+    token = live.encode_session(handle())
+    route_client.transport.request.return_value = httpx.Response(
+        206,
+        content=b"\x00\xffrecording",
+        headers={
+            "content-type": "video/mp4",
+            "content-disposition": "attachment; filename=recording.mp4",
+            "content-range": "bytes 0-10/20",
+        },
+    )
+    result = route_client.client.get(f"/v1/live/sessions/{token}/content")
+    assert result.status_code == 206
+    assert result.content == b"\x00\xffrecording"
+    assert result.headers["content-type"] == "video/mp4"
+    assert result.headers["content-range"] == "bytes 0-10/20"
+
+
+@pytest.mark.asyncio
+async def test_pre_call_preserves_safe_body_options_and_refunds_failed_signaling(monkeypatch):
+    from litellm.proxy import proxy_server
+
+    auth = UserAPIKeyAuth(api_key="owner")
+    authorize = AsyncMock()
+    process = AsyncMock(return_value=({"model": "voice"}, Mock()))
+    release = AsyncMock()
+    monkeypatch.setattr(live, "_authorize", authorize)
+    monkeypatch.setattr(live, "process_codex_request", process)
+    monkeypatch.setattr(live, "release_or_invalidate_budget_reservation", release)
+    monkeypatch.setattr(proxy_server, "proxy_logging_obj", SimpleNamespace(get_proxy_hook=lambda _: None))
+    request = Request({"type": "http", "headers": []})
+    synthetic = live._request(
+        request,
+        {
+            "session": {"model": "voice", "instructions": "safe"},
+            "api_base": "https://untrusted.test",
+            "extra_headers": {"x-admin": "true"},
+        },
+    )
+    with pytest.raises(RuntimeError, match="signaling failed"):
+        async with live._precall(synthetic, auth, "voice"):
+            raise RuntimeError("signaling failed")
+    authorize.assert_awaited_once_with("voice", auth)
+    data = process.await_args.args[1]
+    assert data["session"]["instructions"] == "safe"
+    assert "api_base" not in data and "extra_headers" not in data
+    release.assert_awaited_once()
+
+
+@pytest.mark.asyncio
+async def test_transferred_supervisor_retains_budget_on_client_disconnect(monkeypatch):
+    from litellm.proxy import proxy_server
+
+    monkeypatch.setattr(live, "_authorize", AsyncMock())
+    monkeypatch.setattr(live, "process_codex_request", AsyncMock(return_value=({"model": "voice"}, Mock())))
+    release = AsyncMock()
+    monkeypatch.setattr(live, "release_or_invalidate_budget_reservation", release)
+    monkeypatch.setattr(proxy_server, "proxy_logging_obj", SimpleNamespace(get_proxy_hook=lambda _: None))
+    request = live._request(Request({"type": "http", "headers": []}), {"model": "voice"})
+
+    async def disconnect_after_transfer():
+        async with live._precall(request, UserAPIKeyAuth(api_key="owner"), "voice") as prepared:
+            prepared.transferred = True
+            raise RuntimeError("client disconnected")
+
+    with pytest.raises(RuntimeError, match="client disconnected"):
+        await disconnect_after_transfer()
+    release.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_info_before_started_is_forwarded_without_rejecting_session():
+    info = {"type": "info", "code": "data_channel_permissions", "message": "ready"}
+    started = {"type": "session.started", "session": {"id": "sess_upstream"}}
+    socket = SimpleNamespace(recv=AsyncMock(side_effect=[json.dumps(info), json.dumps(started)]))
+    client = SimpleNamespace(send_json=AsyncMock())
+    assert await live._wait_started(socket, client) == started
+    client.send_json.assert_awaited_once_with(info)
+
+
+@pytest.mark.asyncio
+async def test_upstream_startup_error_is_forwarded_without_creating_session():
+    error = {"type": "error", "error": {"code": "forbidden", "message": "Voice access denied"}}
+    socket = SimpleNamespace(recv=AsyncMock(return_value=json.dumps(error)))
+    client = SimpleNamespace(send_json=AsyncMock())
+    with pytest.raises(HTTPException) as rejected:
+        await live._wait_started(socket, client)
+    assert rejected.value.status_code == 502
+    client.send_json.assert_awaited_once_with(error)
+
+
+@pytest.mark.asyncio
+async def test_session_events_keep_stable_public_id_and_feed_shared_usage_sink():
+    original = handle()
+    token = live.encode_session(original)
+    client = SimpleNamespace(send_text=AsyncMock(), scope={}, headers={})
+    observer = SimpleNamespace(store_message=Mock())
+    frontend = live._PublicSocket(client, original, token, UserAPIKeyAuth(api_key="owner"), observer)
+    event = {
+        "type": "session.updated",
+        "session": {"id": original.session_id},
+        "event": {"type": "response.completed", "response": {"id": "resp_a"}},
+    }
+    for _ in range(2):
+        await frontend.send_text(json.dumps(event))
+        assert json.loads(client.send_text.await_args.args[0])["session"]["id"] == token
+    assert observer.store_message.call_count == 2
+
+
+@pytest.mark.asyncio
+async def test_websocket_delegation_model_update_is_authorized_before_forwarding(monkeypatch):
+    authorize = AsyncMock(side_effect=HTTPException(403, "Model forbidden"))
+    monkeypatch.setattr(live, "_authorize", authorize)
+    message = {"type": "session.update", "session": {"delegation": {"responses": {"model": "unauthorized"}}}}
+    client = SimpleNamespace(receive_text=AsyncMock(return_value=json.dumps(message)), scope={}, headers={})
+    auth = UserAPIKeyAuth(api_key="owner")
+    frontend = live._PublicSocket(client, handle(), "public", auth)
+    with pytest.raises(HTTPException) as rejected:
+        await frontend.receive_text()
+    assert rejected.value.status_code == 403
+    authorize.assert_awaited_once_with("unauthorized", auth)
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "limits",
+    [
+        {"rpm_limit": 1},
+        {"model_max_budget": {"backend": 1}},
+        {"team_tpm_limit": 10},
+        {"team_metadata": {"model_rpm_limit": {"backend": 1}}},
+    ],
+)
+async def test_managed_delegation_fails_closed_for_unenforceable_constraints(limits):
+    body = {"session": {"delegation": {"type": "responses", "responses": {"model": "backend"}}}}
+    with pytest.raises(HTTPException) as rejected:
+        await live._authorize_delegation(body, UserAPIKeyAuth(api_key="owner", **limits))
+    assert rejected.value.status_code == 400
+    assert "client delegation" in rejected.value.detail
+
+
+@pytest.mark.asyncio
+async def test_restricted_webrtc_cannot_change_managed_model_outside_proxy(monkeypatch):
+    monkeypatch.setattr(live, "_authorize", AsyncMock())
+    body = {
+        "session": {"delegation": {"type": "responses", "responses": {"model": "backend"}}},
+        "transport": {"type": "webrtc", "sdp": "offer"},
+    }
+    auth = UserAPIKeyAuth(api_key="owner", models=["voice", "backend"])
+    with pytest.raises(HTTPException) as rejected:
+        await live._authorize_delegation(body, auth)
+    assert rejected.value.status_code == 400
+    body["session"]["client"] = {"data_channel": {"allowed_client_events": ["session.close"]}}
+    await live._authorize_delegation(body, auth)
+
+
+@pytest.mark.asyncio
+async def test_budget_scope_releases_on_ownership_decode_error(monkeypatch):
+    release = AsyncMock()
+    monkeypatch.setattr(live, "release_or_invalidate_budget_reservation", release)
+    auth = UserAPIKeyAuth(api_key="owner")
+    with pytest.raises(HTTPException):
+        async with live._budget_scope(auth):
+            live.decode_session("raw-session-id", live._owner(auth))
+    release.assert_awaited_once()
+
+
+@pytest.mark.asyncio
+async def test_model_authorization_rejection_still_releases_budget(monkeypatch):
+    from litellm.proxy import proxy_server
+
+    monkeypatch.setattr(live, "_authorize", AsyncMock(side_effect=HTTPException(403, "Model forbidden")))
+    release = AsyncMock()
+    monkeypatch.setattr(live, "release_or_invalidate_budget_reservation", release)
+    monkeypatch.setattr(proxy_server, "proxy_logging_obj", SimpleNamespace(get_proxy_hook=lambda _: None))
+    request = live._request(Request({"type": "http", "headers": []}), {"model": "forbidden"})
+    with pytest.raises(HTTPException):
+        async with live._precall(request, UserAPIKeyAuth(api_key="owner"), "forbidden"):
+            pytest.fail("Upstream must not be reached")
+    release.assert_awaited_once()
+
+
+@pytest.mark.asyncio
+async def test_precall_guardrail_mutations_are_used_without_forwarding_routing_options(monkeypatch):
+    from litellm.proxy import proxy_server
+
+    monkeypatch.setattr(live, "_authorize", AsyncMock())
+    monkeypatch.setattr(live, "release_or_invalidate_budget_reservation", AsyncMock())
+    process = AsyncMock(
+        return_value=(
+            {
+                "model": "voice",
+                "session": {"model": "voice", "instructions": "redacted"},
+                "api_base": "https://internal.test",
+            },
+            Mock(),
+        )
+    )
+    monkeypatch.setattr(live, "process_codex_request", process)
+    monkeypatch.setattr(proxy_server, "proxy_logging_obj", SimpleNamespace(get_proxy_hook=lambda _: None))
+    body = {"type": "session.start", "session": {"model": "voice", "instructions": "sensitive"}}
+    request = live._request(Request({"type": "http", "headers": []}), body)
+    async with live._precall(request, UserAPIKeyAuth(api_key="owner"), "voice") as prepared:
+        forwarded = live._processed_body(body, prepared.processed)
+        assert forwarded["session"]["instructions"] == "redacted"
+        assert forwarded["type"] == "session.start"
+        assert "api_base" not in forwarded
+    assert process.await_args.args[1]["session"]["instructions"] == "sensitive"
+
+
+@pytest.mark.asyncio
+async def test_failed_live_signaling_releases_real_parallel_limiter_and_can_retry(monkeypatch):
+    import litellm
+    from litellm.proxy import proxy_server as server
+    from litellm.proxy.common_utils.user_api_key_cache import UserApiKeyCache
+    from litellm.proxy.utils import ProxyLogging
+
+    proxy = ProxyLogging(UserApiKeyCache())
+    monkeypatch.setattr(litellm, "callbacks", [])
+    proxy._add_proxy_hooks()
+    model_list = [{"model_name": "voice", "litellm_params": {"model": "openai/gpt-live-1", "api_key": "upstream-key"}}]
+    monkeypatch.setattr(server, "proxy_logging_obj", proxy)
+    monkeypatch.setattr(server, "general_settings", {})
+    monkeypatch.setattr(server, "llm_model_list", model_list)
+    monkeypatch.setattr(server, "llm_router", litellm.Router(model_list=model_list))
+    auth = UserAPIKeyAuth(api_key="live-limiter-owner", max_parallel_requests=1)
+    authenticate = AsyncMock(return_value=auth)
+    monkeypatch.setattr(live, "user_api_key_auth", authenticate)
+    transport = SimpleNamespace(
+        request=AsyncMock(return_value=httpx.Response(403, json={"error": {"code": "forbidden"}}))
+    )
+    monkeypatch.setattr(live, "LiveTransport", Mock(return_value=transport))
+    supervisor = AsyncMock()
+    monkeypatch.setattr(live, "_supervise", supervisor)
+    for _ in range(2):
+        request = live._request(
+            Request(
+                {
+                    "type": "http",
+                    "method": "POST",
+                    "path": "/v1/live/sessions",
+                    "query_string": b"",
+                    "headers": [(b"content-type", b"application/json")],
+                }
+            ),
+            {"session": {"model": "voice"}, "transport": {"type": "webrtc", "sdp": "offer"}},
+        )
+        result = await live.create_live_session(request)
+        assert result.status_code == 403
+        current = await proxy.internal_usage_cache.async_get_cache(
+            "{api_key:live-limiter-owner}:max_parallel_requests", litellm_parent_otel_span=None, local_only=True
+        )
+        limiter = proxy.get_proxy_hook("parallel_request_limiter")
+        assert limiter._gauge_in_flight_from_cache_value(current) == 0
+    assert transport.request.await_count == 2
+    supervisor.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_inherited_managed_fork_cannot_bypass_new_key_constraints():
+    source = handle().model_copy(
+        update={"policy": {"delegation": {"type": "responses", "responses": {"model": "backend"}}}}
+    )
+    payload = live._policy_body({"session": {}, "transport": {"type": "webrtc", "sdp": "offer"}}, source)
+    assert payload["session"]["delegation"]["responses"]["model"] == "backend"
+    with pytest.raises(HTTPException) as rejected:
+        await live._authorize_delegation(payload, UserAPIKeyAuth(api_key="owner", rpm_limit_per_model={"backend": 1}))
+    assert rejected.value.status_code == 400
+
+
+@pytest.mark.parametrize("protocol", ["http", "websocket"])
+@pytest.mark.parametrize(
+    "startup_policy", [{}, {"delegation": {"type": "responses", "responses": {"model": "allowed"}}}]
+)
+@pytest.mark.parametrize("overrides", [{}, {"delegation": {"responses": {}}}])
+def test_restricted_fork_never_trusts_startup_delegation(route_client, protocol, startup_policy, overrides):
+    from starlette.websockets import WebSocketDisconnect
+
+    # The source may now use a revoked backend, including after an unrestricted WebRTC update.
+    route_client.auth.models = ["voice", "allowed"]
+    token = live.encode_session(handle().model_copy(update={"policy": startup_policy}))
+    path = f"/v1/live/sessions/{token}/fork"
+    if protocol == "http":
+        response = route_client.client.post(path, json={"session": overrides})
+        assert response.status_code == 400
+    else:
+        with route_client.client.websocket_connect(path, headers={"Authorization": "Bearer owner"}) as ws:
+            ws.send_json({"type": "session.start", "session": overrides})
+            with pytest.raises(WebSocketDisconnect) as rejected:
+                ws.receive_json()
+            assert rejected.value.code == 1008
+    route_client.factory.assert_not_called()
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("startup_policy", [{}, {"delegation": {"type": "responses", "responses": {"model": "old"}}}])
+async def test_explicit_fork_backend_is_authorized_even_when_startup_policy_differs(monkeypatch, startup_policy):
+    authorize = AsyncMock()
+    monkeypatch.setattr(live, "_authorize", authorize)
+    source = handle().model_copy(update={"policy": startup_policy})
+    auth = UserAPIKeyAuth(api_key="owner", models=["voice", "allowed"])
+    body = {"session": {"delegation": {"type": "responses", "responses": {"model": "allowed"}}}}
+    await live._authorize_fork_policy(body, source, auth)
+    authorize.assert_awaited_once_with("allowed", auth)
+    authorize.side_effect = HTTPException(403, "Model revoked")
+    with pytest.raises(HTTPException) as rejected:
+        await live._authorize_fork_policy(body, source, auth)
+    assert rejected.value.status_code == 403
+
+
+def test_restricted_fork_can_explicitly_select_client_delegation(route_client):
+    route_client.auth.models = ["voice"]
+    body = {"session": {"delegation": {"type": "client"}}}
+    token = live.encode_session(handle())
+    route_client.transport.request.return_value = httpx.Response(
+        200, json={"session": {"id": "sess_fork"}, "transport": {"type": "webrtc", "sdp": "answer"}}
+    )
+    response = route_client.client.post(f"/v1/live/sessions/{token}/fork", json=body)
+    assert response.status_code == 200
+    assert response.json()["transport"]["sdp"] == "answer"
+    route_client.transport.request.assert_awaited_once_with("POST", "live/sessions/sess_upstream/fork", body=body)
+
+
+@pytest.mark.parametrize("protocol", ["http", "websocket"])
+@pytest.mark.parametrize("limits", [{"rpm_limit": 10}, {"tpm_limit": 100}, {"model_max_budget": {"backend": 1}}])
+def test_fork_with_new_limits_cannot_trust_old_client_policy(route_client, protocol, limits):
+    from starlette.websockets import WebSocketDisconnect
+
+    # An unrestricted source could have switched to managed delegation after its handle was issued.
+    for key, value in limits.items():
+        setattr(route_client.auth, key, value)
+    token = live.encode_session(handle())
+    path = f"/v1/live/sessions/{token}/fork"
+    if protocol == "http":
+        response = route_client.client.post(path, json={"session": {}})
+        assert response.status_code == 400
+    else:
+        with route_client.client.websocket_connect(path, headers={"Authorization": "Bearer owner"}) as ws:
+            ws.send_json({"type": "session.start", "session": {}})
+            with pytest.raises(WebSocketDisconnect) as rejected:
+                ws.receive_json()
+            assert rejected.value.code == 1008
+    route_client.factory.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_explicit_managed_fork_still_rejected_for_new_rate_limits():
+    with pytest.raises(HTTPException) as rejected:
+        await live._authorize_fork_policy(
+            {"session": {"delegation": {"type": "responses", "responses": {"model": "backend"}}}},
+            handle(),
+            UserAPIKeyAuth(api_key="owner", rpm_limit=10),
+        )
+    assert rejected.value.status_code == 400
+    assert "cannot enforce" in rejected.value.detail
+
+
+@pytest.mark.asyncio
+async def test_startup_usage_buffer_preserves_nested_events_until_supervisor_owns_accounting():
+    usage = {
+        "type": "response.event",
+        "event": {
+            "type": "response.completed",
+            "response": {
+                "id": "resp_before_start",
+                "model": "backend",
+                "usage": {"input_tokens": 7, "output_tokens": 2},
+            },
+        },
+    }
+    started = {"type": "session.started", "session": {"id": "sess_upstream"}}
+    backend = SimpleNamespace(recv=AsyncMock(side_effect=[json.dumps(usage), json.dumps(started)]))
+    client = SimpleNamespace(send_json=AsyncMock())
+    startup = live._StartupEvents()
+    assert await live._wait_started(backend, client, startup) == started
+    assert [json.loads(message) for message in startup.messages] == [usage]
+
+
+def test_primary_websocket_authenticates_model_and_keeps_public_event_shape(route_client):
+    from websockets.exceptions import ConnectionClosedOK
+    from websockets.frames import Close
+
+    event = {"type": "future.live.event", "session_id": "sess_upstream", "payload": {"untouched": ["a", 2]}}
+    backend = SimpleNamespace(
+        send=AsyncMock(),
+        close=AsyncMock(),
+        recv=AsyncMock(
+            side_effect=[
+                json.dumps({"type": "info", "code": "ready", "message": "Preparing"}),
+                json.dumps({"type": "session.started", "session": {"id": "sess_upstream"}}),
+                json.dumps(event),
+                ConnectionClosedOK(Close(1000, ""), Close(1000, ""), True),
+            ]
+        ),
+    )
+    route_client.transport.connect = AsyncMock(return_value=backend)
+    observer = SimpleNamespace(store_message=Mock())
+    route_client.supervised.return_value = observer
+    with route_client.client.websocket_connect("/v1/live/sessions", headers={"Authorization": "Bearer owner"}) as ws:
+        ws.send_json({"type": "session.start", "session": {"model": "voice", "instructions": "hello", "unknown": True}})
+        assert ws.receive_json()["type"] == "info"
+        started = ws.receive_json()
+        public_id = started["session"]["id"]
+        assert live.decode_session(public_id, live._owner(route_client.auth)).session_id == "sess_upstream"
+        forwarded = ws.receive_json()
+        assert forwarded == {**event, "session_id": public_id}
+    assert route_client.bodies[0] == {}
+    assert route_client.bodies[1]["model"] == "voice"
+    assert route_client.bodies[1]["session"]["instructions"] == "hello"
+    initial = json.loads(backend.send.await_args.args[0])
+    assert initial == {
+        "type": "session.start",
+        "session": {"model": "gpt-live", "instructions": "hello", "unknown": True},
+    }
+    assert any(json.loads(call.args[0]) == event for call in observer.store_message.call_args_list)
+
+
+@pytest.mark.asyncio
+async def test_successful_live_session_holds_real_parallel_slot_until_supervisor_releases(monkeypatch):
+    import litellm
+    from litellm.proxy import proxy_server as server
+    from litellm.proxy.common_utils.user_api_key_cache import UserApiKeyCache
+    from litellm.proxy.utils import ProxyLogging
+
+    proxy = ProxyLogging(UserApiKeyCache())
+    monkeypatch.setattr(litellm, "callbacks", [])
+    proxy._add_proxy_hooks()
+    models = [{"model_name": "voice", "litellm_params": {"model": "openai/gpt-live-1", "api_key": "upstream-key"}}]
+    monkeypatch.setattr(server, "proxy_logging_obj", proxy)
+    monkeypatch.setattr(server, "general_settings", {})
+    monkeypatch.setattr(server, "llm_model_list", models)
+    monkeypatch.setattr(server, "llm_router", litellm.Router(model_list=models))
+    auth = UserAPIKeyAuth(api_key="live-held-slot", max_parallel_requests=1)
+    monkeypatch.setattr(live, "user_api_key_auth", AsyncMock(return_value=auth))
+    transport = SimpleNamespace(
+        request=AsyncMock(
+            return_value=httpx.Response(
+                201, json={"session": {"id": "sess_upstream"}, "transport": {"type": "webrtc", "sdp": "answer"}}
+            )
+        )
+    )
+    monkeypatch.setattr(live, "LiveTransport", Mock(return_value=transport))
+    leases = []
+
+    async def supervise(request, handle, auth, logger, lease):
+        leases.append(lease)
+
+    monkeypatch.setattr(live, "_supervise", supervise)
+
+    def request():
+        return live._request(
+            Request(
+                {
+                    "type": "http",
+                    "method": "POST",
+                    "path": "/v1/live/sessions",
+                    "query_string": b"",
+                    "headers": [(b"content-type", b"application/json")],
+                }
+            ),
+            {"session": {"model": "voice"}, "transport": {"type": "webrtc", "sdp": "offer"}},
+        )
+
+    try:
+        result = await live.create_live_session(request())
+        assert result.status_code == 201
+        assert leases[0] is not None
+        with pytest.raises(HTTPException) as blocked:
+            await live.create_live_session(request())
+        assert blocked.value.status_code == 429
+        assert transport.request.await_count == 1
+    finally:
+        for lease in leases:
+            if lease is not None:
+                await lease.close()
+    current = await proxy.internal_usage_cache.async_get_cache(
+        "{api_key:live-held-slot}:max_parallel_requests", litellm_parent_otel_span=None, local_only=True
+    )
+    limiter = proxy.get_proxy_hook("parallel_request_limiter")
+    assert limiter._gauge_in_flight_from_cache_value(current) == 0
+
+
+@pytest.mark.asyncio
+async def test_supervisor_logger_keeps_deployment_pricing(monkeypatch):
+    import litellm
+
+    original = handle().model_copy(
+        update={"deployment": {**handle().deployment, "model_id": "deployment-priced"}, "initialization_seconds": 15}
+    )
+    logger = Mock(litellm_params={}, model_call_details={})
+    monkeypatch.setattr(live, "process_codex_request", AsyncMock(return_value=({"model": "voice"}, logger)))
+    monkeypatch.setattr(litellm, "get_model_info", Mock(return_value={"input_cost_per_second": 0.1}))
+    connection = SimpleNamespace(close=AsyncMock())
+    transport = SimpleNamespace(connect=AsyncMock(return_value=connection), request=AsyncMock())
+    monkeypatch.setattr(live, "LiveTransport", Mock(return_value=transport))
+    started = AsyncMock()
+    monkeypatch.setattr(live, "CALL_SUPERVISORS", SimpleNamespace(start=started))
+    request = Request({"type": "http", "method": "POST", "path": "/v1/live/sessions", "headers": []})
+    stream = await live._start_supervisor(request, original, UserAPIKeyAuth(api_key="owner"), None)
+    assert stream.messages[0]["usage"]["seconds"] == 15
+    started.assert_awaited_once()
+    metadata = logger.update_from_kwargs.call_args.kwargs["kwargs"]["litellm_metadata"]
+    assert metadata["model_info"]["id"] == "deployment-priced"
+    assert metadata["model_info"]["input_cost_per_second"] == 0.1
+    connection.close.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_supervisor_startup_failure_closes_observer_and_invalidates_unconfirmed_hangup(monkeypatch):
+    from litellm.proxy.spend_tracking import budget_reservation
+
+    logger = Mock(litellm_params={}, model_call_details={})
+    monkeypatch.setattr(live, "process_codex_request", AsyncMock(return_value=({"model": "voice"}, logger)))
+    connection = SimpleNamespace(close=AsyncMock())
+    transport = SimpleNamespace(
+        connect=AsyncMock(return_value=connection),
+        request=AsyncMock(side_effect=httpx.ConnectError("upstream unavailable")),
+    )
+    monkeypatch.setattr(live, "LiveTransport", Mock(return_value=transport))
+    monkeypatch.setattr(
+        live, "CALL_SUPERVISORS", SimpleNamespace(start=AsyncMock(side_effect=RuntimeError("cannot start")))
+    )
+    invalidate = AsyncMock()
+    monkeypatch.setattr(budget_reservation, "invalidate_budget_reservation_counters", invalidate)
+    request = Request({"type": "http", "method": "POST", "path": "/v1/live/sessions", "headers": []})
+    with pytest.raises(RuntimeError, match="cannot start"):
+        await live._start_supervisor(request, handle(), UserAPIKeyAuth(api_key="owner"), None)
+    connection.close.assert_awaited_once()
+    invalidate.assert_awaited_once()
+
+
+def test_admin_sip_accept_requires_exact_deployment_and_returns_owned_handle(route_client, monkeypatch):
+    from litellm.proxy import proxy_server
+    from litellm.proxy._types import LitellmUserRoles
+
+    route_client.auth.user_role = LitellmUserRoles.PROXY_ADMIN
+    monkeypatch.setattr(proxy_server, "llm_model_list", [{"model_name": "voice"}])
+    route_client.transport.request.return_value = httpx.Response(200, content=b"")
+    body = {"session": {"model": "voice", "type": "live", "instructions": "incoming"}}
+    missing = route_client.client.post("/v1/live/sessions/sess_incoming/accept", json=body)
+    assert missing.status_code == 400
+    route_client.transport.request.assert_not_awaited()
+    accepted = route_client.client.post(
+        "/v1/live/sessions/sess_incoming/accept", json=body, headers={"x-litellm-live-model": "voice"}
+    )
+    assert accepted.status_code == 200 and accepted.content == b""
+    token = accepted.headers["x-litellm-live-session-id"]
+    owned = live.decode_session(token, live._owner(route_client.auth))
+    assert owned.session_id == "sess_incoming" and owned.alias == "voice"
+    assert route_client.transport.request.await_args.kwargs["body"]["session"]["model"] == "gpt-live"
+    route_client.supervised.assert_awaited_once()
+    raw_hangup = route_client.client.post("/v1/live/sessions/sess_incoming/hangup")
+    assert raw_hangup.status_code == 403
+    assert route_client.client.post(f"/v1/live/sessions/{token}/hangup").status_code == 200
+
+
+def test_admin_sip_cannot_enroll_through_alias_with_multiple_accounts(route_client, monkeypatch):
+    from litellm.proxy import proxy_server
+    from litellm.proxy._types import LitellmUserRoles
+
+    route_client.auth.user_role = LitellmUserRoles.PROXY_ADMIN
+    monkeypatch.setattr(proxy_server, "llm_model_list", [{"model_name": "voice"}, {"model_name": "voice"}])
+    response = route_client.client.post(
+        "/v1/live/sessions/sess_incoming/reject", json={"status_code": 603}, headers={"x-litellm-live-model": "voice"}
+    )
+    assert response.status_code == 400
+    route_client.transport.request.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_primary_first_frame_timeout_releases_authenticated_reservation(monkeypatch):
+    import asyncio
+
+    from fastapi import WebSocket
+
+    messages = iter([{"type": "websocket.connect"}])
+
+    async def receive():
+        try:
+            return next(messages)
+        except StopIteration:
+            raise asyncio.TimeoutError("first message timeout")
+
+    sent = []
+
+    async def send(message):
+        sent.append(message)
+
+    websocket = WebSocket(
+        {
+            "type": "websocket",
+            "path": "/v1/live/sessions",
+            "query_string": b"",
+            "headers": [(b"authorization", b"Bearer owner")],
+            "scheme": "ws",
+            "server": ("localhost", 4000),
+        },
+        receive,
+        send,
+    )
+    monkeypatch.setattr(live, "_auth", AsyncMock(return_value=UserAPIKeyAuth(api_key="owner")))
+    release = AsyncMock()
+    transport = Mock()
+    monkeypatch.setattr(live, "release_or_invalidate_budget_reservation", release)
+    monkeypatch.setattr(live, "LiveTransport", transport)
+    await live.websocket_live_session(websocket)
+    release.assert_awaited_once()
+    transport.assert_not_called()
+    assert sent[-1]["type"] == "websocket.close" and sent[-1]["code"] == 1008
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("key_limit,global_limit,allowed", [(None, None, True), (1, None, False), (None, 2, False)])
+async def test_legacy_limiter_only_blocks_sessions_requiring_parallel_leases(
+    monkeypatch, key_limit, global_limit, allowed
+):
+    from litellm.proxy import proxy_server
+    from litellm.proxy.hooks.parallel_request_limiter import _PROXY_MaxParallelRequestsHandler
+
+    legacy = Mock(spec=_PROXY_MaxParallelRequestsHandler)
+    monkeypatch.setattr(proxy_server, "proxy_logging_obj", SimpleNamespace(get_proxy_hook=lambda _: legacy))
+    monkeypatch.setattr(proxy_server, "general_settings", {"global_max_parallel_requests": global_limit})
+    monkeypatch.setattr(live, "_authorize", AsyncMock())
+    processor = AsyncMock(return_value=({"model": "voice"}, Mock()))
+    monkeypatch.setattr(live, "process_codex_request", processor)
+    monkeypatch.setattr(live, "release_or_invalidate_budget_reservation", AsyncMock())
+    request = live._request(Request({"type": "http", "headers": []}), {"model": "voice"})
+    auth = UserAPIKeyAuth(api_key="owner", max_parallel_requests=key_limit)
+    if allowed:
+        async with live._precall(request, auth, "voice") as prepared:
+            assert prepared.lease is None
+        processor.assert_awaited_once()
+        return
+    with pytest.raises(HTTPException) as rejected:
+        async with live._precall(request, auth, "voice"):
+            pytest.fail("Parallel-limited sessions require renewable leases")
+    assert rejected.value.status_code == 400
+    processor.assert_not_awaited()
+
+
+@pytest.mark.parametrize("delegation", ["invalid", {"type": "responses", "responses": "invalid"}])
+def test_malformed_delegation_returns_client_error_before_provider(route_client, delegation):
+    result = route_client.client.post(
+        "/v1/live/sessions",
+        json={"session": {"model": "voice", "delegation": delegation}, "transport": {"type": "webrtc", "sdp": "offer"}},
+    )
+    assert result.status_code == 400
+    route_client.transport.request.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("parallel_reserved", [False, True])
+async def test_attachment_skips_parallel_admission_only_with_existing_session_lease(monkeypatch, parallel_reserved):
+    from litellm.proxy import proxy_server
+    from litellm.proxy.hooks.realtime_call_lease import is_realtime_call_attachment
+
+    attachment = object()
+    observed = []
+
+    async def process(request, data, auth, model, route_type):
+        observed.append(is_realtime_call_attachment(attachment))
+        return data, Mock()
+
+    monkeypatch.setattr(proxy_server, "proxy_logging_obj", SimpleNamespace(get_proxy_hook=lambda _: None))
+    monkeypatch.setattr(live, "_authorize", AsyncMock())
+    monkeypatch.setattr(live, "process_codex_request", process)
+    monkeypatch.setattr(live, "release_or_invalidate_budget_reservation", AsyncMock())
+    request = live._request(Request({"type": "http", "headers": []}), {"model": "voice"})
+    async with live._precall(
+        request, UserAPIKeyAuth(api_key="owner"), "voice", attachment=attachment, parallel_reserved=parallel_reserved
+    ):
+        pass
+    assert observed == [parallel_reserved]
+
+
+@pytest.mark.asyncio
+async def test_live_observer_becomes_ready_without_session_started_event(monkeypatch):
+    import asyncio
+
+    from litellm.proxy.realtime_endpoints.call_supervision import CallSupervisor
+
+    class Observer:
+        def __init__(self):
+            self.queue = asyncio.Queue()
+            self.closed = False
+
+        def __aiter__(self):
+            return self
+
+        async def __anext__(self):
+            return await self.queue.get()
+
+        async def close(self):
+            self.closed = True
+
+    observer = Observer()
+    logger = Mock(litellm_params={}, model_call_details={})
+    monkeypatch.setattr(live, "process_codex_request", AsyncMock(return_value=({"model": "voice"}, logger)))
+    sink = SimpleNamespace(store_message=Mock(), log_messages=AsyncMock())
+    monkeypatch.setattr(live, "RealTimeStreaming", Mock(return_value=sink))
+
+    async def hangup(*args, **kwargs):
+        await observer.queue.put(json.dumps({"type": "session.closed", "usage": {"seconds": 0}}))
+        return httpx.Response(200, request=httpx.Request("POST", "https://upstream.test/hangup"))
+
+    monkeypatch.setattr(
+        live,
+        "LiveTransport",
+        Mock(return_value=SimpleNamespace(connect=AsyncMock(return_value=observer), request=hangup)),
+    )
+    supervisors = []
+
+    def build_supervisor(*args, **kwargs):
+        return CallSupervisor(
+            *args, **kwargs, ready_timeout=0.1, drain_timeout=0.1, termination_timeout=0.2, logging_timeout=0.2
+        )
+
+    async def start(supervisor):
+        supervisors.append(supervisor)
+        await supervisor.start()
+
+    monkeypatch.setattr(live, "CallSupervisor", build_supervisor)
+    monkeypatch.setattr(live, "CALL_SUPERVISORS", SimpleNamespace(start=start))
+    request = Request({"type": "http", "method": "POST", "path": "/v1/live/sessions", "headers": []})
+    try:
+        result = await asyncio.wait_for(
+            live._start_supervisor(request, handle(), UserAPIKeyAuth(api_key="owner"), None), 0.5
+        )
+        assert result is sink
+        sink.store_message.assert_not_called()
+    finally:
+        for supervisor in supervisors:
+            await supervisor.close()
+    assert observer.closed

tests/test_litellm/proxy/realtime_endpoints/test_realtime_webrtc_endpoints.py

@@ -7,7 +7,7 @@ Tests for LiteLLM proxy realtime WebRTC HTTP endpoints:
 import json
 import time
 from collections.abc import Awaitable
-from typing import Protocol
+from typing import Final, Protocol
 from unittest.mock import AsyncMock, MagicMock, patch
 
 import httpx
@@ -15,7 +15,6 @@ import pytest
 from fastapi import FastAPI, HTTPException
 from fastapi.testclient import TestClient
 
-
 from litellm.proxy._types import UserAPIKeyAuth
 from litellm.proxy.auth.user_api_key_auth import user_api_key_auth
 from litellm.proxy.common_utils.encrypt_decrypt_utils import (
@@ -120,6 +119,114 @@ def proxy_app(monkeypatch):
     return proxy_server.app
 
 
+@pytest.mark.parametrize("path", ["/v1/live", "/live", "/openai/v1/live"])
+@pytest.mark.parametrize("query", ["", "?intent=custom&architecture=custom"])
+def test_live_multipart_offer_runs_authenticated_call_pipeline(
+    proxy_app: FastAPI, monkeypatch: pytest.MonkeyPatch, path: str, query: str
+) -> None:
+    from litellm.proxy import proxy_server
+    from litellm.proxy.realtime_endpoints import call_sessions
+
+    monkeypatch.setenv("LITELLM_SALT_KEY", "test-live-signaling-salt")
+    session: Final = {"model": "gpt-live-1-codex", "audio": {"output": {"voice": "sol"}}}
+    authenticate: Final = AsyncMock(wraps=call_sessions.user_api_key_auth)
+    process: Final = AsyncMock(side_effect=lambda request, data, *args: (data, MagicMock()))
+    supervise: Final = AsyncMock()
+    upstream: Final = httpx.Response(
+        201,
+        content=b"v=0\r\nanswer",
+        headers={"Location": "/v1/live/rtc_private"},
+        extensions={"chatgpt_realtime": {"model": "gpt-live-1-codex"}},
+    )
+
+    async def route(**kwargs: object) -> Awaitable[httpx.Response]:
+        assert kwargs["route_type"] == "arealtime_calls"
+        assert isinstance(kwargs["data"], dict)
+        assert kwargs["data"]["sdp_body"] == b"v=0\r\noffer"
+        assert kwargs["data"]["session"] == session
+        assert kwargs["data"]["chatgpt_realtime_client_query"] == (
+            {"architecture": "custom", "intent": "custom"}
+            if query
+            else {"architecture": "avas", "intent": "quicksilver"}
+        )
+
+        async def respond() -> httpx.Response:
+            return upstream
+
+        return respond()
+
+    monkeypatch.setattr(call_sessions, "user_api_key_auth", authenticate)
+    monkeypatch.setattr(call_sessions, "process_codex_request", process)
+    monkeypatch.setattr(call_sessions, "supervise_codex_call", supervise)
+    monkeypatch.setattr(proxy_server, "route_request", route)
+    response: Final = TestClient(proxy_app).post(
+        f"{path}{query}",
+        headers={"Authorization": "Bearer sk-test-master-key"},
+        files={
+            "sdp": (None, "v=0\r\noffer", "application/sdp"),
+            "session": (None, json.dumps(session), "application/json"),
+        },
+    )
+
+    assert response.status_code == 201
+    assert response.content == b"v=0\r\nanswer"
+    assert response.headers["content-type"] == "application/sdp"
+    authenticate.assert_awaited_once()
+    process.assert_awaited_once()
+    assert process.await_args.args[3:] == ("gpt-live-1-codex", "arealtime_calls")
+    assert isinstance(process.await_args.args[2], UserAPIKeyAuth)
+    supervise.assert_awaited_once()
+    assert response.headers["location"].startswith("/v1/live/")
+    token: Final = response.headers["location"].rsplit("/", 1)[-1]
+    call: Final = call_sessions.decode_call(token, "Bearer sk-test-master-key")
+    assert call.call_id == "rtc_private"
+    assert call.alias == "gpt-live-1-codex"
+    assert call.usage_supervised
+
+
+@pytest.mark.parametrize("path", ["/v1/live", "/live", "/openai/v1/live"])
+def test_live_multipart_offer_rejects_invalid_credentials_before_routing(
+    proxy_app: FastAPI, monkeypatch: pytest.MonkeyPatch, path: str
+) -> None:
+    from litellm.proxy import proxy_server
+    from litellm.proxy.realtime_endpoints import call_sessions
+
+    authenticate: Final = AsyncMock(wraps=call_sessions.user_api_key_auth)
+    route: Final = AsyncMock()
+    monkeypatch.setattr(call_sessions, "user_api_key_auth", authenticate)
+    monkeypatch.setattr(proxy_server, "route_request", route)
+    response: Final = TestClient(proxy_app).post(
+        path,
+        files={"sdp": (None, "v=0\r\n"), "session": (None, '{"model":"gpt-live-1-codex"}')},
+    )
+
+    assert response.status_code == 401
+    authenticate.assert_awaited_once()
+    route.assert_not_awaited()
+
+
+def test_live_multipart_offer_rejects_model_outside_key_scope(
+    proxy_app: FastAPI, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    from litellm.proxy import proxy_server
+    from litellm.proxy.realtime_endpoints import call_sessions
+
+    authenticate: Final = AsyncMock(return_value=UserAPIKeyAuth(models=["another-model"]))
+    route: Final = AsyncMock()
+    monkeypatch.setattr(call_sessions, "user_api_key_auth", authenticate)
+    monkeypatch.setattr(proxy_server, "route_request", route)
+    response: Final = TestClient(proxy_app).post(
+        "/v1/live",
+        headers={"Authorization": "Bearer restricted-key"},
+        files={"sdp": (None, "v=0\r\n"), "session": (None, '{"model":"gpt-live-1-codex"}')},
+    )
+
+    assert response.status_code == 403
+    assert "gpt-live-1-codex" in response.text
+    authenticate.assert_awaited_once()
+    route.assert_not_awaited()
+
+
 @pytest.fixture
 def mock_route_request_client_secrets():
     """Mock route_request to return a fake upstream client_secrets response."""

tests/test_litellm/proxy/test_live_route_registration.py

@@ -0,0 +1,57 @@
+from unittest.mock import AsyncMock
+
+import httpx
+import pytest
+from fastapi import HTTPException
+from fastapi.testclient import TestClient
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("prefix", ["/live", "/v1/live", "/openai/v1/live"])
+@pytest.mark.parametrize(
+    ("method", "suffix"),
+    [
+        ("POST", ""),
+        ("POST", "/opaque/fork"),
+        ("POST", "/opaque/accept"),
+        ("POST", "/opaque/reject"),
+        ("POST", "/opaque/refer"),
+        ("POST", "/opaque/hangup"),
+        ("GET", "/opaque/content"),
+    ],
+)
+async def test_public_live_http_routes_reach_live_auth_before_generic_passthrough(monkeypatch, prefix, method, suffix):
+    from litellm.proxy import proxy_server
+    from litellm.proxy.realtime_endpoints import live
+
+    authenticate = AsyncMock(side_effect=HTTPException(401, "Live authentication required"))
+    monkeypatch.setattr(live, "_auth", authenticate)
+    async with httpx.AsyncClient(
+        transport=httpx.ASGITransport(app=proxy_server.app), base_url="http://proxy"
+    ) as client:
+        response = await client.request(
+            method,
+            prefix + "/sessions" + suffix,
+            json={"session": {"model": "voice"}, "transport": {"type": "webrtc", "sdp": "offer"}},
+        )
+    assert response.status_code == 401
+    assert "Live authentication required" in response.text
+    authenticate.assert_awaited_once()
+
+
+@pytest.mark.parametrize("prefix", ["/live", "/v1/live", "/openai/v1/live"])
+@pytest.mark.parametrize("suffix", ["", "/opaque/attach", "/opaque/fork"])
+def test_public_live_websockets_reach_live_auth_before_legacy_sideband(monkeypatch, prefix, suffix):
+    from starlette.websockets import WebSocketDisconnect
+
+    from litellm.proxy import proxy_server
+    from litellm.proxy.realtime_endpoints import live
+
+    authenticate = AsyncMock(side_effect=HTTPException(403, "Live authentication rejected"))
+    monkeypatch.setattr(live, "_auth", authenticate)
+    monkeypatch.setattr(proxy_server, "general_settings", {})
+    with TestClient(proxy_server.app) as client:
+        with pytest.raises(WebSocketDisconnect):
+            with client.websocket_connect(prefix + "/sessions" + suffix, headers={"authorization": "Bearer test"}):
+                pass
+    authenticate.assert_awaited_once()

tests/test_litellm/test_cost_calculator.py

@@ -1,4 +1,3 @@
-
 import json
 from pathlib import Path
 from typing import Final
@@ -1822,7 +1821,6 @@ def test_azure_ai_cache_cost_calculation(_local_model_cost_map):
     ), f"Output cost mismatch: got {output_cost}, expected {expected_output_cost}"
 
 
-
 AZURE_GPT_5_6_MAP_KEYS = (
     "azure/gpt-5.6",
     "azure/gpt-5.6-sol",
@@ -1889,6 +1887,7 @@ def test_azure_gpt_5_6_rates_match_azure_price_page(_local_model_cost_map, model
         for key in token_cost_keys:
             assert entry[key] == pytest.approx(global_entry[key] * 1.1), key
 
+
 def test_vertex_regional_deployment_costs_uplift_over_global(monkeypatch):
     """
     Regression for https://github.com/BerriAI/litellm/issues/34393: two Vertex
@@ -4221,6 +4220,8 @@ def test_completion_cost_together_metadata_only_model_still_uses_size_bucket(_lo
     )
 
     assert cost == pytest.approx((23 + 15) * 8e-07, rel=1e-9)
+
+
 def test_select_model_name_strips_unregistered_alias_prefix(_local_model_cost_map):
     """A router-facing model_name alias containing "/" whose leading segment is NOT a
     registered provider must not be double-prefixed into a non-existent cost key.
@@ -4864,13 +4865,16 @@ def test_live_terminal_duration_uses_configured_second_price(monkeypatch, rate,
     ) == pytest.approx(expected)
 
 
-def test_live_terminal_duration_honors_deployment_override(monkeypatch):
+@pytest.mark.parametrize("public_live", [False, True])
+def test_live_terminal_duration_honors_deployment_override(monkeypatch, public_live):
     monkeypatch.setitem(
         litellm.model_cost,
         "live-deployment-test",
         {"litellm_provider": "chatgpt", "mode": "realtime", "input_cost_per_second": 0.025},
     )
-    result = RealtimeAPITokenUsageProcessor.create_logging_realtime_object(Usage(), [_live_terminal_event()])
+    result = RealtimeAPITokenUsageProcessor.create_logging_realtime_object(
+        Usage(), [{"type": "session.closed", "usage": {"seconds": 4}}] if public_live else [_live_terminal_event()]
+    )
     assert completion_cost(
         completion_response=result,
         model="gpt-live-1",
@@ -5170,3 +5174,216 @@ def test_completion_cost_ocr_ignores_deployment_pricing_without_custom_pricing_f
         litellm_logging_obj=logging_obj,
     )
     assert cost == 0.0
+
+
+@pytest.mark.parametrize("terminal", [False, True])
+def test_public_live_seconds_are_cumulative_and_backend_usage_is_separately_priced(monkeypatch, terminal):
+    monkeypatch.setitem(
+        litellm.model_cost,
+        "live-seconds-test",
+        {
+            "litellm_provider": "openai",
+            "mode": "realtime",
+            "input_cost_per_second": 0.025,
+            "input_cost_per_token": 100,
+            "output_cost_per_token": 100,
+        },
+    )
+    monkeypatch.setitem(
+        litellm.model_cost,
+        "live-backend-test",
+        {
+            "litellm_provider": "openai",
+            "mode": "responses",
+            "input_cost_per_token": 0.001,
+            "output_cost_per_token": 0.002,
+        },
+    )
+    backend = {
+        "type": "response.event",
+        "event": {
+            "type": "response.completed",
+            "response": {
+                "id": "resp_backend",
+                "created_at": 1,
+                "model": "live-backend-test",
+                "output": [],
+                "usage": {"input_tokens": 20, "output_tokens": 10, "total_tokens": 30},
+            },
+        },
+    }
+    events = [
+        {"type": "session.usage.updated", "usage": {"seconds": 15}},
+        {"type": "session.usage.updated", "usage": {"seconds": 30}},
+        backend,
+        backend,
+        {"type": "session.closed" if terminal else "session.usage.updated", "usage": {"seconds": 30}},
+    ]
+    usage = RealtimeAPITokenUsageProcessor.collect_and_combine_usage_from_realtime_stream_results(events)
+    assert usage.total_tokens == 30
+    assert handle_realtime_stream_cost_calculation(events, usage, "openai", "live-seconds-test") == pytest.approx(0.79)
+
+
+@pytest.mark.parametrize("seconds", [-1, True, "30", float("inf"), float("nan"), None])
+def test_public_live_invalid_seconds_are_not_billed(monkeypatch, seconds):
+    monkeypatch.setitem(
+        litellm.model_cost,
+        "live-seconds-test",
+        {
+            "litellm_provider": "openai",
+            "mode": "realtime",
+            "input_cost_per_second": 0.025,
+        },
+    )
+    assert (
+        handle_realtime_stream_cost_calculation(
+            [{"type": "session.closed", "usage": {"seconds": seconds}}], Usage(), "openai", "live-seconds-test"
+        )
+        == 0
+    )
+
+
+@pytest.mark.parametrize("terminal", [False, True])
+def test_live_duration_does_not_regress_when_primary_and_observer_events_interleave(monkeypatch, terminal):
+    monkeypatch.setitem(
+        litellm.model_cost,
+        "live-interleaved-test",
+        {"litellm_provider": "openai", "mode": "realtime", "input_cost_per_second": 0.025},
+    )
+    events = [
+        {"type": "session.closed" if terminal else "session.usage.updated", "usage": {"seconds": 30}},
+        {"type": "session.usage.updated", "usage": {"seconds": 15}},
+    ]
+    assert handle_realtime_stream_cost_calculation(events, Usage(), "openai", "live-interleaved-test") == pytest.approx(
+        0.75
+    )
+
+
+@pytest.mark.parametrize("seconds,expected", [(None, 15), (0, 15), (4, 15), (15, 15), (30, 30)])
+def test_live_webrtc_initialization_is_credited_against_duration(monkeypatch, seconds, expected):
+    monkeypatch.setitem(
+        litellm.model_cost,
+        "live-init-test",
+        {
+            "litellm_provider": "openai",
+            "mode": "realtime",
+            "input_cost_per_second": 0.025,
+        },
+    )
+    events = [{"type": "litellm.live.initialization", "usage": {"seconds": 15}}]
+    if seconds is not None:
+        events.append({"type": "session.closed", "usage": {"seconds": seconds}})
+    assert handle_realtime_stream_cost_calculation(events, Usage(), "openai", "live-init-test") == pytest.approx(
+        expected * 0.025
+    )
+
+
+def test_live_invalid_terminal_retains_last_reported_partial_usage(monkeypatch):
+    monkeypatch.setitem(
+        litellm.model_cost,
+        "live-partial-test",
+        {
+            "litellm_provider": "openai",
+            "mode": "realtime",
+            "input_cost_per_second": 0.025,
+        },
+    )
+    events = [
+        {"type": "litellm.live.initialization", "usage": {"seconds": 15}},
+        {"type": "session.usage.updated", "usage": {"seconds": 30}},
+        {"type": "session.closed", "usage": {"seconds": "invalid"}},
+    ]
+    assert handle_realtime_stream_cost_calculation(events, Usage(), "openai", "live-partial-test") == pytest.approx(
+        0.75
+    )
+
+
+@pytest.mark.parametrize(
+    "nested",
+    [
+        {"type": "response.created", "response": {"model": "still-starting"}},
+        {"type": "response.in_progress", "response": {}},
+        {"type": "future.event", "response": ["unknown", "payload"]},
+        {"type": "response.completed", "response": {"id": "broken", "usage": "invalid"}},
+    ],
+)
+def test_live_partial_or_malformed_backend_events_preserve_duration(monkeypatch, nested):
+    from unittest.mock import MagicMock
+
+    monkeypatch.setitem(
+        litellm.model_cost,
+        "live-resilient-test",
+        {
+            "litellm_provider": "openai",
+            "mode": "realtime",
+            "input_cost_per_second": 0.025,
+        },
+    )
+    logger = MagicMock(spec=Logging)
+    logger.model_call_details = {}
+    events = [
+        {"type": "response.event", "event": nested},
+        {"type": "session.closed", "usage": {"seconds": 30}},
+    ]
+    usage = RealtimeAPITokenUsageProcessor.collect_and_combine_usage_from_realtime_stream_results(events)
+    assert usage.total_tokens == 0
+    assert handle_realtime_stream_cost_calculation(
+        events,
+        usage,
+        "openai",
+        "live-resilient-test",
+        litellm_logging_obj=logger,
+    ) == pytest.approx(0.75)
+    assert bool(logger.model_call_details.get("realtime_backend_accounting_incomplete")) == (
+        nested["type"] == "response.completed"
+    )
+
+
+@pytest.mark.parametrize("missing_usage_duplicate", [False, True])
+def test_live_missing_backend_price_preserves_duration_and_marks_accounting_incomplete(
+    monkeypatch, missing_usage_duplicate
+):
+    from unittest.mock import MagicMock
+
+    monkeypatch.setitem(
+        litellm.model_cost,
+        "live-resilient-test",
+        {
+            "litellm_provider": "openai",
+            "mode": "realtime",
+            "input_cost_per_second": 0.025,
+        },
+    )
+    logger = MagicMock(spec=Logging)
+    logger.model_call_details = {}
+    response = {
+        "id": "resp_unknown",
+        "created_at": 1,
+        "model": "unmapped-live-backend-price-test",
+        "output": [],
+        "usage": {"input_tokens": 20, "output_tokens": 10, "total_tokens": 30},
+    }
+    events = [
+        {"type": "response.event", "event": {"type": "response.completed", "response": response}},
+        *(
+            [
+                {
+                    "type": "response.event",
+                    "event": {"type": "response.completed", "response": {**response, "usage": None}},
+                }
+            ]
+            if missing_usage_duplicate
+            else []
+        ),
+        {"type": "session.closed", "usage": {"seconds": 30}},
+    ]
+    usage = RealtimeAPITokenUsageProcessor.collect_and_combine_usage_from_realtime_stream_results(events)
+    assert usage.total_tokens == 30
+    assert handle_realtime_stream_cost_calculation(
+        events,
+        usage,
+        "openai",
+        "live-resilient-test",
+        litellm_logging_obj=logger,
+    ) == pytest.approx(0.75)
+    assert logger.model_call_details["realtime_backend_accounting_incomplete"] is True