[BENCH-623] TTS TTFA breakdown: persist the roundtrip / leading-silence split and visualize it on Latency Variation

runner/src/coval_bench/db/writer.py

@@ -27,7 +27,7 @@
 from psycopg_pool import AsyncConnectionPool
 
 from coval_bench.db.models import Result, Run, RunStatus
-from coval_bench.registries import Metric
+from coval_bench.registries import SERIES_EXCLUDED_METRICS, Metric
 
 STATS_MATVIEWS: tuple[str, ...] = ("results_24h", "results_7d", "results_30d")
 
@@ -171,7 +171,13 @@ async def refresh_bucket(self, run_id: int, *, period_seconds: int) -> None:
                 # interleave such that the staler recompute commits and the
                 # fresher one aborts on the primary key, dropping a run from
                 # the slot. Released on commit/abort.
-                params = {"bucket": bucket_at, "period": period_seconds}
+                params = {
+                    "bucket": bucket_at,
+                    "period": period_seconds,
+                    # Window-aggregate-only metrics stay out of the series
+                    # rollup; see SERIES_EXCLUDED_METRICS for the why.
+                    "series_excluded": [str(m) for m in SERIES_EXCLUDED_METRICS],
+                }
                 await cur.execute(
                     "SELECT pg_advisory_xact_lock(hashtextextended('results_by_bucket',"
                     " extract(epoch FROM %(bucket)s::timestamptz)::bigint))",
@@ -211,6 +217,7 @@ async def refresh_bucket(self, run_id: int, *, period_seconds: int) -> None:
                     WHERE r.status = 'success'
                       AND rn.status IN ('succeeded', 'partial')
                       AND r.metric_value IS NOT NULL
+                      AND r.metric_type != ALL(%(series_excluded)s)
                       AND (
                           rn.scheduled_at = %(bucket)s
                           OR (

runner/src/coval_bench/providers/base.py

@@ -67,6 +67,9 @@ class TTSResult:
     http_version: str | None = None
     submit_to_headers_ms: float | None = None
     connection_reused: bool | None = None
+    # The leading-silence part of ttfa_ms; None when offset detection didn't
+    # run or failed, in which case ttfa_ms is arrival-only and has no split.
+    leading_silence_ms: float | None = None
 
 
 # ---------------------------------------------------------------------------

runner/src/coval_bench/providers/tts/_common.py

@@ -131,6 +131,7 @@ def finalize_tts_result(
         http_version=http_version,
         submit_to_headers_ms=submit_to_headers_ms,
         connection_reused=connection_reused,
+        leading_silence_ms=offset_ms if ttfa_ms is not None else None,
     )
 
 

runner/src/coval_bench/registries/__init__.py

@@ -10,6 +10,7 @@
 from coval_bench.registries.metrics import (
     METRIC_EXCLUSIONS,
     METRIC_SPECS,
+    SERIES_EXCLUDED_METRICS,
     Metric,
     MetricDirection,
     MetricSpec,
@@ -35,6 +36,7 @@
     "Benchmark",
     "METRIC_EXCLUSIONS",
     "METRIC_SPECS",
+    "SERIES_EXCLUDED_METRICS",
     "MODEL_REGISTRY",
     "Metric",
     "MetricDirection",

runner/src/coval_bench/registries/metrics.py

@@ -24,6 +24,8 @@ class Metric(StrEnum):
     TTFT = "TTFT"
     TTFS = "TTFS"
     TTFA = "TTFA"
+    TTFA_ROUNDTRIP = "TTFARoundtrip"
+    TTFA_LEADING_SILENCE = "TTFALeadingSilence"
     RTF = "RTF"
     AUDIO_TO_FINAL = "AudioToFinal"
     V2V = "V2V"
@@ -78,6 +80,23 @@ class MetricSpec(BaseModel, frozen=True):
         decimals=0,
         benchmarks=frozenset({Benchmark.TTS}),
     ),
+    # Perceived TTFA split: roundtrip (send → first chunk) + leading silence
+    # (stream start → first audible sample). Written only when both are known,
+    # so the two rows always sum back to the TTFA row.
+    Metric.TTFA_ROUNDTRIP: MetricSpec(
+        display_name="TTFA Network Roundtrip",
+        units="milliseconds",
+        direction=MetricDirection.LOWER_IS_BETTER,
+        decimals=0,
+        benchmarks=frozenset({Benchmark.TTS}),
+    ),
+    Metric.TTFA_LEADING_SILENCE: MetricSpec(
+        display_name="TTFA Leading Silence",
+        units="milliseconds",
+        direction=MetricDirection.LOWER_IS_BETTER,
+        decimals=0,
+        benchmarks=frozenset({Benchmark.TTS}),
+    ),
     Metric.RTF: MetricSpec(
         display_name="Real-Time Factor",
         units="ratio",
@@ -115,6 +134,17 @@ class MetricSpec(BaseModel, frozen=True):
     raise RuntimeError(f"METRIC_SPECS is missing specs for: {_missing}")
 
 
+# Metrics kept out of the per-bucket series rollup (results_by_bucket) and
+# therefore out of every aggregates response's `series` array. The TTFA
+# components are consumed as window aggregates only (the Latency Variation
+# breakdown); TTS runs ~48x/day, so carrying them per bucket would double an
+# already multi-MB 30d series payload for rows nothing reads. Remove a metric
+# here if a per-run surface ever ships for it.
+SERIES_EXCLUDED_METRICS: frozenset[Metric] = frozenset(
+    {Metric.TTFA_ROUNDTRIP, Metric.TTFA_LEADING_SILENCE}
+)
+
+
 # (provider, model) pairs whose metric is not comparable with the cohort:
 # TTFT gated by buffering rather than engine speed, TTFS acked without
 # finalizing. The orchestrator skips writing these rows; the API hides

runner/src/coval_bench/runner/orchestrator.py

@@ -775,6 +775,37 @@ async def _run_tts_item(
                 )
             )
 
+            # TTFA component rows, only for a successful TTFA with a known
+            # split, so the two always sum back to the TTFA row. A failed row
+            # (provider error can arrive after audio, leaving ttfa_ms set), a
+            # nulled TTFA (transport gate) or an arrival-only TTFA (offset
+            # detection failed) writes neither.
+            leading_silence_ms = tts_result.leading_silence_ms if tts_result else None
+            if (
+                ttfa_status is ResultStatus.SUCCESS
+                and ttfa_value is not None
+                and leading_silence_ms is not None
+            ):
+                for component, value in (
+                    (Metric.TTFA_ROUNDTRIP, ttfa_value - leading_silence_ms),
+                    (Metric.TTFA_LEADING_SILENCE, leading_silence_ms),
+                ):
+                    results.append(
+                        Result(
+                            run_id=run_id,
+                            provider=entry.provider,
+                            model=entry.model,
+                            voice=voice,
+                            benchmark=Benchmark.TTS,
+                            metric_type=component,
+                            metric_value=value,
+                            metric_units=METRIC_SPECS[component].units,
+                            transcript=transcript,
+                            status=ResultStatus.SUCCESS,
+                            error=None,
+                        )
+                    )
+
             # 2. WER via Whisper transcription of synthesized audio (skip when synth errored)
             if item_error is None and audio_path is not None and audio_path.exists():
                 # Whisper transcription is our measurement instrument, not the provider under

runner/tests/unit/test_db_writer.py

@@ -685,6 +685,64 @@ async def _run() -> None:
         assert float(row["p50"]) == pytest.approx(3.0)
 
 
+def test_refresh_bucket_excludes_series_excluded_metrics(
+    pg_conn: psycopg.Connection[Any],
+) -> None:
+    """TTFA component rows stay out of the series rollup: they are consumed as
+    window aggregates only, and at ~48 TTS runs/day they would double an
+    already multi-MB 30d series payload for rows nothing reads."""
+    _apply_migrations(pg_conn)
+    scheduled = datetime.now(UTC).replace(microsecond=0) - timedelta(hours=1)
+
+    def _tts_result(run_id: int, metric_type: str, value: float) -> Result:
+        return Result(
+            run_id=run_id,
+            provider="elevenlabs",
+            model="eleven_flash_v2_5",
+            benchmark=Benchmark.TTS,
+            metric_type=metric_type,
+            metric_value=value,
+            metric_units="milliseconds",
+            status=ResultStatus.SUCCESS,
+        )
+
+    async def _run() -> None:
+        pool = await _make_pool(pg_conn)
+        try:
+            writer = RunWriter(pool)
+            run = await writer.start_run(
+                runner_sha="abc123",
+                dataset_id="tts-v1",
+                dataset_sha256="deadbeef",
+                scheduled_at=scheduled,
+            )
+            assert run.id is not None
+            await writer.record_results(
+                [
+                    _tts_result(run.id, "TTFA", 170.0),
+                    _tts_result(run.id, "TTFARoundtrip", 140.0),
+                    _tts_result(run.id, "TTFALeadingSilence", 30.0),
+                ]
+            )
+            await writer.finish_run(run.id, status=RunStatus.SUCCEEDED)
+            await writer.refresh_bucket(run.id, period_seconds=1800)
+        finally:
+            await pool.close()
+
+    asyncio.run(_run())
+
+    pg_conn.autocommit = True
+    with pg_conn.cursor() as cur:
+        cur.execute("SELECT DISTINCT metric_type FROM benchmarks_v2.results_by_bucket")
+        bucket_metrics = {row[0] for row in cur.fetchall()}
+        # The raw rows keep the split for the stats matviews.
+        cur.execute("SELECT DISTINCT metric_type FROM benchmarks_v2.results")
+        result_metrics = {row[0] for row in cur.fetchall()}
+
+    assert bucket_metrics == {"TTFA"}
+    assert result_metrics == {"TTFA", "TTFARoundtrip", "TTFALeadingSilence"}
+
+
 def test_refresh_bucket_splits_datasets(pg_conn: psycopg.Connection[Any]) -> None:
     """Two runs on different datasets in one bucket: per-dataset rows split,
     the pooled '__all__' row spans both."""

runner/tests/unit/test_metric_registry.py

@@ -19,6 +19,8 @@ def test_metric_values_match_stored_strings() -> None:
         "TTFT",
         "TTFS",
         "TTFA",
+        "TTFARoundtrip",
+        "TTFALeadingSilence",
         "RTF",
         "AudioToFinal",
         "V2V",
@@ -33,6 +35,8 @@ def test_units_match_stored_strings() -> None:
         Metric.TTFT: "seconds",
         Metric.TTFS: "seconds",
         Metric.TTFA: "milliseconds",
+        Metric.TTFA_ROUNDTRIP: "milliseconds",
+        Metric.TTFA_LEADING_SILENCE: "milliseconds",
         Metric.RTF: "ratio",
         Metric.AUDIO_TO_FINAL: "seconds",
         Metric.V2V: "milliseconds",
@@ -43,7 +47,8 @@ def test_units_match_stored_strings() -> None:
 
 def test_benchmark_coverage() -> None:
     assert METRIC_SPECS[Metric.WER].benchmarks == {Benchmark.STT, Benchmark.TTS}
-    assert METRIC_SPECS[Metric.TTFA].benchmarks == {Benchmark.TTS}
+    for metric in (Metric.TTFA, Metric.TTFA_ROUNDTRIP, Metric.TTFA_LEADING_SILENCE):
+        assert METRIC_SPECS[metric].benchmarks == {Benchmark.TTS}
     for metric in (Metric.TTFT, Metric.TTFS, Metric.RTF, Metric.AUDIO_TO_FINAL):
         assert METRIC_SPECS[metric].benchmarks == {Benchmark.STT}
 

runner/tests/unit/test_orchestrator.py

@@ -864,6 +864,7 @@ async def test_tts_http1_downgrade_nulls_ttfa_row(settings: Settings) -> None:
             error=None,
             http_version="HTTP/1.1",
             submit_to_headers_ms=210.0,
+            leading_silence_ms=45.0,
         )
 
         provider_inst = MagicMock()
@@ -944,6 +945,11 @@ def _fake_get_db_symbols() -> tuple[Any, Any, Any, Any]:
         assert "HTTP/1.1" in ttfa_rows[0].error
         assert ttfa_rows[0].http_version == "HTTP/1.1"
 
+        # A nulled TTFA writes no component rows either.
+        assert not [
+            r for r in recorded if r.metric_type.startswith("TTFA") and r.metric_type != "TTFA"
+        ]
+
         assert len(wer_rows) == 1
         assert wer_rows[0].status == ResultStatus.SUCCESS
 
@@ -1053,6 +1059,111 @@ def _fake_get_db_symbols() -> tuple[Any, Any, Any, Any]:
         assert writer.finish_run.call_args.kwargs["status"] == RunStatus.SUCCEEDED
 
 
+@pytest.mark.asyncio
+async def test_tts_ttfa_component_rows_reconcile(settings: Settings) -> None:
+    """A comparable TTFA with a known split writes component rows that sum back to it."""
+    with tempfile.TemporaryDirectory() as tmpdir:
+        audio_path = Path(tmpdir) / "synth.wav"
+        audio_path.write_bytes(b"\x00" * 512)
+
+        tts_result = TTSResult(
+            provider="elevenlabs",
+            model="eleven_flash_v2_5",
+            voice="IKne3meq5aSn9XLyUdCD",
+            ttfa_ms=120.0,
+            audio_path=audio_path,
+            error=None,
+            http_version="HTTP/2",
+            submit_to_headers_ms=90.0,
+            connection_reused=True,
+            leading_silence_ms=45.0,
+        )
+
+        provider_inst = MagicMock()
+        provider_inst.synthesize = AsyncMock(return_value=tts_result)
+        provider_cls = MagicMock(return_value=provider_inst)
+        provider_cls.warmup = AsyncMock(return_value=None)
+
+        tts_providers = {"elevenlabs": provider_cls}
+
+        run = _make_run()
+        writer = _make_stub_writer(run)
+
+        tts_dataset = MagicMock()
+        tts_dataset.items = [_make_tts_item("hello world")]
+
+        def _load(
+            dataset_id: str, *, settings: Any, sample_size: int | None = None, rng: Any = None
+        ) -> Any:
+            return tts_dataset
+
+        fake_pool = MagicMock()
+
+        @contextlib.asynccontextmanager
+        async def _fake_pool(s: Any) -> AsyncIterator[MagicMock]:
+            yield fake_pool
+
+        models_mod = MagicMock()
+        models_mod.Benchmark = Benchmark
+        models_mod.Result = Result
+        models_mod.ResultStatus = ResultStatus
+        models_mod.RunStatus = RunStatus
+
+        def _fake_get_db_symbols() -> tuple[Any, Any, Any, Any]:
+            return _fake_pool, MagicMock(return_value=writer), RunStatus, models_mod
+
+        compute_wer_real = __import__("coval_bench.metrics", fromlist=["compute_wer"]).compute_wer
+        compute_rtf_real = __import__("coval_bench.metrics", fromlist=["compute_rtf"]).compute_rtf
+
+        matrix = [
+            *_paused_registry(Benchmark.TTS),
+            _tts_entry("elevenlabs", "eleven_flash_v2_5", "IKne3meq5aSn9XLyUdCD"),
+        ]
+
+        with (
+            patch(
+                "coval_bench.runner.orchestrator._get_db_symbols",
+                side_effect=_fake_get_db_symbols,
+            ),
+            patch("coval_bench.runner.orchestrator._get_stt_providers", return_value={}),
+            patch(
+                "coval_bench.runner.orchestrator._get_tts_providers",
+                return_value=tts_providers,
+            ),
+            patch("coval_bench.runner.orchestrator._get_load_dataset", return_value=_load),
+            patch(
+                "coval_bench.runner.orchestrator._get_metrics",
+                return_value=(compute_wer_real, compute_rtf_real),
+            ),
+            patch(
+                "coval_bench.runner.orchestrator._transcribe_with_whisper",
+                return_value="hello world",
+            ),
+        ):
+            await run_benchmarks(
+                settings=settings,
+                benchmark_kind="tts",
+                smoke=True,
+                matrix_overrides=matrix,
+            )
+
+        recorded = [r for call in writer.record_results.call_args_list for r in call.args[0]]
+        ttfa = [r for r in recorded if r.metric_type == "TTFA"]
+        roundtrip = [r for r in recorded if r.metric_type == "TTFARoundtrip"]
+        silence = [r for r in recorded if r.metric_type == "TTFALeadingSilence"]
+
+        assert len(ttfa) == len(roundtrip) == len(silence) == 1
+        assert ttfa[0].metric_value == 120.0
+        assert roundtrip[0].metric_value == 75.0
+        assert silence[0].metric_value == 45.0
+        for row in (roundtrip[0], silence[0]):
+            assert row.status == ResultStatus.SUCCESS
+            assert row.metric_units == "milliseconds"
+            assert row.voice == "IKne3meq5aSn9XLyUdCD"
+
+        assert writer.finish_run.call_args.kwargs["status"] == RunStatus.SUCCEEDED
+
+
 # ---------------------------------------------------------------------------
 # 9. test_matrix_overrides
 # ---------------------------------------------------------------------------
@@ -1866,6 +1977,7 @@ async def test_tts_provider_error_wins_over_contamination(
         audio_path=None,
         error="synth stream closed early",
         http_version="HTTP/1.1",  # would otherwise trigger the contamination message
+        leading_silence_ms=45.0,  # split present — a failed row must not emit components
     )
 
     provider_inst = MagicMock()
@@ -1901,6 +2013,9 @@ async def test_tts_provider_error_wins_over_contamination(
     assert "HTTP/1.1" not in (ttfa.error or "")  # contamination message must not win
     assert ttfa.http_version == "HTTP/1.1"  # diagnostic still recorded
     assert "WER" not in by_metric  # errored synth → no WER scoring
+    # A failed TTFA writes no component rows even though its split was known.
+    assert "TTFARoundtrip" not in by_metric
+    assert "TTFALeadingSilence" not in by_metric
     assert summary.success_count == 0