feat(tts): Gemini SSE + xAI WebSocket streaming providers, tts.streaming.provider knob, docs + E2E tests

Salvaged from PR #47588 and rebased onto the post-campaign streaming core:
the StreamingTTSProvider ABC/registry and the ElevenLabs/OpenAI streamers
already live on main (tools/tts_streaming.py), so this ports the pieces
main lacked:

- GeminiStreamer: streamGenerateContent?alt=sse -> base64 PCM chunks
  (24 kHz mono int16), reusing main's DEFAULT_GEMINI_TTS_* constants.
- XAIStreamer: WebSocket wss://api.x.ai/v1/tts -> binary PCM frames,
  async->sync bridged via the _collect_async test seam.
- tts.streaming.provider config knob: pin one streamer, or 'auto' to
  walk the priority list elevenlabs -> gemini -> openai -> xai. Unset
  keeps the never-swap-the-user's-voice default.
- docs/streaming-tts.md: architecture, capability matrix, how to add
  a provider.
- Unit tests for the knob, SSE parsing, and the WS bridge; key-gated
  E2E tests (skipped without credentials).

Refs: #47588

docs/streaming-tts.md

@@ -0,0 +1,93 @@
+# Streaming TTS
+
+Hermes can stream TTS audio as it arrives from the provider, instead of waiting
+for the full audio before playing. This is used by voice mode (CLI/TUI live
+conversation), the dashboard speak-stream WebSocket, and — via the gateway
+`StreamingTTSConsumer` — any platform adapter that opts into streaming audio.
+Voice replies start speaking after the first clause instead of after full
+generation + synthesis.
+
+## Architecture
+
+The streaming pipeline has four parts:
+
+1. **Producer** — the LLM emits text deltas as it generates a response
+2. **Sentence chunker** — `tools.tts_streaming.SentenceChunker` accumulates
+   deltas, strips `<think>` blocks (even split across deltas), and flushes
+   complete sentences
+3. **TTS provider** — a registered `StreamingTTSProvider` turns each sentence
+   into raw PCM chunks (int16 mono at the provider's declared `sample_rate`)
+4. **Audio sink** — `sounddevice.OutputStream` for local playback
+   (`tools.tts_tool.stream_tts_to_speaker`), or a gateway platform adapter's
+   `write_streaming_tts` seam (`gateway/streaming_tts_consumer.py`)
+
+Providers with no chunked API still get per-*sentence* playback via the proven
+sync `text_to_speech_tool` path, so edge (the default) is conversational too.
+All spoken text is cleaned by `tools.tts_text_normalize.prepare_spoken_text`
+(one cleaner, all paths).
+
+## How to pick a provider
+
+By default the dispatcher streams with the provider you already configured
+(`tts.provider`) when that provider has a chunked API — it never silently
+swaps your voice for a different provider just to get streaming.
+
+To override, set `tts.streaming.provider` in your `config.yaml`:
+
+- a provider name (`elevenlabs`, `gemini`, `openai`, `xai`) pins that streamer
+- `auto` walks the priority list `elevenlabs → gemini → openai → xai` and uses
+  the first one whose credentials resolve — an explicit opt-in to "best
+  chunked voice available"
+
+```yaml
+tts:
+  provider: gemini
+  streaming:
+    provider: gemini      # or "auto"
+  gemini:
+    model: gemini-2.5-flash-preview-tts
+    voice: Kore
+```
+
+## Capability matrix
+
+| Provider    | Transport                             | Chunked PCM | Credentials |
+|-------------|---------------------------------------|-------------|-------------|
+| elevenlabs  | chunked HTTP (`pcm_24000`)            | yes         | `ELEVENLABS_API_KEY` / `tts.elevenlabs` |
+| openai      | chunked HTTP (`with_streaming_response`, `pcm`) | yes | `tts.openai.api_key` → env → managed gateway |
+| gemini      | SSE (`streamGenerateContent?alt=sse`) | yes         | `GEMINI_API_KEY` / `GOOGLE_API_KEY` |
+| xai         | WebSocket (`wss://api.x.ai/v1/tts`)   | yes         | xAI OAuth or `XAI_API_KEY` |
+| edge, piper, kitten, neutts, mistral, minimax, deepinfra, … | — | no (per-sentence sync fallback) | as usual |
+
+All credential lookups go through `resolve_provider_secret()`
+(config > env/.env > credential pool) — never bare env reads. Streamed bodies
+are capped at 16 MiB per sentence, mirroring the sync providers' bounded
+upstream-body invariant.
+
+## Adding a new streaming provider
+
+1. Subclass `StreamingTTSProvider` in `tools/tts_streaming.py`
+2. Set `sample_rate` (and `channels` / `sample_width` if not int16 mono)
+3. Implement `available()` (a pure probe — never install anything) and
+   `stream(self, text) -> Iterator[bytes]` yielding raw PCM chunks
+4. Decorate with `@register("yourname")`
+5. Add tests in `tests/tools/test_tts_streaming.py`
+
+The ABC enforces the contract; the registry makes the provider discoverable;
+the dispatcher (`stream_tts_to_speaker`) and the gateway consumer handle the
+sentence buffer, stop events, and audio sink for free.
+
+## Gateway streaming (platform adapters)
+
+`gateway/streaming_tts_consumer.py` bridges agent deltas to an adapter's
+streaming-audio seam. Adapters opt in by overriding, on
+`BasePlatformAdapter`:
+
+- `supports_streaming_tts(chat_id, audio_format) -> bool`
+- `begin_streaming_tts / write_streaming_tts / finish_streaming_tts /
+  abort_streaming_tts`
+
+All default to unsupported/no-op, so existing adapters are untouched. When a
+turn's streaming audio completes, the whole-file auto-TTS reply for that turn
+is suppressed (no double playback); when streaming fails before any audio was
+audible, the gateway falls back to the legacy whole-file voice reply.

tests/tools/test_tts_streaming.py

@@ -342,3 +342,108 @@ def test_display_callback_fires_without_audio(monkeypatch):
 
     assert seen, "display_callback should fire even on the sync path"
     assert done.is_set()
+
+
+# ── tts.streaming.provider config knob (salvaged from PR #47588) ─────────
+
+
+def test_streaming_provider_knob_pins_streamer(monkeypatch):
+    _register_fake(monkeypatch, "faketts")
+    prov = ts.resolve_streaming_provider(
+        {"provider": "edge", "streaming": {"provider": "faketts"}}
+    )
+    assert isinstance(prov, ts.StreamingTTSProvider)
+
+
+def test_streaming_provider_knob_pin_unusable_returns_none(monkeypatch):
+    _register_fake(monkeypatch, "faketts", available=False)
+    # A pinned-but-unusable streamer must NOT fall through to another
+    # provider — the user asked for that one specifically.
+    _register_fake(monkeypatch, "otherstreamer")
+    assert ts.resolve_streaming_provider(
+        {"provider": "otherstreamer", "streaming": {"provider": "faketts"}}
+    ) is None
+
+
+def test_streaming_provider_auto_walks_priority(monkeypatch):
+    # elevenlabs unavailable, gemini available → auto resolves gemini.
+    _register_fake(monkeypatch, "elevenlabs", available=False)
+    fake_gemini = _register_fake(monkeypatch, "gemini")
+    _register_fake(monkeypatch, "openai")
+    prov = ts.resolve_streaming_provider(
+        {"provider": "edge", "streaming": {"provider": "auto"}}
+    )
+    assert isinstance(prov, fake_gemini)
+
+
+def test_streaming_provider_auto_none_when_nothing_usable(monkeypatch):
+    for name in ts._PROVIDER_PRIORITY:
+        _register_fake(monkeypatch, name, available=False)
+    assert ts.resolve_streaming_provider(
+        {"provider": "edge", "streaming": {"provider": "auto"}}
+    ) is None
+
+
+# ── Gemini SSE parsing ────────────────────────────────────────────────────
+
+
+def test_gemini_streamer_decodes_sse_pcm_chunks(monkeypatch):
+    import base64
+    import json as _json
+    import sys
+    import types
+
+    pcm1, pcm2 = b"\x01\x00" * 40, b"\x02\x00" * 40
+
+    def _event(pcm):
+        return "data: " + _json.dumps({
+            "candidates": [{"content": {"parts": [
+                {"inlineData": {"data": base64.b64encode(pcm).decode()}}
+            ]}}]
+        })
+
+    class _Resp:
+        def __enter__(self):
+            return self
+
+        def __exit__(self, *a):
+            return False
+
+        def raise_for_status(self):
+            pass
+
+        def iter_lines(self, decode_unicode=True):
+            yield _event(pcm1)
+            yield ""  # SSE separator
+            yield ": heartbeat"  # comment line
+            yield _event(pcm2)
+
+    captured = {}
+
+    def _post(url, **kwargs):
+        captured["url"] = url
+        captured["params"] = kwargs.get("params")
+        captured["stream"] = kwargs.get("stream")
+        return _Resp()
+
+    fake_requests = types.ModuleType("requests")
+    fake_requests.post = _post
+    monkeypatch.setitem(sys.modules, "requests", fake_requests)
+    monkeypatch.setattr(ts, "get_env_value", lambda k, *a: "g-key" if k in ("GEMINI_API_KEY", "GOOGLE_API_KEY") else None)
+
+    streamer = ts.GeminiStreamer({}, {"voice": "Kore"})
+    assert list(streamer.stream("Hello there.")) == [pcm1, pcm2]
+    assert captured["params"]["alt"] == "sse"
+    assert captured["params"]["key"] == "g-key"
+    assert captured["stream"] is True, "Gemini SSE must use a bounded streamed body"
+
+
+# ── xAI WebSocket bridge ─────────────────────────────────────────────────
+
+
+def test_xai_streamer_yields_collected_frames(monkeypatch):
+    frames = [b"\x01\x00" * 30, b"\x02\x00" * 30]
+    streamer = ts.XAIStreamer.__new__(ts.XAIStreamer)
+    streamer.tts_config, streamer.section = {}, {}
+    monkeypatch.setattr(streamer, "_collect_async", lambda text: list(frames))
+    assert list(streamer.stream("A sentence.")) == frames

tests/tools/test_tts_streaming_e2e.py

@@ -0,0 +1,109 @@
+"""End-to-end tests for streaming TTS providers. Gated on real API keys.
+
+Salvaged from PR #47588 (@Cdddo) and adapted to the current provider ABC
+(``StreamingTTSProvider(tts_config, section)``). These tests are SKIPPED by
+default — they only run when the relevant credential is present. They're
+useful for catching provider-API drift and verifying the integration
+end-to-end, but they shouldn't run in CI without secrets configured.
+"""
+from __future__ import annotations
+
+import os
+
+import pytest
+
+
+def _has_xai_creds() -> bool:
+    # Plain os.environ, matching the other gates here: the repo's test
+    # conftest scrubs dotenv/pooled secrets, so gating on the richer
+    # resolver would collect a test whose runtime credentials are gone.
+    return bool(os.environ.get("XAI_API_KEY"))
+
+
+# --- ElevenLabs ---
+
+
+@pytest.mark.skipif(
+    not os.environ.get("ELEVENLABS_API_KEY"),
+    reason="ELEVENLABS_API_KEY not set",
+)
+def test_elevenlabs_streaming_real():
+    """Generate audio from the real ElevenLabs API and verify non-empty chunks."""
+    from tools.tts_streaming import ElevenLabsStreamer
+
+    provider = ElevenLabsStreamer({}, {})
+    chunks = list(provider.stream("Hello world, this is a test."))
+    assert len(chunks) > 0
+    total_bytes = sum(len(c) for c in chunks)
+    # 1s of PCM at 24kHz mono int16 = 48000 bytes; expect at least 1k for real audio
+    assert total_bytes > 1000
+
+
+# --- Gemini ---
+
+
+@pytest.mark.skipif(
+    not (os.environ.get("GEMINI_API_KEY") or os.environ.get("GOOGLE_API_KEY")),
+    reason="GEMINI_API_KEY/GOOGLE_API_KEY not set",
+)
+def test_gemini_streaming_real():
+    """Generate audio from the real Gemini SSE API and verify non-empty chunks."""
+    from tools.tts_streaming import GeminiStreamer
+
+    provider = GeminiStreamer({}, {})
+    chunks = list(provider.stream("Hola, esto es una prueba."))
+    assert len(chunks) > 0
+    total_bytes = sum(len(c) for c in chunks)
+    assert total_bytes > 1000
+
+
+# --- OpenAI ---
+
+
+@pytest.mark.skipif(
+    not os.environ.get("OPENAI_API_KEY"),
+    reason="OPENAI_API_KEY not set",
+)
+def test_openai_streaming_real():
+    """Generate audio from the real OpenAI API and verify non-empty chunks."""
+    from tools.tts_streaming import OpenAIStreamer
+
+    provider = OpenAIStreamer({}, {})
+    chunks = list(provider.stream("Hello, this is a test."))
+    assert len(chunks) > 0
+    total_bytes = sum(len(c) for c in chunks)
+    assert total_bytes > 1000
+
+
+# --- xAI ---
+
+
+@pytest.mark.skipif(not _has_xai_creds(), reason="no xAI credentials")
+def test_xai_streaming_real():
+    """Generate audio from the real xAI WebSocket API and verify non-empty frames."""
+    from tools.tts_streaming import XAIStreamer
+
+    provider = XAIStreamer({}, {})
+    chunks = list(provider.stream("Hello, this is a test."))
+    assert len(chunks) > 0
+    total_bytes = sum(len(c) for c in chunks)
+    assert total_bytes > 1000
+
+
+# --- Resolver integration (no network; requires at least one key) ---
+
+
+@pytest.mark.skipif(
+    not (
+        os.environ.get("ELEVENLABS_API_KEY")
+        or os.environ.get("GEMINI_API_KEY")
+        or os.environ.get("GOOGLE_API_KEY")
+        or os.environ.get("OPENAI_API_KEY")
+    ),
+    reason="no streaming-capable TTS key set",
+)
+def test_auto_resolver_finds_a_real_provider():
+    from tools.tts_streaming import StreamingTTSProvider, resolve_streaming_provider
+
+    provider = resolve_streaming_provider({"streaming": {"provider": "auto"}})
+    assert isinstance(provider, StreamingTTSProvider)

tools/tts_streaming.py

@@ -139,17 +139,8 @@ def register(name: str) -> Callable[[type[StreamingTTSProvider]], type[Streaming
     return _wrap
 
 
-def resolve_streaming_provider(
-    tts_config: Dict,
-    preferred: Optional[str] = None,
-) -> Optional[StreamingTTSProvider]:
-    """Return a ready streamer for the *configured* provider, else ``None``.
-
-    ``None`` means "no chunked API for this provider" — the dispatcher then
-    speaks per-sentence via the sync path, preserving the user's chosen voice.
-    We never silently swap to a different provider just to get streaming.
-    """
-    name = (preferred or _get_provider(tts_config)).lower().strip()
+def _try_instantiate(name: str, tts_config: Dict) -> Optional[StreamingTTSProvider]:
+    """Construct the registered streamer *name* if it's usable, else None."""
     cls = _REGISTRY.get(name)
     if cls is None or not cls.available():
         return None
@@ -160,6 +151,47 @@ def resolve_streaming_provider(
         return None
 
 
+# Fallback priority for ``tts.streaming.provider: auto`` — best chunked
+# latency/quality first. Deliberately hard-coded (a UX decision, not a
+# config knob); edge is absent because it has no chunked-PCM API — the
+# dispatcher's per-sentence sync path keeps it conversational instead.
+_PROVIDER_PRIORITY: List[str] = ["elevenlabs", "gemini", "openai", "xai"]
+
+
+def resolve_streaming_provider(
+    tts_config: Dict,
+    preferred: Optional[str] = None,
+) -> Optional[StreamingTTSProvider]:
+    """Return a ready streamer for the *configured* provider, else ``None``.
+
+    Resolution order:
+
+    1. ``tts.streaming.provider`` (config knob) when set:
+       * a provider name pins that exact streamer (or ``None`` if unusable);
+       * ``auto`` walks the priority list (``elevenlabs → gemini → openai
+         → xai``) and returns the first usable streamer — an explicit
+         opt-in to "give me the best chunked voice available".
+    2. Otherwise the *configured* TTS provider (or ``preferred`` override).
+       ``None`` means "no chunked API for this provider" — the dispatcher
+       then speaks per-sentence via the sync path, preserving the user's
+       chosen voice. We never silently swap to a different provider just
+       to get streaming.
+    """
+    streaming_cfg = tts_config.get("streaming") or {}
+    pinned = str(streaming_cfg.get("provider") or "").lower().strip()
+    if pinned == "auto":
+        for name in _PROVIDER_PRIORITY:
+            inst = _try_instantiate(name, tts_config)
+            if inst is not None:
+                return inst
+        return None
+    if pinned:
+        return _try_instantiate(pinned, tts_config)
+
+    name = (preferred or _get_provider(tts_config)).lower().strip()
+    return _try_instantiate(name, tts_config)
+
+
 # ---------------------------------------------------------------------------
 # Providers
 # ---------------------------------------------------------------------------
@@ -238,3 +270,168 @@ class OpenAIStreamer(StreamingTTSProvider):
             response_format="pcm",
         ) as response:
             yield from response.iter_bytes()
+
+
+@register("gemini")
+class GeminiStreamer(StreamingTTSProvider):
+    """Gemini ``streamGenerateContent?alt=sse`` → base64 PCM chunks (24 kHz).
+
+    Each SSE event carries one base64-encoded PCM chunk under
+    ``candidates[0].content.parts[*].inlineData.data``; we decode and yield
+    each one as it arrives.
+    """
+
+    sample_rate = 24000
+
+    @staticmethod
+    def available() -> bool:
+        return bool(
+            get_env_value("GEMINI_API_KEY") or get_env_value("GOOGLE_API_KEY")
+        )
+
+    def stream(self, text: str) -> Iterator[bytes]:
+        import base64
+        import json as _json
+
+        import requests
+
+        from tools.tts_tool import (
+            DEFAULT_GEMINI_TTS_BASE_URL,
+            DEFAULT_GEMINI_TTS_MODEL,
+            DEFAULT_GEMINI_TTS_VOICE,
+        )
+
+        api_key = (
+            get_env_value("GEMINI_API_KEY") or get_env_value("GOOGLE_API_KEY")
+        )
+        model = str(self.section.get("model", DEFAULT_GEMINI_TTS_MODEL)).strip() or DEFAULT_GEMINI_TTS_MODEL
+        voice = str(self.section.get("voice", DEFAULT_GEMINI_TTS_VOICE)).strip() or DEFAULT_GEMINI_TTS_VOICE
+        base_url = str(
+            self.section.get("base_url")
+            or get_env_value("GEMINI_BASE_URL")
+            or DEFAULT_GEMINI_TTS_BASE_URL
+        ).strip().rstrip("/")
+
+        payload = {
+            "contents": [{"parts": [{"text": text}]}],
+            "generationConfig": {
+                "responseModalities": ["AUDIO"],
+                "speechConfig": {
+                    "voiceConfig": {
+                        "prebuiltVoiceConfig": {"voiceName": voice},
+                    },
+                },
+            },
+        }
+        # ``?alt=sse`` flips the response from a single JSON blob to an SSE
+        # feed of base64 PCM chunks — the whole point of this provider.
+        url = f"{base_url}/models/{model}:streamGenerateContent"
+
+        def _sse_chunks() -> Iterator[bytes]:
+            with requests.post(
+                url,
+                params={"alt": "sse", "key": api_key},
+                json=payload,
+                timeout=60,
+                stream=True,
+            ) as response:
+                response.raise_for_status()
+                for line in response.iter_lines(decode_unicode=True):
+                    if not line or not line.startswith("data: "):
+                        continue
+                    try:
+                        event = _json.loads(line[len("data: "):])
+                        parts = event["candidates"][0]["content"]["parts"]
+                    except (ValueError, KeyError, IndexError, TypeError):
+                        continue
+                    for part in parts:
+                        inline = part.get("inlineData") or part.get("inline_data") or {}
+                        b64 = inline.get("data", "")
+                        if not b64:
+                            continue
+                        try:
+                            yield base64.b64decode(b64)
+                        except (ValueError, TypeError) as exc:
+                            logger.warning("Gemini SSE: bad base64 audio: %s", exc)
+
+        yield from _sse_chunks()
+
+
+@register("xai")
+class XAIStreamer(StreamingTTSProvider):
+    """xAI WebSocket TTS → binary PCM frames (24 kHz mono int16).
+
+    xAI's chunked TTS API is WebSocket-only (``wss://api.x.ai/v1/tts``).
+    The async WS loop is bridged to the sync iterator contract via
+    ``_collect_async`` — the seam unit tests monkeypatch.
+    """
+
+    sample_rate = 24000
+
+    @staticmethod
+    def available() -> bool:
+        return bool(get_env_value("XAI_API_KEY"))
+
+    def stream(self, text: str) -> Iterator[bytes]:
+        yield from self._collect_async(text)
+
+    # -- async→sync bridge (test seam) ------------------------------------
+
+    def _collect_async(self, text: str) -> List[bytes]:
+        import asyncio
+
+        return asyncio.run(self._drain_async(text))
+
+    async def _drain_async(self, text: str) -> List[bytes]:
+        frames: List[bytes] = []
+        async for frame in self._async_frames(text):
+            frames.append(frame)
+        return frames
+
+    async def _async_frames(self, text: str):
+        import json as _json
+
+        import websockets
+
+        from tools.tts_tool import DEFAULT_XAI_VOICE_ID
+
+        api_key = (get_env_value("XAI_API_KEY") or "").strip()
+        if not api_key:
+            raise RuntimeError("XAI_API_KEY not set; cannot use xAI streaming TTS")
+        voice = str(self.section.get("voice_id", DEFAULT_XAI_VOICE_ID)).strip() or DEFAULT_XAI_VOICE_ID
+        ws_url = str(
+            self.section.get("streaming_url") or "wss://api.x.ai/v1/tts"
+        ).strip()
+
+        async with websockets.connect(
+            ws_url, extra_headers={"Authorization": f"Bearer {api_key}"}
+        ) as ws:
+            await ws.send(_json.dumps({
+                "text": text,
+                "voice_id": voice,
+                "response_format": "pcm",
+            }))
+            try:
+                while True:
+                    message = await ws.recv()
+                    if isinstance(message, (bytes, bytearray, memoryview)):
+                        yield bytes(message)
+                        continue
+                    try:
+                        envelope = _json.loads(message)
+                    except (ValueError, TypeError):
+                        if message == "done":
+                            return
+                        continue
+                    etype = envelope.get("type")
+                    if etype == "done":
+                        return
+                    if etype == "error":
+                        logger.warning("xAI WS error envelope: %s",
+                                       envelope.get("error") or envelope.get("message") or envelope)
+                        return
+            except Exception as exc:
+                if exc.__class__.__name__ == "ConnectionClosed":
+                    return
+                logger.warning("xAI WS receive failed: %s", exc)
+                return