[V1, Feature] Add OpenAI Realtime WebSocket endpoint (M0)

[PopSoda2002]

Summary

Mounts /v1/realtime, an OpenAI-Realtime-compatible WebSocket API on sglang-omni v1, to enable streaming audio in / streaming transcript deltas out for low-latency voice agents and live transcription/translation. Disabled by default — opt in with --enable-realtime.

Related discussions: vLLM streaming-realtime, sgl-project/sglang#22474 (upstream RFC scoped to transcription only — this PR goes for the OpenAI superset instead).

What's in this PR

New package: sglang_omni_v1/serve/realtime/

• events.py — Pydantic schemas for the OpenAI Realtime client/server event vocabulary
• audio_buffer.py — append-only PCM16 rolling buffer
• session.py — per-WebSocket state machine; dispatches client events, drives the engine via the existing Coordinator.stream(), translates engine deltas back to Realtime server events
• manager.py — in-memory session_id → RealtimeSession registry

Misc

[PopSoda2002]

Ref: SGlang omni: sgl-project/sglang#22848, have a reproducible CI

[PopSoda2002]

realtime, trained, qwen3 omni, encoder, bidirectional, causcal. voxtral realtime https://huggingface.co/mistralai/Voxtral-Mini-4B-Realtime-2602

[zhaochenyang20]

Put these data in the Py files. Do not make a separate JSON file.

[zhaochenyang20]

Baseline JSON has only the manual block — even though the PR title bills server VAD as a core feature. Combined with [P0-BLOCKER] #1, my read is VAD mode hung on the second segment during benchmarking, so only the manual numbers shipped. CI gating off this baseline gives the VAD path zero coverage. Also n=5 with quantiles(vals, n=20)[-1] extrapolates p95 = 303 ms above max = 285 ms — statistically an artifact, visually wrong. Bump the fallback to len >= 20, or fall back to max for n < 20; or switch to numpy.percentile(vals, 95).

[zhaochenyang20]

StreamingVAD.reset() deliberately does not zero self.samples_consumed: line 111-117 only clears leftover_pcm / is_speech / silence_run_samples and sets last_speech_offset = self.samples_consumed (line 115), not 0. But handle_vad_emit treats emit.sample_offset as a byte offset inside the current audio_buffer: vad_byte = max(0, emit.sample_offset * 2) (line 220), then utterance_start_byte = min(vad_byte, audio_buffer.num_bytes) (line 221). The two coordinate systems happen to agree on the first utterance; after auto_commit_utterance → drop_buffer_and_reset_vad (session.py:295) the audio_buffer is cleared while vad.samples_consumed stays. The next emit's sample_offset is now an absolute counter (much larger than the fresh buffer's num_bytes), min(...) clamps to num_bytes, auto_commit_utterance sees start_byte == end_byte (line 285) and returns early — the whole utterance is silently dropped. The client receives speech_started / speech_stopped but never the matching committed / transcription.completed. The wall-clock timestamp_ms = offsets_to_ms(self.buffer_origin_samples + emit.sample_offset) (line 217) also double-counts and drifts by ~2x.

Fix:

# vad.py
def reset(self) -> None:
    self.leftover_pcm.clear()
    self.is_speech = False
    self.silence_run_samples = 0
    self.samples_consumed = 0          # zero the absolute counter
    self.last_speech_offset = 0        # always 0 after reset
    if hasattr(self.vad_model, "reset_states"):
        self.vad_model.reset_states()  # type: ignore[union-attr]

Add a unit test: run two back-to-back utterances with the same StreamingVAD, and assert that after SPEECH_STOPPED of the second utterance, auto_commit_utterance switches to the correct buffer slice and the client receives the corresponding committed event.

[zhaochenyang20]

grep -rn '_send_error\|make_event("error"' sglang_omni_v1/serve/realtime/ returns 0 hits — there is zero structured-error emission in the implementation. Every external-boundary guard is a bare assert:

• session.py:144-145: payload = json.loads(raw); assert isinstance(payload, dict)
• session.py:150-152: assert event_type in SUPPORTED_CLIENT_EVENT_TYPES
• session.py:166-176: input_audio_format / audio modality / semantic_vad asserts
• session.py:204-206: assert err != "overflow"
• session.py:303: assert not self.audio_buffer.is_empty()
• session.py:344-346: response-already-active assert
• session.py:353: audio-output-modality assert

But tests/ contains 6 assertions of the form evt["type"] == "error" + a specific error.code (table above). Each trigger goes assert → AssertionError → ASGI 1006 → TestClient sees WebSocketDisconnect → the test's evt["type"] == "error" assertion fails. Six tests can never pass against the current implementation. The test_websocket_invalid_json_kills_session test (smoke.py:314) is actually the correct one — it explicitly pytest.raises(JSONDecodeError). The other six are the contradiction.

