diff --git a/benchmarks/tts/README.md b/benchmarks/tts/README.md
index c403e5dd0..79a182174 100644
--- a/benchmarks/tts/README.md
+++ b/benchmarks/tts/README.md
@@ -280,3 +280,28 @@ benchmarks/tts/
 - Upstream seed-tts-eval integration: vllm-project/vllm-omni#2558
 - Pipeline + Deploy schema: vllm-project/vllm-omni#2383
 - Concurrency cliff RFC: vllm-project/vllm-omni#272
+
+## Offline streaming latency (`stream_latency_bench.py`)
+
+Measures chunk-streamed generation through `AsyncOmni` without a server, reusing the
+Qwen3-TTS example's prompt builder. Per request it records every chunk arrival and derives:
+
+| Metric | Definition |
+|---|---|
+| `ttfa_ms` | arrival of the first audio-carrying chunk |
+| `playback_start_ms` | earliest underrun-free playback start, `max_i (t_i - audio_before_i)` |
+| `stall_at_ttfa_ms` | total stall if playback starts at TTFA (the 1→25-frame cliff on the default schedule) |
+| `rtf_streamed` / `rtf_total` | wall time over streamed audio / over streamed + tail audio |
+| `gap_ms` | inter-chunk arrival gaps |
+
+```bash
+# one GPU through the scheduler; every example flag is passed through
+gpu run --gpus 1 --timeout 40m --note "stream latency 0.6B" -- env HF_HUB_OFFLINE=1 \
+  python benchmarks/tts/stream_latency_bench.py --model Qwen/Qwen3-TTS-12Hz-0.6B-CustomVoice \
+  --query-type CustomVoice --txt-prompts benchmarks/tts/prompts_12.txt --warmup 1 --repeat 3 \
+  --init-timeout 1500 --stage-init-timeout 900 --tag baseline
+```
+
+Results are written under `benchmarks/tts/edge_results/stream_latency/<model>/<deploy>/`.
+`--init-only` initializes, runs the warm-ups and exits (used by `init_profile.py`);
+`VLLM_OMNI_INIT_TIMELINE=<path>` embeds the initialization timeline JSON when available.
diff --git a/benchmarks/tts/edge_results/README.md b/benchmarks/tts/edge_results/README.md
new file mode 100644
index 000000000..46bb779c1
--- /dev/null
+++ b/benchmarks/tts/edge_results/README.md
@@ -0,0 +1,9 @@
+# Edge benchmark results
+
+Committed JSON produced by the edge branch benchmarks (kept small; WAVs are never committed):
+
+- `stream_latency/<model>/<deploy>/<timestamp>[_tag].json` — `stream_latency_bench.py` output (schema 1): `meta` (model, deploy config, git SHA, GPU), `init` (`init_s`, optional init timeline), `resources` (peak RSS, peak GPU MiB, sampling mode), `requests[]` (per-chunk arrivals, TTFA, playback-start, stall at TTFA, RTF streamed / total) and `summary` (p50/p90/max).
+- `init_profile/<timestamp>/summary.json` — `init_profile.py` comparison of initialization phases per configuration.
+- `hw_emulation/`, `scheduling/` — per-work-package comparison tables.
+
+Metric definitions live in `benchmarks/tts/stream_latency_metrics.py`.
diff --git a/benchmarks/tts/prompts_12.txt b/benchmarks/tts/prompts_12.txt
new file mode 100644
index 000000000..92577a418
--- /dev/null
+++ b/benchmarks/tts/prompts_12.txt
@@ -0,0 +1,12 @@
+Hello, welcome to the voice synthesis benchmark test.
+She said she would be here by noon, but nobody showed up.
+The quick brown fox jumps over the lazy dog near the riverbank.
+I can't believe how beautiful the sunset looks from up here on the mountain.
+Please remember to bring your identification documents to the appointment tomorrow morning.
+Have you ever wondered what it would be like to travel through time and visit ancient civilizations?
+The restaurant on the corner serves the best pasta I have ever tasted in my entire life.
+After the meeting, we should discuss the quarterly results and plan for the next phase.
+Learning a new language takes patience, practice, and a genuine curiosity about other cultures.
+The train leaves at half past seven, so we need to arrive at the station before then.
+Could you please turn down the music a little bit, I'm trying to concentrate on my work.
+It was a dark and stormy night when the old lighthouse keeper heard a knock at the door.
diff --git a/benchmarks/tts/stream_latency_bench.py b/benchmarks/tts/stream_latency_bench.py
new file mode 100644
index 000000000..1a9b9306d
--- /dev/null
+++ b/benchmarks/tts/stream_latency_bench.py
@@ -0,0 +1,388 @@
+#!/usr/bin/env python3
+"""Offline streaming-latency benchmark for chunk-streamed TTS pipelines (AsyncOmni).
+
+Reuses the Qwen3-TTS example's input builder so prompts/templates are identical to
+``examples/offline_inference/text_to_speech/qwen3_tts/end2end.py``; every flag that
+example accepts is passed through (``--query-type``, ``--txt-prompts``,
+``--deploy-config``, ``--init-timeout``, ``--stage-init-timeout``, ...).
+
+Bench-only flags:
+  --model            override the model id chosen by the example (e.g. the 0.6B variants)
+  --warmup N         warm-up requests (excluded from statistics), default 1
+  --repeat K         repeat the prompt list K times, default 1
+  --init-only        initialize the engine, run the warm-ups, write JSON, exit
+  --tag              free-form tag appended to the result file name
+  --output-dir       results root (default benchmarks/tts/edge_results/stream_latency)
+  --save-wav         also write the streamed audio per request
+  --gpu-poll-interval-s  resource poller interval (default 0.2)
+
+Metrics per request: TTFA, chunk arrivals, uninterrupted-playback start, stall at
+TTFA, RTF (streamed and streamed+tail), see ``stream_latency_metrics.py``.
+
+Example (one GPU through the scheduler):
+  gpu run --gpus 1 --timeout 40m --note "wp0 baseline" -- env HF_HUB_OFFLINE=1 \\
+    python benchmarks/tts/stream_latency_bench.py --model Qwen/Qwen3-TTS-12Hz-0.6B-CustomVoice \\
+    --query-type CustomVoice --txt-prompts benchmarks/tts/prompts_12.txt --warmup 1 --repeat 3
+"""
+
+from __future__ import annotations
+
+import argparse
+import asyncio
+import datetime as _dt
+import json
+import os
+import subprocess
+import sys
+import threading
+import time
+from pathlib import Path
+from typing import Any
+
+import torch
+
+BENCH_DIR = Path(__file__).resolve().parent
+REPO_ROOT = BENCH_DIR.parent.parent
+EXAMPLE_DIR = REPO_ROOT / "examples" / "offline_inference" / "text_to_speech" / "qwen3_tts"
+sys.path.insert(0, str(BENCH_DIR))
+sys.path.insert(0, str(EXAMPLE_DIR))
+
+import stream_latency_metrics as metrics  # noqa: E402
+
+SCHEMA_VERSION = 1
+
+
+# --------------------------------------------------------------------------- args
+def parse_bench_args(argv: list[str]) -> tuple[argparse.Namespace, list[str]]:
+    p = argparse.ArgumentParser(add_help=False)
+    p.add_argument("--model", default=None)
+    p.add_argument("--warmup", type=int, default=1)
+    p.add_argument("--repeat", type=int, default=1)
+    p.add_argument("--init-only", action="store_true")
+    p.add_argument("--tag", default="")
+    p.add_argument("--output-dir", default=str(BENCH_DIR / "edge_results" / "stream_latency"))
+    p.add_argument("--save-wav", action="store_true")
+    p.add_argument("--gpu-poll-interval-s", type=float, default=0.2)
+    p.add_argument("--parallel-stage-init", action="store_true", help="Pass parallel_stage_init=True to the engine")
+    p.add_argument("--deploy-profile", default=None, help="Named deploy profile (e.g. edge); forwarded to Omni")
+    p.add_argument("--bench-help", action="store_true")
+    bench, rest = p.parse_known_args(argv)
+    if bench.bench_help:
+        p.print_help()
+        sys.exit(0)
+    return bench, rest
+
+
+def build_inputs(example_args: argparse.Namespace, model_override: str | None) -> tuple[str, list[dict[str, Any]]]:
+    import end2end  # example module, imported lazily so --bench-help works without deps
+
+    model_name, inputs = end2end._build_inputs(example_args)
+    if model_override and model_override != model_name:
+        model_name = model_override
+        rebuilt = []
+        for item in inputs:
+            info = item["additional_information"]
+            rebuilt.append(
+                {
+                    "prompt_token_ids": [0] * end2end._estimate_prompt_len(info, model_name),
+                    "additional_information": info,
+                }
+            )
+        inputs = rebuilt
+    return model_name, inputs
+
+
+# ----------------------------------------------------------------- resources
+class ResourcePoller(threading.Thread):
+    """Samples RSS of this process tree and GPU memory of its processes."""
+
+    def __init__(self, interval_s: float = 0.2):
+        super().__init__(daemon=True)
+        self.interval_s = interval_s
+        self._stop_evt = threading.Event()
+        self.peak_rss_bytes = 0
+        self.peak_gpu_mem_mib = 0
+        self.gpu_mem_samples = 0
+        self.gpu_mem_mode = "none"
+        try:
+            import psutil
+
+            self._proc = psutil.Process()
+        except Exception:  # pragma: no cover - psutil is a vllm dependency
+            self._proc = None
+
+    def _pids(self) -> set[int]:
+        if self._proc is None:
+            return {os.getpid()}
+        try:
+            return {self._proc.pid, *(c.pid for c in self._proc.children(recursive=True))}
+        except Exception:
+            return {os.getpid()}
+
+    def _sample_rss(self, pids: set[int]) -> int:
+        if self._proc is None:
+            return 0
+        import psutil
+
+        total = 0
+        for pid in pids:
+            try:
+                total += psutil.Process(pid).memory_info().rss
+            except Exception:
+                continue
+        return total
+
+    def _sample_gpu(self, pids: set[int]) -> int | None:
+        try:
+            out = subprocess.run(
+                ["nvidia-smi", "--query-compute-apps=pid,used_memory", "--format=csv,noheader,nounits"],
+                capture_output=True,
+                text=True,
+                timeout=5,
+            )
+        except Exception:
+            return None
+        if out.returncode != 0:
+            return None
+        total = 0
+        matched = False
+        for line in out.stdout.strip().splitlines():
+            parts = [x.strip() for x in line.split(",")]
+            if len(parts) != 2:
+                continue
+            try:
+                pid, mem = int(parts[0]), int(parts[1])
+            except ValueError:
+                continue
+            if pid in pids:
+                total += mem
+                matched = True
+        if matched:
+            self.gpu_mem_mode = "per-process"
+            return total
+        # fallback: whole-GPU usage of the first visible device
+        idx = (os.environ.get("CUDA_VISIBLE_DEVICES", "0").split(",")[0] or "0").strip()
+        try:
+            out = subprocess.run(
+                ["nvidia-smi", "--query-gpu=memory.used", "--format=csv,noheader,nounits", "-i", idx],
+                capture_output=True,
+                text=True,
+                timeout=5,
+            )
+            if out.returncode == 0 and out.stdout.strip():
+                self.gpu_mem_mode = "whole-gpu"
+                return int(out.stdout.strip().splitlines()[0])
+        except Exception:
+            return None
+        return None
+
+    def run(self) -> None:
+        while not self._stop_evt.is_set():
+            pids = self._pids()
+            self.peak_rss_bytes = max(self.peak_rss_bytes, self._sample_rss(pids))
+            gpu = self._sample_gpu(pids)
+            if gpu is not None:
+                self.gpu_mem_samples += 1
+                self.peak_gpu_mem_mib = max(self.peak_gpu_mem_mib, gpu)
+            self._stop_evt.wait(self.interval_s)
+
+    def stop(self) -> dict[str, Any]:
+        self._stop_evt.set()
+        self.join(timeout=5)
+        return {
+            "peak_rss_bytes": self.peak_rss_bytes,
+            "peak_gpu_mem_mib": self.peak_gpu_mem_mib,
+            "gpu_mem_samples": self.gpu_mem_samples,
+            "gpu_mem_mode": self.gpu_mem_mode,
+        }
+
+
+def _git_sha(path: Path) -> str | None:
+    try:
+        return (
+            subprocess.run(
+                ["git", "-C", str(path), "rev-parse", "--short", "HEAD"], capture_output=True, text=True, timeout=5
+            ).stdout.strip()
+            or None
+        )
+    except Exception:
+        return None
+
+
+def _gpu_name() -> str | None:
+    try:
+        out = subprocess.run(
+            ["nvidia-smi", "--query-gpu=name", "--format=csv,noheader"], capture_output=True, text=True, timeout=5
+        )
+        return out.stdout.strip().splitlines()[0] if out.returncode == 0 and out.stdout.strip() else None
+    except Exception:
+        return None
+
+
+# --------------------------------------------------------------------- audio
+def _cat_audio(audio: Any) -> torch.Tensor | None:
+    if audio is None:
+        return None
+    if isinstance(audio, list):
+        parts = [a for a in audio if a is not None]
+        if not parts:
+            return None
+        audio = torch.cat([torch.as_tensor(a).reshape(-1) for a in parts], dim=-1)
+    return torch.as_tensor(audio).reshape(-1).float().cpu()
+
+
+def _sample_rate(mm: dict[str, Any] | None, default: int = 24000) -> int:
+    if not mm or mm.get("sr") is None:
+        return default
+    s = mm["sr"]
+    s = s[-1] if isinstance(s, list) and s else s
+    return int(s.item() if hasattr(s, "item") else s)
+
+
+# ----------------------------------------------------------------------- run
+async def run_request(omni: Any, prompt: dict[str, Any], request_id: str) -> tuple[dict[str, Any], torch.Tensor | None]:
+    t_start = time.perf_counter()
+    t_prev = t_start
+    chunks: list[dict[str, Any]] = []
+    streamed: list[torch.Tensor] = []
+    tail: torch.Tensor | None = None
+    sr = 24000
+    t_end = t_start
+    async for so in omni.generate(prompt, request_id=request_id):
+        now = time.perf_counter()
+        mm = so.outputs[0].multimodal_output if so.outputs else None
+        sr = _sample_rate(mm, sr)
+        audio = _cat_audio(mm.get("audio") if mm else None)
+        if not so.finished:
+            n = 0 if audio is None else int(audio.numel())
+            chunks.append(
+                {"idx": len(chunks), "t_ms": (now - t_start) * 1e3, "gap_ms": (now - t_prev) * 1e3, "samples": n}
+            )
+            if audio is not None and n:
+                streamed.append(audio)
+            t_prev = now
+        else:
+            t_end = now
+            tail = audio
+    total_wall_s = t_end - t_start
+    tail_s = (tail.numel() / sr) if tail is not None else 0.0
+    rec = {
+        "request_id": request_id,
+        "sr": sr,
+        "chunks": chunks,
+        "t_final_ms": total_wall_s * 1e3,
+        **metrics.request_metrics(chunks, sr, total_wall_s, tail_audio_s=tail_s),
+    }
+    text = prompt.get("additional_information", {}).get("text")
+    rec["text"] = text[0] if isinstance(text, list) and text else text
+    audio_out = torch.cat(streamed) if streamed else None
+    return rec, audio_out
+
+
+async def main() -> int:
+    bench, rest = parse_bench_args(sys.argv[1:])
+    sys.argv = [sys.argv[0], *rest]
+    import end2end
+
+    example_args = end2end.parse_args()
+    model_name, inputs = build_inputs(example_args, bench.model)
+
+    omni_kwargs = vars(example_args).copy()
+    omni_kwargs["model"] = model_name
+    if bench.parallel_stage_init:
+        omni_kwargs["parallel_stage_init"] = True
+    if bench.deploy_profile:
+        omni_kwargs["deploy_profile"] = bench.deploy_profile
+    from vllm_omni import AsyncOmni
+
+    poller = ResourcePoller(bench.gpu_poll_interval_s)
+    poller.start()
+    t0 = time.perf_counter()
+    omni = AsyncOmni(**omni_kwargs)
+    init_s = time.perf_counter() - t0
+
+    results: list[dict[str, Any]] = []
+    wavs: dict[str, torch.Tensor] = {}
+    plan: list[tuple[str, dict[str, Any]]] = []
+    for i in range(bench.warmup):
+        plan.append((f"w{i}", inputs[i % len(inputs)]))
+    if not bench.init_only:
+        for r in range(bench.repeat):
+            for i, prompt in enumerate(inputs):
+                plan.append((f"r{r}_{i}", prompt))
+    for rid, prompt in plan:
+        rec, audio = await run_request(omni, prompt, rid)
+        rec["warmup"] = rid.startswith("w")
+        results.append(rec)
+        if audio is not None:
+            wavs[rid] = audio
+        print(
+            f"[bench] {rid}: TTFA={rec['ttfa_ms']} ms playback_start={rec['playback_start_ms']} ms "
+            f"stall@ttfa={rec['stall_at_ttfa_ms']} ms chunks={rec['n_chunks']} audio={rec['streamed_audio_s']:.2f}s "
+            f"wall={rec['total_wall_s']:.3f}s rtf_total={rec['rtf_total']}",
+            flush=True,
+        )
+    resources = poller.stop()
+    measured = [r for r in results if not r["warmup"]]
+    summary = metrics.summarize(measured)
+
+    timeline = None
+    tl_path = os.environ.get("VLLM_OMNI_INIT_TIMELINE")
+    if tl_path and os.path.exists(tl_path):
+        try:
+            with open(tl_path) as f:
+                timeline = json.load(f)
+        except Exception:
+            timeline = None
+
+    deploy_cfg = getattr(example_args, "deploy_config", None)
+    model_tag = model_name.split("/")[-1]
+    deploy_tag = (
+        Path(deploy_cfg).stem
+        if deploy_cfg
+        else (f"profile_{bench.deploy_profile}" if bench.deploy_profile else "default")
+    )
+    ts = _dt.datetime.now().strftime("%Y%m%d-%H%M%S")
+    out_dir = Path(bench.output_dir) / model_tag / deploy_tag
+    out_dir.mkdir(parents=True, exist_ok=True)
+    out_path = out_dir / (f"{ts}_{bench.tag}.json" if bench.tag else f"{ts}.json")
+    doc = {
+        "schema": SCHEMA_VERSION,
+        "meta": {
+            "model": model_name,
+            "deploy_config": deploy_cfg,
+            "query_type": getattr(example_args, "query_type", None),
+            "txt_prompts": getattr(example_args, "txt_prompts", None),
+            "git_sha": _git_sha(REPO_ROOT),
+            "gpu_name": _gpu_name(),
+            "cuda_visible_devices": os.environ.get("CUDA_VISIBLE_DEVICES"),
+            "timestamp": ts,
+            "argv": sys.argv[1:],
+            "bench_args": vars(bench),
+            "warmup": bench.warmup,
+            "repeat": bench.repeat,
+            "init_only": bench.init_only,
+        },
+        "init": {"init_s": init_s, "init_timeline": timeline},
+        "resources": resources,
+        "requests": results,
+        "summary": summary,
+    }
+    with open(out_path, "w") as f:
+        json.dump(doc, f, indent=1)
+    if bench.save_wav:
+        import soundfile as sf
+
+        for rid, audio in wavs.items():
+            sf.write(str(out_dir / f"{ts}_{rid}.wav"), audio.numpy(), results[0]["sr"], format="WAV")
+    print(f"[bench] init_s={init_s:.1f} resources={resources}", flush=True)
+    print(f"[bench] SUMMARY {json.dumps(summary)}", flush=True)
+    print(f"[bench] wrote {out_path}", flush=True)
+    shutdown = getattr(omni, "shutdown", None)
+    if callable(shutdown):
+        shutdown()
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(asyncio.run(main()))
diff --git a/benchmarks/tts/stream_latency_metrics.py b/benchmarks/tts/stream_latency_metrics.py
new file mode 100644
index 000000000..8e012c990
--- /dev/null
+++ b/benchmarks/tts/stream_latency_metrics.py
@@ -0,0 +1,160 @@
+"""Pure-Python metrics for streaming TTS latency (no torch / engine imports).
+
+A streamed utterance is a list of chunk records ``{"t_ms": float, "samples": int}``
+where ``t_ms`` is the arrival time of the chunk relative to request submission and
+``samples`` is the number of PCM samples it carries at sample rate ``sr``.
+
+Definitions
+-----------
+TTFA
+    Arrival time of the first chunk that carries audio.
+playback_start_ms
+    The earliest time a player could start such that it never underruns given the
+    observed arrivals: ``max_i (t_i - C_i)`` where ``C_i`` is the audio duration
+    (ms) accumulated before chunk ``i``. Always ``>= TTFA``.
+stall_at_ttfa_ms
+    Total stall a player would suffer if it started at TTFA instead of
+    ``playback_start_ms`` (this is the "1 -> 25 frame cliff" measured on the
+    default Qwen3-TTS schedule).
+RTF
+    Wall time divided by generated audio seconds; lower is faster than real time.
+"""
+
+from __future__ import annotations
+
+import math
+from typing import Any
+
+
+def _audio_chunks(chunks: list[dict[str, Any]]) -> list[dict[str, Any]]:
+    return [c for c in chunks if int(c.get("samples", 0)) > 0]
+
+
+def chunk_durations_ms(chunks: list[dict[str, Any]], sr: int) -> list[float]:
+    """Audio duration in ms of each audio-carrying chunk, in arrival order."""
+    return [1000.0 * int(c["samples"]) / sr for c in _audio_chunks(chunks)]
+
+
+def cumulative_audio_before_ms(chunks: list[dict[str, Any]], sr: int) -> list[float]:
+    """``C_i``: audio (ms) delivered before chunk ``i`` (audio chunks only)."""
+    out: list[float] = []
+    acc = 0.0
+    for d in chunk_durations_ms(chunks, sr):
+        out.append(acc)
+        acc += d
+    return out
+
+
+def ttfa_ms(chunks: list[dict[str, Any]]) -> float | None:
+    audio = _audio_chunks(chunks)
+    return float(audio[0]["t_ms"]) if audio else None
+
+
+def playback_start_ms(chunks: list[dict[str, Any]], sr: int) -> float | None:
+    """Earliest underrun-free playback start; ``None`` when no audio arrived."""
+    audio = _audio_chunks(chunks)
+    if not audio:
+        return None
+    before = cumulative_audio_before_ms(chunks, sr)
+    return max(float(c["t_ms"]) - b for c, b in zip(audio, before))
+
+
+def underruns_if_start_at(chunks: list[dict[str, Any]], sr: int, start_ms: float) -> list[tuple[int, float]]:
+    """Stalls a player starting at ``start_ms`` would suffer: ``[(chunk_index, stall_ms)]``.
+
+    After each stall the playback cursor shifts by the stall, so later chunks are
+    judged against the shifted schedule (this models a real ring-buffer player).
+    """
+    audio = _audio_chunks(chunks)
+    before = cumulative_audio_before_ms(chunks, sr)
+    stalls: list[tuple[int, float]] = []
+    cursor = float(start_ms)
+    for i, (c, b) in enumerate(zip(audio, before)):
+        needed_at = cursor + b
+        t = float(c["t_ms"])
+        if t > needed_at + 1e-9:
+            stall = t - needed_at
+            stalls.append((i, stall))
+            cursor += stall
+    return stalls
+
+
+def stall_at_ttfa_ms(chunks: list[dict[str, Any]], sr: int) -> float | None:
+    t0 = ttfa_ms(chunks)
+    if t0 is None:
+        return None
+    return sum(s for _, s in underruns_if_start_at(chunks, sr, t0))
+
+
+def audio_seconds(chunks: list[dict[str, Any]], sr: int) -> float:
+    return sum(int(c["samples"]) for c in _audio_chunks(chunks)) / sr
+
+
+def rtf(total_wall_s: float, audio_s: float) -> float | None:
+    if audio_s <= 0:
+        return None
+    return total_wall_s / audio_s
+
+
+def inter_chunk_gaps_ms(chunks: list[dict[str, Any]]) -> list[float]:
+    """Arrival gaps between consecutive audio chunks (excludes the first chunk)."""
+    audio = _audio_chunks(chunks)
+    return [float(b["t_ms"]) - float(a["t_ms"]) for a, b in zip(audio, audio[1:])]
+
+
+def percentile(values: list[float], q: float) -> float | None:
+    """Nearest-rank percentile (``q`` in [0, 1]); ``None`` for an empty list."""
+    if not values:
+        return None
+    ordered = sorted(values)
+    rank = max(1, math.ceil(q * len(ordered)))
+    return ordered[min(rank, len(ordered)) - 1]
+
+
+def _stats(values: list[float]) -> dict[str, float | None]:
+    vals = [float(v) for v in values if v is not None]
+    return {
+        "n": len(vals),
+        "p50": percentile(vals, 0.50),
+        "p90": percentile(vals, 0.90),
+        "max": max(vals) if vals else None,
+        "mean": (sum(vals) / len(vals)) if vals else None,
+    }
+
+
+def request_metrics(
+    chunks: list[dict[str, Any]], sr: int, total_wall_s: float, tail_audio_s: float = 0.0
+) -> dict[str, Any]:
+    """Derive all per-request metrics from a chunk timeline."""
+    streamed_s = audio_seconds(chunks, sr)
+    t_last = max((float(c["t_ms"]) for c in _audio_chunks(chunks)), default=None)
+    return {
+        "ttfa_ms": ttfa_ms(chunks),
+        "playback_start_ms": playback_start_ms(chunks, sr),
+        "stall_at_ttfa_ms": stall_at_ttfa_ms(chunks, sr),
+        "n_chunks": len(_audio_chunks(chunks)),
+        "streamed_audio_s": streamed_s,
+        "tail_audio_s": tail_audio_s,
+        "total_wall_s": total_wall_s,
+        "rtf_streamed": rtf(t_last / 1000.0, streamed_s) if t_last is not None else None,
+        "rtf_total": rtf(total_wall_s, streamed_s + tail_audio_s),
+        "gap_ms": inter_chunk_gaps_ms(chunks),
+    }
+
+
+def summarize(records: list[dict[str, Any]]) -> dict[str, Any]:
+    """Aggregate per-request metric dicts (as produced by :func:`request_metrics`)."""
+    gaps: list[float] = []
+    for r in records:
+        gaps.extend(r.get("gap_ms") or [])
+    return {
+        "n_requests": len(records),
+        "ttfa_ms": _stats([r.get("ttfa_ms") for r in records]),
+        "playback_start_ms": _stats([r.get("playback_start_ms") for r in records]),
+        "stall_at_ttfa_ms": _stats([r.get("stall_at_ttfa_ms") for r in records]),
+        "rtf_streamed": _stats([r.get("rtf_streamed") for r in records]),
+        "rtf_total": _stats([r.get("rtf_total") for r in records]),
+        "gap_ms": _stats(gaps),
+        "total_audio_s": sum(float(r.get("streamed_audio_s") or 0.0) for r in records),
+        "total_wall_s": sum(float(r.get("total_wall_s") or 0.0) for r in records),
+    }
diff --git a/tests/benchmarks/test_stream_latency_metrics.py b/tests/benchmarks/test_stream_latency_metrics.py
new file mode 100644
index 000000000..9e4d00827
--- /dev/null
+++ b/tests/benchmarks/test_stream_latency_metrics.py
@@ -0,0 +1,81 @@
+"""Tests for benchmarks/tts/stream_latency_metrics.py (pure Python, CPU)."""
+
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+
+import pytest
+
+sys.path.insert(0, str(Path(__file__).parent.parent.parent / "benchmarks" / "tts"))
+import stream_latency_metrics as m  # noqa: E402
+
+pytestmark = [pytest.mark.core_model, pytest.mark.cpu]
+
+SR = 24000
+FRAME = 1920  # one 12.5 Hz codec frame at 24 kHz = 80 ms
+
+
+def test_measured_default_schedule_has_80ms_cliff():
+    # Shape measured on L20X with the default 1 -> 25 frame schedule.
+    chunks = [
+        {"t_ms": 30.0, "samples": FRAME},
+        {"t_ms": 190.0, "samples": 25 * FRAME},
+        {"t_ms": 348.0, "samples": 25 * FRAME},
+    ]
+    assert m.ttfa_ms(chunks) == 30.0
+    assert m.playback_start_ms(chunks, SR) == pytest.approx(110.0)
+    stalls = m.underruns_if_start_at(chunks, SR, 30.0)
+    assert stalls == [(1, pytest.approx(80.0))]
+    assert m.stall_at_ttfa_ms(chunks, SR) == pytest.approx(80.0)
+
+
+def test_fast_arrivals_start_at_ttfa():
+    chunks = [
+        {"t_ms": 30.0, "samples": 4 * FRAME},
+        {"t_ms": 100.0, "samples": 8 * FRAME},
+        {"t_ms": 200.0, "samples": 25 * FRAME},
+    ]
+    assert m.playback_start_ms(chunks, SR) == 30.0
+    assert m.underruns_if_start_at(chunks, SR, 30.0) == []
+    assert m.stall_at_ttfa_ms(chunks, SR) == 0.0
+
+
+def test_late_middle_chunk_dominates_and_shifts_cursor():
+    chunks = [{"t_ms": 10.0, "samples": FRAME}, {"t_ms": 500.0, "samples": FRAME}, {"t_ms": 520.0, "samples": FRAME}]
+    # C = [0, 80, 160]; t - C = [10, 420, 360] -> 420
+    assert m.playback_start_ms(chunks, SR) == pytest.approx(420.0)
+    stalls = m.underruns_if_start_at(chunks, SR, 10.0)
+    # chunk 1 needed at 90, arrives 500 -> stall 410; cursor 420; chunk 2 needed at 580, arrives 520 -> ok
+    assert stalls == [(1, pytest.approx(410.0))]
+
+
+def test_zero_sample_chunks_are_ignored():
+    chunks = [{"t_ms": 5.0, "samples": 0}, {"t_ms": 30.0, "samples": FRAME}, {"t_ms": 40.0, "samples": 0}]
+    assert m.ttfa_ms(chunks) == 30.0
+    assert m.chunk_durations_ms(chunks, SR) == [pytest.approx(80.0)]
+    assert m.inter_chunk_gaps_ms(chunks) == []
+    assert m.playback_start_ms([], SR) is None
+    assert m.stall_at_ttfa_ms([{"t_ms": 1.0, "samples": 0}], SR) is None
+
+
+def test_request_metrics_and_summary():
+    chunks = [{"t_ms": 30.0, "samples": FRAME}, {"t_ms": 190.0, "samples": 25 * FRAME}]
+    r = m.request_metrics(chunks, SR, total_wall_s=0.4, tail_audio_s=0.4)
+    assert r["n_chunks"] == 2
+    assert r["streamed_audio_s"] == pytest.approx(26 * 0.08)
+    assert r["rtf_streamed"] == pytest.approx(0.190 / (26 * 0.08))
+    assert r["rtf_total"] == pytest.approx(0.4 / (26 * 0.08 + 0.4))
+    assert r["gap_ms"] == [pytest.approx(160.0)]
+    s = m.summarize([r, r, r])
+    assert s["n_requests"] == 3
+    assert s["ttfa_ms"]["p50"] == 30.0
+    assert s["gap_ms"]["n"] == 3
+    assert s["stall_at_ttfa_ms"]["max"] == pytest.approx(80.0)
+
+
+def test_percentile_nearest_rank():
+    assert m.percentile([], 0.5) is None
+    assert m.percentile([1.0], 0.9) == 1.0
+    assert m.percentile([1.0, 2.0, 3.0, 4.0], 0.5) == 2.0
+    assert m.percentile([1.0, 2.0, 3.0, 4.0], 0.9) == 4.0