feat(tts): add Airy streaming provider

.env.example

@@ -48,6 +48,7 @@ HAKIMAI_API_KEY=
 ZOOM_API_KEY=
 ZOOM_API_SECRET=
 PHONELY_API_KEY=
+AIRY_API_KEY=
 
 # --- Phonely LLM proxy (optional) ---
 PHONELY_BASE_URL=https://db.phonely.ai

runner/src/coval_bench/config.py

@@ -90,6 +90,7 @@ def _normalized_dual_write_requires_bucket(self) -> Settings:
     schedule_period_seconds: int = Field(default=1800, gt=0)
 
     # --- Provider API keys (all optional; loaded from Secret Manager at runtime) ---
+    airy_api_key: SecretStr | None = None
     openai_api_key: SecretStr | None = None
     elevenlabs_api_key: SecretStr | None = None
     atlas_api_key: SecretStr | None = None

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

@@ -15,6 +15,7 @@
 from __future__ import annotations
 
 from coval_bench.providers.base import TTSProvider
+from coval_bench.providers.tts.airy import AiryTTSProvider
 from coval_bench.providers.tts.alibaba import AlibabaTTSProvider
 from coval_bench.providers.tts.atlas import AtlasTTSProvider
 from coval_bench.providers.tts.azure import AzureTTSProvider
