[Frontend][Benchmark] Add duplex performance metrics for OmniInteract and Omni-DuplexEval benchmark (#7242)

docs/cli/bench/serve.md

@@ -316,8 +316,12 @@ arrives, checkpointing the response's history position so a later committed user
 outputs are ineligible and omitted from the official manifest; `audio_clipped_bytes` records output beyond the rounded video
 horizon.
 
-TTFT, TTFP, and RTF start at client receipt of `response.created`. TPOT/ITL use engine stage-0 timing; ITL is emitted only when
-every token interval is present.
+Per-response TTFT and TTFP start when the server begins executing the native model-turn request that owns the response. RTF
+continues to use client receipt of `response.created` through the last audio packet, divided by emitted audio duration. Global
+TTFT, TTFP, and RTF cover the complete input-stream window. TPOT/ITL use engine stage-0 timing; ITL is emitted only when every
+token interval is present within a continuous generation segment. Model-unit pacing and gaps between generation segments are
+excluded from TPOT/ITL. Raw request metrics retain `response_created_to_first_text_ms` and
+`response_created_to_first_audio_ms` as client-envelope diagnostics.
 
 The checked-in local performance configuration measures four deterministic cases from each OmniInteract subset (12 videos
 total), with no benchmark warmups and a maximum concurrency of two. Each subset also sends one readiness request before its

tests/benchmarks/metrics/test_metrics.py

@@ -150,16 +150,16 @@ def test_unmeasured_duplex_latency_does_not_add_zero_samples():
     measured.audio_ttfp = 0.2
     measured.audio_rtf = 0.5
     measured.duplex_session_metrics = {
-        "mean_ttft_ms": 100.0,
-        "mean_ttfp_ms": 200.0,
-        "mean_rtf": 0.5,
+        "ttft_ms": 100.0,
+        "ttfp_ms": 200.0,
+        "rtf": 0.5,
     }
     listen_only = _make_output(100, output_tokens=0)
     listen_only.ttft = listen_only.audio_ttfp = listen_only.audio_rtf = 0.0
     listen_only.duplex_session_metrics = {
-        "mean_ttft_ms": None,
-        "mean_ttfp_ms": None,
-        "mean_rtf": None,
+        "ttft_ms": None,
+        "ttfp_ms": None,
+        "rtf": None,
     }
 
     metrics = _calculate_test_metrics([measured, listen_only], {"ttft": 150.0})
@@ -174,9 +174,9 @@ def test_unmeasured_duplex_latency_does_not_add_zero_samples():
 def test_all_unmeasured_duplex_latency_is_not_reported_as_zero():
     listen_only = _make_output(100, output_tokens=0)
     listen_only.duplex_session_metrics = {
-        "mean_ttft_ms": None,
-        "mean_ttfp_ms": None,
-        "mean_rtf": None,
+        "ttft_ms": None,
+        "ttfp_ms": None,
+        "rtf": None,
     }
 
     metrics = _calculate_test_metrics(
@@ -196,13 +196,13 @@ def test_unmeasured_duplex_tpot_does_not_add_zero_or_misalign_goodput():
     missing_tpot.itl = []
     missing_tpot.text_latency = missing_tpot.ttft = 0.1
     missing_tpot.tpot_measured = False
-    missing_tpot.duplex_session_metrics = {"mean_ttft_ms": 100.0}
+    missing_tpot.duplex_session_metrics = {"ttft_ms": 100.0}
 
     slow_ttft = _make_output(100, output_tokens=5)
     slow_ttft.ttft = 1.0
     slow_ttft.text_latency = 1.4
     slow_ttft.itl = [0.1] * 4
-    slow_ttft.duplex_session_metrics = {"mean_ttft_ms": 1000.0}
+    slow_ttft.duplex_session_metrics = {"ttft_ms": 1000.0}
 
     metrics = _calculate_test_metrics(
         [missing_tpot, slow_ttft],
@@ -219,7 +219,7 @@ def test_all_unmeasured_duplex_token_timing_is_not_reported_as_zero():
     output.itl = []
     output.text_latency = output.ttft
     output.tpot_measured = False
-    output.duplex_session_metrics = {"mean_ttft_ms": 100.0}
+    output.duplex_session_metrics = {"ttft_ms": 100.0}
 
     metrics = _calculate_test_metrics([output], {"tpot": float("inf")})
 
@@ -236,7 +236,7 @@ def test_duplex_response_timings_do_not_build_a_session_token_timeline():
         {"response_id": "r1", "stage0_tokens": {"itls_ms": [100.0, 100.0]}},
         {"response_id": "r2", "stage0_tokens": {"itls_ms": [100.0, 100.0]}},
     ]
-    output.duplex_session_metrics = {"mean_ttft_ms": 100.0}
+    output.duplex_session_metrics = {"ttft_ms": 100.0}
 
     metrics = _calculate_test_metrics([output])
 
@@ -252,7 +252,7 @@ def test_unmeasured_tpot_stays_missing_after_tokenizer_fallback():
     output.itl = []
     output.text_latency = output.ttft
     output.tpot_measured = False
-    output.duplex_session_metrics = {"mean_ttft_ms": 100.0}
+    output.duplex_session_metrics = {"ttft_ms": 100.0}
 
     def tokenizer(text, *, add_special_tokens):
         assert text == output.generated_text
@@ -282,7 +282,7 @@ def test_unmeasured_tpot_stays_missing_after_tokenizer_fallback():
 def test_zero_itl_does_not_create_zero_tpot():
     output = _make_output(100, output_tokens=3)
     output.itl = [0.0, 0.0]
-    output.duplex_session_metrics = {"mean_ttft_ms": 100.0}
+    output.duplex_session_metrics = {"ttft_ms": 100.0}
 
     metrics = _calculate_test_metrics([output], {"tpot": 1.0})
 
@@ -357,9 +357,9 @@ def test_single_token_responses_do_not_report_zero_tpot(capsys):
 def test_duplex_goodput_does_not_pair_measurements_from_different_requests():
     text_only, audio_only = _make_output(100), _make_output(100)
     text_only.ttft, text_only.audio_ttfp = 0.1, 0.0
-    text_only.duplex_session_metrics = {"mean_ttft_ms": 100.0, "mean_ttfp_ms": None, "mean_rtf": None}
+    text_only.duplex_session_metrics = {"ttft_ms": 100.0, "ttfp_ms": None, "rtf": None}
     audio_only.ttft, audio_only.audio_ttfp = 0.0, 0.2
-    audio_only.duplex_session_metrics = {"mean_ttft_ms": None, "mean_ttfp_ms": 200.0, "mean_rtf": None}
+    audio_only.duplex_session_metrics = {"ttft_ms": None, "ttfp_ms": 200.0, "rtf": None}
 
     metrics = _calculate_test_metrics([text_only, audio_only], {"ttft": 500.0, "audio_ttft": 500.0})
 

tests/benchmarks/patch/test_patch.py

@@ -287,20 +287,25 @@ async def test_seed_tts_realtime_duplex_exports_per_request_metrics(monkeypatch)
                 "ttft": "conversation.item.create client send to first non-empty text delta",
                 "ttfp": "conversation.item.create client send to first audio packet",
                 "rtf": "request-start-to-last-audio receive time divided by emitted audio duration",
+                "tpot": "Stage-0 engine mean time per output token",
             },
             "ttft_ms": pytest.approx(20.0, abs=2.0),
+            "tpot_ms": 10.0,
             "ttfp_ms": pytest.approx(30.0, abs=2.0),
             "rtf": pytest.approx(0.3, abs=0.03),
             "audio_generation_ms": pytest.approx(30.0, abs=2.0),
             "audio_duration_ms": 100.0,
         }
-    assert output.duplex_session_metrics == {
-        "session_id": session_id,
-        "audio_turn_count": 4,
-        "mean_ttft_ms": pytest.approx(20.0, abs=2.0),
-        "mean_ttfp_ms": pytest.approx(30.0, abs=2.0),
-        "mean_rtf": pytest.approx(0.3, abs=0.03),
-    }
+    session = output.duplex_session_metrics
+    assert session["session_id"] == session_id
+    assert session["audio_turn_count"] == 4
+    assert session["ttft_ms"]["count"] == 4
+    assert session["ttft_ms"]["mean"] == pytest.approx(20.0, abs=2.0)
+    assert session["tpot_ms"] == {"count": 4, "mean": 10.0, "p50": 10.0, "p99": 10.0}
+    assert session["ttfp_ms"]["count"] == 4
+    assert session["ttfp_ms"]["mean"] == pytest.approx(30.0, abs=2.0)
+    assert session["rtf"]["count"] == 4
+    assert session["rtf"]["mean"] == pytest.approx(0.3, abs=0.03)
 
 
 def test_stage0_token_timings_use_weighted_tpot_when_itls_are_incomplete():

tests/benchmarks/test_omniinteract.py

