[Frontend][Voxtral] Add opt-in segment timestamps to /v1/realtime

The realtime WebSocket API streams transcription text with no timing
information, so clients doing diarization, subtitling or alignment have to
guess word timing from message arrival (#39735).

Voxtral realtime emits one token per 80 ms audio frame, so an output token's
index is an index into the audio, and `[STREAMING_WORD]` marks where an
emission group ends. Clients opt in per connection with
`timestamp_granularities: ["segment"]` on `session.update` and get a
`segments` array of `{text, end}` on `transcription.delta` and
`transcription.done`.

Alignment:

    end_s(k) = (len(prompt_tokens) - n_pad_frames - n_delay + k) * 0.08

The left pad is audio and the delay tokens are not, so both must come out of
the prompt length; the formula proposed on the issue subtracts only the delay
and lands every timestamp 2.56 s late. The pad is derived from the pad audio
rather than from `n_left_pad_tokens` - mistral-common builds one from the
other today, and a change that decoupled them would otherwise silently shift
every timestamp - and mismatches raise instead.

Segments are emission groups, not words: the model omits `[STREAMING_WORD]`
for words sharing an emission frame by design (arXiv 2602.11298 sections 3.1
and 6.2), and the marker precedes the subwords of its own group, so a
boundary closes the segment accumulated since the previous boundary.
Segmentation is on token ids, not delta text, because the boundary token
decodes to the empty string under `skip_special_tokens`.

Also adds the `session.updated` acknowledgement, which makes the opt-in
verifiable instead of silent, and replaces three tests' 5 s waits for an
event the server never sent.

Models opt in with `supports_realtime_segment_timestamps`; Qwen3-ASR
realtime inherits `False` since it re-prompts per segment and has no
token-per-frame lockstep. Clients that do not opt in see byte-identical
payloads: the `segments` key is omitted entirely.

Rejected at negotiation, each naming problem, cause and fix: a model that
cannot do it, a model advertising support it has not implemented, `word`
granularity, a tokenizer without the marker, and `--stream-interval > 1`.
That last one is not a timestamp limitation - batching withholds an output
until `stream-interval` tokens accumulate, but the lockstep cannot produce
the next token until the previous one is streamed back, so realtime stalls
whether or not timestamps were requested. The error says so rather than
suggesting the opt-in be dropped.

Signed-off-by: Andrii Pasternak <andriipasternak31@gmail.com>

Author: AndriiPasternak31

docs/serving/online_serving/speech_to_text.md

@@ -181,7 +181,8 @@ Audio must be sent as base64-encoded PCM16 audio at 16kHz sample rate, mono chan
 
 1. Client connects to `ws://host/v1/realtime`
 2. Server sends `session.created` event
-3. Client optionally sends `session.update` with model/params
+3. Client optionally sends `session.update` with model/params, and the server
+    acknowledges with `session.updated`
 4. Client sends `input_audio_buffer.commit` when ready
 5. Client sends `input_audio_buffer.append` events with base64 PCM16 chunks
 6. Server sends `transcription.delta` events with incremental text
@@ -196,17 +197,61 @@ Audio must be sent as base64-encoded PCM16 audio at 16kHz sample rate, mono chan
 | ----- | ----------- |
 | `input_audio_buffer.append` | Send base64-encoded audio chunk: `{"type": "input_audio_buffer.append", "audio": "<base64>"}` |
 | `input_audio_buffer.commit` | Trigger transcription processing or end: `{"type": "input_audio_buffer.commit", "final": bool}` |
-| `session.update` | Configure session: `{"type": "session.update", "model": "model-name"}` |
+| `session.update` | Configure session: `{"type": "session.update", "model": "model-name", "timestamp_granularities": ["segment"]}` |
 
 ### Server → Client Events
 
 | Event | Description |
 | ----- | ----------- |
 | `session.created` | Connection established with session ID and timestamp |
+| `session.updated` | Acknowledgement of `session.update`, echoing the configuration that took effect |
 | `transcription.delta` | Incremental transcription text: `{"type": "transcription.delta", "delta": "text"}` |
 | `transcription.done` | Final transcription with usage stats |
 | `error` | Error notification with message and optional code |
 
+#### Segment-level timestamps
+
+Models that emit one output token per audio frame can report when each piece
+of transcribed audio ended. Opt in per connection:
+
+```json
+{"type": "session.update", "model": "mistralai/Voxtral-Mini-4B-Realtime-2602", "timestamp_granularities": ["segment"]}
+```
+
+If the request was honoured the server replies with `session.updated`, echoing
+`"timestamp_granularities": ["segment"]`. If it cannot be, the server replies
+with `error` instead and the session stays unconfigured, so the opt-in is
+never silently dropped. Once enabled, `transcription.delta` and
+`transcription.done` carry a `segments` array:
+
+```json
+{"type": "transcription.done", "text": " Mary had a little lamb", "segments": [{"text": " Mary had", "end": 1.36}, {"text": " a little lamb", "end": 2.08}]}
+```
+
+Points worth knowing before building on this:
+
+- **End only.** The model marks where a group of audio ends, not where it
+  starts. Deriving a start from the previous `end` is wrong across pauses.
+- **Segments are emission groups, not words.** The model may commit several
+  words at once, and then they share one entry. Expect at most one entry per
+  spoken word.
+- **Granularity is one audio frame**, 80 ms for Voxtral realtime.
+- **The clock is relative to the utterance**, and restarts on every non-final
+  `input_audio_buffer.commit`.
+- **Timestamps trail their text by one event.** The boundary marker itself
+  decodes to no text, so a delta carrying `segments` has an empty `delta`. A
+  client that skips events with empty text will drop them.
+- `transcription.done` repeats every segment of the utterance, including the
+  trailing one, which no delta can carry because generation ends first.
+- Requires `--stream-interval 1` (the default), and the opt-in is rejected
+  otherwise. Batched streaming stalls realtime transcription entirely - the
+  server withholds an output until `stream-interval` tokens accumulate, but
+  the next token cannot be generated until the previous one is fed back - so
+  omitting `timestamp_granularities` is not a way to keep such a server
+  usable.
+- Clients that do not opt in see byte-identical payloads: the `segments` key
+  is omitted entirely.
+
 #### Example Clients
 
 - [openai_realtime_client.py](https://github.com/vllm-project/vllm/tree/main/examples/speech_to_text/realtime/openai_realtime_client.py) - Upload and transcribe an audio file

examples/speech_to_text/realtime/openai_realtime_client.py

@@ -45,7 +45,13 @@ def audio_to_pcm16_base64(audio_path: str) -> str:
     return base64.b64encode(pcm16.tobytes()).decode("utf-8")
 
 
-async def realtime_transcribe(audio_path: str, host: str, port: int, model: str):
+async def realtime_transcribe(
+    audio_path: str,
+    host: str,
+    port: int,
+    model: str,
+    segment_timestamps: bool = False,
+):
     """
     Connect to the Realtime API and transcribe an audio file.
     """
@@ -60,8 +66,21 @@ async def realtime_transcribe(audio_path: str, host: str, port: int, model: str)
             print(f"Unexpected response: {response}")
             return
 
-        # Validate model
-        await ws.send(json.dumps({"type": "session.update", "model": model}))
+        # Validate model, and opt in to segment timestamps if asked
+        session_update = {"type": "session.update", "model": model}
+        if segment_timestamps:
+            session_update["timestamp_granularities"] = ["segment"]
+        await ws.send(json.dumps(session_update))
+
+        # The server echoes the configuration that took effect, so a request
+        # the model cannot honour is never silently dropped.
+        response = json.loads(await ws.recv())
+        if response["type"] == "error":
+            print(f"Session update rejected: {response['error']}")
+            return
+        if segment_timestamps and not response.get("timestamp_granularities"):
+            print("Server did not enable segment timestamps.")
+            return
 
         # Signal ready to start
         await ws.send(json.dumps({"type": "input_audio_buffer.commit"}))
@@ -99,6 +118,10 @@ async def realtime_transcribe(audio_path: str, host: str, port: int, model: str)
                 print(response["delta"], end="", flush=True)
             elif response["type"] == "transcription.done":
                 print(f"\n\nFinal transcription: {response['text']}")
+                for segment in response.get("segments") or []:
+                    # End times only: the model marks where a group of audio
+                    # ends, not where it starts.
+                    print(f"  ends at {segment['end']:>7.2f}s  {segment['text']!r}")
                 if response.get("usage"):
                     print(f"Usage: {response['usage']}")
                 break
@@ -115,7 +138,11 @@ def main(args):
         audio_path = str(AudioAsset("mary_had_lamb").get_local_path())
         print(f"No audio path provided, using default: {audio_path}")
 
-    asyncio.run(realtime_transcribe(audio_path, args.host, args.port, args.model))
+    asyncio.run(
+        realtime_transcribe(
+            audio_path, args.host, args.port, args.model, args.segment_timestamps
+        )
+    )
 
 
 if __name__ == "__main__":
@@ -134,6 +161,11 @@ def main(args):
         default=None,
         help="Path to the audio file to transcribe.",
     )
+    parser.add_argument(
+        "--segment-timestamps",
+        action="store_true",
+        help="Ask the server for the end time of each transcribed segment.",
+    )
     parser.add_argument(
         "--host",
         type=str,

tests/entrypoints/speech_to_text/realtime/test_realtime_validation.py

@@ -3,7 +3,6 @@
 
 import asyncio
 import json
-import warnings
 
 import numpy as np
 import pybase64 as base64
@@ -70,6 +69,138 @@ def mary_had_lamb_audio_chunks() -> list[str]:
     return chunks
 
 
+async def _start_session(
+    ws, model_name: str, timestamp_granularities: list[str] | None = None
+) -> dict:
+    """Open a session and return the session.updated acknowledgement."""
+    event = await receive_event(ws, timeout=30.0)
+    assert event["type"] == "session.created"
+
+    update: dict = {"type": "session.update", "model": model_name}
+    if timestamp_granularities is not None:
+        update["timestamp_granularities"] = timestamp_granularities
+    await send_event(ws, update)
+
+    event = await receive_event(ws, timeout=10.0)
+    assert event["type"] == "session.updated"
+    return event
+
+
+async def _stream_utterance(ws, chunks: list[str], timeout: float = 60.0) -> list[dict]:
+    """Stream one utterance and return every event up to transcription.done."""
+    await send_event(ws, {"type": "input_audio_buffer.commit"})
+    for chunk in chunks:
+        await send_event(ws, {"type": "input_audio_buffer.append", "audio": chunk})
+    await send_event(ws, {"type": "input_audio_buffer.commit", "final": True})
+
+    events = []
+    while True:
+        event = await receive_event(ws, timeout=timeout)
+        if event["type"] == "error":
+            pytest.fail(f"Received error: {event}")
+        events.append(event)
+        if event["type"] == "transcription.done":
+            return events
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("model_name", [MODEL_NAME])
+async def test_segment_timestamps(
+    model_name, mary_had_lamb_audio_chunks, rocm_aiter_fa_attention
+):
+    """Segment timestamps are opt-in, aligned, and invisible when not asked for.
+
+    The alignment bound is the point of this test: the token index of a
+    generated token leads the audio it describes by the streaming prefix
+    minus the left pad minus the transcription delay. Subtracting only the
+    delay - as https://github.com/vllm-project/vllm/issues/39735 proposes -
+    leaves the 32-frame left pad in and puts every timestamp 2.56 s late,
+    which the upper bound catches.
+    """
+    server_args = ["--enforce-eager", "--max-model-len", "2048"]
+
+    if model_name.startswith("mistralai"):
+        server_args += MISTRAL_FORMAT_ARGS
+
+    add_attention_backend(server_args, rocm_aiter_fa_attention)
+
+    chunk_duration_s = 1600 / 16000
+    duration_s = len(mary_had_lamb_audio_chunks) * chunk_duration_s
+
+    with RemoteOpenAIServer(
+        model_name, server_args, env_dict=REALTIME_ENV_OVERRIDES
+    ) as remote_server:
+        ws_url = _get_websocket_url(remote_server)
+
+        # --- Not opted in: the wire must be byte-identical to before ------
+        async with websockets.connect(ws_url) as ws:
+            ack = await _start_session(ws, model_name)
+            assert ack["timestamp_granularities"] == []
+
+            # (ROCm) generous timeout: first use triggers aiter JIT.
+            events = await _stream_utterance(
+                ws, mary_had_lamb_audio_chunks, timeout=600.0
+            )
+            done = events[-1]
+            assert set(done) == {"type", "text", "usage"}
+            assert all("segments" not in event for event in events)
+            baseline_text = done["text"]
+
+        # --- Opted in: same text, plus timestamps -------------------------
+        async with websockets.connect(ws_url) as ws:
+            ack = await _start_session(ws, model_name, ["segment"])
+            assert ack["timestamp_granularities"] == ["segment"]
+
+            events = await _stream_utterance(ws, mary_had_lamb_audio_chunks)
+            done = events[-1]
+            segments = done["segments"]
+
+            # Opting in must not change what was transcribed.
+            assert done["text"] == baseline_text
+            assert segments
+
+            # done repeats the deltas' segments, plus the trailing segment
+            # that generation ended before any delta could carry.
+            streamed = [
+                segment
+                for event in events
+                if event["type"] == "transcription.delta"
+                for segment in event["segments"]
+            ]
+            assert segments[: len(streamed)] == streamed
+            assert 0 <= len(segments) - len(streamed) <= 1
+
+            ends = [segment["end"] for segment in segments]
+            assert ends == sorted(ends)
+            assert all(end >= 0.08 for end in ends)
+            assert all(abs(end / 0.08 - round(end / 0.08)) < 1e-6 for end in ends)
+
+            # Bounded on both sides: +32 frames of left pad would overshoot,
+            # a negative offset would undershoot.
+            assert 0.5 * duration_s <= ends[-1] <= duration_s + 0.5
+
+            # Entries are emission groups, so there are at most as many as
+            # there are words, and together they reconstruct the transcript.
+            assert len(segments) <= len(baseline_text.split())
+            reconstructed = "".join(segment["text"] for segment in segments)
+            assert baseline_text.endswith(reconstructed)
+            assert len(reconstructed) >= 0.9 * len(baseline_text)
+
+        # --- The clock restarts on every utterance ------------------------
+        async with websockets.connect(ws_url) as ws:
+            await _start_session(ws, model_name, ["segment"])
+            short_chunks = mary_had_lamb_audio_chunks[:40]
+
+            first = await _stream_utterance(ws, short_chunks)
+            second = await _stream_utterance(ws, short_chunks)
+
+            assert first[-1]["segments"]
+            assert second[-1]["segments"]
+            # Not "greater than the first utterance's last end": each commit
+            # is a new engine request with a fresh prompt and left pad.
+            assert second[-1]["segments"][0]["end"] < 1.0
+
+
 @pytest.mark.asyncio
 @pytest.mark.parametrize("model_name", [MODEL_NAME])
 async def test_multi_chunk_streaming(
@@ -95,17 +226,8 @@ async def test_multi_chunk_streaming(
             await send_event(ws, {"type": "session.update", "model": model_name})
 
             # Wait for the server to acknowledge the session update.
-            try:
-                while True:
-                    event = await receive_event(ws, timeout=5.0)
-                    if event["type"] == "session.updated":
-                        break
-            except TimeoutError:
-                warnings.warn(
-                    f"session.updated not received within {5.0}s after "
-                    "session.update. The server may not implement this event.",
-                    stacklevel=2,
-                )
+            event = await receive_event(ws, timeout=10.0)
+            assert event["type"] == "session.updated"
 
             # (ROCm) Warm-up: send a non-final commit (required to start
             # transcription) with a small audio chunk to trigger aiter
@@ -203,17 +325,8 @@ async def test_empty_commit_does_not_crash_engine(
 
             await send_event(ws, {"type": "session.update", "model": model_name})
 
-            try:
-                while True:
-                    event = await receive_event(ws, timeout=5.0)
-                    if event["type"] == "session.updated":
-                        break
-            except TimeoutError:
-                warnings.warn(
-                    f"session.updated not received within {5.0}s after "
-                    "session.update. The server may not implement this event.",
-                    stacklevel=2,
-                )
+            event = await receive_event(ws, timeout=10.0)
+            assert event["type"] == "session.updated"
 
             # Start generation without sending any audio
             await send_event(ws, {"type": "input_audio_buffer.commit"})
@@ -239,17 +352,8 @@ async def test_empty_commit_does_not_crash_engine(
 
             await send_event(ws, {"type": "session.update", "model": model_name})
 
-            try:
-                while True:
-                    event = await receive_event(ws, timeout=5.0)
-                    if event["type"] == "session.updated":
-                        break
-            except TimeoutError:
-                warnings.warn(
-                    f"session.updated not received within {5.0}s after "
-                    "session.update. The server may not implement this event.",
-                    stacklevel=2,
-                )
+            event = await receive_event(ws, timeout=10.0)
+            assert event["type"] == "session.updated"
 
             # Start transcription
             await send_event(ws, {"type": "input_audio_buffer.commit"})