@@ -55,6 +56,7 @@
     GOOGLE_TTS_AVAILABLE = False
 
 TTS_PROVIDERS: dict[str, type[TTSProvider]] = {
+    "airy": AiryTTSProvider,
     "openai": OpenAITTSProvider,
     "atlas": AtlasTTSProvider,
     "cartesia": CartesiaTTSProvider,
@@ -88,6 +90,7 @@
     TTS_PROVIDERS["google"] = GoogleTTSProvider
 
 __all__ = [
+    "AiryTTSProvider",
     "TTS_PROVIDERS",
     "HUME_AVAILABLE",
     "GOOGLE_TTS_AVAILABLE",

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

@@ -0,0 +1,154 @@
+# Copyright 2026 The Coval Benchmarks Authors
+# SPDX-License-Identifier: Apache-2.0
+
+"""Airy TTS over HTTP streaming: 24 kHz mono signed 16-bit little-endian PCM.
+
+The stream endpoint emits raw PCM; ``finalize_tts_result`` wraps it in a
+24 kHz WAV and adds leading silence to the first-chunk arrival time.
+Protocol: https://airy.so/cloud-api/docs/api/tts/speech-synthesis-stream
+"""
+
+from __future__ import annotations
+
+import time
+
+import httpx
+import structlog
+
+from coval_bench.config import Settings
+from coval_bench.providers._http_session import (
+    connection_reused,
+    get_shared_client,
+    submit_to_headers_ms,
+)
+from coval_bench.providers.base import TTSProvider, TTSResult
+from coval_bench.providers.tts._common import finalize_tts_result
+
+logger: structlog.BoundLogger = structlog.get_logger(__name__)
+
+SAMPLE_RATE = 24000
+
+_BASE_URL = "https://api.airy.so"
+_STREAM_PATH = "/v1/audio/speech/stream"
+
+
+class AiryTTSProvider(TTSProvider):
+    """Synthesize English benchmark prompts with airy-tts-v1 in normal style."""
+
+    _VALID_MODELS = frozenset({"airy-tts-v1"})
+
+    def __init__(self, settings: Settings, model: str, voice: str) -> None:
+        if not self._model_supported(model):
+            raise ValueError(f"Unsupported Airy model: {model!r}")
+        api_key_secret = settings.airy_api_key
+        if api_key_secret is None or not api_key_secret.get_secret_value():
+            raise ValueError("airy_api_key is required in Settings")
+
+        self._model = model
+        self._voice = voice
+        self._api_key = api_key_secret.get_secret_value()
+        # HTTP header encoding errors can include the entire Authorization value.
+        if any(not 33 <= ord(char) <= 126 for char in self._api_key):
+            raise ValueError("airy_api_key must contain only visible ASCII without whitespace")
+        self._client = get_shared_client("airy", _BASE_URL)
+
+    @property
+    def name(self) -> str:
+        return f"airy-{self._model}"
+
+    @property
+    def model(self) -> str:
+        return self._model
+
+    @classmethod
+    async def warmup(cls, settings: Settings) -> None:
+        """Warm the HTTP connection with HEAD, without submitting synthesis.
+
+        Even a 401/405 response establishes a connection. The shared pool and
+        per-request diagnostics follow the other HTTP TTS providers.
+        """
+        client = get_shared_client("airy", _BASE_URL)
+        start = time.monotonic()
+        response = await client.head(_STREAM_PATH)
+        logger.info(
+            "airy_prewarm",
+            warmup_ms=round((time.monotonic() - start) * 1000, 1),
+            http_version=response.http_version,
+        )
+        if response.http_version != "HTTP/2":
+            logger.warning("airy_prewarm_no_http2", http_version=response.http_version)
+
+    async def synthesize(self, text: str) -> TTSResult:
+        audio_chunks: list[bytes] = []
+        first_chunk_at: float | None = None
+        status_code: int | None = None
+        http_version: str | None = None
+        setup_ms: float | None = None
+        reused: bool | None = None
+        error: str | None = None
+
+        payload = {
+            "model": self._model,
+            "input": text,
+            "voice_id": self._voice,
+            "language": "en",
+            "style": "normal",
+        }
+        start = time.monotonic()
+        try:
+            async with self._client.stream(
+                "POST",
+                _STREAM_PATH,
+                headers={"Authorization": f"Bearer {self._api_key}"},
+                json=payload,
+            ) as response:
+                status_code = response.status_code
+                http_version = response.http_version
+                setup_ms = submit_to_headers_ms(response.request)
+                reused = connection_reused(response.request)
+                response.raise_for_status()
+                _validate_audio_headers(response.headers)
+                # No chunk_size: buffering to a fixed byte count would inflate TTFA.
+                async for chunk in response.aiter_bytes():
+                    if chunk:
+                        if first_chunk_at is None:
+                            first_chunk_at = time.monotonic()
+                        audio_chunks.append(chunk)
+        except Exception as exc:
+            logger.warning("airy_tts_error", provider="airy", model=self._model, exc_info=exc)
+            error = str(exc) or type(exc).__name__
+            # A partial stream must not be saved/scored as complete synthesis.
+            audio_chunks.clear()
+
+        return finalize_tts_result(
+            provider="airy",
+            model=self._model,
+            voice=self._voice,
+            pcm=b"".join(audio_chunks),
+            sample_rate=SAMPLE_RATE,
+            audio_synthesis_start=start,
+            first_audio_chunk_at=first_chunk_at,
+            error=error,
+            status_code=status_code,
+            http_version=http_version,
+            submit_to_headers_ms=setup_ms,
+            connection_reused=reused,
+        )
+
+
+def _validate_audio_headers(headers: httpx.Headers) -> None:
+    content_type = headers.get("Content-Type", "").split(";", 1)[0].strip().lower()
+    if content_type != "audio/pcm":
+        raise ValueError(f"Expected Airy audio/pcm response, got {content_type!r}")
+
+    # These are the model's pinned defaults. Reject a declared mismatch instead
+    # of resampling or silently writing PCM with the wrong WAV metadata.
+    expected = {
+        "X-Audio-Sample-Rate": str(SAMPLE_RATE),
+        "X-Audio-Channels": "1",
+        "X-Audio-Sample-Format": "s16le",
+    }
+    for name, value in expected.items():
+        actual = headers.get(name, value)
+        if actual != value:
+            raise ValueError(f"Expected Airy {name}={value}, got {actual!r}")

runner/src/coval_bench/registries/provider_keys.py

@@ -43,6 +43,7 @@ def provider_names(kind: Literal["stt", "tts"]) -> frozenset[str]:
 
 
 PROVIDER_ENV: dict[str, str] = {
+    "airy": "AIRY_API_KEY",
     "openai": "OPENAI_API_KEY",
     "cartesia": "CARTESIA_API_KEY",
     "elevenlabs": "ELEVENLABS_API_KEY",

runner/tests/providers/tts/test_airy.py

@@ -0,0 +1,181 @@
+# Copyright 2026 The Coval Benchmarks Authors
+# SPDX-License-Identifier: Apache-2.0
+
+from __future__ import annotations
+
+import json
+import wave
+from collections.abc import AsyncIterator
+from types import SimpleNamespace
+
+import httpx
+import pytest
+from pydantic import SecretStr
+
+from coval_bench.config import Settings
+from coval_bench.providers.tts import airy
+
+from .conftest import make_pcm_bytes
+
+_VOICE = "a597bb7a98fc9ec1"
+_HEADERS = {
+    "Content-Type": "audio/pcm",
+    "X-Audio-Sample-Rate": "24000",
+    "X-Audio-Channels": "1",
+    "X-Audio-Sample-Format": "s16le",
+}
+_PCM = make_pcm_bytes()
+
+
+class AudioStream(httpx.AsyncByteStream):
+    def __init__(self, chunks: list[bytes], *, broken: bool = False) -> None:
+        self.chunks = chunks
+        self.broken = broken
+        self.now = 10.0
+
+    async def __aiter__(self) -> AsyncIterator[bytes]:
+        for index, chunk in enumerate(self.chunks):
+            self.now = 10.025 if index == 0 else 12.0
+            yield chunk
+        if self.broken:
+            raise httpx.ReadError("stream interrupted")
+
+
+@pytest.fixture()
+def airy_settings(monkeypatch: pytest.MonkeyPatch) -> Settings:
+    monkeypatch.setenv("AIRY_API_KEY", "test-airy-key")
+    settings = Settings(_env_file=None)
+    assert settings.airy_api_key is not None
+    assert settings.airy_api_key.get_secret_value() == "test-airy-key"
+    return settings
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("voice", [_VOICE, "new-custom-voice"])
+async def test_stream_request_and_first_chunk_timing_produce_24khz_wav(
+    airy_settings: Settings, monkeypatch: pytest.MonkeyPatch, voice: str
+) -> None:
+    stream = AudioStream([_PCM[:240], _PCM[240:]])
+
+    def handle(request: httpx.Request) -> httpx.Response:
+        assert request.method == "POST"
+        assert str(request.url) == "https://api.airy.so/v1/audio/speech/stream"
+        assert request.headers["Authorization"] == "Bearer test-airy-key"
+        assert json.loads(request.content) == {
+            "model": "airy-tts-v1",
+            "input": "Hello from Airy.",
+            "voice_id": voice,
+            "language": "en",
+            "style": "normal",
+        }
+        request.extensions.update(__t_submit=10.0, __t_headers=10.005, __connection_reused=True)
+        return httpx.Response(
+            200, headers=_HEADERS, stream=stream, extensions={"http_version": b"HTTP/2"}
+        )
+
+    async with httpx.AsyncClient(
+        base_url="https://api.airy.so", transport=httpx.MockTransport(handle)
+    ) as client:
+        monkeypatch.setattr(airy, "get_shared_client", lambda *args: client)
+        monkeypatch.setattr(airy, "time", SimpleNamespace(monotonic=lambda: stream.now))
+        provider = airy.AiryTTSProvider(airy_settings, model="airy-tts-v1", voice=voice)
+        result = await provider.synthesize("Hello from Airy.")
+
+    assert result.error is None
+    assert result.provider == "airy"
+    assert result.model == "airy-tts-v1"
+    assert result.voice == voice
+    assert result.ttfa_ms == pytest.approx(25.0)
+    assert result.http_version == "HTTP/2"
+    assert result.submit_to_headers_ms == pytest.approx(5.0)
+    assert result.connection_reused is True
+    assert result.audio_path is not None
+    try:
+        with wave.open(str(result.audio_path), "rb") as wav:
+            assert wav.getframerate() == 24000
+            assert wav.getnchannels() == 1
+            assert wav.getsampwidth() == 2
+            assert wav.readframes(wav.getnframes()) == _PCM
+    finally:
+        result.audio_path.unlink()
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    ("status", "headers", "chunks", "broken"),
+    [
+        (401, {"Content-Type": "application/json"}, [b'{"error":"unauthorized"}'], False),
+        (429, {"Content-Type": "application/json"}, [b'{"error":"rate limited"}'], False),
+        (200, _HEADERS, [], False),
+        (200, {**_HEADERS, "X-Audio-Sample-Rate": "48000"}, [_PCM], False),
+        (200, {**_HEADERS, "X-Audio-Channels": "2"}, [_PCM], False),
+        (200, {**_HEADERS, "X-Audio-Sample-Format": "f32le"}, [_PCM], False),
+        (200, {**_HEADERS, "Content-Type": "audio/wav"}, [_PCM], False),
+        (200, _HEADERS, [_PCM], True),
+    ],
+    ids=["auth", "rate-limit", "empty", "sample-rate", "channels", "format", "wav", "broken"],
+)
+async def test_failed_or_incompatible_stream_never_saves_audio(
+    airy_settings: Settings,
+    monkeypatch: pytest.MonkeyPatch,
+    status: int,
+    headers: dict[str, str],
+    chunks: list[bytes],
+    broken: bool,
+) -> None:
+    def handle(request: httpx.Request) -> httpx.Response:
+        return httpx.Response(status, headers=headers, stream=AudioStream(chunks, broken=broken))
+
+    async with httpx.AsyncClient(
+        base_url="https://api.airy.so", transport=httpx.MockTransport(handle)
+    ) as client:
+        monkeypatch.setattr(airy, "get_shared_client", lambda *args: client)
+        provider = airy.AiryTTSProvider(airy_settings, model="airy-tts-v1", voice=_VOICE)
+        result = await provider.synthesize("Hello.")
+
+    assert result.error
+    assert result.audio_path is None
+    if status >= 400:
+        assert result.status_code == status
+
+
+@pytest.mark.parametrize("api_key", [None, SecretStr("")])
+def test_missing_api_key_is_rejected(api_key: SecretStr | None) -> None:
+    settings = Settings(_env_file=None, airy_api_key=api_key)
+    with pytest.raises(ValueError, match="airy_api_key"):
+        airy.AiryTTSProvider(settings, model="airy-tts-v1", voice=_VOICE)
+
+
+@pytest.mark.parametrize("suffix", ["\n", "\x00", "é"])
+def test_malformed_key_is_rejected_without_exposing_it(suffix: str) -> None:
+    settings = Settings(_env_file=None, airy_api_key=SecretStr("test-private-key" + suffix))
+    with pytest.raises(ValueError, match="airy_api_key") as exc:
+        airy.AiryTTSProvider(settings, model="airy-tts-v1", voice=_VOICE)
+    assert "test-private-key" not in str(exc.value)
+
+
+def test_invalid_model_is_rejected(airy_settings: Settings) -> None:
+    with pytest.raises(ValueError, match="model"):
+        airy.AiryTTSProvider(airy_settings, model="unknown-model", voice=_VOICE)
+
+
+@pytest.mark.asyncio
+async def test_warmup_uses_head_without_synthesizing(
+    airy_settings: Settings, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    requests: list[httpx.Request] = []
+
+    def handle(request: httpx.Request) -> httpx.Response:
+        requests.append(request)
+        return httpx.Response(405)
+
+    async with httpx.AsyncClient(
+        base_url="https://api.airy.so", transport=httpx.MockTransport(handle)
+    ) as client:
+        monkeypatch.setattr(airy, "get_shared_client", lambda *args: client)
+        await airy.AiryTTSProvider.warmup(airy_settings)
+
+    assert len(requests) == 1
+    assert requests[0].method == "HEAD"
+    assert str(requests[0].url) == "https://api.airy.so/v1/audio/speech/stream"
+    assert requests[0].content == b""