[Frontend][Benchmark] Add duplex performance metrics to refactored full-duplex framework (#7714)

Co-authored-by: amy-why-3459 <wuhaiyan17@huawei.com>

tests/benchmarks/metrics/test_metrics.py

@@ -17,10 +17,24 @@ from vllm_omni.benchmarks.metrics.metrics import (
     print_stage_durations_metrics,
 )
 from vllm_omni.benchmarks.patch.patch import MixRequestFuncOutput
+from vllm_omni.metrics.definitions import stage_modality_flags
 
 pytestmark = [pytest.mark.core_model, pytest.mark.benchmark, pytest.mark.cpu]
 
 
+def test_stage_modality_flags_match_print_stage():
+    text = stage_modality_flags("text", "")
+    audio = stage_modality_flags("", "audio")
+    stream = stage_modality_flags("latent", "stream")
+    video_frames = stage_modality_flags("video", "image")
+
+    assert text.is_text_stage and not text.is_audio_stage
+    assert audio.is_audio_stage and not audio.is_internal_stream_stage
+    assert stream.is_internal_stream_stage and not stream.is_text_stage
+    assert video_frames.is_video_stage and not video_frames.is_image_stage
+    assert not stage_modality_flags(None, None).is_text_stage
+
+
 def test_tpot_matches_mean_itl_per_request():
     """TPOT and pooled ITL should agree when output_len tracks ITL samples."""
     output = MixRequestFuncOutput()

tests/benchmarks/test_omniinteract.py

@@ -198,6 +198,12 @@ def test_response_metrics_include_engine_tpot_and_stream_window():
         "tpot_ms": {"count": 2, "mean": 15.0, "p50": 10.0, "p99": 20.0},
         "ttfp_ms": {"count": 2, "mean": 300.0, "p50": 200.0, "p99": 400.0},
         "rtf": {"count": 2, "mean": 2.0, "p50": 2.0, "p99": 2.0},
+        "stages": {
+            "0": {
+                "ttft_ms": {"count": 2, "mean": 0.0, "p50": 0.0, "p99": 0.0},
+                "tpot_ms": {"count": 2, "mean": 15.0, "p50": 10.0, "p99": 20.0},
+            }
+        },
         "stream_ttft_ms": 200.0,
         "stream_ttfp_ms": 300.0,
         "stream_rtf": 8.333333,

tests/clients/test_duplex_client.py

@@ -42,6 +42,7 @@ from vllm_omni.clients.duplex import (
     read_pcm16_wav,
     reference_audio_data_url,
     summarize_session_request_metrics,
+    summarize_stage_metrics,
     write_pcm16_wav,
 )
 from vllm_omni.clients.inline_duplex import InlineDuplexClient
@@ -1066,6 +1067,7 @@ def test_event_collector_reports_engine_token_and_audio_intervals():
             "max": 18.0,
         },
     }
+    assert timing["stages"] == {"0": timing["stage0_tokens"]}
     assert timing["audio_output"] == {
         "source": "client_monotonic_receive",
         "chunk_count": 3,
@@ -1321,6 +1323,152 @@ def test_summarize_session_request_metrics_averages_audio_turns():
     }
 
 
+def test_event_collector_reports_all_engine_stages():
+    collector = EventCollector()
+    collector.add({"type": "response.created", "response": {"id": "resp-a"}}, received_at_s=10.0)
+    collector.add(
+        {
+            "type": "response.output_audio.delta",
+            "response_id": "resp-a",
+            "delta": base64.b64encode(b"audio").decode("ascii"),
+            "sample_rate_hz": 16_000,
+            "metadata": {
+                "audio_duration_ms": 80,
+                "vllm_omni": {
+                    "stage_metrics": {
+                        "0": {
+                            "final_output_type": "text",
+                            "num_tokens_out": 4,
+                            "vllm_ttft_ms": 120.0,
+                            "vllm_tpot_ms": 15.0,
+                            "vllm_itls_ms": [10.0, 20.0],
+                        },
+                        "1": {
+                            "output_unit_type": "stream",
+                            "output_unit_count": 8,
+                            "serving_time_to_first_output_ms": 40.0,
+                            "time_per_output_unit_ms": 5.0,
+                            "inter_output_latencies_ms": [5.0],
+                        },
+                    }
+                },
+            },
+        },
+        received_at_s=10.2,
+    )
+    collector.add(
+        {
+            "type": "response.output_audio.delta",
+            "response_id": "resp-a",
+            "delta": base64.b64encode(b"audio").decode("ascii"),
+            "sample_rate_hz": 16_000,
+            "metadata": {
+                "audio_duration_ms": 160,
+                "vllm_omni": {
+                    "stage_metrics": {
+                        "2": {
+                            "final_output_type": "audio",
+                            "output_unit_type": "audio",
+                            "serving_time_to_first_output_ms": 80.0,
+                            "output_unit_count": 2,
+                            "audio_generated_frames": 2,
+                            "audio_duration_s": 0.08,
+                        }
+                    }
+                },
+            },
+        },
+        received_at_s=10.3,
+    )
+
+    timing = collector.timing_summary(after_s=10.0, response_id="resp-a")
+
+    assert list(timing["stages"]) == ["0", "1", "2"]
+    assert timing["stages"]["0"] is timing["stage0_tokens"]
+    assert timing["stages"]["0"]["ttft_ms"] == 120.0
+    assert "ttft_ms" not in timing["stages"]["1"]
+    assert timing["stages"]["1"]["ttfc_ms"] == 40.0
+    assert timing["stages"]["1"]["tpop_ms"] == 5.0
+    assert "ttft_ms" not in timing["stages"]["2"]
+    assert timing["stages"]["2"]["ttfp_ms"] == 80.0
+
+
+def test_audio_stage_omits_missing_or_zero_ttfp():
+    def _audio_stage_timing(serving_time: object) -> dict[str, object]:
+        collector = EventCollector()
+        collector.add({"type": "response.created", "response": {"id": "resp-a"}}, received_at_s=10.0)
+        stage_2: dict[str, object] = {
+            "final_output_type": "audio",
+            "output_unit_type": "audio",
+            "output_unit_count": 1,
+            "audio_generated_frames": 1,
+            "audio_duration_s": 0.08,
+        }
+        if serving_time is not None:
+            stage_2["serving_time_to_first_output_ms"] = serving_time
+        collector.add(
+            {
+                "type": "response.output_audio.delta",
+                "response_id": "resp-a",
+                "delta": base64.b64encode(b"audio").decode("ascii"),
+                "sample_rate_hz": 16_000,
+                "metadata": {
+                    "audio_duration_ms": 80,
+                    "vllm_omni": {"stage_metrics": {"2": stage_2}},
+                },
+            },
+            received_at_s=10.2,
+        )
+        return collector.timing_summary(after_s=10.0, response_id="resp-a")["stages"]["2"]
+
+    assert "ttfp_ms" not in _audio_stage_timing(None)
+    assert "ttfp_ms" not in _audio_stage_timing(0.0)
+    assert _audio_stage_timing(80.0)["ttfp_ms"] == 80.0
+
+
+def test_summarize_session_request_metrics_groups_stages():
+    summary = summarize_session_request_metrics(
+        [
+            {
+                "ttft_ms": 100.0,
+                "stages": {
+                    "0": {"ttft_ms": 10.0, "tpot_ms": 2.0},
+                    "1": {"ttfc_ms": 20.0, "tpop_ms": 0.0},
+                },
+            },
+            {
+                "ttft_ms": 300.0,
+                "stages": {
+                    "0": {"ttft_ms": 30.0, "tpot_ms": 6.0},
+                    "1": {"ttfc_ms": 40.0, "tpop_ms": 8.0},
+                    "2": {"ttfp_ms": 50.0},
+                },
+            },
+        ],
+        session_id="sess-1",
+    )
+
+    assert summary["stages"] == {
+        "0": {
+            "ttft_ms": {"count": 2, "mean": 20.0, "p50": 10.0, "p99": 30.0},
+            "tpot_ms": {"count": 2, "mean": 4.0, "p50": 2.0, "p99": 6.0},
+        },
+        "1": {
+            "ttfc_ms": {"count": 2, "mean": 30.0, "p50": 20.0, "p99": 40.0},
+            "tpop_ms": {"count": 1, "mean": 8.0, "p50": 8.0, "p99": 8.0},
+        },
+        "2": {
+            "ttfp_ms": {"count": 1, "mean": 50.0, "p50": 50.0, "p99": 50.0},
+        },
+    }
+    assert summarize_stage_metrics([{"stage0_tokens": {"ttft_ms": 12.0, "tpot_ms": 3.0}}]) == {
+        "0": {
+            "ttft_ms": {"count": 1, "mean": 12.0, "p50": 12.0, "p99": 12.0},
+            "tpot_ms": {"count": 1, "mean": 3.0, "p50": 3.0, "p99": 3.0},
+        }
+    }
+
+
 def test_pcm16_wav_round_trip(tmp_path):
     path = tmp_path / "audio.wav"
     pcm16 = b"\x01\x00\x02\x00"

tests/e2e/features/fullduplex/test_omni_duplex_eval_cli.py

@@ -417,6 +417,63 @@ def test_collect_duplex_session_metrics_matches_omniinteract_window():
     assert report["duplex_stream_ttfp_ms"]["mean"] == 300.0
 
 
+def test_collect_duplex_session_metrics_keeps_all_engine_stages():
+    collector = EventCollector()
+    audio = {
+        "type": "response.output_audio.delta",
+        "response_id": "r1",
+        "format": "pcm16",
+        "delta": base64.b64encode(bytes((1, 0)) * 2400).decode(),
+        "sample_rate_hz": 24_000,
+        "metadata": {
+            "audio_duration_ms": 100,
+            "vllm_omni": {
+                "stage_metrics": {
+                    "0": {
+                        "final_output_type": "text",
+                        "num_tokens_out": 4,
+                        "vllm_ttft_ms": 12.0,
+                        "vllm_tpot_ms": 3.0,
+                        "vllm_itls_ms": [3.0],
+                    },
+                    "1": {
+                        "output_unit_type": "stream",
+                        "output_unit_count": 8,
+                        "serving_time_to_first_output_ms": 40.0,
+                        "time_per_output_unit_ms": 5.0,
+                        "inter_output_latencies_ms": [5.0],
+                    },
+                    "2": {
+                        "final_output_type": "audio",
+                        "serving_time_to_first_output_ms": 80.0,
+                    },
+                }
+            },
+        },
+    }
+    collector.add({"type": "response.created", "response": {"id": "r1"}}, received_at_s=10.1)
+    collector.add(
+        {"type": "response.output_text.delta", "response_id": "r1", "delta": "Done."},
+        received_at_s=10.2,
+    )
+    collector.add(audio, received_at_s=10.3)
+    collector.add({"type": "response.done", "response": {"id": "r1"}}, received_at_s=10.4)
+    bundle = collect_duplex_session_metrics(collector, stream_start=10.0, session_id="sample")
+    stages = bundle.request_metrics[0]["stages"]
+    assert list(stages) == ["0", "1", "2"]
+    assert stages["0"] == bundle.request_metrics[0]["stage0_tokens"]
+    assert stages["1"]["ttfc_ms"] == 40.0
+    assert stages["1"]["tpop_ms"] == 5.0
+    assert stages["2"]["ttfp_ms"] == 80.0
+    assert bundle.session_metrics["stages"]["0"]["tpot_ms"]["mean"] == 3.0
+    assert bundle.session_metrics["stages"]["2"]["ttfp_ms"]["mean"] == 80.0
+    report = build_duplex_metrics_report(
+        request_metrics=bundle.request_metrics,
+        session_metrics=[bundle.session_metrics],
+    )
+    assert "duplex_stream_stages" not in report
+
+
 def test_judge_exercises_openai_http_schema():
     requests = []
 

tests/engine/duplex/test_duplex_events.py