@@ -148,6 +148,72 @@ def _successful_output(
     return case, collector, result
 
 
+def test_response_metrics_include_engine_tpot_and_stream_window():
+    first_audio = _audio("r1")
+    first_audio["metadata"] = {
+        "audio_duration_ms": 100,
+        "vllm_omni": {
+            "stage_metrics": {
+                "0": {
+                    "num_tokens_out": 3,
+                    "vllm_tpot_ms": 10.0,
+                    "vllm_itls_ms": [10.0, 10.0],
+                }
+            }
+        },
+    }
+    second_audio = _audio("r2")
+    second_audio["metadata"] = {
+        "audio_duration_ms": 200,
+        "vllm_omni": {
+            "stage_metrics": {
+                "0": {
+                    "num_tokens_out": 2,
+                    "vllm_tpot_ms": 20.0,
+                    "vllm_itls_ms": [20.0],
+                }
+            }
+        },
+    }
+    collector = _collector(
+        (_created("r1"), 10.1),
+        (_text("r1", "first"), 10.2),
+        (first_audio, 10.3),
+        (_done("r1"), 10.4),
+        (_created("r2"), 12.1),
+        (_text("r2", "second"), 12.2),
+        (second_audio, 12.5),
+        (_done("r2"), 12.6),
+    )
+    result = oi.OmniInteractCaseResult("1q1a", "video.mp4", "", session_id="session")
+
+    oi._populate_response_metrics(result, collector, stream_start=10.0)
+
+    assert result.output_tokens == 5
+    assert [metric["tpot_ms"] for metric in result.duplex_request_metrics] == [10.0, 20.0]
+    assert result.duplex_session_metrics == {
+        "session_id": "session",
+        "audio_turn_count": 2,
+        "ttft_ms": {"count": 2, "mean": 100.0, "p50": 100.0, "p99": 100.0},
+        "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},
+        "stream_ttft_ms": 200.0,
+        "stream_ttfp_ms": 300.0,
+        "stream_rtf": 8.333333,
+        "stream_audio_generation_ms": 2500.0,
+        "stream_audio_duration_ms": 300.0,
+        "stream_measurement_origin": {
+            "ttft": "input stream start to first non-empty text delta",
+            "ttfp": "input stream start to first audio packet",
+            "rtf": (
+                "input stream start-to-last-audio receive time divided by total emitted audio duration; "
+                "includes concurrent realtime input"
+            ),
+        },
+    }
+
+
 def _write_success(
     root: Path,
     case: data.OmniInteractCase,
@@ -645,7 +711,7 @@ async def test_adapter_requires_and_forwards_exact_prepared_payload(tmp_path: Pa
             transcript="timing metadata missing",
             output_tokens=0,
             duplex_request_metrics=[{"request_metrics": {"ttft_ms": 1.0}}],
-            duplex_session_metrics={"mean_ttft_ms": 10.0},
+            duplex_session_metrics={"ttft_ms": 10.0},
         )
 
     monkeypatch.setattr(benchmark_patch, "run_omniinteract_case", run)
@@ -713,7 +779,7 @@ async def test_adapter_reports_exact_or_weighted_token_timing(
             success=True,
             output_tokens=5,
             duplex_request_metrics=request_metrics,
-            duplex_session_metrics={"mean_ttft_ms": 100.0},
+            duplex_session_metrics={"ttft_ms": 100.0},
         )
 
     monkeypatch.setattr(benchmark_patch, "run_omniinteract_case", run)

tests/clients/test_duplex_client.py

@@ -736,11 +736,162 @@ def test_event_collector_reports_engine_token_and_audio_intervals():
         "measurement_origin": {
             "ttft": "input_audio_buffer.commit client send to first non-empty text delta",
             "ttfp": "input_audio_buffer.commit client send to first audio packet",
+            "tpot": "Stage-0 engine mean time per output token",
         },
         "ttft_ms": 250.0,
         "ttfp_ms": 300.0,
         "audio_generation_ms": 460.0,
         "audio_duration_ms": 240.0,
+        "tpot_ms": 15.0,
+    }
+
+
+def test_session_summary_omits_unmeasured_zero_tpot():
+    summary = summarize_session_request_metrics(
+        [
+            {"ttft_ms": 100.0, "tpot_ms": 0.0},
+            {"ttft_ms": 200.0, "tpot_ms": None},
+        ],
+        session_id="session",
+    )
+
+    assert summary["ttft_ms"] == {
+        "count": 2,
+        "mean": 150.0,
+        "p50": 100.0,
+        "p99": 200.0,
+    }
+    assert "tpot_ms" not in summary
+
+
+def test_event_collector_reports_global_metrics_across_responses():
+    collector = EventCollector()
+    for response_id, created_at_s, text_at_s, audio_times, durations in (
+        ("resp-a", 10.1, 10.2, (10.3, 10.4), (100, 200)),
+        ("resp-b", 12.1, 12.2, (12.3, 12.5), (150, 300)),
+    ):
+        collector.add(
+            {"type": "response.created", "response": {"id": response_id}},
+            received_at_s=created_at_s,
+        )
+        collector.add(
+            {
+                "type": "response.audio_transcript.delta",
+                "response_id": response_id,
+                "delta": response_id,
+            },
+            received_at_s=text_at_s,
+        )
+        for received_at_s, duration_ms in zip(audio_times, durations, strict=True):
+            collector.add(
+                {
+                    "type": "response.audio.delta",
+                    "response_id": response_id,
+                    "delta": base64.b64encode(b"audio").decode("ascii"),
+                    "sample_rate_hz": 16_000,
+                    "metadata": {"audio_duration_ms": duration_ms},
+                },
+                received_at_s=received_at_s,
+            )
+
+    metrics = collector.global_timing_summary(
+        after_s=10.0,
+        window_started_at_s=10.0,
+        response_ids=["resp-a", "resp-b"],
+        measurement_origin={
+            "ttft": "input stream start to first non-empty text delta",
+            "ttfp": "input stream start to first audio packet",
+            "rtf": "input stream start-to-last-audio receive time divided by total emitted audio duration",
+        },
+    )
+
+    assert metrics == {
+        "source": "client_monotonic_receive",
+        "response_ids": ["resp-a", "resp-b"],
+        "measurement_origin": {
+            "ttft": "input stream start to first non-empty text delta",
+            "ttfp": "input stream start to first audio packet",
+            "rtf": "input stream start-to-last-audio receive time divided by total emitted audio duration",
+        },
+        "ttft_ms": 200.0,
+        "ttfp_ms": 300.0,
+        "audio_generation_ms": 2500.0,
+        "audio_duration_ms": 500.0,
+    }
+
+
+def test_event_collector_prefers_server_request_start_for_ttf():
+    collector = EventCollector()
+    collector.add(
+        {
+            "type": "response.created",
+            "response": {
+                "id": "resp-a",
+                "metadata": {
+                    "duplex_event": {
+                        "response_request_metrics": {
+                            "source": "server_monotonic_request_start",
+                            "measurement_origin": {
+                                "ttft": "native request start to first text output",
+                                "ttfp": "native request start to first audio output",
+                            },
+                            "ttft_ms": 125.0,
+                            "ttfp_ms": 175.0,
+                        }
+                    }
+                },
+            },
+        },
+        received_at_s=10.0,
+    )
+    stage_metrics = {
+        "0": {
+            "num_tokens_out": 4,
+            "vllm_ttft_ms": 120.0,
+            "vllm_tpot_ms": 15.0,
+            "vllm_itl_ms": 14.0,
+            "vllm_itls_ms": [10.0, 14.0, 18.0],
+        }
+    }
+    for received_at_s, cumulative_audio_ms in ((10.2, 80), (10.25, 160), (10.36, 240)):
+        collector.add(
+            {
+                "type": "response.audio.delta",
+                "response_id": "resp-a",
+                "delta": base64.b64encode(b"audio").decode("ascii"),
+                "sample_rate_hz": 16_000,
+                "metadata": {
+                    "audio_duration_ms": cumulative_audio_ms,
+                    "vllm_omni": {"stage_metrics": stage_metrics},
+                },
+            },
+            received_at_s=received_at_s,
+        )
+    collector.add(
+        {"type": "response.audio_transcript.delta", "response_id": "resp-a", "delta": "hello"},
+        received_at_s=10.15,
+    )
+
+    timing = collector.timing_summary(
+        after_s=10.0,
+        input_committed_at_s=9.9,
+        response_id="resp-a",
+    )
+
+    assert timing["request_metrics"] == {
+        "source": "server_request_start_and_client_receive",
+        "measurement_origin": {
+            "ttft": "native request start to first text output",
+            "ttfp": "native request start to first audio output",
+            "tpot": "Stage-0 engine mean time per output token",
+        },
+        "ttft_ms": 125.0,
+        "ttfp_ms": 175.0,
+        "audio_generation_ms": 460.0,
+        "audio_duration_ms": 240.0,
+        "response_created_to_first_text_ms": 150.0,
+        "response_created_to_first_audio_ms": 200.0,
+        "tpot_ms": 15.0,
     }
 
 
@@ -806,9 +957,9 @@ def test_summarize_session_request_metrics_averages_audio_turns():
     assert summary == {
         "session_id": "sess-1",
         "audio_turn_count": 2,
-        "mean_ttft_ms": 200.0,
-        "mean_ttfp_ms": 300.0,
-        "mean_rtf": 0.6,
+        "ttft_ms": {"count": 2, "mean": 200.0, "p50": 100.0, "p99": 300.0},
+        "ttfp_ms": {"count": 2, "mean": 300.0, "p50": 200.0, "p99": 400.0},
+        "rtf": {"count": 2, "mean": 0.6, "p50": 0.5, "p99": 0.7},
     }
 
 

tests/e2e/features/fullduplex/test_omni_duplex_eval_cli.py

@@ -15,6 +15,15 @@ import pytest
 from vllm_omni.benchmarks.duplex import omni_duplex_eval_runner as runner
 from vllm_omni.benchmarks.duplex.omni_duplex_eval_dataset import DuplexSample
 from vllm_omni.benchmarks.duplex.omni_duplex_eval_judge import DuplexJudge
+from vllm_omni.benchmarks.duplex.omni_duplex_eval_runner import GenerateSampleResult
+from vllm_omni.benchmarks.duplex_session_metrics import (
+    DUPLEX_METRICS_FILENAME,
+    build_duplex_metrics_report,
+    collect_duplex_session_metrics,
+    merge_duplex_metrics_report,
+    read_duplex_metrics_report,
+)
+from vllm_omni.clients.duplex import EventCollector
 from vllm_omni.entrypoints.cli.benchmark import omni_duplex_eval as cli
 from vllm_omni.entrypoints.cli.benchmark.omni_duplex_eval import OmniDuplexEvalSubcommand
 
@@ -44,7 +53,25 @@ def test_cli_generate_evaluate_summarize_flow(tmp_path: Path, monkeypatch: pytes
         output.parent.mkdir(parents=True, exist_ok=True)
         output.write_text(json.dumps([{"sentence": "It moved.", "start": 0, "end": 1}]), encoding="utf-8")
         output.with_name(output.stem + ".meta.json").write_text(json.dumps({"clock": "media"}), encoding="utf-8")
-        return output
+        return GenerateSampleResult(
+            output=output,
+            request_metrics=[
+                {
+                    "sample_id": sample.id,
+                    "split": sample.split,
+                    "ttft_ms": 10.0,
+                    "ttfp_ms": 20.0,
+                    "rtf": 0.5,
+                }
+            ],
+            session_metrics={
+                "sample_id": sample.id,
+                "split": sample.split,
+                "stream_ttft_ms": 10.0,
+                "stream_ttfp_ms": 20.0,
+                "stream_rtf": 0.5,
+            },
+        )
 
     class FakeJudge:
         def __init__(self, *args, **kwargs):
@@ -75,6 +102,19 @@ def test_cli_generate_evaluate_summarize_flow(tmp_path: Path, monkeypatch: pytes
         ]
     )
     OmniDuplexEvalSubcommand.cmd(generate)
+    duplex_metrics = json.loads((response_root / DUPLEX_METRICS_FILENAME).read_text(encoding="utf-8"))
+    assert duplex_metrics["duplex_request_metrics"] == [
+        {
+            "sample_id": "sample-1",
+            "split": "PR_correction",
+            "ttft_ms": 10.0,
+            "ttfp_ms": 20.0,
+            "rtf": 0.5,
+        }
+    ]
+    assert duplex_metrics["duplex_stream_ttft_ms"]["mean"] == 10.0
+    assert duplex_metrics["duplex_stream_ttfp_ms"]["mean"] == 20.0
+    assert duplex_metrics["duplex_stream_rtf"]["mean"] == 0.5
     evaluate = parser.parse_args(
         [
             "evaluate",
@@ -144,7 +184,7 @@ async def test_generate_exercises_realtime_socket_and_media_clock(tmp_path: Path
 
     async with websockets.serve(handler, "127.0.0.1", 0) as server:
         port = server.sockets[0].getsockname()[1]
-        output = await runner.generate_sample(
+        result = await runner.generate_sample(
             sample,
             url=f"ws://127.0.0.1:{port}/v1/realtime?duplex=1",
             model="mock",
@@ -152,13 +192,231 @@ async def test_generate_exercises_realtime_socket_and_media_clock(tmp_path: Path
             output_root=tmp_path / "responses",
         )
 
-    assert json.loads(output.read_text(encoding="utf-8")) == [{"sentence": "Done.", "start": 0.8, "end": 0.8}]
-    meta = json.loads(output.with_name("sample.meta.json").read_text(encoding="utf-8"))
+    assert json.loads(result.output.read_text(encoding="utf-8")) == [{"sentence": "Done.", "start": 0.8, "end": 0.8}]
+    meta = json.loads(result.output.with_name("sample.meta.json").read_text(encoding="utf-8"))
     assert meta["response_done"] is True
     assert meta["drain_timeout"] is None
+    assert "duplex_request_metrics" not in meta
+    assert result.request_metrics
+    assert result.request_metrics[0]["sample_id"] == "sample"
+    assert result.session_metrics["stream_ttfp_ms"] is not None
     assert any(event["type"] == "playback.ack" for event in received)
 
 
+def _pr_manifest(tmp_path: Path, sample_ids: tuple[str, ...]) -> Path:
+    manifest = tmp_path / "manifest.json"
+    manifest.write_text(
+        json.dumps(
+            [
+                {
+                    "id": sample_id,
+                    "split": "PR_correction",
+                    "question_text": "What changed?",
+                    "answer1": "The object moved.",
+                }
+                for sample_id in sample_ids
+            ]
+        ),
+        encoding="utf-8",
+    )
+    return manifest
+
+
+def _run_generate_cli(manifest: Path, response_root: Path) -> None:
+    parser = argparse.ArgumentParser()
+    OmniDuplexEvalSubcommand.add_cli_args(parser)
+    OmniDuplexEvalSubcommand.cmd(
+        parser.parse_args(
+            [
+                "generate",
+                "--dataset",
+                str(manifest),
+                "--family",
+                "pr",
+                "--model",
+                "mock",
+                "--ref-audio",
+                str(manifest),
+                "--response-root",
+                str(response_root),
+            ]
+        )
+    )
+
+
+def _metrics_for(sample_id: str, *, ttft_ms: float, ttfp_ms: float, rtf: float) -> GenerateSampleResult:
+    split = "PR_correction"
+    return GenerateSampleResult(
+        output=Path(split) / f"{sample_id}.json",
+        request_metrics=[{"sample_id": sample_id, "split": split, "ttft_ms": ttft_ms, "ttfp_ms": ttfp_ms, "rtf": rtf}],
+        session_metrics={
+            "sample_id": sample_id,
+            "split": split,
+            "stream_ttft_ms": ttft_ms,
+            "stream_ttfp_ms": ttfp_ms,
+            "stream_rtf": rtf,
+        },
+    )
+
+
+def _fake_generate(results: dict[str, GenerateSampleResult | None]):
+    async def generate(
+        sample: DuplexSample,
+        *,
+        output_root: str | Path,
+        url: str,
+        model: str,
+        ref_audio: str | Path,
+        fps: float = 1.0,
+        mix: str = "question",
+        pace: str = "realtime",
+        clock: str = "media",
+        overwrite: bool = False,
+        unit_ms: int = 1000,
+    ) -> GenerateSampleResult:
+        _ = (url, model, ref_audio, fps, mix, pace, clock, overwrite, unit_ms)
+        result = results[sample.id]
+        if result is None:
+            return GenerateSampleResult(output=Path(output_root) / sample.split / f"{sample.id}.json")
+        return result
+
+    return generate
+
+
+def test_merge_duplex_metrics_report_replaces_only_incoming_sample_keys():
+    existing = build_duplex_metrics_report(
+        request_metrics=[
+            {"sample_id": "kept", "split": "PR_correction", "ttft_ms": 11.0},
+            {"sample_id": "replaced", "split": "PR_correction", "ttft_ms": 12.0},
+        ],
+        session_metrics=[
+            {"sample_id": "kept", "split": "PR_correction", "stream_ttft_ms": 11.0},
+            {"sample_id": "replaced", "split": "PR_correction", "stream_ttft_ms": 12.0},
+        ],
+    )
+    merged = merge_duplex_metrics_report(
+        existing,
+        request_metrics=[{"sample_id": "replaced", "split": "PR_correction", "ttft_ms": 99.0}],
+        session_metrics=[{"sample_id": "replaced", "split": "PR_correction", "stream_ttft_ms": 99.0}],
+    )
+    assert [row["sample_id"] for row in merged["duplex_request_metrics"]] == ["kept", "replaced"]
+    assert merged["duplex_request_metrics"][1]["ttft_ms"] == 99.0
+    assert merged["duplex_stream_ttft_ms"]["mean"] == 55.0
+
+
+def test_read_duplex_metrics_report_returns_none_when_missing(tmp_path: Path):
+    assert read_duplex_metrics_report(tmp_path / DUPLEX_METRICS_FILENAME) is None
+
+
+def test_read_duplex_metrics_report_rejects_invalid_json(tmp_path: Path):
+    path = tmp_path / DUPLEX_METRICS_FILENAME
+    path.write_text("{not-json", encoding="utf-8")
+    with pytest.raises(json.JSONDecodeError):
+        read_duplex_metrics_report(path)
+
+
+def test_merge_duplex_metrics_report_without_existing_is_passthrough():
+    merged = merge_duplex_metrics_report(
+        None,
+        request_metrics=[{"sample_id": "a", "split": "PR_correction", "ttft_ms": 1.0}],
+        session_metrics=[{"sample_id": "a", "split": "PR_correction", "stream_ttft_ms": 1.0}],
+    )
+    assert merged["duplex_request_metrics"][0]["ttft_ms"] == 1.0
+    assert merged["duplex_stream_ttft_ms"]["mean"] == 1.0
+
+
+def test_merge_duplex_metrics_report_rejects_non_object_rows():
+    with pytest.raises(ValueError, match="duplex_request_metrics entries must be JSON objects"):
+        merge_duplex_metrics_report(
+            {"duplex_request_metrics": ["bad"]},
+            request_metrics=[],
+            session_metrics=[],
+        )
+
+
+def test_cli_generate_all_skipped_keeps_existing_metrics(tmp_path: Path, monkeypatch: pytest.MonkeyPatch):
+    manifest = _pr_manifest(tmp_path, ("sample-1",))
+    response_root = tmp_path / "responses"
+    monkeypatch.setattr(
+        cli,
+        "generate_sample",
+        _fake_generate({"sample-1": _metrics_for("sample-1", ttft_ms=10.0, ttfp_ms=20.0, rtf=0.5)}),
+    )
+    _run_generate_cli(manifest, response_root)
+    monkeypatch.setattr(cli, "generate_sample", _fake_generate({"sample-1": None}))
+    _run_generate_cli(manifest, response_root)
+    duplex_metrics = json.loads((response_root / DUPLEX_METRICS_FILENAME).read_text(encoding="utf-8"))
+    assert duplex_metrics["duplex_request_metrics"][0]["ttft_ms"] == 10.0
+    assert duplex_metrics["duplex_stream_ttft_ms"]["mean"] == 10.0
+    assert duplex_metrics["duplex_stream_ttfp_ms"]["mean"] == 20.0
+    assert duplex_metrics["duplex_stream_rtf"]["mean"] == 0.5
+
+
+def test_cli_generate_partial_resume_merges_metrics(tmp_path: Path, monkeypatch: pytest.MonkeyPatch):
+    manifest = _pr_manifest(tmp_path, ("sample-1", "sample-2"))
+    response_root = tmp_path / "responses"
+    monkeypatch.setattr(
+        cli,
+        "generate_sample",
+        _fake_generate(
+            {
+                "sample-1": _metrics_for("sample-1", ttft_ms=10.0, ttfp_ms=20.0, rtf=0.5),
+                "sample-2": _metrics_for("sample-2", ttft_ms=30.0, ttfp_ms=40.0, rtf=0.25),
+            }
+        ),
+    )
+    _run_generate_cli(manifest, response_root)
+    monkeypatch.setattr(
+        cli,
+        "generate_sample",
+        _fake_generate(
+            {
+                "sample-1": None,
+                "sample-2": _metrics_for("sample-2", ttft_ms=90.0, ttfp_ms=80.0, rtf=1.5),
+            }
+        ),
+    )
+    _run_generate_cli(manifest, response_root)
+    duplex_metrics = json.loads((response_root / DUPLEX_METRICS_FILENAME).read_text(encoding="utf-8"))
+    by_id = {row["sample_id"]: row for row in duplex_metrics["duplex_request_metrics"]}
+    assert by_id["sample-1"]["ttft_ms"] == 10.0
+    assert by_id["sample-2"]["ttft_ms"] == 90.0
+    assert duplex_metrics["duplex_stream_ttft_ms"]["mean"] == 50.0
+    assert duplex_metrics["duplex_stream_ttfp_ms"]["mean"] == 50.0
+    assert duplex_metrics["duplex_stream_rtf"]["mean"] == 1.0
+
+
+def test_collect_duplex_session_metrics_matches_omniinteract_window():
+    collector = EventCollector()
+    audio = {
+        "type": "response.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},
+    }
+    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")
+    assert bundle.request_metrics[0]["ttft_ms"] == 100.0
+    assert bundle.request_metrics[0]["ttfp_ms"] == 200.0
+    assert bundle.request_metrics[0]["rtf"] == 2.0
+    assert bundle.session_metrics["stream_ttft_ms"] == 200.0
+    assert bundle.session_metrics["stream_ttfp_ms"] == 300.0
+    report = build_duplex_metrics_report(
+        request_metrics=bundle.request_metrics,
+        session_metrics=[bundle.session_metrics],
+    )
+    assert report["duplex_stream_ttft_ms"]["mean"] == 200.0
+    assert report["duplex_stream_ttfp_ms"]["mean"] == 300.0
+
+
 def test_judge_exercises_openai_http_schema():
     requests = []
 

tests/e2e/online_serving/test_minicpmo_4_5_duplex.py

@@ -30,7 +30,7 @@ from tests.e2e.online_serving.run_minicpmo_realtime_duplex_multi_session import
     run_multi_session,
 )
 from tests.helpers.mark import hardware_test
-from vllm_omni.clients.duplex import build_realtime_url
+from vllm_omni.clients.duplex import build_realtime_url, metric_mean
 from vllm_omni.experimental.fullduplex.video_stacking import concat_frames_b64
 
 pytestmark = pytest.mark.omni
@@ -59,9 +59,12 @@ def _assert_session_metrics(metrics: object, *, expected_count: int) -> None:
     assert isinstance(metrics, dict)
     assert isinstance(metrics["session_id"], str)
     assert metrics["audio_turn_count"] == expected_count
-    assert metrics["mean_ttft_ms"] is not None and metrics["mean_ttft_ms"] >= 0
-    assert metrics["mean_ttfp_ms"] is not None and metrics["mean_ttfp_ms"] >= 0
-    assert metrics["mean_rtf"] is not None and metrics["mean_rtf"] >= 0
+    ttft_ms = metric_mean(metrics["ttft_ms"])
+    ttfp_ms = metric_mean(metrics["ttfp_ms"])
+    rtf = metric_mean(metrics["rtf"])
+    assert ttft_ms is not None and ttft_ms >= 0
+    assert ttfp_ms is not None and ttfp_ms >= 0
+    assert rtf is not None and rtf >= 0
 
 
 async def _receive_protocol_events(ws, required_types: set[str], *, timeout_s: float) -> list[dict[str, object]]:

tests/e2e/online_serving/test_minicpmo_realtime_duplex_drivers.py

@@ -1209,7 +1209,9 @@ def test_realtime_duplex_demo_partitions_timing_by_response_identity():
                 "ttft": "input_audio_buffer.commit client send to first non-empty text delta",
                 "ttfp": "input_audio_buffer.commit client send to first audio packet",
                 "rtf": "commit-to-last-audio receive time divided by emitted audio duration",
+                "tpot": "Stage-0 engine mean time per output token",
             },
+            "tpot_ms": None,
         },
         {
             "session_id": "seed-tts-session",
@@ -1225,15 +1227,17 @@ def test_realtime_duplex_demo_partitions_timing_by_response_identity():
                 "ttft": "input_audio_buffer.commit client send to first non-empty text delta",
                 "ttfp": "input_audio_buffer.commit client send to first audio packet",
                 "rtf": "commit-to-last-audio receive time divided by emitted audio duration",
+                "tpot": "Stage-0 engine mean time per output token",
             },
+            "tpot_ms": None,
         },
     ]
     assert session_metrics == {
         "session_id": "seed-tts-session",
         "audio_turn_count": 2,
-        "mean_ttft_ms": 150.0,
-        "mean_ttfp_ms": 250.0,
-        "mean_rtf": 3.125,
+        "ttft_ms": {"count": 2, "mean": 150.0, "p50": 150.0, "p99": 150.0},
+        "ttfp_ms": {"count": 2, "mean": 250.0, "p50": 200.0, "p99": 300.0},
+        "rtf": {"count": 2, "mean": 3.125, "p50": 2.5, "p99": 3.75},
     }
 
 

tests/engine/test_output_processor.py

@@ -2,6 +2,7 @@
 # SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project
 """Regression tests for OmniRequestState multimodal DELTA drain and consolidation guard."""
 
+from dataclasses import dataclass
 from types import SimpleNamespace
 from unittest.mock import MagicMock
 
@@ -38,6 +39,21 @@ _DETOK = MagicMock(
 )
 _LOGPROBS = MagicMock(logprobs=None, cumulative_logprob=None, prompt_logprobs=None)
 
+
+@dataclass
+class _StreamingUpdate:
+    final: bool
+    prompt: object | None
+    prompt_token_ids: list[int]
+    arrival_time: float
+
+
+@dataclass
+class _FinishedRequestMetric:
+    request_id: str
+    mean_time_per_output_token: float
+
+
 _DEFAULT_STATE_KWARGS = dict(
     request_id="r",
     external_req_id="r",
@@ -130,6 +146,119 @@ def test_native_text_metrics_include_segment_generation_token_count(monkeypatch)
     assert processor.pop_native_text_metrics("r")["num_generation_tokens"] == 27
 
 
+def test_native_text_metrics_exclude_cross_segment_gap(monkeypatch):
+    monkeypatch.setattr(VLLMOutputProcessor, "_update_stats_from_output", lambda *args, **kwargs: None)
+    processor = object.__new__(MultimodalOutputProcessor)
+    processor._native_text_metrics_by_request = {}
+    processor.lora_states = {}
+    state = _make_state(RequestOutputKind.DELTA)
+    iteration_stats = MagicMock()
+
+    def update_segment(_output, timestamp, _was_prefilling, native_stats, *_args):
+        native_stats.num_generation_tokens = 2 if timestamp == 10.03 else 1
+        native_stats.first_token_ts = 10.02 if timestamp == 10.03 else timestamp
+        native_stats.last_token_ts = timestamp
+        native_stats.first_token_latency = 0.01
+
+    iteration_stats.update_from_output.side_effect = update_segment
+    processor._update_stats_from_output(
+        state,
+        MagicMock(),
+        10.0,
+        iteration_stats,
+    )
+    first_segment = processor.pop_native_text_metrics("r")
+    assert first_segment["vllm_tpot_ms"] == 0.0
+    assert first_segment["vllm_itls_ms"] == []
+
+    state.apply_streaming_update(
+        _StreamingUpdate(
+            final=False,
+            prompt=None,
+            prompt_token_ids=[],
+            arrival_time=10.01,
+        )
+    )
+    state.is_prefilling = True
+    processor._update_stats_from_output(
+        state,
+        MagicMock(),
+        10.02,
+        iteration_stats,
+    )
+    state.is_prefilling = False
+    processor._update_stats_from_output(
+        state,
+        MagicMock(),
+        10.03,
+        iteration_stats,
+    )
+    second_segment = processor.pop_native_text_metrics("r")
+
+    assert second_segment["num_generation_tokens"] == 2
+    assert second_segment["vllm_itls_ms"] == pytest.approx([10.0])
+    assert second_segment["vllm_tpot_ms"] == pytest.approx(10.0)
+
+
+def test_native_text_tpot_weights_multi_token_engine_output(monkeypatch):
+    monkeypatch.setattr(VLLMOutputProcessor, "_update_stats_from_output", lambda *args, **kwargs: None)
+    processor = object.__new__(MultimodalOutputProcessor)
+    processor._native_text_metrics_by_request = {}
+    processor.lora_states = {}
+    state = _make_state(RequestOutputKind.DELTA)
+    iteration_stats = MagicMock()
+
+    def update_segment(_output, timestamp, _was_prefilling, native_stats, *_args):
+        native_stats.num_generation_tokens = 1 if timestamp == 10.0 else 4
+        native_stats.first_token_ts = 10.0
+        native_stats.last_token_ts = timestamp
+        native_stats.first_token_latency = 0.01
+
+    iteration_stats.update_from_output.side_effect = update_segment
+    state.is_prefilling = True
+    processor._update_stats_from_output(state, MagicMock(), 10.0, iteration_stats)
+    state.is_prefilling = False
+    processor._update_stats_from_output(state, MagicMock(), 10.03, iteration_stats)
+    record = processor.pop_native_text_metrics("r")
+
+    assert record["num_generation_tokens"] == 4
+    assert record["vllm_itls_ms"] == pytest.approx([30.0])
+    assert record["vllm_itl_ms"] == pytest.approx(30.0)
+    assert record["vllm_tpot_ms"] == pytest.approx(10.0)
+    assert "_tpot_elapsed_ms" not in record
+    assert "_tpot_intervals" not in record
+
+
+@pytest.mark.parametrize(
+    ("finished_tpot_s", "expected_tpot_ms"),
+    [(0.0, 19.0), (0.023, 23.0)],
+)
+def test_native_text_tpot_only_accepts_positive_finished_metric(
+    monkeypatch,
+    finished_tpot_s,
+    expected_tpot_ms,
+):
+    monkeypatch.setattr(VLLMOutputProcessor, "_update_stats_from_finished", lambda *args, **kwargs: None)
+    processor = object.__new__(MultimodalOutputProcessor)
+    processor._native_text_metrics_by_request = {"r": {"vllm_tpot_ms": 19.0}}
+    state = _make_state(RequestOutputKind.DELTA)
+    iteration_stats = MagicMock()
+    iteration_stats.finished_requests = [
+        _FinishedRequestMetric(
+            request_id="r",
+            mean_time_per_output_token=finished_tpot_s,
+        )
+    ]
+
+    processor._update_stats_from_finished(
+        state,
+        FinishReason.STOP,
+        iteration_stats,
+    )
+
+    assert processor.pop_native_text_metrics("r")["vllm_tpot_ms"] == expected_tpot_ms
+
+
 def test_delta_drains_output_modality_per_step():
     """DELTA drains the mm_type key (output modality) but preserves hidden-state keys."""
     s = _make_state(RequestOutputKind.DELTA)

tests/entrypoints/openai_api/test_duplex_handler.py

@@ -9007,19 +9007,82 @@ async def test_minicpmo_native_duplex_continuous_speak_reuses_active_response_un
     assert session.turn_id == 1
 
 
+def test_response_timing_binds_latest_request_start_for_model_turn():
+    session = DuplexSession(
+        session_id="sid-response-request-timing",
+        config=DuplexSessionConfig(extra_body={"auto_response": True}),
+    )
+    session.mark_model_turn_request_started(0, 10.0)
+    session.mark_model_turn_request_started(0, 11.0)
+    session.begin_response(turn_id=0)
+
+    assert session.mark_response_first_outputs(
+        observed_at_s=11.2,
+        has_text=True,
+        has_audio=False,
+    )["ttft_ms"] == pytest.approx(200.0)
+
+    session.mark_model_turn_request_started(0, 12.0)
+    assert session.mark_response_first_outputs(
+        observed_at_s=12.3,
+        has_text=False,
+        has_audio=True,
+    )["ttfp_ms"] == pytest.approx(1300.0)
+
+
+def test_response_tpot_fallback_ignores_single_token_segment():
+    session = DuplexSession(
+        session_id="sid-response-tpot",
+        config=DuplexSessionConfig(extra_body={"auto_response": True}),
+    )
+    session.begin_response(turn_id=0)
+
+    first_metrics = session.accumulate_response_stage_metrics({"0": {"num_tokens_out": 1, "vllm_tpot_ms": 900.0}})
+    assert "vllm_tpot_ms" not in first_metrics["0"]
+
+    second_metrics = session.accumulate_response_stage_metrics({"0": {"num_tokens_out": 2, "vllm_tpot_ms": 15.0}})
+    assert second_metrics["0"]["num_tokens_out"] == 3
+    assert second_metrics["0"]["vllm_tpot_ms"] == 15.0
+
+
+def test_response_tpot_keeps_token_weighted_value_when_itls_exist():
+    session = DuplexSession(
+        session_id="sid-response-tpot-multi-token",
+        config=DuplexSessionConfig(extra_body={"auto_response": True}),
+    )
+    session.begin_response(turn_id=0)
+    metrics = session.accumulate_response_stage_metrics(
+        {
+            "0": {
+                "num_tokens_out": 4,
+                "vllm_tpot_ms": 10.0,
+                "vllm_itls_ms": [30.0],
+            }
+        }
+    )
+    assert metrics["0"]["vllm_itls_ms"] == [30.0]
+    assert metrics["0"]["vllm_itl_ms"] == 30.0
+    assert metrics["0"]["vllm_tpot_ms"] == 10.0
+
+
 @pytest.mark.asyncio
-async def test_continuous_response_metrics_accumulate_only_owned_model_units():
+async def test_continuous_response_metrics_accumulate_only_owned_model_units(monkeypatch):
     handler = OmniDuplexSessionHandler(chat_service=FakeChatService(FakeEngineClient()))
     session = DuplexSession(
         session_id="sid-response-metrics",
         config=DuplexSessionConfig(extra_body={"auto_response": True}),
     )
+    session.mark_model_turn_request_started(0, 10.0)
+    monkeypatch.setattr(
+        "vllm_omni.entrypoints.duplex.runtime_bridge.time.monotonic",
+        lambda: 10.25,
+    )
     sent: list[dict[str, Any]] = []
 
     async def send_json(payload: dict[str, Any]) -> None:
         sent.append(payload)
 
-    async def emit(*, text: str, audio: str, tokens: int, ttft_ms: float, itls: list[float]) -> None:
+    async def emit(*, text: str, audio: str, tokens: int, ttft_ms: float, itls: list[float], tpot_ms: float) -> None:
         await handler._send_one_native_duplex_event(
             send_json,
             {
@@ -9038,7 +9101,7 @@ async def test_continuous_response_metrics_accumulate_only_owned_model_units():
                         "num_tokens_out": tokens,
                         "stage_gen_time_ms": 80.0,
                         "vllm_ttft_ms": ttft_ms,
-                        "vllm_tpot_ms": 9.0,
+                        "vllm_tpot_ms": tpot_ms,
                         "vllm_itls_ms": itls,
                     }
                 },
@@ -9046,21 +9109,32 @@ async def test_continuous_response_metrics_accumulate_only_owned_model_units():
             session=session,
         )
 
-    await emit(text="first", audio="audio-a", tokens=4, ttft_ms=50.0, itls=[10.0, 11.0])
-    await emit(text="second", audio="audio-b", tokens=6, ttft_ms=70.0, itls=[12.0, 13.0, 14.0])
+    await emit(text="first", audio="audio-a", tokens=3, ttft_ms=50.0, itls=[10.0, 11.0], tpot_ms=10.5)
+    await emit(text="second", audio="audio-b", tokens=4, ttft_ms=70.0, itls=[12.0, 13.0, 14.0], tpot_ms=13.0)
 
     deltas = [payload for payload in sent if payload.get("type") == "response.output_audio.delta"]
     assert len(deltas) == 2
     first_metrics = deltas[0]["vllm_omni"]["stage_metrics"]["0"]
     second_metrics = deltas[1]["vllm_omni"]["stage_metrics"]["0"]
-    assert first_metrics["num_tokens_out"] == 4
-    assert second_metrics["num_tokens_out"] == 10
+    assert first_metrics["num_tokens_out"] == 3
+    assert first_metrics["vllm_tpot_ms"] == pytest.approx(10.5)
+    assert deltas[0]["vllm_omni"]["response_request_metrics"] == {
+        "source": "server_monotonic_request_start",
+        "measurement_origin": {
+            "ttft": "native model-turn request execution start to first non-empty text output",
+            "ttfp": "native model-turn request execution start to first audio output",
+        },
+        "ttft_ms": 250.0,
+        "ttfp_ms": 250.0,
+    }
+    assert second_metrics["num_tokens_out"] == 7
     assert second_metrics["vllm_ttft_ms"] == 50.0
     assert second_metrics["vllm_itls_ms"] == [10.0, 11.0, 12.0, 13.0, 14.0]
+    assert second_metrics["vllm_tpot_ms"] == pytest.approx(12.0)
 
     first_response_id = deltas[0]["response_id"]
     session.end_response()
-    await emit(text="third", audio="audio-c", tokens=3, ttft_ms=90.0, itls=[15.0, 16.0])
+    await emit(text="third", audio="audio-c", tokens=3, ttft_ms=90.0, itls=[15.0, 16.0], tpot_ms=15.5)
 
     third_delta = [payload for payload in sent if payload.get("type") == "response.output_audio.delta"][-1]
     third_metrics = third_delta["vllm_omni"]["stage_metrics"]["0"]
@@ -9068,6 +9142,7 @@ async def test_continuous_response_metrics_accumulate_only_owned_model_units():
     assert third_metrics["num_tokens_out"] == 3
     assert third_metrics["vllm_ttft_ms"] == 90.0
     assert third_metrics["vllm_itls_ms"] == [15.0, 16.0]
+    assert third_metrics["vllm_tpot_ms"] == pytest.approx(15.5)
 
 
 @pytest.mark.asyncio

vllm_omni/benchmarks/duplex/omni_duplex_eval_runner.py

@@ -8,24 +8,58 @@ from __future__ import annotations
 import asyncio
 import hashlib
 import json
+import time
+from dataclasses import dataclass, field
 from pathlib import Path
-from typing import Any
+from typing import TYPE_CHECKING
 
 import pybase64 as base64
 
-from vllm_omni.experimental.fullduplex.client import RealtimeDuplexClient, build_realtime_url, wait_for
+from vllm_omni.benchmarks.duplex.omni_duplex_eval_dataset import DuplexSample
+from vllm_omni.experimental.fullduplex.client import (
+    RealtimeDuplexClient,
+    RealtimeEventCollector,
+    build_realtime_url,
+    wait_for,
+)
+
+if TYPE_CHECKING:
+    from vllm_omni.clients.duplex import EventCollector
 
 from .omni_duplex_eval_clock import extract_timed_sentences
 from .omni_duplex_eval_media import iter_av_units, iter_jpegs, materialize_media, read_audio_pcm16, video_duration
 
 
+@dataclass(frozen=True)
+class GenerateSampleResult:
+    """Timed-sentence path plus duplex metrics for one generate sample."""
+
+    output: Path
+    request_metrics: list[dict[str, object]] = field(default_factory=list)
+    session_metrics: dict[str, object] = field(default_factory=dict)
+
+
 def _ref_audio(path: str | Path) -> str:
     value = Path(path).expanduser().read_bytes()
     return "data:audio/wav;base64," + base64.b64encode(value).decode("ascii")
 
 
+def _event_collector_from_realtime(source: RealtimeEventCollector) -> EventCollector:
+    """Replay experimental collector events onto the public EventCollector."""
+    from vllm_omni.clients.duplex import EventCollector
+
+    collector = EventCollector()
+    for event, received_at_s in zip(source.events, source.event_received_at_s, strict=True):
+        collector.add(event, received_at_s=received_at_s)
+    return collector
+
+
+def _with_sample_identity(payload: dict[str, object], sample: DuplexSample) -> dict[str, object]:
+    return {"sample_id": sample.id, "split": sample.split, **payload}
+
+
 async def generate_sample(
-    sample: Any,
+    sample: DuplexSample,
     *,
     url: str,
     model: str,
@@ -37,11 +71,17 @@ async def generate_sample(
     clock: str = "media",
     overwrite: bool = False,
     unit_ms: int = 1000,
-) -> Path:
+) -> GenerateSampleResult:
+    """Generate one sample's timed sentences and collect duplex session metrics.
+
+    Timed sentences and ``*.meta.json`` stay the quality-eval artifacts. Duplex
+    request/session metrics are returned for the CLI to write
+    ``duplex_metrics.json``; they are not written into meta.
+    """
     output = Path(output_root) / sample.split / f"{sample.id}.json"
     meta_path = output.with_name(output.stem + ".meta.json")
     if output.exists() and not overwrite:
-        return output
+        return GenerateSampleResult(output=output)
     if mix != "question":
         raise NotImplementedError("v1 supports mix=question; soundtrack mixing is reserved for P1")
     media_dir = output.parent / ".media"
@@ -57,10 +97,12 @@ async def generate_sample(
     response_done = False
     drain_timeout = None
     close_timeout = None
+    stream_start: float | None = None
     async with client:
         await client.configure(model, ref_audio=_ref_audio(ref_audio), instructions="Streaming Omni Conversation.")
         ack_task = asyncio.create_task(_ack_playback(client))
         try:
+            stream_start = time.monotonic()
             await client.stream_av_units(iter_av_units(pcm, frames, unit_ms=unit_ms), realtime=realtime)
             await client.commit()
             try:
@@ -84,6 +126,7 @@ async def generate_sample(
         except TimeoutError as exc:
             close_timeout = str(exc)
         events = list(client.events.events)
+        collector = _event_collector_from_realtime(client.events)
     timed = [sentence.as_dict() for sentence in extract_timed_sentences(events, clock=clock)]
     output.parent.mkdir(parents=True, exist_ok=True)
     output.write_text(json.dumps(timed, ensure_ascii=False, indent=2), encoding="utf-8")
@@ -102,7 +145,18 @@ async def generate_sample(
         "ref_audio_sha256": hashlib.sha256(Path(ref_audio).read_bytes()).hexdigest(),
     }
     meta_path.write_text(json.dumps(meta, indent=2), encoding="utf-8")
-    return output
+    from vllm_omni.benchmarks.duplex_session_metrics import collect_duplex_session_metrics
+
+    bundle = collect_duplex_session_metrics(
+        collector,
+        stream_start=stream_start if stream_start is not None else 0.0,
+        session_id=sample.id,
+    )
+    return GenerateSampleResult(
+        output=output,
+        request_metrics=[_with_sample_identity(metric, sample) for metric in bundle.request_metrics],
+        session_metrics=_with_sample_identity(bundle.session_metrics, sample),
+    )
 
 
 async def _ack_playback(client: RealtimeDuplexClient) -> None:

vllm_omni/benchmarks/duplex_session_metrics.py

@@ -0,0 +1,229 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project
+
+"""Assemble duplex serve metrics from a public EventCollector.
+
+This module does not instrument the server. It reads the same
+``EventCollector.timing_summary`` / ``global_timing_summary`` surfaces that
+OmniInteract already uses, derives RTF with ``compute_audio_rtf``, and folds
+per-session ``stream_*`` values into a run-level report.
+"""
+
+from __future__ import annotations
+
+import json
+import math
+from collections.abc import Mapping, Sequence
+from dataclasses import dataclass
+from pathlib import Path
+from typing import TYPE_CHECKING
+
+from vllm_omni.metrics.definitions import compute_audio_rtf
+
+if TYPE_CHECKING:
+    from vllm_omni.clients.duplex import EventCollector
+
+DUPLEX_METRICS_FILENAME = "duplex_metrics.json"
+
+_REQUEST_MEASUREMENT_ORIGIN = {
+    "tpot": "Stage-0 engine mean time per output token",
+    "rtf": "response.created client receive to last audio packet divided by emitted audio duration",
+}
+_STREAM_MEASUREMENT_ORIGIN = {
+    "ttft": "input stream start to first non-empty text delta",
+    "ttfp": "input stream start to first audio packet",
+    "rtf": (
+        "input stream start-to-last-audio receive time divided by total emitted audio duration; "
+        "includes concurrent realtime input"
+    ),
+}
+
+
+@dataclass(frozen=True)
+class DuplexSessionMetricBundle:
+    """One session's request rows plus the session/stream summary."""
+
+    request_metrics: list[dict[str, object]]
+    session_metrics: dict[str, object]
+    output_tokens: int
+
+
+def audio_rtf_from_raw_metric(raw_metric: Mapping[str, object]) -> float | None:
+    """Derive audio RTF from client-reported generation and duration milliseconds."""
+    generation_ms = raw_metric.get("audio_generation_ms")
+    duration_ms = raw_metric.get("audio_duration_ms")
+    if not isinstance(generation_ms, int | float) or not isinstance(duration_ms, int | float) or duration_ms <= 0:
+        return None
+    return round(compute_audio_rtf(float(generation_ms) / 1000.0, float(duration_ms) / 1000.0), 6)
+
+
+def collect_duplex_session_metrics(
+    collector: EventCollector,
+    *,
+    stream_start: float,
+    session_id: str | None,
+) -> DuplexSessionMetricBundle:
+    """Build per-response and session/stream metrics for one duplex session.
+
+    ``stream_start`` is the client monotonic time just before input media is
+    pushed. Per-response TTFT/TTFP prefer server ``response_request_metrics``
+    when present; RTF and the session ``stream_*`` window stay client-receive.
+    """
+    from vllm_omni.clients.duplex import summarize_session_request_metrics
+
+    request_metrics: list[dict[str, object]] = []
+    output_tokens = 0
+    for request_index, response_id in enumerate(collector.response_ids):
+        timing = collector.timing_summary(
+            after_s=stream_start,
+            input_committed_at_s=None,
+            response_id=response_id,
+            measurement_origin=_REQUEST_MEASUREMENT_ORIGIN,
+        )
+        raw_metric = timing.get("request_metrics")
+        stage0 = timing.get("stage0_tokens")
+        metric: dict[str, object] = {
+            "session_id": session_id,
+            "request_index": request_index,
+            "response_id": response_id,
+        }
+        if isinstance(raw_metric, dict):
+            metric.update(raw_metric)
+            metric["rtf"] = audio_rtf_from_raw_metric(raw_metric)
+        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):
+            request_metrics.append(metric)
+    session_metrics = summarize_session_request_metrics(
+        request_metrics,
+        session_id=session_id,
+    )
+    stream_metrics = collector.global_timing_summary(
+        after_s=stream_start,
+        window_started_at_s=stream_start,
+        response_ids=list(collector.response_ids),
+        measurement_origin=_STREAM_MEASUREMENT_ORIGIN,
+    )
+    if stream_metrics:
+        session_metrics.update(
+            {
+                "stream_ttft_ms": stream_metrics.get("ttft_ms"),
+                "stream_ttfp_ms": stream_metrics.get("ttfp_ms"),
+                "stream_rtf": audio_rtf_from_raw_metric(stream_metrics),
+                "stream_audio_generation_ms": stream_metrics.get("audio_generation_ms"),
+                "stream_audio_duration_ms": stream_metrics.get("audio_duration_ms"),
+                "stream_measurement_origin": stream_metrics.get("measurement_origin"),
+            }
+        )
+    return DuplexSessionMetricBundle(
+        request_metrics=request_metrics,
+        session_metrics=session_metrics,
+        output_tokens=output_tokens,
+    )
+
+
+def duplex_stream_metrics(session_metrics: Sequence[Mapping[str, object]]) -> dict[str, object]:
+    """Summarize finite, non-negative per-session ``stream_*`` values."""
+    from vllm_omni.clients.duplex import distribution_summary
+
+    result: dict[str, object] = {}
+    for session_key, result_key, digits in (
+        ("stream_ttft_ms", "duplex_stream_ttft_ms", 3),
+        ("stream_ttfp_ms", "duplex_stream_ttfp_ms", 3),
+        ("stream_rtf", "duplex_stream_rtf", 6),
+    ):
+        values = [
+            float(value)
+            for metric in session_metrics
+            if isinstance((value := metric.get(session_key)), int | float)
+            and not isinstance(value, bool)
+            and math.isfinite(value)
+            and value >= 0
+        ]
+        summary = distribution_summary(values, digits=digits)
+        if summary is not None:
+            result[result_key] = summary
+    return result
+
+
+def build_duplex_metrics_report(
+    *,
+    request_metrics: Sequence[Mapping[str, object]],
+    session_metrics: Sequence[Mapping[str, object]],
+) -> dict[str, object]:
+    """Flatten this generate run into the ``duplex_metrics.json`` payload."""
+    report: dict[str, object] = {
+        "duplex_request_metrics": [dict(metric) for metric in request_metrics],
+        "duplex_session_metrics": [dict(metric) for metric in session_metrics],
+    }
+    report.update(duplex_stream_metrics(session_metrics))
+    return report
+
+
+def sample_metric_key(metric: Mapping[str, object]) -> tuple[str, str] | None:
+    """Identity used to merge resume rows: ``(split, sample_id)``."""
+    split = metric.get("split")
+    sample_id = metric.get("sample_id")
+    if isinstance(split, str) and split and isinstance(sample_id, str) and sample_id:
+        return (split, sample_id)
+    return None
+
+
+def merge_metric_rows(
+    existing: Sequence[Mapping[str, object]],
+    incoming: Sequence[Mapping[str, object]],
+) -> list[dict[str, object]]:
+    """Replace existing rows that share an incoming ``(split, sample_id)``.
+
+    Incoming rows without a key are appended. Existing rows without a key are
+    kept because they cannot be matched.
+    """
+    incoming_keys = {key for metric in incoming if (key := sample_metric_key(metric)) is not None}
+    merged = [dict(metric) for metric in existing if (key := sample_metric_key(metric)) not in incoming_keys]
+    merged.extend(dict(metric) for metric in incoming)
+    return merged
+
+
+def read_duplex_metrics_report(path: Path) -> dict[str, object] | None:
+    """Load an on-disk report. Missing files return ``None``; corrupt JSON raises."""
+    if not path.is_file():
+        return None
+    payload = json.loads(path.read_text(encoding="utf-8"))
+    if not isinstance(payload, dict):
+        raise ValueError(f"{path} must contain a JSON object")
+    return payload
+
+
+def _metric_rows(value: object, *, field_name: str) -> list[Mapping[str, object]]:
+    if value is None:
+        return []
+    if not isinstance(value, list):
+        raise ValueError(f"{field_name} must be a JSON list")
+    rows: list[Mapping[str, object]] = []
+    for item in value:
+        if not isinstance(item, dict):
+            raise ValueError(f"{field_name} entries must be JSON objects")
+        rows.append(item)
+    return rows
+
+
+def merge_duplex_metrics_report(
+    existing: Mapping[str, object] | None,
+    *,
+    request_metrics: Sequence[Mapping[str, object]],
+    session_metrics: Sequence[Mapping[str, object]],
+) -> dict[str, object]:
+    """Merge this run into a previous ``duplex_metrics.json`` payload."""
+    previous_requests = _metric_rows(
+        None if existing is None else existing.get("duplex_request_metrics"),
+        field_name="duplex_request_metrics",
+    )
+    previous_sessions = _metric_rows(
+        None if existing is None else existing.get("duplex_session_metrics"),
+        field_name="duplex_session_metrics",
+    )
+    return build_duplex_metrics_report(
+        request_metrics=merge_metric_rows(previous_requests, request_metrics),
+        session_metrics=merge_metric_rows(previous_sessions, session_metrics),
+    )

vllm_omni/benchmarks/metrics/metrics.py

@@ -916,11 +916,9 @@ def calculate_metrics(
             all_tpots.append(tpot)
             itls += outputs[i].itl
             session_metrics = getattr(outputs[i], "duplex_session_metrics", None)
-            ttft_measured = not isinstance(session_metrics, dict) or session_metrics.get("mean_ttft_ms") is not None
-            audio_ttfp_measured = (
-                not isinstance(session_metrics, dict) or session_metrics.get("mean_ttfp_ms") is not None
-            )
-            audio_rtf_measured = not isinstance(session_metrics, dict) or session_metrics.get("mean_rtf") is not None
+            ttft_measured = not isinstance(session_metrics, dict) or session_metrics.get("ttft_ms") is not None
+            audio_ttfp_measured = not isinstance(session_metrics, dict) or session_metrics.get("ttfp_ms") is not None
+            audio_rtf_measured = not isinstance(session_metrics, dict) or session_metrics.get("rtf") is not None
             if ttft_measured:
                 ttfts.append(outputs[i].ttft)
             if audio_ttfp_measured:

vllm_omni/benchmarks/omniinteract.py

@@ -946,13 +946,9 @@ class _RealtimeSession:
 
 def _audio_rtf_from_raw_metric(raw_metric: dict[str, object]) -> float | None:
     """Derive the audio RTF from a client-reported raw request metric."""
-    from vllm_omni.metrics.definitions import compute_audio_rtf
+    from vllm_omni.benchmarks.duplex_session_metrics import audio_rtf_from_raw_metric
 
-    generation_ms = raw_metric.get("audio_generation_ms")
-    duration_ms = raw_metric.get("audio_duration_ms")
-    if not isinstance(generation_ms, int | float) or not isinstance(duration_ms, int | float) or duration_ms <= 0:
-        return None
-    return round(compute_audio_rtf(float(generation_ms) / 1000.0, float(duration_ms) / 1000.0), 6)
+    return audio_rtf_from_raw_metric(raw_metric)
 
 
 def _populate_response_metrics(
@@ -961,45 +957,16 @@ def _populate_response_metrics(
     *,
     stream_start: float,
 ) -> None:
-    measurement_origin = {
-        "ttft": "response.created client receive to first non-empty text delta",
-        "ttfp": "response.created client receive to first audio packet",
-        "rtf": "response.created client receive to last audio packet divided by emitted audio duration",
-    }
-    request_metrics: list[dict[str, object]] = []
-    output_tokens = 0
-    for request_index, response_id in enumerate(collector.response_ids):
-        timing = collector.timing_summary(
-            after_s=stream_start,
-            input_committed_at_s=None,
-            response_id=response_id,
-            measurement_origin=measurement_origin,
-        )
-        raw_metric = timing.get("request_metrics")
-        stage0 = timing.get("stage0_tokens")
-        metric = {
-            "session_id": result.session_id,
-            "request_index": request_index,
-            "response_id": response_id,
-            **(raw_metric if isinstance(raw_metric, dict) else {}),
-        }
-        if isinstance(raw_metric, dict):
-            # The client reports raw data only; the RTF is derived here with
-            # the canonical server-side metric definition.
-            metric["rtf"] = _audio_rtf_from_raw_metric(raw_metric)
-        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):
-            request_metrics.append(metric)
-    result.output_tokens = output_tokens
-    result.duplex_request_metrics = request_metrics
-    from vllm_omni.clients.duplex import summarize_session_request_metrics
+    from vllm_omni.benchmarks.duplex_session_metrics import collect_duplex_session_metrics
 
-    result.duplex_session_metrics = summarize_session_request_metrics(
-        request_metrics,
+    bundle = collect_duplex_session_metrics(
+        collector,
+        stream_start=stream_start,
         session_id=result.session_id,
     )
+    result.output_tokens = bundle.output_tokens
+    result.duplex_request_metrics = bundle.request_metrics
+    result.duplex_session_metrics = bundle.session_metrics
 
 
 async def run_omniinteract_case(

vllm_omni/benchmarks/patch/patch.py

@@ -2272,9 +2272,11 @@ async def _async_request_omniinteract(
         output.audio_duration = case_result.audio_bytes / (24_000 * 2)
         output.audio_frames = case_result.audio_bytes // 2
         session_metrics = case_result.duplex_session_metrics
-        output.ttft = float(session_metrics.get("mean_ttft_ms") or 0.0) / 1000.0
-        output.audio_ttfp = float(session_metrics.get("mean_ttfp_ms") or 0.0) / 1000.0
-        output.audio_rtf = float(session_metrics.get("mean_rtf") or 0.0)
+        from vllm_omni.clients.duplex import metric_mean
+
+        output.ttft = (metric_mean(session_metrics.get("ttft_ms")) or 0.0) / 1000.0
+        output.audio_ttfp = (metric_mean(session_metrics.get("ttfp_ms")) or 0.0) / 1000.0
+        output.audio_rtf = metric_mean(session_metrics.get("rtf")) or 0.0
         token_timing_measured = _apply_stage0_token_timings(
             output,
             [request_metric.get("stage0_tokens") for request_metric in case_result.duplex_request_metrics],
@@ -2524,9 +2526,11 @@ async def async_request_openai_realtime_duplex(
             await client.close_session(timeout_s=30.0)
 
             output.generated_text = " ".join(filter(None, turn_transcripts))
-            output.ttft = float(session_metrics.get("mean_ttft_ms") or 0.0) / 1000.0
-            output.audio_ttfp = float(session_metrics.get("mean_ttfp_ms") or 0.0) / 1000.0
-            output.audio_rtf = float(session_metrics.get("mean_rtf") or 0.0)
+            from vllm_omni.clients.duplex import metric_mean
+
+            output.ttft = (metric_mean(session_metrics.get("ttft_ms")) or 0.0) / 1000.0
+            output.audio_ttfp = (metric_mean(session_metrics.get("ttfp_ms")) or 0.0) / 1000.0
+            output.audio_rtf = metric_mean(session_metrics.get("rtf")) or 0.0
             output.audio_duration = (
                 sum(float(metric.get("audio_duration_ms") or 0.0) for metric in turn_metrics) / 1000.0
             )
@@ -2926,7 +2930,7 @@ async def benchmark(
 
         def measured_ttft(output: RequestFuncOutput) -> float | None:
             session_metrics = getattr(output, "duplex_session_metrics", None)
-            if isinstance(session_metrics, dict) and session_metrics.get("mean_ttft_ms") is None:
+            if isinstance(session_metrics, dict) and session_metrics.get("ttft_ms") is None:
                 return None
             return output.ttft
 
@@ -2998,6 +3002,24 @@ async def benchmark(
     ]
     if duplex_session_metrics:
         result["duplex_session_metrics"] = duplex_session_metrics
+        from vllm_omni.clients.duplex import distribution_summary
+
+        for session_key, result_key, digits in (
+            ("stream_ttft_ms", "duplex_stream_ttft_ms", 3),
+            ("stream_ttfp_ms", "duplex_stream_ttfp_ms", 3),
+            ("stream_rtf", "duplex_stream_rtf", 6),
+        ):
+            values = [
+                float(value)
+                for metric in duplex_session_metrics
+                if isinstance((value := metric.get(session_key)), int | float)
+                and not isinstance(value, bool)
+                and np.isfinite(value)
+                and value >= 0
+            ]
+            summary = distribution_summary(values, digits=digits)
+            if summary is not None:
+                result[result_key] = summary
     if omniinteract_summary is not None:
         result["omniinteract"] = omniinteract_summary
 

vllm_omni/clients/duplex.py

@@ -35,7 +35,7 @@ import math
 import random
 import time
 import wave
-from collections.abc import AsyncIterator, Awaitable, Callable, Iterator, Sequence
+from collections.abc import AsyncIterator, Awaitable, Callable, Iterator, Mapping, Sequence
 from dataclasses import dataclass, field
 from pathlib import Path
 from typing import Protocol
@@ -80,6 +80,8 @@ __all__ = [
     "image_data_url",
     "read_pcm16_wav",
     "reference_audio_data_url",
+    "distribution_summary",
+    "metric_mean",
     "summarize_session_request_metrics",
     "wait_for_condition",
     "write_pcm16_wav",
@@ -1248,6 +1250,15 @@ 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:
@@ -1266,6 +1277,36 @@ def _interval_summary(values: list[float]) -> dict[str, float | int]:
     }
 
 
+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")
@@ -1285,6 +1326,34 @@ def _event_stage_metrics(event: dict[str, object]) -> dict[str, object] | None:
     return None
 
 
+def _event_response_request_metrics(event: dict[str, object]) -> dict[str, object] | None:
+    direct_metrics = event.get("response_request_metrics")
+    if isinstance(direct_metrics, dict):
+        return direct_metrics
+    candidates: list[object] = [event.get("vllm_omni")]
+    metadata = event.get("metadata")
+    if isinstance(metadata, dict):
+        candidates.extend((metadata, metadata.get("vllm_omni"), metadata.get("duplex_event")))
+    response = event.get("response")
+    if isinstance(response, dict):
+        response_metadata = response.get("metadata")
+        if isinstance(response_metadata, dict):
+            candidates.extend(
+                (
+                    response_metadata,
+                    response_metadata.get("vllm_omni"),
+                    response_metadata.get("duplex_event"),
+                )
+            )
+    for candidate in candidates:
+        if not isinstance(candidate, dict):
+            continue
+        request_metrics = candidate.get("response_request_metrics")
+        if isinstance(request_metrics, dict):
+            return request_metrics
+    return None
+
+
 class EventCollector:
     """Accumulate events for assertions and latency summaries (tests/benchmarks).
 
@@ -1379,6 +1448,85 @@ class EventCollector:
                 return received_at_s
         return None
 
+    def global_timing_summary(
+        self,
+        *,
+        after_s: float,
+        window_started_at_s: float,
+        response_ids: list[str],
+        measurement_origin: dict[str, str],
+    ) -> dict[str, object]:
+        """Summarize one client-observed timing window across responses.
+
+        Returns raw measurements only. Derived metrics such as RTF are computed
+        by the caller — e.g. with ``vllm_omni.metrics.definitions.compute_audio_rtf``.
+        """
+        selected_response_ids = set(response_ids)
+        first_text_received_at_s: float | None = None
+        audio_received_at_s: list[float] = []
+        response_audio_duration_ms: dict[str, float] = {}
+
+        for event, received_at_s in zip(self.events, self.event_received_at_s, strict=True):
+            if received_at_s < after_s:
+                continue
+            response_id = self.response_id(event)
+            if response_id is None or response_id not in selected_response_ids:
+                continue
+            if (
+                event.get("type")
+                in {
+                    "response.audio_transcript.delta",
+                    "response.output_text.delta",
+                    "response.text.delta",
+                }
+                and isinstance(event.get("delta"), str)
+                and bool(event["delta"])
+                and first_text_received_at_s is None
+            ):
+                first_text_received_at_s = received_at_s
+            if event.get("type") not in _AUDIO_DELTA_TYPES:
+                continue
+            delta = event.get("delta") or event.get("audio")
+            if not isinstance(delta, str) or not delta:
+                continue
+            audio_received_at_s.append(received_at_s)
+            metadata = event.get("metadata")
+            duration_ms = metadata.get("audio_duration_ms") if isinstance(metadata, dict) else None
+            parsed_duration_ms = _finite_number(duration_ms, nonnegative=True)
+            if parsed_duration_ms is not None:
+                response_audio_duration_ms[response_id] = max(
+                    response_audio_duration_ms.get(response_id, 0.0),
+                    parsed_duration_ms,
+                )
+
+        if not audio_received_at_s:
+            return {}
+
+        audio_duration_ms = sum(
+            response_audio_duration_ms.get(
+                response_id,
+                self._format_for_bytes().duration_ms(len(self.audio_bytes(response_id))),
+            )
+            for response_id in response_ids
+        )
+        audio_generation_ms = max(
+            0.0,
+            (audio_received_at_s[-1] - window_started_at_s) * 1000.0,
+        )
+        return {
+            "source": "client_monotonic_receive",
+            "response_ids": list(response_ids),
+            "measurement_origin": dict(measurement_origin),
+            "ttft_ms": (
+                _rounded_ms((first_text_received_at_s - window_started_at_s) * 1000.0)
+                if first_text_received_at_s is not None
+                else None
+            ),
+            "ttfp_ms": _rounded_ms((audio_received_at_s[0] - window_started_at_s) * 1000.0),
+            "audio_generation_ms": _rounded_ms(audio_generation_ms),
+            "audio_duration_ms": _rounded_ms(audio_duration_ms),
+        }
+
     def timing_summary(
         self,
         *,
@@ -1389,6 +1537,7 @@ class EventCollector:
     ) -> dict[str, object]:
         """Summarize engine token metrics and client-observed audio cadence."""
         stage0_metrics: dict[str, object] | None = None
+        response_request_metrics: dict[str, object] = {}
         response_created_at_s: float | None = None
         first_text_received_at_s: float | None = None
         audio_received_at_s: list[float] = []
@@ -1419,6 +1568,10 @@ class EventCollector:
             if isinstance(stage0, dict):
                 stage0_metrics = stage0
 
+            event_request_metrics = _event_response_request_metrics(event)
+            if event_request_metrics is not None:
+                response_request_metrics.update(event_request_metrics)
+
             if event.get("type") not in _AUDIO_DELTA_TYPES:
                 continue
             delta = event.get("delta") or event.get("audio")
@@ -1478,6 +1631,31 @@ class EventCollector:
             }
             request_started_at_s = input_committed_at_s if input_committed_at_s is not None else response_created_at_s
             if request_started_at_s is not None:
+                default_measurement_origin = (
+                    {
+                        "ttft": "input_audio_buffer.commit client send to first non-empty text delta",
+                        "ttfp": "input_audio_buffer.commit client send to first audio packet",
+                    }
+                    if input_committed_at_s is not None
+                    else {
+                        "ttft": "response.created client receive to first non-empty text delta",
+                        "ttfp": "response.created client receive to first audio packet",
+                    }
+                )
+                resolved_measurement_origin = {
+                    **default_measurement_origin,
+                    **(measurement_origin or {}),
+                }
+                if stage0_metrics is not None:
+                    resolved_measurement_origin["tpot"] = "Stage-0 engine mean time per output token"
+                server_ttft_ms = _finite_number(response_request_metrics.get("ttft_ms"), nonnegative=True)
+                server_ttfp_ms = _finite_number(response_request_metrics.get("ttfp_ms"), nonnegative=True)
+                server_origins = response_request_metrics.get("measurement_origin")
+                if isinstance(server_origins, dict):
+                    if server_ttft_ms is not None and isinstance(server_origins.get("ttft"), str):
+                        resolved_measurement_origin["ttft"] = server_origins["ttft"]
+                    if server_ttfp_ms is not None and isinstance(server_origins.get("ttfp"), str):
+                        resolved_measurement_origin["ttfp"] = server_origins["ttfp"]
                 audio_duration_ms = (
                     max(cumulative_audio_ms)
                     if cumulative_audio_ms
@@ -1493,21 +1671,49 @@ class EventCollector:
                 # compute_audio_rtf — so metric definitions stay out of this
                 # dependency-free client module.
                 result["request_metrics"] = {
-                    "source": "client_monotonic_receive",
-                    "measurement_origin": measurement_origin
-                    or {
-                        "ttft": "input_audio_buffer.commit client send to first non-empty text delta",
-                        "ttfp": "input_audio_buffer.commit client send to first audio packet",
-                    },
+                    "source": (
+                        "server_request_start_and_client_receive"
+                        if server_ttft_ms is not None or server_ttfp_ms is not None
+                        else "client_monotonic_receive"
+                    ),
+                    "measurement_origin": resolved_measurement_origin,
                     "ttft_ms": (
-                        _rounded_ms((first_text_received_at_s - request_started_at_s) * 1000.0)
-                        if first_text_received_at_s is not None
-                        else None
+                        server_ttft_ms
+                        if server_ttft_ms is not None
+                        else (
+                            _rounded_ms((first_text_received_at_s - request_started_at_s) * 1000.0)
+                            if first_text_received_at_s is not None
+                            else None
+                        )
+                    ),
+                    "ttfp_ms": (
+                        server_ttfp_ms
+                        if server_ttfp_ms is not None
+                        else _rounded_ms((audio_received_at_s[0] - request_started_at_s) * 1000.0)
                     ),
-                    "ttfp_ms": _rounded_ms((audio_received_at_s[0] - request_started_at_s) * 1000.0),
                     "audio_generation_ms": _rounded_ms(audio_generation_ms),
                     "audio_duration_ms": _rounded_ms(audio_duration_ms),
                 }
+                if server_ttft_ms is not None or server_ttfp_ms is not None:
+                    request_metrics = result["request_metrics"]
+                    assert isinstance(request_metrics, dict)
+                    request_metrics["response_created_to_first_text_ms"] = (
+                        _rounded_ms((first_text_received_at_s - response_created_at_s) * 1000.0)
+                        if first_text_received_at_s is not None and response_created_at_s is not None
+                        else None
+                    )
+                    request_metrics["response_created_to_first_audio_ms"] = (
+                        _rounded_ms((audio_received_at_s[0] - response_created_at_s) * 1000.0)
+                        if response_created_at_s is not None
+                        else None
+                    )
+                if stage0_metrics is not None:
+                    request_metrics = result["request_metrics"]
+                    assert isinstance(request_metrics, dict)
+                    request_metrics["tpot_ms"] = _finite_number(
+                        stage0_metrics.get("vllm_tpot_ms"),
+                        nonnegative=True,
+                    )
         return result
 
     def _format_for_bytes(self) -> AudioFormat:
@@ -1640,30 +1846,38 @@ def summarize_session_request_metrics(
     *,
     session_id: str | None,
 ) -> dict[str, object]:
-    """Average client-observed metrics across turns that emitted audio.
+    """Summarize client- and engine-observed metrics across audio responses.
 
     ``request_metrics`` entries are caller-assembled dicts; keys that are
     absent or non-numeric in an entry are simply skipped. ``rtf`` is not
     produced by :meth:`EventCollector.timing_summary` (which reports raw data
-    only) — callers that want ``mean_rtf`` add an ``rtf`` value per turn,
-    e.g. via ``vllm_omni.metrics.definitions.compute_audio_rtf``.
+    only) — callers that want session ``rtf`` add an ``rtf`` value per turn,
+    e.g. via ``vllm_omni.metrics.definitions.compute_audio_rtf``. Zero or
+    missing ``tpot_ms`` values are omitted from session ``tpot_ms``.
+
+    Aggregatable fields are nested as ``{count, mean, p50, p99}``.
     """
 
-    def mean(metric: str, *, digits: int = 3) -> float | None:
-        values = [
+    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 math.isfinite(float(request[metric]))
+            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)
         ]
-        return round(sum(values) / len(values), digits) if values else None
 
-    return {
+    summary: dict[str, object] = {
         "session_id": session_id,
         "audio_turn_count": len(request_metrics),
-        "mean_ttft_ms": mean("ttft_ms"),
-        "mean_ttfp_ms": mean("ttfp_ms"),
-        "mean_rtf": mean("rtf", digits=6),
+        "ttft_ms": distribution_summary(values("ttft_ms")),
+        "ttfp_ms": distribution_summary(values("ttfp_ms")),
+        "rtf": distribution_summary(values("rtf"), digits=6),
     }
+    if (tpot := distribution_summary(values("tpot_ms", positive=True))) is not None:
+        summary["tpot_ms"] = tpot
+    return summary
 
 
 async def acknowledge_collected_playback(client: DuplexClient, collector: EventCollector) -> None:

vllm_omni/entrypoints/cli/benchmark/omni_duplex_eval.py

@@ -1,16 +1,29 @@
 # SPDX-License-Identifier: Apache-2.0
 # SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project
 
+"""CLI for Omni-DuplexEval generate / evaluate / summarize.
+
+``generate`` still writes per-sample timed sentences and ``*.meta.json``.
+Duplex performance metrics are written to ``<response-root>/duplex_metrics.json``.
+Skipped samples keep previously recorded rows; incoming ``(split, sample_id)``
+rows replace the matching ones, then ``duplex_stream_*`` is recomputed.
+"""
+
 import argparse
 import asyncio
 import concurrent.futures
 import json
 from pathlib import Path
 
-from vllm_omni.benchmarks.duplex.omni_duplex_eval_dataset import DEFAULT_DATASET, load_samples
+from vllm_omni.benchmarks.duplex.omni_duplex_eval_dataset import DEFAULT_DATASET, DuplexSample, load_samples
 from vllm_omni.benchmarks.duplex.omni_duplex_eval_eval import evaluate_sample, summarize_scores
 from vllm_omni.benchmarks.duplex.omni_duplex_eval_judge import DuplexJudge
-from vllm_omni.benchmarks.duplex.omni_duplex_eval_runner import generate_sample
+from vllm_omni.benchmarks.duplex.omni_duplex_eval_runner import GenerateSampleResult, generate_sample
+from vllm_omni.benchmarks.duplex_session_metrics import (
+    DUPLEX_METRICS_FILENAME,
+    merge_duplex_metrics_report,
+    read_duplex_metrics_report,
+)
 from vllm_omni.entrypoints.cli.benchmark.base import OmniBenchmarkSubcommandBase
 
 
@@ -54,6 +67,26 @@ def add_cli_args(parser: argparse.ArgumentParser) -> None:
     summarize.add_argument("--score-root", required=True)
 
 
+def _write_duplex_metrics(response_root: str | Path, results: list[GenerateSampleResult]) -> Path:
+    request_metrics = [metric for result in results for metric in result.request_metrics]
+    session_metrics = [result.session_metrics for result in results if result.session_metrics]
+    path = Path(response_root) / DUPLEX_METRICS_FILENAME
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text(
+        json.dumps(
+            merge_duplex_metrics_report(
+                read_duplex_metrics_report(path),
+                request_metrics=request_metrics,
+                session_metrics=session_metrics,
+            ),
+            ensure_ascii=False,
+            indent=2,
+        ),
+        encoding="utf-8",
+    )
+    return path
+
+
 def run(args: argparse.Namespace) -> int:
     if args.action == "summarize":
         print(json.dumps(summarize_scores(args.score_root), ensure_ascii=False, indent=2))
@@ -71,12 +104,12 @@ def run(args: argparse.Namespace) -> int:
         if args.concurrency < 1:
             raise ValueError("--concurrency must be at least 1")
 
-        async def generate() -> None:
+        async def generate() -> list[GenerateSampleResult]:
             semaphore = asyncio.Semaphore(args.concurrency)
 
-            async def generate_one(sample) -> None:
+            async def generate_one(sample: DuplexSample) -> GenerateSampleResult:
                 async with semaphore:
-                    await generate_sample(
+                    return await generate_sample(
                         sample,
                         url=args.url,
                         model=args.model,
@@ -89,16 +122,16 @@ def run(args: argparse.Namespace) -> int:
                         overwrite=args.overwrite,
                     )
 
-            await asyncio.gather(*(generate_one(sample) for sample in samples))
+            return list(await asyncio.gather(*(generate_one(sample) for sample in samples)))
 
-        asyncio.run(generate())
+        _write_duplex_metrics(args.response_root, asyncio.run(generate()))
         return 0
 
     if args.eval_workers < 1:
         raise ValueError("--eval-workers must be at least 1")
     judge = DuplexJudge(args.judge_base_url, args.judge_model, api_key=args.judge_api_key)
 
-    def evaluate_one(sample) -> None:
+    def evaluate_one(sample: DuplexSample) -> None:
         response_path = Path(args.response_root) / sample.split / f"{sample.id}.json"
         score_path = Path(args.score_root) / sample.split / f"{sample.id}.json"
         if score_path.exists() and not args.overwrite:

vllm_omni/entrypoints/duplex/protocol.py

@@ -413,6 +413,10 @@ class ResponseState:
     active_request_id: str | None = None
     active_response_id: str | None = None
     active_response_turn_id: int | None = None
+    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
     active_response_input_commit_seq: int | None = None
     active_response_awaits_input_commit: bool = False
     last_response_id: str | None = None
@@ -656,6 +660,45 @@ class DuplexSession:
     def bind_response_turn(self, turn_id: int | None) -> None:
         self._response.active_response_turn_id = turn_id
 
+    def mark_model_turn_request_started(self, turn_id: int, started_at_s: float) -> None:
+        """Record the latest native request start that can own one model turn."""
+        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 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)
+        if has_text and self._response.active_response_ttft_ms is None:
+            self._response.active_response_ttft_ms = elapsed_ms
+        if has_audio and self._response.active_response_ttfp_ms is None:
+            self._response.active_response_ttfp_ms = elapsed_ms
+        metrics: dict[str, object] = {
+            "source": "server_monotonic_request_start",
+            "measurement_origin": {
+                "ttft": "native model-turn request execution start to first non-empty text output",
+                "ttfp": "native model-turn request execution start to first audio output",
+            },
+        }
+        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 active_response_accepts_model_turn(self, turn_id: int | None) -> bool:
         if self._response.active_response_id is None:
             return False
@@ -940,6 +983,11 @@ class DuplexSession:
     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
 
@@ -976,6 +1024,12 @@ class DuplexSession:
         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._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_ttft_ms = None
+        self._response.active_response_ttfp_ms = 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
@@ -1057,17 +1111,22 @@ class DuplexSession:
 
             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
-                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
-                )
-                self._response.stage_metric_tpot_weight[stage_id] = (
-                    self._response.stage_metric_tpot_weight.get(stage_id, 0) + weight
-                )
-                current["vllm_tpot_ms"] = self._response.stage_metric_tpot_weighted_ms[stage_id] / float(
-                    self._response.stage_metric_tpot_weight[stage_id]
+            if isinstance(tpot_ms, int | float) and not isinstance(tpot_ms, bool) and tpot_ms > 0:
+                weight = (
+                    max(int(token_count) - 1, 0)
+                    if isinstance(token_count, int | float) and not isinstance(token_count, bool)
+                    else 1
                 )
+                if weight > 0:
+                    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
+                    )
+                    self._response.stage_metric_tpot_weight[stage_id] = (
+                        self._response.stage_metric_tpot_weight.get(stage_id, 0) + weight
+                    )
+                    current["vllm_tpot_ms"] = self._response.stage_metric_tpot_weighted_ms[stage_id] / float(
+                        self._response.stage_metric_tpot_weight[stage_id]
+                    )
 
             for name, value in raw_values.items():
                 if name not in handled_fields:
@@ -1272,6 +1331,9 @@ class DuplexSession:
             self._response.active_request_id = None
         self._response.active_response_id = None
         self._response.active_response_turn_id = None
+        self._response.active_response_request_started_at_s = None
+        self._response.active_response_ttft_ms = None
+        self._response.active_response_ttfp_ms = None
         self._response.active_response_input_commit_seq = None
         self._response.active_response_awaits_input_commit = False
         self._clear_response_metrics()
@@ -1589,6 +1651,10 @@ class DuplexSession:
         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._response.active_response_request_started_at_s = None
+        self._response.active_response_ttft_ms = None
+        self._response.active_response_ttfp_ms = None
         self._response.active_response_input_commit_seq = None
         self._response.active_response_awaits_input_commit = False
         self._clear_response_metrics()
@@ -1604,6 +1670,10 @@ class DuplexSession:
         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._response.active_response_request_started_at_s = None
+        self._response.active_response_ttft_ms = None
+        self._response.active_response_ttfp_ms = None
         self._response.active_response_input_commit_seq = None
         self._response.active_response_awaits_input_commit = False
         self.clear_server_vad_audio()

vllm_omni/entrypoints/duplex/runtime_bridge.py

@@ -6,6 +6,7 @@ from __future__ import annotations
 import asyncio
 import base64
 import inspect
+import time
 from collections.abc import Mapping
 from dataclasses import asdict, is_dataclass
 from enum import Enum
@@ -1113,13 +1114,20 @@ class NativeRuntimeBridgeMixin:
         if response_id is None:
             response_id = session.begin_response(turn_id=model_turn_id)
             response_created = True
-            await send_json(
-                self._response_created_payload(
-                    session,
-                    response_id,
-                    epoch=session.epoch,
-                )
+        response_request_metrics = session.mark_response_first_outputs(
+            observed_at_s=time.monotonic(),
+            has_text=has_text,
+            has_audio=has_audio,
+        )
+        if response_created:
+            response_created_payload = self._response_created_payload(
+                session,
+                response_id,
+                epoch=session.epoch,
             )
+            if response_request_metrics:
+                response_created_payload["response_request_metrics"] = response_request_metrics
+            await send_json(response_created_payload)
         response_stage_metrics = session.accumulate_response_stage_metrics(
             native_result.get("stage_metrics") if isinstance(native_result.get("stage_metrics"), Mapping) else None
         )
@@ -1137,6 +1145,7 @@ class NativeRuntimeBridgeMixin:
                 speak_payload,
                 native_result,
                 stage_metrics=response_stage_metrics,
+                response_request_metrics=response_request_metrics,
             )
             await send_json(speak_payload)
         previous_sent_ms = session.playback.sent_ms
@@ -1205,6 +1214,7 @@ class NativeRuntimeBridgeMixin:
             payload,
             native_result,
             stage_metrics=response_stage_metrics,
+            response_request_metrics=response_request_metrics,
         )
         await send_json(payload)
         if (
@@ -1365,6 +1375,7 @@ class NativeRuntimeBridgeMixin:
         native_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 = native_result.get("runtime_impl")
@@ -1390,6 +1401,8 @@ class NativeRuntimeBridgeMixin:
                 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/entrypoints/duplex/session_runner.py

@@ -6,6 +6,7 @@ from __future__ import annotations
 import asyncio
 import binascii
 import json
+import time
 import uuid
 from collections.abc import Callable
 from contextlib import suppress
@@ -494,6 +495,7 @@ class DuplexSessionRunnerMixin:
             async def _run() -> bool:
                 nonlocal runtime_closed
                 try:
+                    session.mark_model_turn_request_started(append_turn_id, time.monotonic())
                     append_ok, emitted_response = await self._append_runtime_input(
                         session,
                         payload,

vllm_omni/outputs/output_processor.py

@@ -59,6 +59,10 @@ def _modality_to_type_string(modality: OutputModality) -> str:
     return "text"
 
 
+_TPOT_ELAPSED_MS = "_tpot_elapsed_ms"
+_TPOT_INTERVALS = "_tpot_intervals"
+
+
 def _mean_time_per_output_token_ms(stats: RequestStateStats) -> float:
     output_intervals = stats.num_generation_tokens - 1
     if output_intervals <= 0:
@@ -67,6 +71,17 @@ def _mean_time_per_output_token_ms(stats: RequestStateStats) -> float:
     return decode_time_s * 1000.0 / float(output_intervals)
 
 
+def _accumulate_segment_tpot(record: dict[str, object], *, elapsed_ms: float, new_tokens: int) -> None:
+    """Weight one decode step by its token count. ITL stays one sample per step."""
+    if new_tokens <= 0:
+        return
+    total_elapsed_ms = float(record.get(_TPOT_ELAPSED_MS) or 0.0) + elapsed_ms
+    total_intervals = int(record.get(_TPOT_INTERVALS) or 0) + new_tokens
+    record[_TPOT_ELAPSED_MS] = total_elapsed_ms
+    record[_TPOT_INTERVALS] = total_intervals
+    record["vllm_tpot_ms"] = total_elapsed_ms / float(total_intervals)
+
+
 class OmniRequestState(RequestState):
     """Request state for omni models, tracking multimodal outputs.
 
@@ -430,7 +445,10 @@ class MultimodalOutputProcessor(VLLMOutputProcessor):
         )
 
     def pop_native_text_metrics(self, request_id: str) -> dict[str, Any]:
-        return self._native_text_metrics_by_request.pop(request_id, {})
+        record = self._native_text_metrics_by_request.pop(request_id, {})
+        record.pop(_TPOT_ELAPSED_MS, None)
+        record.pop(_TPOT_INTERVALS, None)
+        return record
 
     def abort_requests(self, request_ids, internal: bool) -> list[str]:
         aborted_ids, _outputs = self.abort_requests_collecting_outputs(request_ids, internal=internal)
@@ -770,6 +788,7 @@ class MultimodalOutputProcessor(VLLMOutputProcessor):
         was_prefilling = req_state.is_prefilling
         native_stats = req_state.native_text_stats if isinstance(req_state, OmniRequestState) else None
         previous_last_token_ts = native_stats.last_token_ts if native_stats is not None else 0.0
+        previous_num_generation_tokens = int(native_stats.num_generation_tokens) if native_stats is not None else 0
 
         # NOTE: We pass ``None`` for  *iteration_stats* to the parent so that
         # the upstream's ``_update_stats_from_output`` logs stats via its own
@@ -808,7 +827,13 @@ class MultimodalOutputProcessor(VLLMOutputProcessor):
             itls_ms = record.setdefault("vllm_itls_ms", [])
             itls_ms.append(itl_ms)
             record["vllm_itl_ms"] = sum(itls_ms) / float(len(itls_ms))
-        record["vllm_tpot_ms"] = _mean_time_per_output_token_ms(native_stats)
+            _accumulate_segment_tpot(
+                record,
+                elapsed_ms=itl_ms,
+                new_tokens=max(int(native_stats.num_generation_tokens) - previous_num_generation_tokens, 0),
+            )
+        elif not record["vllm_itls_ms"]:
+            record["vllm_tpot_ms"] = _mean_time_per_output_token_ms(native_stats)
 
     def _update_stats_from_finished(
         self,
@@ -836,6 +861,7 @@ class MultimodalOutputProcessor(VLLMOutputProcessor):
         finished_request = iteration_stats.finished_requests[-1]
         if finished_request.request_id != req_state.external_req_id:
             return
-        self._native_text_metric_record(req_state.external_req_id)["vllm_tpot_ms"] = (
-            float(finished_request.mean_time_per_output_token) * 1000.0
-        )
+        finished_tpot_ms = float(finished_request.mean_time_per_output_token) * 1000.0
+        record = self._native_text_metric_record(req_state.external_req_id)
+        if finished_tpot_ms > 0 and not record.get("vllm_itls_ms"):
+            record["vllm_tpot_ms"] = finished_tpot_ms