[Feature] WebSocket streaming audio input for ASR (#22848)

Co-authored-by: Yihao Wang <42559837+AgainstEntropy@users.noreply.github.com>

python/sglang/srt/entrypoints/http_server.py

@@ -54,6 +54,7 @@ from fastapi import (
     Query,
     Request,
     UploadFile,
+    WebSocket,
 )
 from fastapi.exceptions import RequestValidationError
 from fastapi.middleware.cors import CORSMiddleware
@@ -1605,6 +1606,17 @@ async def openai_v1_audio_transcriptions(
     )
 
 
+@app.websocket("/v1/realtime")
+async def openai_v1_realtime_transcription(ws: WebSocket):
+    """OpenAI Realtime transcription WebSocket endpoint."""
+    # /v1/realtime is OpenAI's unified Realtime URL covering transcription +
+    # chat modes. This handler implements the transcription subset only;
+    # chat-mode session.update payloads are rejected by the
+    # `Literal["transcription"]` constraint on TranscriptionSessionConfig.type
+    # (see realtime/protocol.py).
+    await ws.app.state.openai_serving_transcription.handle_websocket(ws)
+
+
 @app.get("/v1/models", response_class=ORJSONResponse)
 async def available_models():
     """Show available models. OpenAI-compatible endpoint."""

python/sglang/srt/entrypoints/openai/realtime/__init__.py

@@ -0,0 +1,5 @@
+"""Realtime transcription WebSocket package. Exposes the FastAPI WS entry."""
+
+from sglang.srt.entrypoints.openai.realtime.handler import (
+    handle_realtime_transcription as handle_realtime_transcription,
+)

python/sglang/srt/entrypoints/openai/realtime/handler.py

@@ -0,0 +1,119 @@
+"""WebSocket entry for realtime transcription.
+
+Handles accept, concurrency, cleanup. Event loop is in session.py.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+
+from fastapi import WebSocket, WebSocketDisconnect
+from openai.types.realtime import RealtimeErrorEvent
+from openai.types.realtime.realtime_error import RealtimeError
+
+from sglang.srt.entrypoints.openai.realtime.session import RealtimeConnection
+from sglang.srt.entrypoints.openai.transcription_adapters.base import (
+    TranscriptionAdapter,
+)
+from sglang.srt.managers.tokenizer_manager import TokenizerManager
+from sglang.srt.server_args import ServerArgs
+from sglang.srt.utils import random_uuid
+
+logger = logging.getLogger(__name__)
+
+
+async def _safe_send(websocket: WebSocket, text: str) -> None:
+    try:
+        await websocket.send_text(text)
+    except (WebSocketDisconnect, RuntimeError) as e:
+        logger.debug("[realtime] send failed (peer gone): %s", e)
+
+
+async def _safe_close(websocket: WebSocket) -> None:
+    try:
+        await websocket.close()
+    except (WebSocketDisconnect, RuntimeError) as e:
+        logger.debug("[realtime] close failed (already closed): %s", e)
+
+
+async def _reject_before_session(
+    websocket: WebSocket,
+    code: str,
+    message: str,
+    *,
+    error_type: str = "invalid_request_error",
+) -> None:
+    """Reject path that runs before acquiring the session semaphore, so
+    unsupported / over-capacity peers don't hold a session slot."""
+    try:
+        await websocket.accept()
+    except (WebSocketDisconnect, RuntimeError) as e:
+        logger.debug("[realtime] reject: accept failed: %s", e)
+        return
+    logger.info("[realtime] rejected (%s)", code)
+    envelope = RealtimeErrorEvent(
+        event_id=f"event_{random_uuid()}",
+        type="error",
+        error=RealtimeError(type=error_type, code=code, message=message),
+    )
+    await _safe_send(websocket, envelope.model_dump_json())
+    await _safe_close(websocket)
+
+
+async def handle_realtime_transcription(
+    websocket: WebSocket,
+    tokenizer_manager: TokenizerManager,
+    adapter: TranscriptionAdapter,
+    server_args: ServerArgs,
+    session_semaphore: asyncio.Semaphore,
+) -> None:
+    """WS endpoint for /v1/realtime. Pre-session validation runs before
+    the semaphore so rejects don't consume a session slot; the
+    ``async with`` then guarantees the slot is released even if
+    RealtimeConnection raises."""
+    if not adapter.supports_chunked_streaming:
+        await _reject_before_session(
+            websocket,
+            "not_supported",
+            "Model does not support streaming ASR",
+        )
+        return
+
+    if session_semaphore.locked():
+        await _reject_before_session(
+            websocket,
+            "too_many_sessions",
+            f"Maximum concurrent sessions reached "
+            f"({server_args.asr_max_concurrent_sessions}).",
+            error_type="rate_limit_exceeded",
+        )
+        return
+
+    async with session_semaphore:
+        try:
+            try:
+                await websocket.accept()
+            except (WebSocketDisconnect, RuntimeError) as e:
+                logger.debug("[realtime] accept failed: %s", e)
+                return
+            connection = RealtimeConnection(
+                websocket, tokenizer_manager, adapter, server_args
+            )
+            await connection.run()
+        except WebSocketDisconnect:
+            logger.info("[realtime] client disconnected (normal)")
+        except Exception:
+            logger.exception("[realtime] unexpected error in session")
+            envelope = RealtimeErrorEvent(
+                event_id=f"event_{random_uuid()}",
+                type="error",
+                error=RealtimeError(
+                    type="server_error",
+                    code="inference_failed",
+                    message="Internal server error",
+                ),
+            )
+            await _safe_send(websocket, envelope.model_dump_json())
+        finally:
+            await _safe_close(websocket)

python/sglang/srt/entrypoints/openai/realtime/protocol.py

@@ -0,0 +1,78 @@
+"""Wire schema for Realtime WS transcription sessions."""
+
+from __future__ import annotations
+
+from typing import Literal, Optional, Union
+
+from openai.types.realtime import SessionUpdateEvent as _SessionUpdateEvent
+from openai.types.realtime.audio_transcription import (
+    AudioTranscription as _AudioTranscription,
+)
+from openai.types.realtime.realtime_audio_formats import AudioPCM as _AudioPCM
+from openai.types.realtime.realtime_audio_formats import AudioPCMA as _AudioPCMA
+from openai.types.realtime.realtime_audio_formats import AudioPCMU as _AudioPCMU
+from openai.types.realtime.realtime_transcription_session_audio import (
+    RealtimeTranscriptionSessionAudio as _AudioCfg,
+)
+from openai.types.realtime.realtime_transcription_session_audio_input import (
+    RealtimeTranscriptionSessionAudioInput as _AudioInputCfg,
+)
+from openai.types.realtime.realtime_transcription_session_create_request import (
+    RealtimeTranscriptionSessionCreateRequest as _SessionCfg,
+)
+from pydantic import Field
+from typing_extensions import Annotated
+
+# Fallback rate when the client omits `audio.input.format.rate`. SDK pins
+# `AudioPCM.rate` to Literal[24000], so this matches the only value the SDK
+# accepts when the field is present.
+DEFAULT_INPUT_SAMPLE_RATE = 24000
+
+# Wire rates we accept on `audio.input.format.rate` and resample to
+# `adapter.model_sample_rate` server-side. 24000 matches the SDK pin;
+# 16000 and 48000 widen it to cover common ASR-client and consumer-audio
+# rates. Add a value here only after verifying transcription quality.
+SUPPORTED_INPUT_SAMPLE_RATES = (16000, 24000, 48000)
+
+
+class AudioPCM(_AudioPCM):
+    type: Literal["audio/pcm"] = "audio/pcm"
+    rate: Optional[int] = None
+
+
+class AudioPCMU(_AudioPCMU):
+    type: Literal["audio/pcmu"] = "audio/pcmu"
+
+
+class AudioPCMA(_AudioPCMA):
+    type: Literal["audio/pcma"] = "audio/pcma"
+
+
+AudioInputFormat = Annotated[
+    Union[AudioPCM, AudioPCMU, AudioPCMA],
+    Field(discriminator="type"),
+]
+
+
+class AudioTranscription(_AudioTranscription):
+    # SDK pins model to Literal["whisper-1", "gpt-4o-*-transcribe", ...];
+    # sglang serves arbitrary ASR models (Qwen3-ASR, etc.) and treats the
+    # client-supplied name as echo-only.
+    model: Optional[str] = None
+
+
+class TranscriptionSessionAudioInput(_AudioInputCfg):
+    format: Optional[AudioInputFormat] = None
+    transcription: Optional[AudioTranscription] = None
+
+
+class TranscriptionSessionAudio(_AudioCfg):
+    input: Optional[TranscriptionSessionAudioInput] = None
+
+
+class TranscriptionSessionConfig(_SessionCfg):
+    audio: Optional[TranscriptionSessionAudio] = None
+
+
+class SessionUpdateEvent(_SessionUpdateEvent):
+    session: TranscriptionSessionConfig

python/sglang/srt/entrypoints/openai/realtime/session.py

@@ -0,0 +1,740 @@
+"""WebSocket session for realtime ASR.
+
+Pre-commit deltas reference the reserved current_item_id that the
+subsequent input_audio_buffer.committed and conversation.item.created
+events will announce — sglang-specific, deviates from OpenAI's
+commit-only delta emission.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import io
+import json
+import logging
+import math
+from dataclasses import dataclass, field
+from typing import Any, Dict, List, Optional
+
+import numpy as np
+import pybase64
+import soundfile as sf
+from fastapi import WebSocket, WebSocketDisconnect
+from openai.types.realtime import (
+    ConversationItemCreatedEvent,
+    InputAudioBufferAppendEvent,
+    InputAudioBufferClearedEvent,
+    InputAudioBufferClearEvent,
+    InputAudioBufferCommitEvent,
+    InputAudioBufferCommittedEvent,
+    RealtimeErrorEvent,
+    SessionCreatedEvent,
+    SessionUpdatedEvent,
+)
+from openai.types.realtime.conversation_item_input_audio_transcription_completed_event import (
+    ConversationItemInputAudioTranscriptionCompletedEvent,
+    UsageTranscriptTextUsageDuration,
+)
+from openai.types.realtime.conversation_item_input_audio_transcription_delta_event import (
+    ConversationItemInputAudioTranscriptionDeltaEvent,
+)
+from openai.types.realtime.conversation_item_input_audio_transcription_failed_event import (
+    ConversationItemInputAudioTranscriptionFailedEvent,
+)
+from openai.types.realtime.conversation_item_input_audio_transcription_failed_event import (
+    Error as TranscriptionFailedError,
+)
+from openai.types.realtime.realtime_conversation_item_user_message import (
+    Content as InputAudioContent,
+)
+from openai.types.realtime.realtime_conversation_item_user_message import (
+    RealtimeConversationItemUserMessage,
+)
+from openai.types.realtime.realtime_error import RealtimeError
+from pydantic import BaseModel, ValidationError
+
+from sglang.srt.entrypoints.openai.protocol import TranscriptionRequest
+from sglang.srt.entrypoints.openai.realtime.protocol import (
+    DEFAULT_INPUT_SAMPLE_RATE,
+    SUPPORTED_INPUT_SAMPLE_RATES,
+    AudioPCM,
+    SessionUpdateEvent,
+    TranscriptionSessionAudioInput,
+    TranscriptionSessionConfig,
+)
+from sglang.srt.entrypoints.openai.streaming_asr import (
+    StreamingASRState,
+    needs_space,
+    normalize_whitespace,
+    process_asr_chunk,
+)
+from sglang.srt.entrypoints.openai.transcription_adapters.base import (
+    TranscriptionAdapter,
+)
+from sglang.srt.managers.tokenizer_manager import TokenizerManager
+from sglang.srt.server_args import ServerArgs
+from sglang.srt.utils import random_uuid
+
+logger = logging.getLogger(__name__)
+
+
+# PCM16: 16-bit samples → 2 bytes each. Used for frame-length validation
+# and bytes/sec arithmetic against `np.frombuffer(..., dtype=np.int16)` below.
+_SAMPLE_WIDTH = 2
+
+
+def _resample_to_target_rate(pcm: bytes, src_rate: int, target_rate: int) -> bytes:
+    if src_rate == target_rate or not pcm:
+        return pcm
+    import torch
+    import torchaudio
+
+    samples = np.frombuffer(pcm, dtype=np.int16).astype(np.float32) / 32768.0
+    audio = torch.from_numpy(samples).unsqueeze(0)
+    audio = torchaudio.functional.resample(
+        audio, orig_freq=src_rate, new_freq=target_rate
+    )
+    samples = audio.squeeze(0).numpy()
+    # Clip to int16 range via 2^15 - 1 so a clipped 1.0 stays representable.
+    return (np.clip(samples, -1.0, 1.0) * 32767.0).astype(np.int16).tobytes()
+
+
+def _pcm_to_wav(pcm: bytes, sample_rate: int) -> bytes:
+    samples = np.frombuffer(pcm, dtype=np.int16)
+    buf = io.BytesIO()
+    sf.write(buf, samples, sample_rate, format="WAV")
+    return buf.getvalue()
+
+
+_CLIENT_EVENT_TYPES: Dict[str, type] = {
+    "session.update": SessionUpdateEvent,
+    "input_audio_buffer.append": InputAudioBufferAppendEvent,
+    "input_audio_buffer.commit": InputAudioBufferCommitEvent,
+    "input_audio_buffer.clear": InputAudioBufferClearEvent,
+}
+
+
+def _parse_client_event(raw: Dict[str, Any]) -> Optional[BaseModel]:
+    """Parse, returning None if type is unknown. Raises ValidationError on
+    a malformed payload of a known type."""
+    cls = _CLIENT_EVENT_TYPES.get(raw.get("type"))
+    if cls is None:
+        return None
+    return cls.model_validate(raw)
+
+
+@dataclass
+class _SessionConfig:
+    """Session-level configuration negotiated via session.update: audio
+    input format, requested language, sampling params. Persists until
+    the session ends; ``configured`` gates audio-frame handling so the
+    server doesn't run inference on PCM sent before session.update."""
+
+    input_sample_rate: int = DEFAULT_INPUT_SAMPLE_RATE
+    language: Optional[str] = None
+    client_model: Optional[str] = None
+    sampling_params: Optional[Dict[str, Any]] = None
+    configured: bool = False
+
+
+@dataclass
+class _AudioState:
+    """Per-item audio state: PCM buffer accumulated from
+    input_audio_buffer.append, the chunked ASR rollback state, and the
+    static buffer-size limits set at __init__. pcm_buffer / state /
+    last_inference_offset reset on commit-roll and clear; the size limits
+    stay constant for the session's lifetime."""
+
+    max_buffer_bytes: int
+    chunk_size_bytes: int
+    state: StreamingASRState
+    pcm_buffer: bytearray = field(default_factory=bytearray)
+    last_inference_offset: int = 0
+
+
+@dataclass
+class _ItemState:
+    """Per-item conversation-item ids and the wire-formatted deltas
+    emitted so far for the current item. current_item_id is reserved at
+    __init__ and only announced to the client by
+    input_audio_buffer.committed."""
+
+    current_item_id: str
+    previous_item_id: Optional[str] = None
+    emitted_deltas: List[str] = field(default_factory=list)
+
+
+class RealtimeConnection:
+    """One realtime transcription session. Drives the WS receive loop,
+    dispatches typed client events to the matching _on_* handler, and
+    triggers chunked ASR inference at audio buffer thresholds."""
+
+    def __init__(
+        self,
+        websocket: WebSocket,
+        tokenizer_manager: TokenizerManager,
+        adapter: TranscriptionAdapter,
+        server_args: ServerArgs,
+    ) -> None:
+        self.websocket = websocket
+        self.tokenizer_manager = tokenizer_manager
+        self.adapter = adapter
+        self.server_args = server_args
+
+        self.session_id = f"sess_{random_uuid()}"
+        self._current_client_event_id: Optional[str] = None
+
+        self.model_sample_rate = adapter.model_sample_rate
+        self.bytes_per_second = self.model_sample_rate * _SAMPLE_WIDTH
+        self.max_buffer_seconds = server_args.asr_max_buffer_seconds
+
+        self.config = _SessionConfig()
+
+        state = StreamingASRState(**adapter.chunked_streaming_config)
+        chunk_size_bytes = int(state.chunk_size_sec * self.bytes_per_second)
+        if chunk_size_bytes <= 0:
+            raise RuntimeError(
+                f"adapter.chunked_streaming_config produced non-positive "
+                f"chunk_size_sec; got {state.chunk_size_sec!r}"
+            )
+        self.audio = _AudioState(
+            max_buffer_bytes=self.max_buffer_seconds * self.bytes_per_second,
+            chunk_size_bytes=chunk_size_bytes,
+            state=state,
+        )
+
+        self.item = _ItemState(current_item_id=f"item_{random_uuid()}")
+
+    async def run(self) -> None:
+        await self._send(
+            SessionCreatedEvent(
+                event_id=f"event_{random_uuid()}",
+                type="session.created",
+                session=self._build_session_info(),
+            )
+        )
+
+        try:
+            await self._run_loop()
+        except WebSocketDisconnect:
+            logger.info("[realtime] client disconnected: %s", self.session_id)
+        except Exception:
+            logger.exception("[realtime] unexpected error: %s", self.session_id)
+            try:
+                await self._send_error(
+                    "inference_failed",
+                    "Internal server error",
+                    error_type="server_error",
+                )
+            except (WebSocketDisconnect, RuntimeError) as e:
+                logger.debug(
+                    "[realtime] failed to notify client of unexpected error: %s",
+                    e,
+                )
+
+    async def _run_loop(self) -> None:
+        """Receive-and-dispatch loop. Validation errors emit an error event
+        and continue; fatal append-path errors (buffer overflow, append-time
+        inference failure) close the WebSocket and terminate the loop.
+        """
+        while True:
+            self._current_client_event_id = None
+            message = await self.websocket.receive()
+            if message["type"] == "websocket.disconnect":
+                return
+
+            text = message.get("text")
+            if not text:
+                if message.get("bytes") is not None:
+                    # OpenAI Realtime is base64 PCM in JSON; binary frames aren't supported.
+                    await self._send_error(
+                        "invalid_payload",
+                        "Binary frames are not supported on /v1/realtime; "
+                        "use input_audio_buffer.append with base64 audio.",
+                    )
+                continue
+
+            try:
+                raw = json.loads(text)
+            except json.JSONDecodeError:
+                await self._send_error("invalid_payload", "Invalid JSON")
+                continue
+            if not isinstance(raw, dict):
+                await self._send_error(
+                    "invalid_payload", "Top-level event must be a JSON object"
+                )
+                continue
+
+            self._current_client_event_id = raw.get("event_id")
+            try:
+                event = _parse_client_event(raw)
+            except ValidationError as e:
+                # Report first error only; matches OpenAI server behavior.
+                err = e.errors()[0]
+                loc = ".".join(str(x) for x in err["loc"])
+                await self._send_error(
+                    "invalid_value",
+                    err.get("msg") or "Invalid payload",
+                    param=loc or None,
+                )
+                continue
+            if event is None:
+                await self._send_error(
+                    "unknown_event",
+                    f"Unknown event type: {raw.get('type')!r}",
+                )
+                continue
+            terminate = await self._dispatch(event)
+            if terminate:
+                return
+
+    async def _dispatch(self, event: BaseModel) -> bool:
+        """Returns True if the session should terminate."""
+        if isinstance(event, InputAudioBufferAppendEvent):
+            return await self._on_input_audio_buffer_append(event)
+        if isinstance(event, SessionUpdateEvent):
+            await self._on_session_update(event)
+        elif isinstance(event, InputAudioBufferCommitEvent):
+            await self._on_input_audio_buffer_commit(event)
+        elif isinstance(event, InputAudioBufferClearEvent):
+            await self._on_input_audio_buffer_clear(event)
+        return False
+
+    async def _on_session_update(self, event: SessionUpdateEvent) -> None:
+        cfg = event.session
+
+        # Normalize audio to an empty input cfg if absent so downstream
+        # `audio.X is not None` reads as a business rule, not an existence check.
+        # transcription stays nullable so partial-update can detect whether
+        # the client sent the block.
+        audio = (
+            cfg.audio.input if cfg.audio else None
+        ) or TranscriptionSessionAudioInput()
+        transcription = audio.transcription
+
+        # Validate first, then mutate config only after the whole update is accepted.
+        if audio.turn_detection is not None:
+            await self._send_error(
+                "not_supported",
+                "Server-side VAD is not implemented; "
+                "set audio.input.turn_detection: null and commit explicitly.",
+                param="session.audio.input.turn_detection",
+            )
+            return
+        if audio.noise_reduction is not None:
+            await self._send_error(
+                "not_supported",
+                "audio.input.noise_reduction is not supported; set to null.",
+                param="session.audio.input.noise_reduction",
+            )
+            return
+        if transcription is not None and transcription.prompt is not None:
+            await self._send_error(
+                "not_supported",
+                "audio.input.transcription.prompt is not supported.",
+                param="session.audio.input.transcription.prompt",
+            )
+            return
+        if (
+            transcription is not None
+            and transcription.model
+            and transcription.model != self.server_args.served_model_name
+        ):
+            await self._send_error(
+                "not_supported",
+                f"Model {transcription.model!r} is not served by this endpoint "
+                f"(serving {self.server_args.served_model_name!r}); set "
+                f"transcription.model to null or to the server's model name.",
+                param="session.audio.input.transcription.model",
+            )
+            return
+
+        new_rate = self.config.input_sample_rate  # default: keep current
+        fmt = audio.format
+        if fmt is not None:
+            if not isinstance(fmt, AudioPCM):
+                # G.711 (pcmu / pcma): not implemented.
+                await self._send_error(
+                    "not_supported",
+                    f"audio.input.format.type must be 'audio/pcm'; "
+                    f"{fmt.type!r} is not implemented",
+                    param="session.audio.input.format.type",
+                )
+                return
+            if fmt.rate is not None and fmt.rate not in SUPPORTED_INPUT_SAMPLE_RATES:
+                await self._send_error(
+                    "invalid_value",
+                    f"audio.input.format.rate must be one of "
+                    f"{SUPPORTED_INPUT_SAMPLE_RATES}, got {fmt.rate}",
+                    param="session.audio.input.format.rate",
+                )
+                return
+            new_rate = fmt.rate or DEFAULT_INPUT_SAMPLE_RATE
+            # Changing the rate mid-item would leave already-buffered PCM
+            # at the old rate mixed with new audio at the new rate, so
+            # require the client to commit or clear before switching.
+            if new_rate != self.config.input_sample_rate and self.audio.pcm_buffer:
+                await self._send_error(
+                    "invalid_state",
+                    "Cannot change audio.input.format.rate while audio is "
+                    "buffered; commit or clear the current item first.",
+                    param="session.audio.input.format.rate",
+                )
+                return
+
+        # Mutation pass — no early returns past this point.
+        self.config.input_sample_rate = new_rate
+        if transcription is not None:
+            self.config.client_model = transcription.model
+            self.config.language = transcription.language
+        self.config.sampling_params = self.adapter.build_sampling_params(
+            TranscriptionRequest(language=self.config.language)
+        )
+        self.config.configured = True
+
+        # Side effects: log + ack.
+        if cfg.include:
+            logger.info(
+                "[realtime] %s: include[] received but not implemented; ignoring: %s",
+                self.session_id,
+                cfg.include,
+            )
+        if self.config.input_sample_rate != self.model_sample_rate:
+            logger.info(
+                "[realtime] %s configured: resample %d→%d (ratio %.2f), language=%s",
+                self.session_id,
+                self.config.input_sample_rate,
+                self.model_sample_rate,
+                self.config.input_sample_rate / self.model_sample_rate,
+                self.config.language,
+            )
+        await self._send(
+            SessionUpdatedEvent(
+                event_id=f"event_{random_uuid()}",
+                type="session.updated",
+                session=self._build_session_info(),
+            )
+        )
+
+    async def _on_input_audio_buffer_append(
+        self, event: InputAudioBufferAppendEvent
+    ) -> bool:
+        """Returns True if the session should terminate (buffer overflow or
+        append-time inference failure)."""
+        if not self.config.configured:
+            await self._send_error(
+                "invalid_state", "Send session.update before audio frames"
+            )
+            return False
+
+        # Empty audio is a no-op (heartbeat frames); skip b64decode.
+        if not event.audio:
+            return False
+
+        try:
+            data = pybase64.b64decode(event.audio, validate=True)
+        except (ValueError, TypeError):
+            await self._send_error(
+                "invalid_audio", "audio field is not valid base64", param="audio"
+            )
+            return False
+
+        if len(data) % _SAMPLE_WIDTH != 0:
+            await self._send_error(
+                "invalid_audio_format",
+                f"PCM16 frame length must be a multiple of {_SAMPLE_WIDTH} bytes",
+            )
+            return False
+
+        # Estimate post-resample size before resampling so oversized frames fail early.
+        src_samples = len(data) // _SAMPLE_WIDTH
+        target_samples = math.ceil(
+            src_samples * self.model_sample_rate / self.config.input_sample_rate
+        )
+        if (
+            len(self.audio.pcm_buffer) + target_samples * _SAMPLE_WIDTH
+            > self.audio.max_buffer_bytes
+        ):
+            # Close 1009 ("message too big") so clients can distinguish
+            # session-resource exhaustion from a normal close.
+            await self._send_error_and_close(
+                "buffer_overflow",
+                f"Accumulated audio exceeded {self.max_buffer_seconds}s; "
+                f"client is sending faster than inference can keep up",
+                close_code=1009,
+            )
+            return True
+
+        if self.config.input_sample_rate != self.model_sample_rate:
+            data = await asyncio.to_thread(
+                _resample_to_target_rate,
+                data,
+                self.config.input_sample_rate,
+                self.model_sample_rate,
+            )
+        self.audio.pcm_buffer.extend(data)
+
+        new_audio_bytes = len(self.audio.pcm_buffer) - self.audio.last_inference_offset
+        if new_audio_bytes >= self.audio.chunk_size_bytes:
+            ok = await self._run_inference(is_last=False)
+            if not ok:
+                # WS already closed inside _run_inference.
+                return True
+        return False
+
+    async def _on_input_audio_buffer_commit(
+        self, event: InputAudioBufferCommitEvent
+    ) -> None:
+        if not self.config.configured:
+            await self._send_error("invalid_state", "Send session.update before commit")
+            return
+        if not self.audio.pcm_buffer and not self.audio.state.full_transcript:
+            await self._send_error(
+                "invalid_state", "Cannot commit an empty audio buffer"
+            )
+            return
+
+        has_new_audio = len(self.audio.pcm_buffer) > self.audio.last_inference_offset
+        item_id = self.item.current_item_id
+        prev_item_id = self.item.previous_item_id
+
+        partial_transcript = normalize_whitespace("".join(self.item.emitted_deltas))
+
+        await self._send(
+            InputAudioBufferCommittedEvent(
+                event_id=f"event_{random_uuid()}",
+                type="input_audio_buffer.committed",
+                item_id=item_id,
+                previous_item_id=prev_item_id,
+            )
+        )
+        await self._send(
+            ConversationItemCreatedEvent(
+                event_id=f"event_{random_uuid()}",
+                type="conversation.item.created",
+                previous_item_id=prev_item_id,
+                item=RealtimeConversationItemUserMessage(
+                    id=item_id,
+                    type="message",
+                    role="user",
+                    status="completed",
+                    content=[
+                        InputAudioContent(
+                            type="input_audio", transcript=partial_transcript
+                        )
+                    ],
+                ),
+            )
+        )
+
+        # Capture pcm duration before `_start_next_item()` runs: starting
+        # the next item clears pcm_buffer, so reading it after gives 0.
+        pcm_duration_seconds = len(self.audio.pcm_buffer) / self.bytes_per_second
+
+        if has_new_audio:
+            ok = await self._run_inference(is_last=True)
+            if not ok:
+                # _run_inference already emitted transcription.failed and
+                # rolled the item; don't also emit completed.
+                return
+        elif self.audio.state.full_transcript:
+            # Audio length was exactly a chunk_size_bytes multiple. Flush
+            # the tail tokens update() held back.
+            tail = self.audio.state.finalize()
+            await self._emit_transcription_delta(tail)
+
+        # Build from emitted_deltas, not state.full_transcript: prefix injection
+        # means the last chunk's full_transcript is only the continuation tail.
+        transcript = normalize_whitespace("".join(self.item.emitted_deltas))
+
+        await self._send(
+            ConversationItemInputAudioTranscriptionCompletedEvent(
+                event_id=f"event_{random_uuid()}",
+                type="conversation.item.input_audio_transcription.completed",
+                item_id=item_id,
+                content_index=0,
+                transcript=transcript,
+                usage=UsageTranscriptTextUsageDuration(
+                    type="duration", seconds=pcm_duration_seconds
+                ),
+            )
+        )
+
+        self._start_next_item()
+
+    async def _on_input_audio_buffer_clear(
+        self, event: InputAudioBufferClearEvent
+    ) -> None:
+        # Reserve a fresh current_item_id so post-clear pre-commit deltas
+        # don't share an item_id with deltas the client already received
+        # for the abandoned audio. previous_item_id is NOT touched — the
+        # cleared item was never committed, so the prior-commit chain
+        # shouldn't include it.
+        self._reset_inference_state()
+        self.item.current_item_id = f"item_{random_uuid()}"
+        await self._send(
+            InputAudioBufferClearedEvent(
+                event_id=f"event_{random_uuid()}", type="input_audio_buffer.cleared"
+            )
+        )
+
+    async def _run_inference(self, is_last: bool) -> bool:
+        """Run ASR on the current cumulative buffer. Returns False on failure:
+        commit-time emits transcription.failed and rolls the item; append-time
+        emits a generic error envelope and closes the WebSocket."""
+        wav_data = await asyncio.to_thread(
+            _pcm_to_wav, bytes(self.audio.pcm_buffer), self.model_sample_rate
+        )
+        try:
+            delta = await process_asr_chunk(
+                tokenizer_manager=self.tokenizer_manager,
+                adapter=self.adapter,
+                state=self.audio.state,
+                audio_data=wav_data,
+                sampling_params=self.config.sampling_params,
+                is_last=is_last,
+            )
+        except Exception:
+            logger.exception(
+                "[realtime] inference failed: session=%s item=%s buffer_bytes=%d",
+                self.session_id,
+                self.item.current_item_id,
+                len(self.audio.pcm_buffer),
+            )
+            if is_last:
+                # Commit-time failure: committed + created already emitted,
+                # so the item exists client-side and transcription.failed
+                # can reference it. Wire message is hardcoded "Transcription
+                # failed" — don't leak backend traces to the client; full
+                # error is in the logger.exception above.
+                await self._send(
+                    ConversationItemInputAudioTranscriptionFailedEvent(
+                        event_id=f"event_{random_uuid()}",
+                        type="conversation.item.input_audio_transcription.failed",
+                        item_id=self.item.current_item_id,
+                        content_index=0,
+                        error=TranscriptionFailedError(
+                            type="server_error",
+                            code="inference_failed",
+                            message="Transcription failed",
+                        ),
+                    )
+                )
+                self._start_next_item()
+            else:
+                # Append-time failure: the item isn't visible client-side
+                # yet (committed/created fire at commit), so
+                # transcription.failed would reference a ghost id.
+                await self._send_error_and_close(
+                    "inference_failed",
+                    "Transcription failed",
+                    close_code=1011,
+                )
+            return False
+
+        self.audio.last_inference_offset = len(self.audio.pcm_buffer)
+        await self._emit_transcription_delta(delta)
+        return True
+
+    async def _emit_transcription_delta(self, delta: str) -> None:
+        """emitted_deltas stores wire-formatted text (with leading
+        boundary spaces baked in), so "".join(...) reconstructs the
+        cumulative transcript verbatim."""
+        if not delta:
+            return
+        for word in delta.split(" "):
+            if not word:
+                continue
+            prev = self.item.emitted_deltas[-1] if self.item.emitted_deltas else ""
+            formatted = f" {word}" if needs_space(prev, word) else word
+            self.item.emitted_deltas.append(formatted)
+            await self._send(
+                ConversationItemInputAudioTranscriptionDeltaEvent(
+                    event_id=f"event_{random_uuid()}",
+                    type="conversation.item.input_audio_transcription.delta",
+                    item_id=self.item.current_item_id,
+                    content_index=0,
+                    delta=formatted,
+                )
+            )
+
+    def _start_next_item(self) -> None:
+        self.item.previous_item_id = self.item.current_item_id
+        self.item.current_item_id = f"item_{random_uuid()}"
+        self._reset_inference_state()
+
+    def _reset_inference_state(self) -> None:
+        """Missing any of these resets leaks state across items."""
+        self.audio.state = StreamingASRState(**self.adapter.chunked_streaming_config)
+        self.audio.pcm_buffer.clear()  # in-place; reuses the buffer's allocation
+        self.item.emitted_deltas.clear()
+        self.audio.last_inference_offset = 0
+
+    def _build_session_info(self) -> TranscriptionSessionConfig:
+        # id / object aren't SDK fields; round-trip via extra='allow' so
+        # dumps emit them like the real server.
+        return TranscriptionSessionConfig.model_validate(
+            {
+                "type": "transcription",
+                "id": self.session_id,
+                "object": "realtime.transcription_session",
+                "audio": {
+                    "input": {
+                        "format": {
+                            "type": "audio/pcm",
+                            "rate": self.config.input_sample_rate,
+                        },
+                        "transcription": {
+                            "model": self.config.client_model,
+                            "language": self.config.language,
+                        },
+                        "noise_reduction": None,
+                        "turn_detection": None,
+                    }
+                },
+            }
+        )
+
+    async def _send(self, event: BaseModel) -> None:
+        await self.websocket.send_text(event.model_dump_json())
+
+    async def _send_error(
+        self,
+        code: str,
+        message: str,
+        *,
+        error_type: str = "invalid_request_error",
+        param: Optional[str] = None,
+    ) -> None:
+        envelope = RealtimeErrorEvent(
+            event_id=f"event_{random_uuid()}",
+            type="error",
+            error=RealtimeError(
+                type=error_type,
+                code=code,
+                message=message,
+                param=param,
+                event_id=self._current_client_event_id,
+            ),
+        )
+        await self.websocket.send_text(envelope.model_dump_json())
+
+    async def _send_error_and_close(
+        self,
+        code: str,
+        message: str,
+        *,
+        close_code: int,
+        error_type: str = "server_error",
+    ) -> None:
+        # Independent try-blocks: a failed send must not skip the close.
+        # We still need to release local starlette socket state even when
+        # the wire send doesn't reach the peer.
+        try:
+            await self._send_error(code, message, error_type=error_type)
+        except (WebSocketDisconnect, RuntimeError) as e:
+            logger.debug("[realtime] send error %s before close failed: %s", code, e)
+        try:
+            await self.websocket.close(code=close_code)
+        except (WebSocketDisconnect, RuntimeError) as e:
+            logger.debug("[realtime] close %d after %s failed: %s", close_code, code, e)

python/sglang/srt/entrypoints/openai/serving_transcription.py

@@ -29,7 +29,7 @@ import time
 import uuid
 from typing import TYPE_CHECKING, AsyncGenerator, List, Optional, Union
 
-from fastapi import Request
+from fastapi import Request, WebSocket
 from fastapi.responses import ORJSONResponse, Response, StreamingResponse
 
 from sglang.srt.entrypoints.openai.protocol import (
@@ -42,9 +42,14 @@ from sglang.srt.entrypoints.openai.protocol import (
     TranscriptionUsage,
     TranscriptionVerboseResponse,
 )
+from sglang.srt.entrypoints.openai.realtime import (
+    handle_realtime_transcription,
+)
 from sglang.srt.entrypoints.openai.serving_base import OpenAIServingBase
 from sglang.srt.entrypoints.openai.streaming_asr import (
     StreamingASRState,
+    needs_space,
+    process_asr_chunk,
     split_audio_chunks,
 )
 from sglang.srt.entrypoints.openai.transcription_adapters import resolve_adapter
@@ -65,6 +70,11 @@ class OpenAIServingTranscription(OpenAIServingBase):
         self._adapter = resolve_adapter(
             getattr(model_config.hf_config, "architectures", [])
         )
+        # Cap concurrent /v1/realtime sessions. The Semaphore is bound to the
+        # event loop on first acquire (uvicorn's loop in normal serving).
+        self._session_semaphore = asyncio.Semaphore(
+            tokenizer_manager.server_args.asr_max_concurrent_sessions
+        )
 
     def _request_id_prefix(self) -> str:
         return "trsc-"
@@ -190,7 +200,7 @@ class OpenAIServingTranscription(OpenAIServingBase):
 
         # For fused auto-detect, parse_fused_output returns the scrubbed
         # user-visible text. On parse failure (FSM abort, truncation) it
-        # returns (None, None) and we fall back to a best-effort scrub —
+        # returns (None, None) and we fall back to strip_special_tokens —
         # the language stays unset rather than reporting a bogus detection.
         if getattr(request, "_fused_autodetect", False):
             lang, visible = self._adapter.parse_fused_output(
@@ -375,13 +385,14 @@ class OpenAIServingTranscription(OpenAIServingBase):
         - Token-level streaming within chunks (stream=True)
         - Encoder window caching across chunks
         - Cross-chunk KV cache reuse
-        - WebSocket endpoint for real-time audio input
         """
         created_time = int(time.time())
         request_id = f"{self._request_id_prefix()}{uuid.uuid4().hex}"
         model = request.model
         state = StreamingASRState(**self._adapter.chunked_streaming_config)
-        first_word = True
+        # Track only the trailing char of the cumulative emit; `needs_space`
+        # uses prev[-1] / cur[0] so we don't need to keep the full buffer.
+        last_char = ""
 
         try:
             chunks = split_audio_chunks(request.audio_data, state.chunk_size_sec)
@@ -391,49 +402,24 @@ class OpenAIServingTranscription(OpenAIServingBase):
                     logger.info("[streaming_asr] client disconnected, stopping")
                     break
                 is_last = i == len(chunks) - 1
-                prompt = self._adapter.prompt_template + state.get_prefix_text()
 
-                chunk_request = GenerateReqInput(
-                    text=prompt,
+                delta = await process_asr_chunk(
+                    tokenizer_manager=self.tokenizer_manager,
+                    adapter=self._adapter,
+                    state=state,
                     audio_data=chunk_audio,
                     sampling_params=adapted_request.sampling_params,
-                    stream=False,
-                    modalities=["audio"],
+                    is_last=is_last,
+                    raw_request=raw_request,
                     routing_key=self.extract_routing_key(raw_request),
                 )
 
-                try:
-                    ret = None
-                    async for ret in self.tokenizer_manager.generate_request(
-                        chunk_request, raw_request
-                    ):
-                        break
-                except asyncio.CancelledError:
-                    raise
-                except ValueError as e:
-                    logger.warning(
-                        "[streaming_asr] chunk %d failed with ValueError: %s", i, e
-                    )
-                    continue
-
-                if ret is None:
-                    logger.warning("[streaming_asr] empty response for chunk %d", i)
-                    continue
-
-                text = self._adapter.postprocess_text(ret.get("text", ""))
-
-                if is_last:
-                    state.full_transcript = text
-                    delta = state.finalize()
-                else:
-                    delta = state.update(text)
-
                 if delta:
                     for word in delta.split(" "):
                         if not word:
                             continue
-                        content = word if first_word else " " + word
-                        first_word = False
+                        content = f" {word}" if needs_space(last_char, word) else word
+                        last_char = content[-1]
                         chunk_resp = TranscriptionStreamResponse(
                             id=request_id,
                             created=created_time,
@@ -469,3 +455,12 @@ class OpenAIServingTranscription(OpenAIServingBase):
             yield f"data: {error}\n\n"
 
         yield "data: [DONE]\n\n"
+
+    async def handle_websocket(self, websocket: WebSocket) -> None:
+        await handle_realtime_transcription(
+            websocket,
+            tokenizer_manager=self.tokenizer_manager,
+            adapter=self._adapter,
+            server_args=self.tokenizer_manager.server_args,
+            session_semaphore=self._session_semaphore,
+        )

python/sglang/srt/entrypoints/openai/streaming_asr.py

@@ -1,8 +1,26 @@
+import asyncio
 import io
+import logging
+import re
 from dataclasses import dataclass
-from typing import List
+from typing import Any, Dict, List, Optional
 
 import soundfile as sf
+from fastapi import Request
+
+from sglang.srt.entrypoints.openai.transcription_adapters.base import (
+    TranscriptionAdapter,
+)
+from sglang.srt.managers.io_struct import GenerateReqInput
+from sglang.srt.managers.tokenizer_manager import TokenizerManager
+
+logger = logging.getLogger(__name__)
+
+
+# Collapse whitespace before punctuation so batched-inference token
+# boundary jitter (" ," vs ",") doesn't leak into deltas. Covers both
+# ASCII punctuation and the CJK / fullwidth equivalents.
+_PUNCT_WS_RE = re.compile(r"\s+([,.;:!?，。！？；：、])")
 
 
 @dataclass
@@ -22,13 +40,23 @@ class StreamingASRState:
     unfixed_chunk_num: int
     unfixed_token_num: int
     confirmed_text: str = ""
+    # Monotonic accumulator; used as prompt prefix so the model sees a
+    # natural continuation point, not the rolled-back ``confirmed_text``.
+    emitted_text: str = ""
     full_transcript: str = ""
     chunk_index: int = 0
 
     def get_prefix_text(self) -> str:
-        if self.chunk_index < self.unfixed_chunk_num or not self.confirmed_text:
+        if self.chunk_index < self.unfixed_chunk_num or not self.emitted_text:
             return ""
-        return self.confirmed_text
+        return self.emitted_text
+
+    def _record_emit(self, delta: str) -> str:
+        if delta:
+            self.emitted_text = (
+                f"{self.emitted_text} {delta}".strip() if self.emitted_text else delta
+            )
+        return delta
 
     def update(self, new_transcript: str) -> str:
         old_confirmed = self.confirmed_text
@@ -40,7 +68,7 @@ class StreamingASRState:
         self.full_transcript = new_transcript
         self.chunk_index += 1
         if self.confirmed_text.startswith(old_confirmed):
-            return self.confirmed_text[len(old_confirmed) :].strip()
+            return self._record_emit(self.confirmed_text[len(old_confirmed) :].strip())
         # Model revised earlier text, use word level common prefix to avoid
         # re-emitting already-sent content and cutting mid-word.
         old_words = old_confirmed.split()
@@ -50,7 +78,7 @@ class StreamingASRState:
             if ow != nw:
                 break
             common_count += 1
-        return " ".join(new_words[common_count:])
+        return self._record_emit(" ".join(new_words[common_count:]))
 
     def finalize(self) -> str:
         confirmed_words = self.confirmed_text.split()
@@ -64,8 +92,8 @@ class StreamingASRState:
             common_count += 1
         self.confirmed_text = self.full_transcript
         if common_count == 0 and confirmed_words and all_words:
-            return self.full_transcript
-        return " ".join(all_words[common_count:])
+            return self._record_emit(self.full_transcript)
+        return self._record_emit(" ".join(all_words[common_count:]))
 
 
 def split_audio_chunks(audio_data: bytes, chunk_size_sec: float) -> List[bytes]:
@@ -91,3 +119,91 @@ def split_audio_chunks(audio_data: bytes, chunk_size_sec: float) -> List[bytes]:
         sf.write(buf, data[:end], sample_rate, format="WAV")
         chunks.append(buf.getvalue())
     return chunks
+
+
+def normalize_whitespace(text: str) -> str:
+    return _PUNCT_WS_RE.sub(r"\1", text)
+
+
+_NO_SPACE_BEFORE = frozenset(".,!?;:%)]}，。！？；：、）】》」』")
+_NO_SPACE_AFTER = frozenset("([{（【《「『")
+
+
+def _is_cjk(c: str) -> bool:
+    """Whether char is a CJK-context glyph that doesn't take inter-word
+    spaces — ideographs, Japanese kana, CJK punctuation, fullwidth forms.
+    Excludes Hangul / Devanagari / Arabic etc., which are non-ASCII but
+    space-separated and need the normal boundary space."""
+    cp = ord(c)
+    return (
+        0x3000 <= cp <= 0x303F  # CJK Symbols and Punctuation (，。、《》「」…)
+        or 0x3040 <= cp <= 0x309F  # Hiragana
+        or 0x30A0 <= cp <= 0x30FF  # Katakana
+        or 0x3400 <= cp <= 0x4DBF  # CJK Unified Ideographs Ext A
+        or 0x4E00 <= cp <= 0x9FFF  # CJK Unified Ideographs
+        or 0xFF00 <= cp <= 0xFFEF  # Halfwidth & Fullwidth Forms (fullwidth ASCII)
+    )
+
+
+def needs_space(prev: str, cur: str) -> bool:
+    """Return whether a boundary space is needed between emitted deltas.
+
+    Avoid spaces around punctuation and between adjacent CJK-context glyphs.
+    Shared by the realtime WS and HTTP SSE chunked streaming paths.
+    """
+    if not prev or not cur:
+        return False
+    if prev[-1].isspace() or cur[0].isspace():
+        return False
+    if cur[0] in _NO_SPACE_BEFORE or prev[-1] in _NO_SPACE_AFTER:
+        return False
+    if _is_cjk(prev[-1]) and _is_cjk(cur[0]):
+        return False
+    return True
+
+
+async def process_asr_chunk(
+    tokenizer_manager: TokenizerManager,
+    adapter: TranscriptionAdapter,
+    state: StreamingASRState,
+    audio_data: bytes,
+    sampling_params: Dict[str, Any],
+    is_last: bool,
+    raw_request: Optional[Request] = None,
+    routing_key: Optional[str] = None,
+) -> str:
+    """Run inference on one audio chunk. Shared by the HTTP and WebSocket paths."""
+    prompt = adapter.prompt_template + state.get_prefix_text()
+
+    chunk_request = GenerateReqInput(
+        text=prompt,
+        audio_data=audio_data,
+        sampling_params=sampling_params,
+        stream=False,
+        modalities=["audio"],
+    )
+    if routing_key is not None:
+        chunk_request.routing_key = routing_key
+
+    try:
+        ret = None
+        async for ret in tokenizer_manager.generate_request(chunk_request, raw_request):
+            break
+    except asyncio.CancelledError:
+        raise
+    except ValueError:
+        logger.warning(
+            "[streaming_asr] chunk %d failed", state.chunk_index, exc_info=True
+        )
+        raise
+
+    if ret is None:
+        logger.warning("[streaming_asr] empty response for chunk %d", state.chunk_index)
+        return ""
+
+    text = normalize_whitespace(adapter.postprocess_text(ret.get("text", "")))
+
+    if is_last:
+        state.full_transcript = text
+        return state.finalize()
+    return state.update(text)

python/sglang/srt/entrypoints/openai/transcription_adapters/base.py

@@ -81,6 +81,14 @@ class TranscriptionAdapter(ABC):
         """Whether this model uses chunk-based streaming instead of token-level streaming."""
         return False
 
+    @property
+    def model_sample_rate(self) -> int:
+        """Target sample rate in Hz the model expects. Realtime WS path
+        resamples client PCM to this rate before chunking. Default 16000
+        matches Whisper / Qwen3-ASR; override for models expecting other rates.
+        """
+        return 16000
+
     @property
     def prompt_template(self) -> str:
         """Prompt template for chunked streaming requests.

python/sglang/srt/server_args.py

@@ -512,6 +512,8 @@ class ServerArgs:
     tool_call_parser: Optional[str] = None
     tool_server: Optional[str] = None
     sampling_defaults: str = "model"
+    asr_max_buffer_seconds: int = 60
+    asr_max_concurrent_sessions: int = 32
 
     # Data parallelism
     dp_size: int = 1
@@ -864,6 +866,8 @@ class ServerArgs:
         self._handle_multimodal()
         # Validate SSL arguments early (before dummy-model short-circuit).
         self._handle_ssl_validation()
+        # Validate transcription/ASR-specific server args (model-independent).
+        self._handle_asr_validation()
 
         # Validate PD disaggregation flags early (before dummy-model short-circuit).
         from sglang.srt.arg_groups.pd_disaggregation_hook import (
@@ -4143,6 +4147,19 @@ class ServerArgs:
             )
             self.enable_mixed_chunk = False
 
+    def _handle_asr_validation(self):
+        """Validate transcription/ASR-specific server args."""
+        if self.asr_max_buffer_seconds <= 0:
+            raise ValueError(
+                f"--asr-max-buffer-seconds must be positive "
+                f"(got {self.asr_max_buffer_seconds})."
+            )
+        if self.asr_max_concurrent_sessions <= 0:
+            raise ValueError(
+                f"--asr-max-concurrent-sessions must be positive "
+                f"(got {self.asr_max_concurrent_sessions})."
+            )
+
     def _handle_other_validations(self):
         # Handle model inference tensor dump.
         if self.debug_tensor_dump_output_folder is not None:
@@ -5192,6 +5209,24 @@ class ServerArgs:
             "'model' uses the model's generation_config.json to get the recommended "
             "sampling parameters if available. Default is 'model'.",
         )
+        parser.add_argument(
+            "--asr-max-buffer-seconds",
+            type=int,
+            default=ServerArgs.asr_max_buffer_seconds,
+            help="Maximum seconds of PCM audio the streaming ASR WebSocket handler "
+            "will accumulate before closing the session with a buffer_overflow "
+            "error. Guards against OOM when a client streams audio faster than "
+            "inference can consume it. Default 60s.",
+        )
+        parser.add_argument(
+            "--asr-max-concurrent-sessions",
+            type=int,
+            default=ServerArgs.asr_max_concurrent_sessions,
+            help="Maximum number of concurrent realtime ASR WebSocket sessions "
+            "served by /v1/realtime. New connections beyond this cap are "
+            "accepted, sent an error{code:too_many_sessions} frame, and closed. "
+            "Default 32.",
+        )
 
         # Data parallelism
         parser.add_argument(