Fix: Add the _send_error helper at the implementation layer, refactor all external boundary guards to emit structured error events followed by graceful shutdown. This complies with the OpenAI Realtime protocol contract and directly resolves the below [P0-BLOCKER] issue regarding buffer overflow DoS vulnerability.

async def _send_error(self, code: str, message: str, event_id: str | None = None) -> None:
    await self.send(make_event(
        "error",
        error={
            "type": "invalid_request_error",
            "code": code,
            "message": message,
            "event_id": event_id,
            "param": None,
        },
    ))

# handle_audio_append:
if err == "overflow":
    await self._send_error(
        "input_audio_buffer_too_large",
        f"buffer would exceed {self.audio_buffer.max_bytes} bytes",
    )
    await self.websocket.close(code=1009)
    self.closed = True
    return

# handle_session_update / dispatch / handle_audio_commit / handle_response_create
# Apply the same refactoring: replace each assert with the corresponding _send_error + close logic.

Either implement the error event at the implementation layer, or modify the tests to use pytest.raises(WebSocketDisconnect) — yet the former is the correct approach aligned with the protocol specification.

[zhaochenyang20]

File: sglang_omni_v1/serve/openai_api.py:469-476，sglang_omni_v1/serve/realtime/session.py:571-592，sglang_omni_v1/serve/realtime/manager.py:37-39

The WebSocket route's finally block only calls manager.close(session.session_id) (manager.py:37-39 just pops from a dict). RealtimeSession.teardown() at session.py:571-592 — which does Client.abort(active_request_id), cancels active_task, cancels queue_drainer, closes the websocket — is dead code. After client disconnect the drainer coroutine is still parked on await self.transcription_queue.get(), active_task is still streaming, no Client.abort() is issued — the pipeline keeps the in-flight request running to completion. One leaked GPU inference + one leaked asyncio task per unexpected disconnect. Under sustained load this is leak-to-OOM.

Fix:

@app.websocket("/v1/realtime")
async def realtime(websocket: WebSocket) -> None:
    await websocket.accept()
    session = manager.open(websocket)
    try:
        await session.run()
    finally:
        await session.teardown()           # MUST be awaited
        manager.close(session.session_id)

[zhaochenyang20]

File: tests/test_v1_realtime_hardening.py:81, 124, 128, 131, 212，sglang_omni_v1/serve/realtime/vad.py:8, 50

All three monkeypatch.setattr(vad_module, "load_silero_model", ...) calls reference a non-existent attribute. vad.py:8 imports load_silero_vad (not load_silero_model); vad.py:50 calls load_silero_vad(onnx=True) directly — the module has no load_silero_model name. monkeypatch.setattr raises AttributeError on a missing attr, so all three tests fail at setup. Likewise a.model is not b.model (line 128) and constructed == [a.model, b.model] (line 131) reference model, but the implementation uses vad_model. test_session_update_propagates_vad_load_failure / test_streaming_vad_constructs_a_distinct_model_per_instance / test_manual_commit_resets_vad_and_origin either crash at import or silently skip in CI.

Fix: Expose the import at the module level with a unified alias in vad.py, and rename the attribute vad_model to model:

# vad.py
from silero_vad import load_silero_vad as load_silero_model

class StreamingVAD:
    def __init__(self, config: VADConfig | None = None) -> None:
        ...
        self.model = load_silero_model()

This allows direct patching in tests. This fix also naturally enables the "silero ORT singletonization" mentioned in [P1-PERF] below — the model parameter can be injected externally, and after singletonization, Test #2 should be revised to "validate the per-process model is shared".

[zhaochenyang20]

handle_item_create silently drops input_audio content parts

Lines 314-318 keep only input_text and text; the comment claims audio attachments belong on input_audio_buffer.* but the OpenAI spec also lets conversation.item.create carry input_audio (base64 PCM) inline. M0 may legitimately defer, but the current impl silently discards without error — clients never know. Per house style: emit a structured error event when input_audio content is present.

[zhaochenyang20]

Assumes the Usage type has a to_dict method. If CompletionStreamChunk.usage ever migrates to e.g. dataclasses.asdict, this raises AttributeError. Tighten to usage = dataclasses.asdict(chunk.usage) if chunk.usage else None, or formalize a Usage protocol.

[zhaochenyang20]

remove smoke test

[zhaochenyang20]

Both VAD tests skip when /tmp/realtime_fixture.wav is absent (lines 96-100 and 205-207). The PR doesn't document how to generate it, so in CI both tests skip silently — the only two real end-to-end VAD tests are de facto absent from the CI matrix.

[zhaochenyang20]

Counting source-code substring occurrences as "the helper is called from ≥ 3 sites" is fragile — any refactor breaks the test without a real correctness regression. Replace with a behavioral test: drive manual commit + auto commit + clear through a real session, assert all three paths leave the VAD in the post-reset state. This naturally merges with the regression test for [P0-BLOCKER] #1.