@@ -33,7 +33,9 @@ from vllm_omni.engine.duplex.events import (
     SessionReplaced,
     SessionResumed,
     SessionResyncRequired,
+    Speak,
     TextDelta,
+    TranscriptDelta,
     TranscriptDone,
     TurnEvent,
     error_event,
@@ -119,6 +121,18 @@ def test_optional_wire_fields_are_omitted_when_none():
     assert delta["output_index"] == 0
     assert delta["content_index"] == 0
 
+    transcript = TranscriptDelta(response_id="resp_1", item_id="item_resp_1", delta="hi").to_realtime()
+    assert "metadata" not in transcript
+    assert (
+        TranscriptDelta(
+            response_id="resp_1",
+            item_id="item_resp_1",
+            delta="hi",
+            metadata={"vllm_omni": {"response_request_metrics": {"ttft_ms": 200.0}}},
+        ).to_realtime()["metadata"]["vllm_omni"]["response_request_metrics"]["ttft_ms"]
+        == 200.0
+    )
+
     anonymous_listen = Listen(session_id="sid", epoch=0, details={"reason": "silence"}).to_realtime()
     assert "response_id" not in anonymous_listen
     assert "id" not in anonymous_listen["response"]
@@ -478,3 +492,87 @@ def test_projection_leaves_error_to_typed_emit_sites():
     assert isinstance(projected[0], DuplexRawEvent)
     assert projected[0].type == "duplex.error"
     assert projected[0].to_realtime()["event"]["code"] == "bad_event"
+
+
+def test_response_speak_projection_keeps_vllm_omni_request_metrics():
+    state = RealtimeProjectionState(session_id="duplex-proj", model="test-model")
+    project_internal_event(state, {"type": "response.created", "response_id": "resp_1"})
+    metrics = {
+        "source": "server_monotonic_request_start",
+        "ttft_ms": 200.0,
+        "measurement_origin": {"ttft": "text", "ttfp": "audio"},
+    }
+
+    events = project_internal_event(
+        state,
+        {
+            "type": "response.speak",
+            "response_id": "resp_1",
+            "session_id": "duplex-proj",
+            "epoch": 0,
+            "model_speak": True,
+            "vllm_omni": {"response_request_metrics": metrics},
+        },
+    )
+
+    assert _types(events) == ["response.speak"]
+    assert isinstance(events[0], Speak)
+    assert events[0].to_realtime()["metadata"]["vllm_omni"]["response_request_metrics"]["ttft_ms"] == 200.0
+
+
+def test_text_then_audio_projects_ttft_on_transcript_and_ttfp_on_audio_delta():
+    state = RealtimeProjectionState(session_id="duplex-proj", model="test-model")
+    project_internal_event(
+        state, {"type": "response.created", "response_id": "resp_1", "modalities": ["audio", "text"]}
+    )
+    ttft = {
+        "source": "server_monotonic_request_start",
+        "ttft_ms": 200.0,
+        "measurement_origin": {"ttft": "text", "ttfp": "audio"},
+    }
+
+    text_only = project_internal_event(
+        state,
+        {
+            "type": "response.output_audio.delta",
+            "response_id": "resp_1",
+            "audio": "",
+            "text": "hi",
+            "session_id": "duplex-proj",
+            "epoch": 0,
+            "model_speak": True,
+            "vllm_omni": {"response_request_metrics": ttft},
+        },
+    )
+
+    assert _types(text_only) == ["response.output_audio_transcript.delta"]
+    assert isinstance(text_only[0], TranscriptDelta)
+    text_metrics = text_only[0].to_realtime()["metadata"]["vllm_omni"]["response_request_metrics"]
+    assert text_metrics["ttft_ms"] == 200.0
+    assert "ttfp_ms" not in text_metrics
+
+    pcm = base64.b64encode(b"\x00\x10" * 8).decode("ascii")
+    ttfp = {
+        "source": "server_monotonic_request_start",
+        "ttft_ms": 200.0,
+        "ttfp_ms": 400.0,
+        "measurement_origin": {"ttft": "text", "ttfp": "audio"},
+    }
+    audio = project_internal_event(
+        state,
+        {
+            "type": "response.output_audio.delta",
+            "response_id": "resp_1",
+            "audio": pcm,
+            "text": "",
+            "format": "pcm16",
+            "vllm_omni": {"response_request_metrics": ttfp},
+        },
+    )
+
+    assert "response.output_audio.delta" in _types(audio)
+    audio_delta = next(event for event in audio if event.type == "response.output_audio.delta")
+    assert isinstance(audio_delta, AudioDelta)
+    audio_metrics = audio_delta.to_realtime()["metadata"]["vllm_omni"]["response_request_metrics"]
+    assert audio_metrics["ttft_ms"] == 200.0
+    assert audio_metrics["ttfp_ms"] == 400.0

tests/engine/duplex/test_engine_session.py

@@ -5,6 +5,8 @@
 
 from __future__ import annotations
 
+import time
+from collections.abc import Callable
 from dataclasses import FrozenInstanceError
 
 import pytest
@@ -20,8 +22,13 @@ from vllm_omni.engine.duplex.config import (
 )
 from vllm_omni.engine.duplex.contracts import DuplexFence
 from vllm_omni.engine.duplex.events import TurnEvent
-from vllm_omni.engine.duplex.session.engine_session import DuplexEngineSession, DuplexFenceMismatchError
+from vllm_omni.engine.duplex.session.engine_session import (
+    RESPONSE_REQUEST_MEASUREMENT_ORIGIN,
+    DuplexEngineSession,
+    DuplexFenceMismatchError,
+)
 from vllm_omni.engine.duplex.session.playback_ledger import apply_playback_ack
+from vllm_omni.metrics.stats import DUPLEX_STAGE_TABLE_EXCLUDE, OrchestratorAggregator, StageRequestStats, StageStats
 from vllm_omni.model_executor.models.minicpmo_4_5.duplex.capabilities import (
     minicpmo45_native_capabilities,
 )
@@ -29,8 +36,51 @@ from vllm_omni.model_executor.models.minicpmo_4_5.duplex.capabilities import (
 pytestmark = [pytest.mark.core_model, pytest.mark.cpu]
 
 
-def _session(session_id: str = "duplex-test", config: DuplexSessionConfig | None = None) -> DuplexEngineSession:
-    return DuplexEngineSession(session_id=session_id, config=config or DuplexSessionConfig(model="test-model"))
+def _session(
+    session_id: str = "duplex-test",
+    config: DuplexSessionConfig | None = None,
+    *,
+    num_stages: int = 1,
+    log_stats: bool = False,
+    clock: Callable[[], float] | None = None,
+) -> DuplexEngineSession:
+    session = DuplexEngineSession(
+        session_id=session_id,
+        config=config or DuplexSessionConfig(model="test-model"),
+        num_stages=num_stages,
+        log_stats=log_stats,
+    )
+    if clock is not None:
+        session._clock = clock
+    return session
+
+
+def _stage_stats(
+    *,
+    stage_id: int,
+    request_id: str = "stage-req",
+    num_tokens_out: int = 3,
+    vllm_ttft_ms: float = 12.0,
+    vllm_tpot_ms: float = 0.0,
+    serving_time_to_first_output_ms: float = 0.0,
+) -> StageRequestStats:
+    return StageRequestStats(
+        batch_id=0,
+        batch_size=1,
+        num_tokens_in=7,
+        num_tokens_out=num_tokens_out,
+        stage_gen_time_ms=120.0,
+        rx_transfer_bytes=0,
+        rx_decode_time_ms=0.0,
+        rx_in_flight_time_ms=0.0,
+        stage_stats=StageStats(),
+        stage_id=stage_id,
+        request_id=request_id,
+        final_output_type="text",
+        vllm_ttft_ms=vllm_ttft_ms,
+        vllm_tpot_ms=vllm_tpot_ms,
+        serving_time_to_first_output_ms=serving_time_to_first_output_ms,
+    )
 
 
 def test_commit_audio_input_does_not_advance_model_turn_identity():
@@ -733,6 +783,222 @@ def test_minicpmo_native_capabilities_do_not_overclaim_single_session_deployment
     assert caps["supports_multi_session_same_replica"] is False
 
 
+def test_response_timing_binds_latest_request_start_for_model_turn():
+    session = _session()
+    session.mark_model_turn_request_started(0, 10.0)
+    session.mark_model_turn_request_started(0, 11.0)
+    session.begin_response(turn_id=0)
+
+    first = session.mark_response_first_outputs(
+        observed_at_s=11.2,
+        has_text=True,
+        has_audio=False,
+    )
+    assert first["ttft_ms"] == pytest.approx(200.0)
+    assert first["measurement_origin"] == RESPONSE_REQUEST_MEASUREMENT_ORIGIN
+    assert "ttfp_ms" not in first
+
+    # After begin_response the origin is frozen at the pending start (11.0).
+    # The 12.0 append must not rebind, so TTFP is 12.3-11.0 = 1300 ms, not 300.
+    session.mark_model_turn_request_started(0, 12.0)
+    audio = session.mark_response_first_outputs(
+        observed_at_s=12.3,
+        has_text=False,
+        has_audio=True,
+    )
+    assert audio["ttft_ms"] == pytest.approx(200.0)
+    assert audio["ttfp_ms"] == pytest.approx(1300.0)
+
+
+def test_response_timing_attaches_metrics_only_when_newly_observed():
+    session = _session()
+    session.mark_model_turn_request_started(0, 10.0)
+    session.begin_response(turn_id=0)
+
+    first = session.mark_response_first_outputs(observed_at_s=10.2, has_text=True, has_audio=False)
+    assert first["ttft_ms"] == pytest.approx(200.0)
+    assert session.mark_response_first_outputs(observed_at_s=10.3, has_text=True, has_audio=False) == {}
+    audio = session.mark_response_first_outputs(observed_at_s=10.4, has_text=False, has_audio=True)
+    assert audio["ttft_ms"] == pytest.approx(200.0)
+    assert audio["ttfp_ms"] == pytest.approx(400.0)
+    assert session.mark_response_first_outputs(observed_at_s=10.5, has_text=True, has_audio=True) == {}
+
+
+def test_response_timing_is_cleared_on_end_barge_in_and_close():
+    session = _session()
+    session.mark_model_turn_request_started(0, 10.0)
+    session.begin_response(turn_id=0)
+    assert session.mark_response_first_outputs(observed_at_s=10.2, has_text=True, has_audio=False)
+    session.end_response()
+    assert session.mark_response_first_outputs(observed_at_s=10.4, has_text=True, has_audio=False) == {}
+
+    session.mark_model_turn_request_started(0, 11.0)
+    session.barge_in()
+    session.begin_response(turn_id=0)
+    assert session.mark_response_first_outputs(observed_at_s=11.2, has_text=True, has_audio=False) == {}
+
+    session.mark_model_turn_request_started(0, 12.0)
+    session.close()
+    session.begin_response(turn_id=0)
+    assert session.mark_response_first_outputs(observed_at_s=12.2, has_text=True, has_audio=False) == {}
+
+
+def test_complete_model_turn_drops_request_starts_for_finished_turns():
+    session = _session()
+    session.mark_model_turn_request_started(0, 10.0)
+    session.mark_model_turn_request_started(1, 11.0)
+    session.complete_model_turn(0)
+    session.begin_response(turn_id=1)
+    assert session.mark_response_first_outputs(
+        observed_at_s=11.25,
+        has_text=True,
+        has_audio=False,
+    )["ttft_ms"] == pytest.approx(250.0)
+    session.end_response()
+    session.begin_response(turn_id=0)
+    assert session.mark_response_first_outputs(observed_at_s=11.5, has_text=True, has_audio=False) == {}
+
+
+def test_log_stats_off_does_not_open_a_response_aggregator():
+    session = _session()
+    session.begin_response()
+    session.observe_stage_request_stats(0, _stage_stats(stage_id=0))
+    assert session._response_aggregator is None
+    session.end_response()
+    assert session._response_aggregator is None
+
+
+def test_end_response_logs_pending_and_active_stage_request_stats(monkeypatch: pytest.MonkeyPatch):
+    monkeypatch.setattr(OrchestratorAggregator, "build_and_log_summary", lambda self: {})
+    session = _session(num_stages=2, log_stats=True)
+    wire_stats = _stage_stats(stage_id=0, request_id="stage0-req", num_tokens_out=3)
+    session.observe_stage_request_stats(0, wire_stats)
+    assert session._response_aggregator is None
+    assert wire_stats.request_id == "stage0-req"
+
+    response_id = session.begin_response()
+    aggregator = session._response_aggregator
+    assert aggregator is not None
+    assert aggregator.num_stages == 2
+    recorded = aggregator.stage_events[response_id]
+    assert recorded[0].num_tokens_out == 3
+    assert recorded[0].request_id == response_id
+    assert wire_stats.request_id == "stage0-req"
+
+    session.observe_stage_request_stats(1, _stage_stats(stage_id=1, num_tokens_out=5, vllm_ttft_ms=0.0))
+    assert [event.stage_id for event in aggregator.stage_events[response_id]] == [0, 1]
+
+    session.end_response()
+    assert session._response_aggregator is None
+    assert response_id in aggregator.e2e_done
+
+
+def test_end_response_prints_one_column_per_stage(monkeypatch: pytest.MonkeyPatch):
+    logged: list[OrchestratorAggregator] = []
+    monkeypatch.setattr(OrchestratorAggregator, "build_and_log_summary", lambda self: logged.append(self) or {})
+    session = _session(num_stages=3, log_stats=True)
+    session.observe_stage_request_stats(0, _stage_stats(stage_id=0, num_tokens_out=3, vllm_ttft_ms=40.0))
+    response_id = session.begin_response()
+    session.observe_stage_request_stats(0, _stage_stats(stage_id=0, num_tokens_out=4, vllm_ttft_ms=12.0))
+    session.observe_stage_request_stats(1, _stage_stats(stage_id=1, num_tokens_out=5, vllm_ttft_ms=20.0))
+    session.observe_stage_request_stats(1, _stage_stats(stage_id=1, num_tokens_out=6, vllm_ttft_ms=0.0))
+    session.observe_stage_request_stats(2, _stage_stats(stage_id=2, num_tokens_out=0, vllm_ttft_ms=0.0))
+    session.end_response()
+
+    assert len(logged) == 1
+    rows = logged[0].stage_events[response_id]
+    assert [event.stage_id for event in rows] == [0, 1, 2]
+    assert rows[0].num_tokens_out == 7
+    assert rows[0].vllm_ttft_ms == pytest.approx(40.0)
+    assert rows[1].num_tokens_out == 11
+    assert rows[1].vllm_ttft_ms == pytest.approx(20.0)
+    assert logged[0].stage_table_exclude == DUPLEX_STAGE_TABLE_EXCLUDE
+
+
+def test_end_response_omits_serving_time_to_first_output_from_stage_table(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    original = OrchestratorAggregator.build_and_log_summary
+    captured: list[tuple[OrchestratorAggregator, dict[str, object]]] = []
+
+    def _capture(self: OrchestratorAggregator) -> dict[str, object]:
+        summary = original(self)
+        captured.append((self, summary))
+        return summary
+
+    monkeypatch.setattr(OrchestratorAggregator, "build_and_log_summary", _capture)
+    session = _session(num_stages=3, log_stats=True)
+    response_id = session.begin_response()
+    session.observe_stage_request_stats(0, _stage_stats(stage_id=0, serving_time_to_first_output_ms=80.631))
+    session.observe_stage_request_stats(1, _stage_stats(stage_id=1, serving_time_to_first_output_ms=418.829))
+    session.observe_stage_request_stats(
+        2, _stage_stats(stage_id=2, vllm_ttft_ms=0.0, serving_time_to_first_output_ms=487.109)
+    )
+    session.end_response()
+
+    assert len(captured) == 1
+    aggregator, summary = captured[0]
+    stage_table = summary.get("stage_table")
+    assert isinstance(stage_table, list) and stage_table
+    rows = stage_table[0]["stages"]
+    assert isinstance(rows, list)
+    assert [row["stage_id"] for row in rows] == [0, 1, 2]
+    assert all("serving_time_to_first_output_ms" not in row for row in rows)
+    assert all("vllm_ttft_ms" in row for row in rows)
+    events = aggregator.stage_events[response_id]
+    assert [event.serving_time_to_first_output_ms for event in events] == pytest.approx([80.631, 418.829, 487.109])
+
+
+def test_http_stage_table_keeps_serving_time_to_first_output() -> None:
+    agg = OrchestratorAggregator(num_stages=2, log_stats=True, wall_start_ts=0.0, final_stage_id_for_e2e=1)
+    agg.on_stage_metrics(0, "r1", _stage_stats(stage_id=0, serving_time_to_first_output_ms=80.0))
+    agg.on_stage_metrics(1, "r1", _stage_stats(stage_id=1, serving_time_to_first_output_ms=418.0))
+    agg.on_finalize_request(1, "r1", req_start_ts=0.0)
+
+    summary = agg.build_and_log_summary()
+    rows = summary["stage_table"][0]["stages"]
+    assert [row["serving_time_to_first_output_ms"] for row in rows] == [80.0, 418.0]
+
+
+def test_logged_e2e_includes_wait_before_first_output(monkeypatch: pytest.MonkeyPatch) -> None:
+    mono = {"t": 100.0}
+    wall = {"t": 1_000.0}
+    monkeypatch.setattr(time, "time", lambda: wall["t"])
+    logged: list[OrchestratorAggregator] = []
+    monkeypatch.setattr(OrchestratorAggregator, "build_and_log_summary", lambda self: logged.append(self) or {})
+    session = _session(log_stats=True, clock=lambda: mono["t"])
+    session.mark_model_turn_request_started(0, 100.0)
+
+    mono["t"] = 102.0
+    wall["t"] = 1_002.0
+    response_id = session.begin_response(turn_id=0)
+    first = session.mark_response_first_outputs(observed_at_s=102.0, has_text=True, has_audio=True)
+    assert first["ttft_ms"] == pytest.approx(2000.0)
+
+    mono["t"] = 102.1
+    wall["t"] = 1_002.1
+    session.end_response()
+
+    assert logged[0].e2e_events[0].request_id == response_id
+    assert logged[0].e2e_events[0].e2e_total_ms == pytest.approx(2100.0)
+
+
+def test_logged_tpot_matches_client_weighted_aggregation(monkeypatch: pytest.MonkeyPatch) -> None:
+    logged: list[OrchestratorAggregator] = []
+    monkeypatch.setattr(OrchestratorAggregator, "build_and_log_summary", lambda self: logged.append(self) or {})
+    session = _session(log_stats=True)
+    response_id = session.begin_response()
+    session.observe_stage_request_stats(0, _stage_stats(stage_id=0, num_tokens_out=11, vllm_tpot_ms=10.0))
+    session.observe_stage_request_stats(0, _stage_stats(stage_id=0, num_tokens_out=3, vllm_tpot_ms=100.0))
+    session.accumulate_response_stage_metrics({"0": {"num_tokens_out": 11, "vllm_tpot_ms": 10.0}})
+    client = session.accumulate_response_stage_metrics({"0": {"num_tokens_out": 3, "vllm_tpot_ms": 100.0}})
+    expected = client["0"]["vllm_tpot_ms"]
+    session.end_response()
+
+    assert expected == pytest.approx(25.0)
+    assert logged[0].stage_events[response_id][0].vllm_tpot_ms == pytest.approx(expected)
+
+
 def test_draining_response_stays_ack_admissible_after_next_begin_response() -> None:
     from vllm_omni.engine.duplex.events import ErrorEvent
     from vllm_omni.engine.duplex.session.playback_ledger import apply_playback_ack

tests/engine/duplex/test_session_runner.py

@@ -46,9 +46,10 @@ from vllm_omni.engine.duplex.messages import (
     DuplexSessionEventMessage,
     OpenDuplexSessionMessage,
 )
+from vllm_omni.engine.duplex.session.engine_session import RESPONSE_REQUEST_MEASUREMENT_ORIGIN
 from vllm_omni.engine.duplex.session.manager import DuplexSessionManager
 from vllm_omni.engine.duplex.session.runner import DuplexSessionRunner
-from vllm_omni.metrics.stats import StageRequestStats, StageStats
+from vllm_omni.metrics.stats import OrchestratorAggregator, StageRequestStats, StageStats
 from vllm_omni.model_executor.models.minicpmo_4_5.duplex.plugin import MiniCPMO45DuplexPlugin
 
 pytestmark = [pytest.mark.core_model, pytest.mark.cpu]
@@ -202,6 +203,7 @@ async def open_harness(
     runtime_config: DuplexSessionRuntimeConfig | None = None,
     stage_count: int = 2,
     clock: Any = None,
+    log_stats: bool = False,
 ) -> Harness:
     plugin = MiniCPMO45DuplexPlugin(_fake_encode_audio)
     port = RecordingStagePort(stage_count=stage_count)
@@ -214,6 +216,7 @@ async def open_harness(
         result_sink=results,
         runtime_config=runtime_config or DuplexSessionRuntimeConfig(),
         model_config=None,
+        log_stats=log_stats,
         clock=clock,
     )
     body: dict[str, object] = {"auto_response": auto_response, **(extra_body or {})}
@@ -1398,6 +1401,144 @@ async def test_stage0_metrics_from_several_units_are_summed_into_one_response()
         await close_harness(h)
 
 
+@pytest.mark.asyncio
+async def test_response_done_logs_orchestrator_stage_table(monkeypatch: pytest.MonkeyPatch) -> None:
+    logged: list[OrchestratorAggregator] = []
+
+    def _capture(self: OrchestratorAggregator) -> dict[str, object]:
+        logged.append(self)
+        return {}
+
+    monkeypatch.setattr(OrchestratorAggregator, "build_and_log_summary", _capture)
+    h = await open_harness(log_stats=True)
+    try:
+        assert h.session.log_stats is True
+        assert h.session.num_stages == 2
+        await h.run(append_audio())
+        request_id = h.stage0_request_id()
+        h.deliver(
+            SimpleNamespace(
+                request_id=request_id,
+                finished=False,
+                outputs=[SimpleNamespace(text="hi", token_ids=[11], multimodal_output={})],
+                multimodal_output={},
+            ),
+            stage_id=0,
+            metrics=stage_stats(stage_id=0, request_id=request_id, num_tokens_out=3),
+        )
+        assert await h.settle() == []
+
+        events = await h.deliver_and_settle(
+            tts_output(request_id, samples=24000, text="hi", turn_end=True),
+            metrics=stage_stats(stage_id=1, request_id=request_id, num_tokens_out=4),
+        )
+        done = find(events, "response.done")
+        assert done.status == "completed"
+        assert len(logged) == 1
+        aggregator = logged[0]
+        assert aggregator.num_stages == 2
+        stage_ids = [event.stage_id for event in aggregator.stage_events[done.response_id]]
+        assert stage_ids == [0, 1]
+        assert done.response_id in aggregator.e2e_done
+    finally:
+        await close_harness(h)
+
+
+@pytest.mark.asyncio
+async def test_duplex_stage_request_stamps_wall_clock_request_timestamp() -> None:
+    """serving_time_to_first_output_ms is (first_output_ts - request_timestamp)*1000.
+
+    Duplex used to leave request_timestamp at 0, so the table printed unix_ts*1000.
+    """
+    from tests.engine.test_duplex_orchestrator import (
+        SESSION_ID as ORCH_SESSION_ID,
+    )
+    from tests.engine.test_duplex_orchestrator import (
+        _build,
+        _close,
+        _open,
+        _stage0_request_id,
+    )
+    from vllm_omni.engine.duplex_orchestrator import DuplexOrchestratorRequestState
+
+    orchestrator, _, rpc_q, _ = _build()
+    try:
+        result = await _open(orchestrator, rpc_q)
+        assert result.ok
+        state = orchestrator.request_states[_stage0_request_id()]
+        assert isinstance(state, DuplexOrchestratorRequestState)
+        assert state.request_timestamp > 1_000_000_000.0
+    finally:
+        if ORCH_SESSION_ID in orchestrator.session_manager.runners:
+            await _close(orchestrator, rpc_q)
+        await orchestrator.session_manager.shutdown()
+
+
+def _response_request_metrics_of(event: object) -> dict[str, object]:
+    """Server request-start clocks as the client reads them off one wire event."""
+    payload = event.to_realtime()
+    metadata = payload.get("metadata")
+    assert isinstance(metadata, dict), payload
+    vllm_omni = metadata.get("vllm_omni")
+    assert isinstance(vllm_omni, dict), metadata
+    metrics = vllm_omni.get("response_request_metrics")
+    assert isinstance(metrics, dict), vllm_omni
+    return metrics
+
+
+@pytest.mark.asyncio
+async def test_first_audio_delta_carries_server_request_start_metrics() -> None:
+    clock = {"now": 1000.0}
+    h = await open_harness(clock=lambda: clock["now"])
+    try:
+        await h.run(append_audio())
+        request_id = h.stage0_request_id()
+        clock["now"] = 1001.3
+        events = await h.deliver_and_settle(tts_output(request_id, samples=24000, text="hi"))
+        metrics = _response_request_metrics_of(find(events, "response.output_audio.delta"))
+        assert metrics["source"] == "server_monotonic_request_start"
+        assert metrics["measurement_origin"] == dict(RESPONSE_REQUEST_MEASUREMENT_ORIGIN)
+        assert metrics["ttft_ms"] == pytest.approx(1300.0)
+        assert metrics["ttfp_ms"] == pytest.approx(1300.0)
+        speak_metrics = _response_request_metrics_of(find(events, "response.speak"))
+        assert speak_metrics["ttft_ms"] == pytest.approx(1300.0)
+        assert speak_metrics["ttfp_ms"] == pytest.approx(1300.0)
+
+        clock["now"] = 1002.0
+        later = await h.deliver_and_settle(tts_output(request_id, samples=48000, text="hello"))
+        later_payload = find(later, "response.output_audio.delta").to_realtime()
+        later_metadata = later_payload.get("metadata")
+        later_vllm_omni = later_metadata.get("vllm_omni") if isinstance(later_metadata, dict) else None
+        later_metrics = later_vllm_omni.get("response_request_metrics") if isinstance(later_vllm_omni, dict) else None
+        assert later_metrics is None
+    finally:
+        await close_harness(h)
+
+
+@pytest.mark.asyncio
+async def test_stale_epoch_append_does_not_own_request_start() -> None:
+    clock = {"now": 1000.0}
+    h = await open_harness(clock=lambda: clock["now"])
+    try:
+        await h.run(append_audio())
+        request_id = h.stage0_request_id()
+        clock["now"] = 1000.5
+        append_ok, emitted = await h.runner.model.append_runtime_input(
+            {"duplex_turn_id": 0},
+            final=False,
+            expected_epoch=h.session.epoch + 1,
+        )
+        assert append_ok is True
+        assert emitted is False
+        clock["now"] = 1001.3
+        events = await h.deliver_and_settle(tts_output(request_id, samples=24000, text="hi"))
+        metrics = _response_request_metrics_of(find(events, "response.output_audio.delta"))
+        assert metrics["ttft_ms"] == pytest.approx(1300.0)
+        assert metrics["ttfp_ms"] == pytest.approx(1300.0)
+    finally:
+        await close_harness(h)
+
+
 # --------------------------------------------------------------------------- #
 # Server VAD                                                                  #
 # --------------------------------------------------------------------------- #

vllm_omni/benchmarks/duplex_session_metrics.py

@@ -81,6 +81,7 @@ def collect_duplex_session_metrics(
             measurement_origin=_REQUEST_MEASUREMENT_ORIGIN,
         )
         raw_metric = timing.get("request_metrics")
+        stages = timing.get("stages")
         stage0 = timing.get("stage0_tokens")
         metric: dict[str, object] = {
             "session_id": session_id,
@@ -90,10 +91,17 @@ def collect_duplex_session_metrics(
         if isinstance(raw_metric, dict):
             metric.update(raw_metric)
             metric["rtf"] = audio_rtf_from_raw_metric(raw_metric)
+        copied_stages: dict[str, dict[str, object]] = {}
+        if isinstance(stages, dict):
+            for stage_id, stage_snapshot in stages.items():
+                if isinstance(stage_snapshot, dict):
+                    copied_stages[str(stage_id)] = dict(stage_snapshot)
+        if copied_stages:
+            metric["stages"] = copied_stages
         if isinstance(stage0, dict):
             metric["stage0_tokens"] = dict(stage0)
             output_tokens += int(stage0.get("output_token_count") or 0)
-        if isinstance(raw_metric, dict) or isinstance(stage0, dict):
+        if isinstance(raw_metric, dict) or copied_stages or isinstance(stage0, dict):
             request_metrics.append(metric)
     session_metrics = summarize_session_request_metrics(
         request_metrics,

vllm_omni/benchmarks/metrics/metrics.py

@@ -131,14 +131,6 @@ def _p_label(p: float) -> str:
     return str(pf)
 
 
-_STREAMING_OUTPUT_UNIT_TYPES = frozenset(
-    {
-        "text",
-        "stream",
-        "audio",
-    }
-)
-
 _AGGREGATE_PERCENTILE_FIELD_NAMES = {
     defs.AUDIO_TTFP: (
         defs.MEAN_AUDIO_TTFP_MS,
@@ -187,28 +179,6 @@ def _wants_stream_icl(metrics: list[str]) -> bool:
     return "icl" in metrics
 
 
-def _stage_modality_flags(
-    final_output_type: str,
-    output_unit_type: str,
-) -> tuple[bool, bool, bool, bool, bool]:
-    is_text_stage = final_output_type == "text" or output_unit_type == "text"
-    is_audio_stage = final_output_type == "audio" or output_unit_type == "audio"
-    is_video_stage = final_output_type in {"video", "videos"} or output_unit_type == "video"
-    # Video diffusion may still report output_unit_type="image" when frames are
-    # stored in ``images``; prefer video when final_output_type says so.
-    is_image_stage = (not is_video_stage) and (final_output_type in {"image", "images"} or output_unit_type == "image")
-    is_internal_stream_stage = (
-        output_unit_type in _STREAMING_OUTPUT_UNIT_TYPES and not is_text_stage and not is_audio_stage
-    )
-    return (
-        is_text_stage,
-        is_audio_stage,
-        is_image_stage,
-        is_video_stage,
-        is_internal_stream_stage,
-    )
-
-
 def has_metric_samples(metrics: object, metric_name: str) -> bool:
     sample_count_attr = {
         "ttft": "num_ttft_samples",
@@ -736,7 +706,7 @@ def print_stage_metrics(
         is_image_stage,
         is_video_stage,
         is_internal_stream_stage,
-    ) = _stage_modality_flags(getattr(sm, "final_output_type"), getattr(sm, "output_unit_type"))
+    ) = defs.stage_modality_flags(getattr(sm, "final_output_type"), getattr(sm, "output_unit_type"))
 
     print("{s:{c}^{n}}".format(s=title, n=50, c="="))
     if is_video_stage:
@@ -786,7 +756,7 @@ def _build_stage_metrics_from_outputs(
         ]
         output_unit_type = output_unit_types[0] if output_unit_types else "other"
 
-        is_text_stage, is_audio_stage, _, _, is_internal_stream_stage = _stage_modality_flags(
+        is_text_stage, is_audio_stage, _, _, is_internal_stream_stage = defs.stage_modality_flags(
             final_output_type, output_unit_type
         )
         total_output = 0

vllm_omni/clients/duplex.py

@@ -40,7 +40,7 @@ import random
 import time
 import wave
 from abc import ABC, abstractmethod
-from collections.abc import AsyncIterator, Awaitable, Callable, Iterator, Mapping, Sequence
+from collections.abc import AsyncIterator, Awaitable, Callable, Iterator, Sequence
 from dataclasses import dataclass, field
 from pathlib import Path
 from urllib.parse import parse_qsl, urlencode, urlsplit, urlunsplit
@@ -48,6 +48,18 @@ from uuid import uuid4
 
 import pybase64 as base64
 
+from vllm_omni.clients.utils import (
+    _finite_metric_values,
+    _finite_number,
+    _interval_summary,
+    _rounded_ms,
+    _stage_engine_metrics_block,
+    _stage_id_sort_key,
+    distribution_summary,
+    metric_mean,
+    summarize_stage_metrics,
+)
+
 __all__ = [
     "DUPLEX_FIRST_UNIT_MS",
     "DUPLEX_UNIT_MS",
@@ -89,6 +101,7 @@ __all__ = [
     "distribution_summary",
     "metric_mean",
     "summarize_session_request_metrics",
+    "summarize_stage_metrics",
     "wait_for_condition",
     "write_pcm16_wav",
 ]
@@ -1339,67 +1352,6 @@ class DuplexClient(DuplexClientBase):
 # Test/benchmark collector
 
 
-def _rounded_ms(value: float) -> float:
-    return round(float(value), 3)
-
-
-def _finite_number(value: object, *, nonnegative: bool = False) -> float | None:
-    if not isinstance(value, int | float) or isinstance(value, bool) or not math.isfinite(float(value)):
-        return None
-    number = float(value)
-    if nonnegative and number < 0:
-        return None
-    return number
-
-
-def _interval_summary(values: list[float]) -> dict[str, float | int]:
-    clean = sorted(_rounded_ms(value) for value in values if math.isfinite(value) and value >= 0)
-    if not clean:
-        return {"count": 0, "mean": 0.0, "p50": 0.0, "p95": 0.0, "max": 0.0}
-
-    def nearest_rank(percentile: float) -> float:
-        index = max(0, math.ceil(percentile * len(clean)) - 1)
-        return clean[min(index, len(clean) - 1)]
-
-    return {
-        "count": len(clean),
-        "mean": _rounded_ms(sum(clean) / len(clean)),
-        "p50": nearest_rank(0.50),
-        "p95": nearest_rank(0.95),
-        "max": clean[-1],
-    }
-
-
-def distribution_summary(values: Sequence[float], *, digits: int = 3) -> dict[str, float | int] | None:
-    """Summarize values as ``{count, mean, p50, p99}`` for duplex report fields."""
-    clean = sorted(float(value) for value in values if math.isfinite(float(value)))
-    if not clean:
-        return None
-
-    def nearest_rank(percentile: float) -> float:
-        index = max(0, math.ceil(percentile * len(clean)) - 1)
-        return clean[min(index, len(clean) - 1)]
-
-    return {
-        "count": len(clean),
-        "mean": round(sum(clean) / len(clean), digits),
-        "p50": round(nearest_rank(0.50), digits),
-        "p99": round(nearest_rank(0.99), digits),
-    }
-
-
-def metric_mean(value: object) -> float | None:
-    """Read a scalar mean, or the ``mean`` field of a distribution summary."""
-    if isinstance(value, Mapping):
-        nested = value.get("mean")
-        if isinstance(nested, int | float) and not isinstance(nested, bool) and math.isfinite(float(nested)):
-            return float(nested)
-        return None
-    if isinstance(value, int | float) and not isinstance(value, bool) and math.isfinite(float(value)):
-        return float(value)
-    return None
-
-
 def _event_stage_metrics(event: dict[str, object]) -> dict[str, object] | None:
     candidates: list[object] = [event.get("vllm_omni")]
     metadata = event.get("metadata")
@@ -1629,7 +1581,7 @@ class EventCollector:
         measurement_origin: dict[str, str] | None = None,
     ) -> dict[str, object]:
         """Summarize engine token metrics and client-observed audio cadence."""
-        stage0_metrics: dict[str, object] | None = None
+        observed_stage_metrics: dict[str, dict[str, object]] = {}
         response_request_metrics: dict[str, object] = {}
         response_created_at_s: float | None = None
         first_text_received_at_s: float | None = None
@@ -1657,9 +1609,10 @@ class EventCollector:
                 first_text_received_at_s = received_at_s
 
             stage_metrics = _event_stage_metrics(event)
-            stage0 = stage_metrics.get("0") if isinstance(stage_metrics, dict) else None
-            if isinstance(stage0, dict):
-                stage0_metrics = stage0
+            if isinstance(stage_metrics, dict):
+                for stage_id, stage_snapshot in stage_metrics.items():
+                    if isinstance(stage_snapshot, dict):
+                        observed_stage_metrics[str(stage_id)] = stage_snapshot
 
             event_request_metrics = _event_response_request_metrics(event)
             if event_request_metrics is not None:
@@ -1677,21 +1630,18 @@ class EventCollector:
                 cumulative_audio_ms.append(max(0.0, float(duration_ms)))
 
         result: dict[str, object] = {}
-        if stage0_metrics is not None:
-            raw_itls = stage0_metrics.get("vllm_itls_ms")
-            itls = (
-                [float(value) for value in raw_itls if isinstance(value, int | float)]
-                if isinstance(raw_itls, list)
-                else []
-            )
-            result["stage0_tokens"] = {
-                "source": "engine_stage_metrics",
-                "output_token_count": int(stage0_metrics.get("num_tokens_out") or 0),
-                "ttft_ms": float(stage0_metrics.get("vllm_ttft_ms") or 0.0),
-                "tpot_ms": float(stage0_metrics.get("vllm_tpot_ms") or 0.0),
-                "itls_ms": itls,
-                "inter_token_interval_ms": _interval_summary(itls),
+        stage0_metrics = observed_stage_metrics.get("0")
+        if observed_stage_metrics:
+            stages = {
+                stage_id: _stage_engine_metrics_block(stage_snapshot)
+                for stage_id, stage_snapshot in sorted(
+                    observed_stage_metrics.items(),
+                    key=lambda item: _stage_id_sort_key(item[0]),
+                )
             }
+            result["stages"] = stages
+            if "0" in stages:
+                result["stage0_tokens"] = stages["0"]
 
         if audio_received_at_s:
             intervals_ms = [
@@ -1943,26 +1893,17 @@ def summarize_session_request_metrics(
 
     Aggregatable fields are nested as ``{count, mean, p50, p99}``.
     """
-
-    def values(metric: str, *, positive: bool = False) -> list[float]:
-        return [
-            float(request[metric])
-            for request in request_metrics
-            if isinstance(request.get(metric), int | float)
-            and not isinstance(request.get(metric), bool)
-            and math.isfinite(float(request[metric]))
-            and (not positive or float(request[metric]) > 0)
-        ]
-
     summary: dict[str, object] = {
         "session_id": session_id,
         "audio_turn_count": len(request_metrics),
-        "ttft_ms": distribution_summary(values("ttft_ms")),
-        "ttfp_ms": distribution_summary(values("ttfp_ms")),
-        "rtf": distribution_summary(values("rtf"), digits=6),
+        "ttft_ms": distribution_summary(_finite_metric_values(request_metrics, "ttft_ms")),
+        "ttfp_ms": distribution_summary(_finite_metric_values(request_metrics, "ttfp_ms")),
+        "rtf": distribution_summary(_finite_metric_values(request_metrics, "rtf"), digits=6),
     }
-    if (tpot := distribution_summary(values("tpot_ms", positive=True))) is not None:
+    if (tpot := distribution_summary(_finite_metric_values(request_metrics, "tpot_ms", positive=True))) is not None:
         summary["tpot_ms"] = tpot
+    if (stages := summarize_stage_metrics(request_metrics)) is not None:
+        summary["stages"] = stages
     return summary
 
 

vllm_omni/clients/utils.py

@@ -0,0 +1,264 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project
+"""Client-local helpers for duplex JSON metric blocks."""
+
+from __future__ import annotations
+
+import math
+from collections.abc import Mapping, Sequence
+from typing import NamedTuple
+
+__all__ = [
+    "STREAMING_OUTPUT_UNIT_TYPES",
+    "StageModalityFlags",
+    "distribution_summary",
+    "metric_mean",
+    "stage_modality_flags",
+    "summarize_stage_metrics",
+]
+
+# Intermediate streaming units that are not the pipeline's text or audio
+# final output (e.g. MiniCPM Talker codec tokens).
+STREAMING_OUTPUT_UNIT_TYPES = frozenset({"text", "stream", "audio"})
+
+
+class StageModalityFlags(NamedTuple):
+    """How duplex JSON classifies one engine stage."""
+
+    is_text_stage: bool
+    is_audio_stage: bool
+    is_image_stage: bool
+    is_video_stage: bool
+    is_internal_stream_stage: bool
+
+
+def stage_modality_flags(
+    final_output_type: object,
+    output_unit_type: object,
+) -> StageModalityFlags:
+    """Classify a stage from wire ``final_output_type`` / ``output_unit_type``."""
+    final_type = final_output_type if isinstance(final_output_type, str) else ""
+    unit_type = output_unit_type if isinstance(output_unit_type, str) else ""
+    is_text_stage = final_type == "text" or unit_type == "text"
+    is_audio_stage = final_type == "audio" or unit_type == "audio"
+    is_video_stage = final_type in {"video", "videos"} or unit_type == "video"
+    # Video diffusion may still report output_unit_type="image" when frames are
+    # stored in ``images``; prefer video when final_output_type says so.
+    is_image_stage = (not is_video_stage) and (final_type in {"image", "images"} or unit_type == "image")
+    is_internal_stream_stage = unit_type in STREAMING_OUTPUT_UNIT_TYPES and not is_text_stage and not is_audio_stage
+    return StageModalityFlags(
+        is_text_stage=is_text_stage,
+        is_audio_stage=is_audio_stage,
+        is_image_stage=is_image_stage,
+        is_video_stage=is_video_stage,
+        is_internal_stream_stage=is_internal_stream_stage,
+    )
+
+
+def _rounded_ms(value: float) -> float:
+    return round(float(value), 3)
+
+
+def _finite_number(value: object, *, nonnegative: bool = False) -> float | None:
+    if not isinstance(value, int | float) or isinstance(value, bool) or not math.isfinite(float(value)):
+        return None
+    number = float(value)
+    if nonnegative and number < 0:
+        return None
+    return number
+
+
+def _interval_summary(values: list[float]) -> dict[str, float | int]:
+    clean = sorted(_rounded_ms(value) for value in values if math.isfinite(value) and value >= 0)
+    if not clean:
+        return {"count": 0, "mean": 0.0, "p50": 0.0, "p95": 0.0, "max": 0.0}
+
+    def nearest_rank(percentile: float) -> float:
+        index = max(0, math.ceil(percentile * len(clean)) - 1)
+        return clean[min(index, len(clean) - 1)]
+
+    return {
+        "count": len(clean),
+        "mean": _rounded_ms(sum(clean) / len(clean)),
+        "p50": nearest_rank(0.50),
+        "p95": nearest_rank(0.95),
+        "max": clean[-1],
+    }
+
+
+def distribution_summary(values: Sequence[float], *, digits: int = 3) -> dict[str, float | int] | None:
+    """Summarize values as ``{count, mean, p50, p99}`` for duplex report fields."""
+    clean = sorted(float(value) for value in values if math.isfinite(float(value)))
+    if not clean:
+        return None
+
+    def nearest_rank(percentile: float) -> float:
+        index = max(0, math.ceil(percentile * len(clean)) - 1)
+        return clean[min(index, len(clean) - 1)]
+
+    return {
+        "count": len(clean),
+        "mean": round(sum(clean) / len(clean), digits),
+        "p50": round(nearest_rank(0.50), digits),
+        "p99": round(nearest_rank(0.99), digits),
+    }
+
+
+def metric_mean(value: object) -> float | None:
+    """Read a scalar mean, or the ``mean`` field of a distribution summary."""
+    if isinstance(value, Mapping):
+        nested = value.get("mean")
+        if isinstance(nested, int | float) and not isinstance(nested, bool) and math.isfinite(float(nested)):
+            return float(nested)
+        return None
+    if isinstance(value, int | float) and not isinstance(value, bool) and math.isfinite(float(value)):
+        return float(value)
+    return None
+
+
+def _stage_id_sort_key(stage_id: str) -> tuple[int, int | str]:
+    try:
+        return (0, int(stage_id))
+    except ValueError:
+        return (1, stage_id)
+
+
+def _stage_type_labels(stage_metrics: Mapping[str, object]) -> dict[str, object]:
+    labels: dict[str, object] = {}
+    for key in ("final_output_type", "output_unit_type"):
+        value = stage_metrics.get(key)
+        if isinstance(value, str) and value:
+            labels[key] = value
+    return labels
+
+
+def _stage_ms_list(stage_metrics: Mapping[str, object], field_name: str) -> list[float]:
+    raw = stage_metrics.get(field_name)
+    return [float(value) for value in raw if isinstance(value, int | float)] if isinstance(raw, list) else []
+
+
+def _has_legacy_vllm_token_metrics(stage_metrics: Mapping[str, object]) -> bool:
+    if stage_metrics.get("vllm_ttft_ms") or stage_metrics.get("vllm_tpot_ms"):
+        return True
+    raw_itls = stage_metrics.get("vllm_itls_ms")
+    return isinstance(raw_itls, list) and bool(raw_itls)
+
+
+def _text_stage_engine_metrics_block(stage_metrics: Mapping[str, object]) -> dict[str, object]:
+    itls = _stage_ms_list(stage_metrics, "vllm_itls_ms")
+    return {
+        "source": "engine_stage_metrics",
+        **_stage_type_labels(stage_metrics),
+        "output_token_count": int(stage_metrics.get("num_tokens_out") or 0),
+        "ttft_ms": float(stage_metrics.get("vllm_ttft_ms") or 0.0),
+        "tpot_ms": float(stage_metrics.get("vllm_tpot_ms") or 0.0),
+        "itls_ms": itls,
+        "inter_token_interval_ms": _interval_summary(itls),
+    }
+
+
+def _audio_stage_engine_metrics_block(stage_metrics: Mapping[str, object]) -> dict[str, object]:
+    block: dict[str, object] = {
+        "source": "engine_stage_metrics",
+        **_stage_type_labels(stage_metrics),
+        "output_unit_count": int(stage_metrics.get("output_unit_count") or 0),
+        "audio_generated_frames": int(stage_metrics.get("audio_generated_frames") or 0),
+        "audio_duration_s": float(stage_metrics.get("audio_duration_s") or 0.0),
+    }
+    serving_time = _finite_number(
+        stage_metrics.get("serving_time_to_first_output_ms"),
+        nonnegative=True,
+    )
+    if serving_time:
+        block["ttfp_ms"] = serving_time
+    return block
+
+
+def _stream_stage_engine_metrics_block(stage_metrics: Mapping[str, object]) -> dict[str, object]:
+    icls = _stage_ms_list(stage_metrics, "inter_output_latencies_ms")
+    return {
+        "source": "engine_stage_metrics",
+        **_stage_type_labels(stage_metrics),
+        "output_unit_count": int(stage_metrics.get("output_unit_count") or 0),
+        "ttfc_ms": float(stage_metrics.get("serving_time_to_first_output_ms") or 0.0),
+        "tpop_ms": float(stage_metrics.get("time_per_output_unit_ms") or 0.0),
+        "icls_ms": icls,
+        "inter_chunk_interval_ms": _interval_summary(icls),
+    }
+
+
+def _stage_engine_metrics_block(stage_metrics: Mapping[str, object]) -> dict[str, object]:
+    flags = stage_modality_flags(
+        stage_metrics.get("final_output_type"),
+        stage_metrics.get("output_unit_type"),
+    )
+    if flags.is_text_stage or (
+        not flags.is_audio_stage
+        and not flags.is_internal_stream_stage
+        and not flags.is_image_stage
+        and not flags.is_video_stage
+        and _has_legacy_vllm_token_metrics(stage_metrics)
+    ):
+        return _text_stage_engine_metrics_block(stage_metrics)
+    if flags.is_audio_stage:
+        return _audio_stage_engine_metrics_block(stage_metrics)
+    if flags.is_internal_stream_stage:
+        return _stream_stage_engine_metrics_block(stage_metrics)
+    block: dict[str, object] = {"source": "engine_stage_metrics", **_stage_type_labels(stage_metrics)}
+    gen_time_ms = stage_metrics.get("stage_gen_time_ms")
+    if isinstance(gen_time_ms, int | float) and not isinstance(gen_time_ms, bool):
+        block["stage_gen_time_ms"] = float(gen_time_ms)
+    return block
+
+
+def _finite_metric_values(
+    rows: Sequence[Mapping[str, object]],
+    metric: str,
+    *,
+    positive: bool = False,
+) -> list[float]:
+    return [
+        float(row[metric])
+        for row in rows
+        if isinstance(row.get(metric), int | float)
+        and not isinstance(row.get(metric), bool)
+        and math.isfinite(float(row[metric]))
+        and (not positive or float(row[metric]) > 0)
+    ]
+
+
+def summarize_stage_metrics(
+    request_metrics: Sequence[Mapping[str, object]],
+) -> dict[str, dict[str, object]] | None:
+    """Roll per-response engine stage blocks into ``{count, mean, p50, p99}``.
+
+    Reads ``stages`` when present, otherwise ``stage0_tokens`` as stage ``"0"``.
+    Zero or missing ``tpot_ms`` / ``tpop_ms`` values are omitted, matching
+    session ``tpot_ms``.
+    """
+    buckets: dict[str, list[Mapping[str, object]]] = {}
+    for request in request_metrics:
+        stages = request.get("stages")
+        if not isinstance(stages, dict):
+            stage0 = request.get("stage0_tokens")
+            stages = {"0": stage0} if isinstance(stage0, dict) else {}
+        for stage_id, stage_snapshot in stages.items():
+            if isinstance(stage_snapshot, dict):
+                buckets.setdefault(str(stage_id), []).append(stage_snapshot)
+    summary: dict[str, dict[str, object]] = {}
+    for stage_id in sorted(buckets, key=_stage_id_sort_key):
+        rows = buckets[stage_id]
+        stage_summary: dict[str, object] = {}
+        for metric, positive in (
+            ("ttft_ms", False),
+            ("tpot_ms", True),
+            ("ttfc_ms", False),
+            ("tpop_ms", True),
+            ("ttfp_ms", False),
+        ):
+            rolled = distribution_summary(_finite_metric_values(rows, metric, positive=positive))
+            if rolled is not None:
+                stage_summary[metric] = rolled
+        if stage_summary:
+            summary[stage_id] = stage_summary
+    return summary or None

vllm_omni/engine/duplex/events.py

@@ -4,8 +4,8 @@
 """The event vocabulary one duplex session speaks, assembled from two layers.
 
 The classes themselves no longer live here. ``vllm_omni.protocol.duplex.events``
-carries the whole vocabulary --- 22 Tier 1 classes re-exported from
-``vllm_omni.protocol.realtime.events``, 8 Tier 2 ones and 12 Tier 3 ones --- so
+carries the whole vocabulary --- 21 Tier 1 classes re-exported from
+``vllm_omni.protocol.realtime.events``, 9 Tier 2 ones and 12 Tier 3 ones --- so
 a non-duplex Realtime consumer can take the Tier 1 half and its wire rendering
 without the duplex session control plane (RFC #6592 P0a). The engine imports
 the duplex module and never the Tier 1 one directly.
@@ -74,6 +74,7 @@ from vllm_omni.protocol.duplex.events import (
     new_event_id,
 )
 
+#: Internal events that terminate a response/session and must never be dropped as stale.
 DOMAIN_TERMINAL_EVENTS = frozenset(
     {
         "response.done",

vllm_omni/engine/duplex/realtime_events.py

@@ -476,7 +476,7 @@ def _response_created_event(event: Mapping[str, object]) -> ResponseCreated:
 
 
 def _response_speak_metadata(event: Mapping[str, object]) -> dict[str, object]:
-    return {key: event[key] for key in ("session_id", "epoch", "model_speak") if key in event}
+    return {key: event[key] for key in ("session_id", "epoch", "model_speak", "vllm_omni") if key in event}
 
 
 def _realtime_audio_delta_events(
@@ -839,17 +839,19 @@ def _project(state: RealtimeProjectionState, event: dict[str, object]) -> list[D
             _refresh_in_progress_response_item(state, response_id)
         text = event.get("text")
         has_text = isinstance(text, str) and bool(text)
+        has_audio_delta = isinstance(audio, str) and bool(audio)
         if has_text:
             _append_response_transcript(state, response_id, cast("str", text))
             _refresh_in_progress_response_item(state, response_id)
         # Keep the audio.delta + transcript.delta pair invariant even for
         # text-less units so clients that treat the pair as unit-complete work.
-        if has_text or (isinstance(audio, str) and bool(audio)):
+        if has_text or has_audio_delta:
             events.append(
                 TranscriptDelta(
                     response_id=_str_or_none(response_id),
                     item_id=_response_item_id(state, response_id),
                     delta=cast("str", text) if has_text else "",
+                    metadata=_response_speak_metadata(event) if has_text and not has_audio_delta else None,
                 )
             )
         if event.get("end_of_turn") is True:

vllm_omni/engine/duplex/session/engine_session.py

@@ -43,6 +43,14 @@ from vllm_omni.engine.duplex.session.lease import (
     DuplexLeaseConfig,
     DuplexLeaseState,
 )
+from vllm_omni.metrics.stats import (
+    DUPLEX_STAGE_TABLE_EXCLUDE,
+    OrchestratorAggregator,
+    StageRequestStats,
+    _one_row_per_stage,
+    _tpot_interval_weight,
+)
+from vllm_omni.metrics.utils import _as_int, _as_optional_int
 
 if TYPE_CHECKING:
     from vllm_omni.engine.duplex.plugin import DuplexModelSessionState
@@ -57,26 +65,6 @@ def _object_dict(**fields: object) -> dict[str, object]:
     return dict(fields)
 
 
-def _as_int(value: object, default: int = 0) -> int:
-    if isinstance(value, bool):
-        return default
-    if isinstance(value, int):
-        return value
-    if isinstance(value, float):
-        return int(value)
-    return default
-
-
-def _as_optional_int(value: object) -> int | None:
-    if value is None or isinstance(value, bool):
-        return None
-    if isinstance(value, int):
-        return value
-    if isinstance(value, float):
-        return int(value)
-    return None
-
-
 def _copy_mapping(value: object) -> dict[str, object] | None:
     if not isinstance(value, dict):
         return None
@@ -116,6 +104,22 @@ class ResponseState:
     stage_metrics: dict[str, dict[str, object]] = field(default_factory=dict)
     stage_metric_tpot_weighted_ms: dict[str, float] = field(default_factory=dict)
     stage_metric_tpot_weight: dict[str, int] = field(default_factory=dict)
+    request_started_at_s_by_turn: dict[int, float] = field(default_factory=dict)
+    active_response_request_started_at_s: float | None = None
+    active_response_ttft_ms: float | None = None
+    active_response_ttfp_ms: float | None = None
+
+
+RESPONSE_REQUEST_MEASUREMENT_ORIGIN: dict[str, str] = {
+    "ttft": (
+        "accepted native-append start to first non-empty text output; "
+        "pending turns keep the latest append before first output, so overlapping user speech shortens TTFT"
+    ),
+    "ttfp": (
+        "accepted native-append start to first audio output; "
+        "pending turns keep the latest append before first output, so overlapping user speech shortens TTFP"
+    ),
+}
 
 
 @dataclass
@@ -217,6 +221,14 @@ class DuplexEngineSession:
     model_state: DuplexModelSessionState | None = field(default=None, repr=False)
     projector: RealtimeProjectionState | None = field(default=None, repr=False)
     created_monotonic: float = field(default_factory=time.monotonic)
+    #: Pipeline width for the per-response ``OrchestratorAggregator`` table.
+    num_stages: int = 1
+    #: Same serve ``log_stats`` gate as HTTP ``build_and_log_summary``.
+    log_stats: bool = False
+    _response_aggregator: OrchestratorAggregator | None = field(default=None, repr=False)
+    #: ``StageRequestStats`` that arrived before ``begin_response`` (stage 0
+    #: feeding TTS). Replayed onto the aggregator when the response opens.
+    _pending_stage_request_stats: list[tuple[int, StageRequestStats]] = field(default_factory=list, repr=False)
 
     def __post_init__(self) -> None:
         if self.accepted_fence is None:
@@ -743,6 +755,11 @@ class DuplexEngineSession:
     def complete_model_turn(self, turn_id: int) -> None:
         """Advance the model-owned output identity after its terminal signal."""
         completed_turn_id = int(turn_id)
+        self._response.request_started_at_s_by_turn = {
+            pending_turn_id: started_at_s
+            for pending_turn_id, started_at_s in self._response.request_started_at_s_by_turn.items()
+            if pending_turn_id > completed_turn_id
+        }
         if completed_turn_id >= self.turn_id:
             self.turn_id = completed_turn_id + 1
             self.sync_fence()
@@ -827,10 +844,17 @@ class DuplexEngineSession:
         return "".join(self._response.assistant_text_buffer)
 
     def begin_response(self, *, turn_id: int | None = None) -> str:
+        if self._response_aggregator is not None:
+            self._log_response_aggregator()
         self._activate_response_options()
         response_id = f"resp-{self.session_id}-{self.epoch}-{uuid4().hex[:8]}"
         self._response.active_response_id = response_id
         self._response.active_response_turn_id = self.turn_id if turn_id is None else int(turn_id)
+        self._clear_response_request_timing()
+        self._response.active_response_request_started_at_s = self._response.request_started_at_s_by_turn.pop(
+            self._response.active_response_turn_id,
+            None,
+        )
         self._response.active_response_input_commit_seq = self.input_commit_seq
         self._response.active_response_awaits_input_commit = self.turn_state == DuplexTurnState.USER_SPEAKING
         self._response.last_response_id = response_id
@@ -842,6 +866,7 @@ class DuplexEngineSession:
         self._playback.current = DuplexPlaybackCursor()
         self._playback.by_response[response_id] = self._playback.current
         self.turn_state = DuplexTurnState.ASSISTANT_GENERATING
+        self._start_response_aggregator()
         return response_id
 
     def _clear_response_metrics(self) -> None:
@@ -849,6 +874,120 @@ class DuplexEngineSession:
         self._response.stage_metric_tpot_weighted_ms.clear()
         self._response.stage_metric_tpot_weight.clear()
 
+    def _response_aggregator_wall_start_ts(self) -> float:
+        now_wall_s = time.time()
+        started_at_s = self._response.active_response_request_started_at_s
+        if started_at_s is None:
+            return now_wall_s
+        elapsed_s = max(0.0, self._clock() - started_at_s)
+        return now_wall_s - elapsed_s
+
+    def _new_response_aggregator(self) -> OrchestratorAggregator:
+        num_stages = max(int(self.num_stages), 1)
+        return OrchestratorAggregator(
+            num_stages=num_stages,
+            log_stats=True,
+            wall_start_ts=self._response_aggregator_wall_start_ts(),
+            final_stage_id_for_e2e=num_stages - 1,
+            stage_table_exclude=DUPLEX_STAGE_TABLE_EXCLUDE,
+        )
+
+    def _start_response_aggregator(self) -> None:
+        if not self.log_stats:
+            self._pending_stage_request_stats.clear()
+            return
+        self._response_aggregator = self._new_response_aggregator()
+        pending, self._pending_stage_request_stats = self._pending_stage_request_stats, []
+        for stage_id, stats in pending:
+            self._record_stage_request_stats(stage_id, stats)
+
+    def _record_stage_request_stats(self, stage_id: int, metrics: StageRequestStats) -> None:
+        aggregator = self._response_aggregator
+        request_id = self.active_response_id
+        if aggregator is None or request_id is None:
+            return
+        if stage_id < 0 or stage_id >= aggregator.num_stages:
+            return
+        aggregator.on_stage_metrics(stage_id, request_id, metrics, metrics.final_output_type)
+
+    def observe_stage_request_stats(self, stage_id: int, metrics: StageRequestStats) -> None:
+        """Feed one engine ``StageRequestStats`` into the response's logger table."""
+        if not self.log_stats:
+            return
+        event = copy.copy(metrics)
+        if event.stage_id is None:
+            event.stage_id = stage_id
+        if self.active_response_id is None:
+            self._pending_stage_request_stats.append((int(stage_id), event))
+            return
+        if self._response_aggregator is None:
+            self._start_response_aggregator()
+        self._record_stage_request_stats(int(stage_id), event)
+
+    def _log_response_aggregator(self) -> None:
+        aggregator = self._response_aggregator
+        self._response_aggregator = None
+        self._pending_stage_request_stats.clear()
+        if aggregator is None:
+            return
+        response_id = self._response.active_response_id
+        if response_id is not None and str(response_id) not in aggregator.e2e_done:
+            final_stage = aggregator.num_stages - 1 if aggregator.num_stages > 0 else 0
+            aggregator.on_finalize_request(final_stage, response_id, aggregator.wall_start_ts)
+        for rid, events in aggregator.stage_events.items():
+            aggregator.stage_events[rid] = _one_row_per_stage(events)
+        aggregator.build_and_log_summary()
+
+    def _clear_response_request_timing(self) -> None:
+        self._response.active_response_request_started_at_s = None
+        self._response.active_response_ttft_ms = None
+        self._response.active_response_ttfp_ms = None
+
+    def mark_model_turn_request_started(self, turn_id: int, started_at_s: float) -> None:
+        """Record the native request start that can own one model turn.
+
+        Pending turns keep the latest accepted append. After ``begin_response``
+        the first accepted append wins; later appends must not rebind.
+        """
+        turn_id = int(turn_id)
+        started_at_s = float(started_at_s)
+        if self.active_response_turn_id == turn_id:
+            if self._response.active_response_request_started_at_s is None:
+                self._response.active_response_request_started_at_s = started_at_s
+            return
+        self._response.request_started_at_s_by_turn[turn_id] = started_at_s
+
+    def mark_response_first_outputs(
+        self,
+        *,
+        observed_at_s: float,
+        has_text: bool,
+        has_audio: bool,
+    ) -> dict[str, object]:
+        """Return server-monotonic TTF metrics newly observed for the active response."""
+        started_at_s = self._response.active_response_request_started_at_s
+        if started_at_s is None:
+            return {}
+        elapsed_ms = max(0.0, (float(observed_at_s) - started_at_s) * 1000.0)
+        newly_observed = False
+        if has_text and self._response.active_response_ttft_ms is None:
+            self._response.active_response_ttft_ms = elapsed_ms
+            newly_observed = True
+        if has_audio and self._response.active_response_ttfp_ms is None:
+            self._response.active_response_ttfp_ms = elapsed_ms
+            newly_observed = True
+        if not newly_observed:
+            return {}
+        metrics: dict[str, object] = {
+            "source": "server_monotonic_request_start",
+            "measurement_origin": dict(RESPONSE_REQUEST_MEASUREMENT_ORIGIN),
+        }
+        if self._response.active_response_ttft_ms is not None:
+            metrics["ttft_ms"] = self._response.active_response_ttft_ms
+        if self._response.active_response_ttfp_ms is not None:
+            metrics["ttfp_ms"] = self._response.active_response_ttfp_ms
+        return metrics
+
     def stash_stage_metrics(self, stage_metrics: Mapping[Any, Any] | None) -> None:
         """Hold a stage snapshot until a response exists to attribute it to.
 
@@ -945,7 +1084,7 @@ class DuplexEngineSession:
             tpot_ms = raw_values.get("vllm_tpot_ms")
             token_count = raw_values.get("num_tokens_out")
             if isinstance(tpot_ms, int | float) and tpot_ms > 0:
-                weight = max(int(token_count) - 1, 1) if isinstance(token_count, int | float) else 1
+                weight = _tpot_interval_weight(token_count)
                 self._response.stage_metric_tpot_weighted_ms[stage_id] = (
                     self._response.stage_metric_tpot_weighted_ms.get(stage_id, 0.0) + float(tpot_ms) * weight
                 )
@@ -1160,6 +1299,7 @@ class DuplexEngineSession:
         playback_commit_policy: str | None = None,
         preserve_request: bool = False,
     ) -> dict[str, object] | None:
+        self._log_response_aggregator()
         response_id = self._response.active_response_id
         response_input_commit_seq = self._response.active_response_input_commit_seq
         if response_input_commit_seq is None:
@@ -1205,6 +1345,7 @@ class DuplexEngineSession:
             self._response.active_request_id = None
         self._response.active_response_id = None
         self._response.active_response_turn_id = None
+        self._clear_response_request_timing()
         self._response.active_response_input_commit_seq = None
         self._response.active_response_awaits_input_commit = False
         # Keep draining bindings for other responses (older TTS may still play).
@@ -1534,6 +1675,7 @@ class DuplexEngineSession:
         return self._playback.by_response.get(item_id.removeprefix("item_"))
 
     def barge_in(self) -> int:
+        self._log_response_aggregator()
         self.epoch += 1
         self.sync_fence()
         self._response.assistant_text_buffer.clear()
@@ -1541,6 +1683,8 @@ class DuplexEngineSession:
         self._response.active_request_id = None
         self._response.active_response_id = None
         self._response.active_response_turn_id = None
+        self._response.request_started_at_s_by_turn.clear()
+        self._clear_response_request_timing()
         self._response.active_response_input_commit_seq = None
         self._response.active_response_awaits_input_commit = False
         self._response.draining_response_by_request.clear()
@@ -1554,9 +1698,12 @@ class DuplexEngineSession:
             self.state = DuplexSessionState.CLOSING
 
     def close(self) -> None:
+        self._log_response_aggregator()
         self.state = DuplexSessionState.CLOSED
         self.turn_state = DuplexTurnState.IDLE
         self._response.active_response_turn_id = None
+        self._response.request_started_at_s_by_turn.clear()
+        self._clear_response_request_timing()
         self._response.active_response_input_commit_seq = None
         self._response.active_response_awaits_input_commit = False
         self._clear_response_metrics()

vllm_omni/engine/duplex/session/manager.py

@@ -99,6 +99,7 @@ class DuplexSessionManager:
         result_sink: janus.AsyncQueue[EngineQueueMessage],
         runtime_config: DuplexSessionRuntimeConfig,
         model_config: ModelConfig | None,
+        log_stats: bool = False,
         clock: Callable[[], float] | None = None,
         executor: concurrent.futures.ThreadPoolExecutor | None = None,
     ) -> None:
@@ -106,6 +107,7 @@ class DuplexSessionManager:
         self.plugin = plugin
         self.stage_port = stage_port
         self.model_config = model_config
+        self.log_stats = bool(log_stats)
         self.runtime_config = runtime_config
         self._output_sink = output_sink
         self._result_sink = result_sink
@@ -499,6 +501,8 @@ class DuplexSessionManager:
                 lease=DuplexLeaseState(config=self._lease_config, generation=0, last_activity=self._clock()),
                 _clock=self._clock,
                 _runtime_config=dict(runtime_config),
+                num_stages=self.stage_port.stage_count,
+                log_stats=self.log_stats,
             )
             # Validates the plugin's sampling policy for this runtime config before admission.
             self.sampling_params_for(session)

vllm_omni/engine/duplex/session/model_channel.py

@@ -46,6 +46,7 @@ from vllm_omni.engine.duplex.contracts import (
 from vllm_omni.engine.duplex.plugin import (
     DuplexRuntimeConfigError,
     coerce_int,
+    payload_turn_id,
 )
 from vllm_omni.engine.duplex.session import helpers
 from vllm_omni.engine.duplex.session.context import DuplexSessionContext, StageOutput
@@ -192,6 +193,7 @@ class ModelChannel:
             return True, False
         if expected_epoch is not None and session.epoch != expected_epoch:
             return True, False
+        self._mark_accepted_append_request_start(payload)
         # Commit timing state before returned output events can clear the
         # silence-continuation chain (e.g. a terminal turn-end).
         if on_append_accepted is not None:
@@ -208,6 +210,16 @@ class ModelChannel:
             return False, emitted_response
         return True, emitted_response
 
+    def _mark_accepted_append_request_start(self, payload: object) -> None:
+        """Stamp TTFT/TTFP origin only after the data plane accepted this append."""
+        session = self._ctx.session
+        turn_id = payload_turn_id(payload)
+        if turn_id is None:
+            turn_id = (
+                session.active_response_turn_id if session.active_response_turn_id is not None else session.turn_id
+            )
+        session.mark_model_turn_request_started(turn_id, session._clock())
+
     async def _append_via_data_plane(
         self,
         payload: object,
@@ -471,8 +483,9 @@ class ModelChannel:
             context=context,
         )
 
-    @staticmethod
-    def stage_metrics_snapshot(stage_id: int, metrics: object, output: object) -> dict[str, dict[str, object]] | None:
+    def stage_metrics_snapshot(
+        self, stage_id: int, metrics: object, output: object
+    ) -> dict[str, dict[str, object]] | None:
         if not isinstance(metrics, StageRequestStats):
             return None
         event = metrics
@@ -482,6 +495,7 @@ class ModelChannel:
             final_output_type = getattr(output, "final_output_type", None)
             if isinstance(final_output_type, str):
                 event = replace(event, final_output_type=final_output_type)
+        self._ctx.session.observe_stage_request_stats(stage_id, event)
         try:
             merged = OrchestratorAggregator._merge_stage_metric_event(None, event)
         except Exception:
@@ -927,7 +941,16 @@ class ModelChannel:
         if response_id is None:
             response_id = session.begin_response(turn_id=model_turn_id)
             response_created = True
-            self._out.emit(self.response_created_payload(response_id, epoch=session.epoch))
+        response_request_metrics = session.mark_response_first_outputs(
+            observed_at_s=session._clock(),
+            has_text=has_text,
+            has_audio=has_audio,
+        )
+        if response_created:
+            created_payload = self.response_created_payload(response_id, epoch=session.epoch)
+            if response_request_metrics:
+                created_payload["response_request_metrics"] = response_request_metrics
+            self._out.emit(created_payload)
         stage_metrics = model_result.get("stage_metrics")
         response_stage_metrics = session.accumulate_response_stage_metrics(
             stage_metrics if isinstance(stage_metrics, Mapping) else None
@@ -942,7 +965,12 @@ class ModelChannel:
                 "end_of_turn": end_of_turn,
                 "model_speak": True,
             }
-            self._attach_runtime_metadata(speak_payload, model_result, stage_metrics=response_stage_metrics)
+            self._attach_runtime_metadata(
+                speak_payload,
+                model_result,
+                stage_metrics=response_stage_metrics,
+                response_request_metrics=response_request_metrics,
+            )
             self._out.emit(speak_payload)
         target_id = draining_response_id if draining_response_id not in (None, session.active_response_id) else None
         previous_sent_ms = session.playback_for_response(target_id).sent_ms
@@ -1013,7 +1041,12 @@ class ModelChannel:
         sample_rate_hz = model_result.get("sample_rate_hz") or model_result.get("audio_sample_rate_hz")
         if isinstance(sample_rate_hz, int | float) and int(sample_rate_hz) > 0:
             payload["sample_rate_hz"] = int(sample_rate_hz)
-        self._attach_runtime_metadata(payload, model_result, stage_metrics=response_stage_metrics)
+        self._attach_runtime_metadata(
+            payload,
+            model_result,
+            stage_metrics=response_stage_metrics,
+            response_request_metrics=response_request_metrics,
+        )
         self._out.emit(payload)
         if (
             not end_of_turn
@@ -1138,6 +1171,7 @@ class ModelChannel:
         model_result: dict[str, object],
         *,
         stage_metrics: Mapping[str, object] | None = None,
+        response_request_metrics: Mapping[str, object] | None = None,
     ) -> None:
         metadata: dict[str, object] = {}
         runtime_impl = model_result.get("runtime_impl")
@@ -1160,6 +1194,8 @@ class ModelChannel:
                 for stage_id, values in effective_stage_metrics.items()
                 if isinstance(values, Mapping)
             }
+        if response_request_metrics:
+            metadata["response_request_metrics"] = dict(response_request_metrics)
         if metadata:
             payload["vllm_omni"] = metadata
 

vllm_omni/engine/duplex_orchestrator.py

@@ -90,6 +90,7 @@ class DuplexOrchestrator(Orchestrator, DuplexStagePort):
             result_sink=self.rpc_async_queue,
             runtime_config=self.duplex_session_config,
             model_config=model_config,
+            log_stats=self.log_stats,
         )
 
     # ------------------------------------------------------------------ #
@@ -133,6 +134,8 @@ class DuplexOrchestrator(Orchestrator, DuplexStagePort):
             logger.debug("[DuplexOrchestrator] stale stage binding ignored for %s stage-%s", request_id, stage_id)
             return
         req_state.stage_submit_ts[stage_id] = _time.time()
+        if req_state.request_timestamp <= 0.0:
+            req_state.request_timestamp = req_state.stage_submit_ts[stage_id]
         self._register_running_request(req_state)
         self.session_manager.register_request(request_id, req_state.session_id)
 
@@ -320,6 +323,7 @@ class DuplexOrchestrator(Orchestrator, DuplexStagePort):
                 prompt=None,
                 sampling_params_list=list(context.sampling_params),
                 final_stage_id=context.final_stage_id,
+                request_timestamp=_time.time(),
                 session_owned=True,
                 session_id=context.session_id,
                 fence=context.fence,
@@ -332,6 +336,8 @@ class DuplexOrchestrator(Orchestrator, DuplexStagePort):
                 request_state.config_generation = context.config_generation
             request_state.session_id = context.session_id
             request_state.fence = context.fence
+            if request_state.request_timestamp <= 0.0:
+                request_state.request_timestamp = _time.time()
         else:
             raise RuntimeError(f"request {context.request_id} is not session-owned")
         self._sync_bridge_state(request_state, context)
@@ -408,7 +414,9 @@ class DuplexOrchestrator(Orchestrator, DuplexStagePort):
                         f"async-chunk prewarm failed for duplex request {context.request_id}; the request was aborted"
                     )
         request_state.stage_fences[context.stage_id] = context.fence
-        request_state.stage_submit_ts[context.stage_id] = _time.time()
+        now = _time.time()
+        request_state.stage_submit_ts[context.stage_id] = now
+        request_state.request_timestamp = now
         self._register_running_request(request_state)
         return DuplexStageSubmissionResult(
             request_id=context.request_id,

vllm_omni/metrics/definitions.py

@@ -17,6 +17,7 @@ time-bearing metrics use the ``_s`` suffix (values in seconds), counters use
 import logging
 import os
 from collections.abc import Mapping, Sequence
+from typing import NamedTuple
 
 from vllm_omni.metrics.utils import resolve_int_by_sequential_keys
 
@@ -341,6 +342,43 @@ AUDIO_CHANNELS_ENV = "VLLM_OMNI_BENCH_AUDIO_CHANNELS"
 # the commonly-cited "audible gap" threshold for streaming TTS.
 AUDIO_CONTINUITY_DEFAULT_THRESHOLD_S = 0.1
 
+# Intermediate streaming units that are not the pipeline's text or audio
+# final output (e.g. MiniCPM Talker codec tokens).
+STREAMING_OUTPUT_UNIT_TYPES = frozenset({"text", "stream", "audio"})
+
+
+class StageModalityFlags(NamedTuple):
+    """How ``--print-stage`` classifies one engine stage."""
+
+    is_text_stage: bool
+    is_audio_stage: bool
+    is_image_stage: bool
+    is_video_stage: bool
+    is_internal_stream_stage: bool
+
+
+def stage_modality_flags(
+    final_output_type: object,
+    output_unit_type: object,
+) -> StageModalityFlags:
+    """Classify a stage from wire ``final_output_type`` / ``output_unit_type``."""
+    final_type = final_output_type if isinstance(final_output_type, str) else ""
+    unit_type = output_unit_type if isinstance(output_unit_type, str) else ""
+    is_text_stage = final_type == "text" or unit_type == "text"
+    is_audio_stage = final_type == "audio" or unit_type == "audio"
+    is_video_stage = final_type in {"video", "videos"} or unit_type == "video"
+    # Video diffusion may still report output_unit_type="image" when frames are
+    # stored in ``images``; prefer video when final_output_type says so.
+    is_image_stage = (not is_video_stage) and (final_type in {"image", "images"} or unit_type == "image")
+    is_internal_stream_stage = unit_type in STREAMING_OUTPUT_UNIT_TYPES and not is_text_stage and not is_audio_stage
+    return StageModalityFlags(
+        is_text_stage=is_text_stage,
+        is_audio_stage=is_audio_stage,
+        is_image_stage=is_image_stage,
+        is_video_stage=is_video_stage,
+        is_internal_stream_stage=is_internal_stream_stage,
+    )
+
 
 # ============================================================================
 # Formula helpers (shared by server-side observe and bench-side calculation)

vllm_omni/metrics/stats.py

@@ -3,9 +3,10 @@
 
 from __future__ import annotations
 
+import copy
 import time
 from collections import defaultdict
-from collections.abc import Callable
+from collections.abc import Callable, Sequence
 from contextlib import contextmanager
 from dataclasses import dataclass
 from typing import TYPE_CHECKING, Any
@@ -13,7 +14,7 @@ from typing import TYPE_CHECKING, Any
 from vllm.logger import init_logger
 
 from vllm_omni.metrics import definitions as defs
-from vllm_omni.metrics.utils import _build_field_defs, _build_row, _format_table
+from vllm_omni.metrics.utils import _as_float, _as_float_list, _as_int, _build_field_defs, _build_row, _format_table
 
 if TYPE_CHECKING:
     from vllm_omni.metrics.transfer import OmniTransferMetrics
@@ -130,6 +131,11 @@ STAGE_EXCLUDE = {
     "finish_reason",
     "pipeline_timings",
 }
+# Duplex logs one StageRequestStats table per response. Chunk submits refresh
+# ``request_timestamp``, so serving_time_to_first_output_ms is often a clamped
+# 0 on the audio column and is not a useful TTFP. HTTP ``--print-stage`` keeps
+# the row.
+DUPLEX_STAGE_TABLE_EXCLUDE = frozenset({defs.SERVING_TIME_TO_FIRST_OUTPUT_MS})
 TRANSFER_EXCLUDE = {"from_stage", "to_stage", "request_id", "used_shm"}
 E2E_EXCLUDE = {"request_id"}
 
@@ -140,6 +146,86 @@ TRANSFER_FIELDS = _build_field_defs(TransferEdgeStats, TRANSFER_EXCLUDE, FIELD_T
 E2E_FIELDS = _build_field_defs(RequestE2EStats, E2E_EXCLUDE, FIELD_TRANSFORMS)
 
 
+def _tpot_interval_weight(token_count: object) -> int:
+    if isinstance(token_count, int | float) and not isinstance(token_count, bool):
+        return max(int(token_count) - 1, 1)
+    return 1
+
+
+def _weighted_tpot_ms(events: Sequence[StageRequestStats]) -> float | None:
+    weighted_ms = 0.0
+    weight = 0
+    for event in events:
+        tpot_ms = float(event.vllm_tpot_ms)
+        if tpot_ms <= 0:
+            continue
+        chunk_weight = _tpot_interval_weight(event.num_tokens_out)
+        weighted_ms += tpot_ms * chunk_weight
+        weight += chunk_weight
+    if weight <= 0:
+        return None
+    return weighted_ms / float(weight)
+
+
+def _apply_merged_stage_stats(template: StageRequestStats, merged: dict[str, object]) -> StageRequestStats:
+    """Write one ``_merge_stage_metric_event`` snapshot back onto a stats row."""
+    stats = copy.copy(template)
+    stats.stage_id = _as_int(merged.get("stage_id"), default=template.stage_id or 0)
+    stats.final_output_type = (
+        str(merged["final_output_type"])
+        if isinstance(merged.get("final_output_type"), str)
+        else template.final_output_type
+    )
+    stats.finish_reason = str(merged["finish_reason"]) if isinstance(merged.get("finish_reason"), str) else None
+    stats.num_tokens_in = _as_int(merged.get(defs.NUM_TOKENS_IN))
+    stats.num_tokens_out = _as_int(merged.get(defs.NUM_TOKENS_OUT))
+    stats.stage_gen_time_ms = _as_float(merged.get(defs.STAGE_GEN_TIME_MS))
+    stats.postprocess_time_ms = _as_float(merged.get(defs.POSTPROCESS_TIME_MS))
+    stats.audio_generated_frames = _as_int(merged.get(defs.AUDIO_FRAMES))
+    stats.audio_sample_rate = _as_int(merged.get(defs.AUDIO_SAMPLE_RATE))
+    stats.audio_duration_s = _as_float(merged.get(f"{defs.AUDIO_DURATION}_s"))
+    stats.image_pixels = _as_int(merged.get(defs.IMAGE_PIXELS))
+    stats.denoise_step_latency_ms = _as_float(merged.get(defs.DENOISE_STEP_LATENCY_MS))
+    stats.output_unit_type = (
+        str(merged["output_unit_type"])
+        if isinstance(merged.get("output_unit_type"), str)
+        else template.output_unit_type
+    )
+    stats.output_unit_count = _as_int(merged.get(defs.OUTPUT_UNIT_COUNT))
+    stats.serving_time_to_first_output_ms = _as_float(merged.get(defs.SERVING_TIME_TO_FIRST_OUTPUT_MS))
+    stats.image_time_to_first_output_ms = _as_float(merged.get(defs.IMAGE_TIME_TO_FIRST_OUTPUT_MS))
+    stats.time_per_output_unit_ms = _as_float(merged.get(defs.TIME_PER_OUTPUT_UNIT_MS))
+    stats.inter_output_latencies_ms = _as_float_list(merged.get(defs.INTER_OUTPUT_LATENCIES_MS))
+    stats.inter_output_latency_ms = _as_float(merged.get(defs.INTER_OUTPUT_LATENCY_MS))
+    stats.vllm_ttft_ms = _as_float(merged.get(defs.VLLM_TTFT_MS))
+    stats.vllm_tpot_ms = _as_float(merged.get(defs.VLLM_TPOT_MS))
+    stats.vllm_itls_ms = _as_float_list(merged.get(defs.VLLM_ITLS_MS))
+    stats.vllm_itl_ms = _as_float(merged.get(defs.VLLM_ITL_MS))
+    return stats
+
+
+def _one_row_per_stage(events: list[StageRequestStats]) -> list[StageRequestStats]:
+    """Fold chunk snapshots so the logger table has one column per stage."""
+    merged_by_stage: dict[int, dict[str, object]] = {}
+    templates: dict[int, StageRequestStats] = {}
+    chunks_by_stage: dict[int, list[StageRequestStats]] = {}
+    for evt in events:
+        if evt.stage_id is None:
+            continue
+        sid = int(evt.stage_id)
+        templates.setdefault(sid, evt)
+        chunks_by_stage.setdefault(sid, []).append(evt)
+        merged_by_stage[sid] = OrchestratorAggregator._merge_stage_metric_event(merged_by_stage.get(sid), evt)
+    rows: list[StageRequestStats] = []
+    for sid in sorted(merged_by_stage):
+        merged = merged_by_stage[sid]
+        tpot_ms = _weighted_tpot_ms(chunks_by_stage[sid])
+        if tpot_ms is not None:
+            merged[defs.VLLM_TPOT_MS] = tpot_ms
+        rows.append(_apply_merged_stage_stats(templates[sid], merged))
+    return rows
+
+
 class OrchestratorAggregator:
     def __init__(
         self,
@@ -150,10 +236,12 @@ class OrchestratorAggregator:
         *,
         transfer_emitter: OmniTransferMetrics | None = None,
         replica_resolver: Callable[[int, str], int | None] | None = None,
+        stage_table_exclude: frozenset[str] = frozenset(),
     ) -> None:
         self.num_stages = int(num_stages)
         self.log_stats = bool(log_stats)
         self.final_stage_id_for_e2e = final_stage_id_for_e2e
+        self.stage_table_exclude = frozenset(stage_table_exclude)
         self.init_run_state(wall_start_ts)
         self.stage_events: dict[str, list[StageRequestStats]] = {}
         self.transfer_events: dict[
@@ -880,7 +968,8 @@ class OrchestratorAggregator:
             # === Stage table (columns = stage_id) ===
             # if any stage has diffusion_metrics, remove postprocess_time_ms field
             # because it is already included in diffusion_metrics
-            local_exclude = STAGE_EXCLUDE.copy()
+            local_exclude: set[str] = set(STAGE_EXCLUDE)
+            local_exclude.update(self.stage_table_exclude)
             has_diffusion_metrics = any(getattr(evt, "diffusion_metrics", None) for evt in stage_evts)
             if has_diffusion_metrics:
                 local_exclude.add("postprocess_time_ms")

vllm_omni/metrics/utils.py

@@ -20,6 +20,40 @@ FAILURE_REASONS = frozenset({"client_abort", "client_disconnect", "stage_error",
 DIFFUSION_METRICS_ONLY_REQUEST_ID = "__vllm_omni_diffusion_metrics__"
 
 
+def _as_int(value: object, default: int = 0) -> int:
+    if isinstance(value, bool):
+        return default
+    if isinstance(value, int):
+        return value
+    if isinstance(value, float):
+        return int(value)
+    return default
+
+
+def _as_optional_int(value: object) -> int | None:
+    if value is None or isinstance(value, bool):
+        return None
+    if isinstance(value, int):
+        return value
+    if isinstance(value, float):
+        return int(value)
+    return None
+
+
+def _as_float(value: object, default: float = 0.0) -> float:
+    if isinstance(value, bool):
+        return default
+    if isinstance(value, int | float):
+        return float(value)
+    return default
+
+
+def _as_float_list(value: object) -> list[float]:
+    if not isinstance(value, list):
+        return []
+    return [item for item in value if isinstance(item, int | float) and not isinstance(item, bool)]
+
+
 def extract_queue_wait_s(pipeline_timings: Mapping[str, float] | None) -> float | None:
     if pipeline_timings is None or "queue_wait_ms" not in pipeline_timings:
         return None

vllm_omni/protocol/duplex/events.py

@@ -55,7 +55,6 @@ from vllm_omni.protocol.realtime.events import (
     SpeechStopped,
     TextDelta,
     TextDone,
-    TranscriptDelta,
     TranscriptDone,
     new_event_id,
     wire_value,
@@ -116,6 +115,14 @@ class AudioDelta(realtime_events.AudioDelta):
     metadata: Mapping[str, object] | None = None
 
 
+@dataclass(frozen=True, slots=True, kw_only=True)
+class TranscriptDelta(realtime_events.TranscriptDelta):
+    """Tier 2: OpenAI's ``TranscriptDelta`` plus duplex request-start metrics on text-only units."""
+
+    optional_wire_fields = frozenset({"metadata"})
+    metadata: Mapping[str, object] | None = None
+
+
 @dataclass(frozen=True, slots=True, kw_only=True)
 class InputCommitted(realtime_events.InputCommitted):
     """Tier 2: OpenAI's ``InputCommitted`` plus the native ``input.committed`` event echoed under ``event``."""
@@ -358,8 +365,8 @@ def error_event(
     )
 
 
-#: The complete event vocabulary a duplex client may receive: 22 Tier 1
-#: classes re-exported unchanged, the 8 Tier 2 ones defined above, and 12
+#: The complete event vocabulary a duplex client may receive: 21 Tier 1
+#: classes re-exported unchanged, the 9 Tier 2 ones defined above, and 12
 #: Tier 3 ones. A duplex consumer imports from here and never reaches past
 #: this module into ``vllm_omni.protocol.realtime``.
 __all__ = [
@@ -384,7 +391,6 @@ __all__ = [
     "SpeechStopped",
     "TextDelta",
     "TextDone",
-    "TranscriptDelta",
     "TranscriptDone",
     # Tier 2 --- an OpenAI event plus duplex-only fields.
     "AudioDelta",
@@ -395,6 +401,7 @@ __all__ = [
     "ResponseCreated",
     "ResponseDone",
     "SessionCreated",
+    "TranscriptDelta",
     # Tier 3 --- vLLM-Omni only.
     "DuplexRawEvent",
     "Listen",