diff --git a/tests/tools/test_tts_streaming.py b/tests/tools/test_tts_streaming.py
new file mode 100644
index 000000000000..77241ea485b7
--- /dev/null
+++ b/tests/tools/test_tts_streaming.py
@@ -0,0 +1,228 @@
+"""Tests for the provider-agnostic streaming TTS backend (tools.tts_streaming)
+and its dispatch through tools.tts_tool.stream_tts_to_speaker.
+
+No live audio or network: the ElevenLabs/OpenAI SDKs, sounddevice, and the sync
+synth path are all mocked. Covers the registry/resolver, provider availability,
+the chunked-streamer playback path, and the universal per-sentence sync fallback.
+"""
+
+import queue
+import threading
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+import tools.tts_streaming as ts
+
+pytest.importorskip("numpy")
+
+
+# ── SentenceChunker ──────────────────────────────────────────────────────
+
+
+class TestSentenceChunker:
+    def test_cuts_sentence_the_moment_its_boundary_arrives(self):
+        c = ts.SentenceChunker()
+        assert c.feed("This is the first full") == []
+        assert c.feed(" sentence of it all. And") == ["This is the first full sentence of it all. "]
+        assert c.flush() == ["And"]
+
+    def test_short_fragment_rides_with_the_next_sentence(self):
+        c = ts.SentenceChunker()
+        # "Ha! " alone is under min_len — it must not become its own clip.
+        assert c.feed("Ha! ") == []
+        assert c.feed("That was a good one, honestly. ") == [
+            "Ha! That was a good one, honestly. "
+        ]
+
+    def test_think_blocks_are_stripped_even_across_deltas(self):
+        c = ts.SentenceChunker()
+        assert c.feed("<think>secret reason") == []
+        assert c.feed("ing</think>The actual spoken answer. ") == ["The actual spoken answer. "]
+
+    def test_flush_drains_the_tail(self):
+        c = ts.SentenceChunker()
+        c.feed("no boundary here")
+        assert c.flush() == ["no boundary here"]
+        assert c.flush() == []
+
+    def test_paragraph_break_is_a_boundary(self):
+        c = ts.SentenceChunker()
+        assert c.feed("A paragraph without punctuation\n\nnext one") == [
+            "A paragraph without punctuation\n\n"
+        ]
+
+
+# ── Registry + resolver ──────────────────────────────────────────────────
+
+
+def _register_fake(monkeypatch, name, available=True, chunks=(b"\x00\x00",)):
+    class _Fake(ts.StreamingTTSProvider):
+        sample_rate = 24000
+
+        @staticmethod
+        def available():
+            return available
+
+        def stream(self, text):
+            yield from chunks
+
+    monkeypatch.setitem(ts._REGISTRY, name, _Fake)
+    return _Fake
+
+
+def test_resolve_returns_configured_streamer(monkeypatch):
+    _register_fake(monkeypatch, "faketts")
+    prov = ts.resolve_streaming_provider({"provider": "faketts"})
+    assert isinstance(prov, ts.StreamingTTSProvider)
+
+
+def test_resolve_none_for_unregistered_provider(monkeypatch):
+    # edge is a sync provider — not registered — so the dispatcher keeps its voice.
+    assert ts.resolve_streaming_provider({"provider": "edge"}) is None
+
+
+def test_resolve_none_when_provider_unavailable(monkeypatch):
+    _register_fake(monkeypatch, "faketts", available=False)
+    assert ts.resolve_streaming_provider({"provider": "faketts"}) is None
+
+
+def test_resolve_honors_preferred_override(monkeypatch):
+    _register_fake(monkeypatch, "faketts")
+    prov = ts.resolve_streaming_provider({"provider": "edge"}, preferred="faketts")
+    assert isinstance(prov, ts.StreamingTTSProvider)
+
+
+def test_never_swaps_provider_for_streaming(monkeypatch):
+    # A registered streamer must NOT be substituted when the user picked another
+    # (non-streaming) provider — that would silently change their voice.
+    _register_fake(monkeypatch, "elevenlabs")
+    assert ts.resolve_streaming_provider({"provider": "edge"}) is None
+
+
+# ── Built-in provider availability ───────────────────────────────────────
+
+
+def test_elevenlabs_available_reflects_key(monkeypatch):
+    monkeypatch.setattr(ts, "get_env_value", lambda k, *a: "key" if k == "ELEVENLABS_API_KEY" else None)
+    assert ts.ElevenLabsStreamer.available() is True
+    monkeypatch.setattr(ts, "get_env_value", lambda k, *a: None)
+    assert ts.ElevenLabsStreamer.available() is False
+
+
+def test_openai_available_reflects_key(monkeypatch):
+    monkeypatch.setattr(ts, "get_env_value", lambda k, *a: "key" if k == "OPENAI_API_KEY" else None)
+    assert ts.OpenAIStreamer.available() is True
+
+
+# ── Dispatch: chunked streamer path ──────────────────────────────────────
+
+
+def _drain_queue(sentences):
+    q = queue.Queue()
+    for s in sentences:
+        q.put(s)
+    q.put(None)
+    return q
+
+
+def _sd_mock():
+    sd = MagicMock()
+    out = MagicMock()
+    sd.OutputStream.return_value = out
+    return sd, out
+
+
+def test_streamer_path_writes_pcm_to_output(monkeypatch):
+    from tools import tts_tool
+
+    class _Fake(ts.StreamingTTSProvider):
+        sample_rate = 24000
+
+        @staticmethod
+        def available():
+            return True
+
+        def stream(self, text):
+            yield b"\x01\x00" * 50
+            yield b"\x02\x00" * 50
+
+    sd, out = _sd_mock()
+    q = _drain_queue(["Hello there, this is a full sentence."])
+    stop, done = threading.Event(), threading.Event()
+
+    with patch("tools.tts_streaming.resolve_streaming_provider", return_value=_Fake({}, {})), \
+         patch.object(tts_tool, "_import_sounddevice", return_value=sd):
+        tts_tool.stream_tts_to_speaker(q, stop, done)
+
+    assert out.write.called, "expected PCM chunks written to the output stream"
+    assert done.is_set()
+
+
+def test_stop_event_aborts_streaming(monkeypatch):
+    from tools import tts_tool
+
+    class _Fake(ts.StreamingTTSProvider):
+        sample_rate = 24000
+
+        @staticmethod
+        def available():
+            return True
+
+        def stream(self, text):
+            for _ in range(1000):
+                yield b"\x00\x00" * 50
+
+    sd, out = _sd_mock()
+    stop, done = threading.Event(), threading.Event()
+    stop.set()  # pre-set: no audio should be written
+    q = _drain_queue(["A complete sentence here."])
+
+    with patch("tools.tts_streaming.resolve_streaming_provider", return_value=_Fake({}, {})), \
+         patch.object(tts_tool, "_import_sounddevice", return_value=sd):
+        tts_tool.stream_tts_to_speaker(q, stop, done)
+
+    assert not out.write.called
+    assert done.is_set()
+
+
+# ── Dispatch: universal per-sentence sync fallback ───────────────────────
+
+
+def test_sync_fallback_speaks_each_sentence(monkeypatch):
+    from tools import tts_tool
+
+    spoken = []
+    monkeypatch.setattr(tts_tool, "text_to_speech_tool",
+                        lambda text, output_path: spoken.append(text))
+    played = []
+    fake_vm = MagicMock()
+    fake_vm.play_audio_file.side_effect = lambda p: played.append(p)
+    monkeypatch.setitem(__import__("sys").modules, "tools.voice_mode", fake_vm)
+    monkeypatch.setattr("os.path.getsize", lambda p: 100)
+    monkeypatch.setattr("os.path.isfile", lambda p: True)
+
+    q = _drain_queue(["First full sentence here. ", "Second full sentence here. "])
+    stop, done = threading.Event(), threading.Event()
+
+    with patch("tools.tts_streaming.resolve_streaming_provider", return_value=None):
+        tts_tool.stream_tts_to_speaker(q, stop, done)
+
+    assert len(spoken) == 2, f"expected both sentences synthesized, got {spoken}"
+    assert len(played) == 2
+    assert done.is_set()
+
+
+def test_display_callback_fires_without_audio(monkeypatch):
+    from tools import tts_tool
+
+    seen = []
+    monkeypatch.setattr(tts_tool, "text_to_speech_tool", lambda text, output_path: None)
+    q = _drain_queue(["A sentence to display aloud."])
+    stop, done = threading.Event(), threading.Event()
+
+    with patch("tools.tts_streaming.resolve_streaming_provider", return_value=None):
+        tts_tool.stream_tts_to_speaker(q, stop, done, display_callback=seen.append)
+
+    assert seen, "display_callback should fire even on the sync path"
+    assert done.is_set()
diff --git a/tools/tts_streaming.py b/tools/tts_streaming.py
new file mode 100644
index 000000000000..8d773d954f59
--- /dev/null
+++ b/tools/tts_streaming.py
@@ -0,0 +1,193 @@
+"""Provider-agnostic streaming TTS: sentence text → int16 PCM chunk iterator.
+
+The keystone of Hermes' conversational voice UX. `stream_tts_to_speaker`
+(``tools.tts_tool``) owns the sentence buffer, sounddevice output, and
+stop/queue protocol; this module owns the *provider* half — turning one
+sentence into audio the moment it's ready, so playback starts on sentence one
+instead of after the whole reply.
+
+Two provider shapes, one contract (int16 mono PCM at ``sample_rate``):
+
+* **True streamers** (`StreamingTTSProvider.stream`) — chunked APIs
+  (ElevenLabs pcm_24000, OpenAI pcm, …) that yield audio as it synthesizes.
+  Lowest time-to-first-audio.
+* **Everyone else** — providers with no chunked API still get per-*sentence*
+  playback via the proven sync `text_to_speech_tool` path (handled by the
+  dispatcher, not here), so edge (the default) is conversational too.
+
+Adding a streamer is `@register("name")` on a `StreamingTTSProvider` subclass;
+the dispatcher, config gate (`tts.<name>.streaming`), and resolver come free.
+"""
+
+from __future__ import annotations
+
+import logging
+import re
+from abc import ABC, abstractmethod
+from typing import Callable, Dict, Iterator, List, Optional
+
+from tools.tts_tool import _get_provider, get_env_value
+
+logger = logging.getLogger(__name__)
+
+# Sentence boundary: after .!? followed by whitespace, or a blank line.
+SENTENCE_BOUNDARY_RE = re.compile(r"(?<=[.!?])(?:\s|\n)|(?:\n\n)")
+_THINK_BLOCK_RE = re.compile(r"<think[\s>].*?</think>", flags=re.DOTALL)
+
+
+class SentenceChunker:
+    """Incremental sentence cutter for LLM token deltas.
+
+    Shared by the speaker pipeline (`stream_tts_to_speaker`) and the
+    speak-stream WebSocket so every surface cuts speech identically. Strips
+    ``<think>`` blocks (even split across deltas) and merges fragments shorter
+    than *min_len* into the following sentence, so "Ha!" rides along with the
+    sentence after it instead of stalling as a tiny clip.
+    """
+
+    def __init__(self, min_len: int = 20):
+        self.min_len = min_len
+        self.buf = ""
+
+    def feed(self, delta: str) -> List[str]:
+        """Absorb *delta*; return every complete sentence now ready to speak."""
+        self.buf = _THINK_BLOCK_RE.sub("", self.buf + delta)
+        if "<think" in self.buf and "</think>" not in self.buf:
+            return []  # open think tag — the closing tag may arrive next delta
+        out: List[str] = []
+        start = 0  # skip boundaries that would leave the head too short
+        while m := SENTENCE_BOUNDARY_RE.search(self.buf, start):
+            head = self.buf[: m.end()]
+            if len(head.strip()) < self.min_len:
+                start = m.end()
+                continue
+            out.append(head)
+            self.buf = self.buf[m.end():]
+            start = 0
+        return out
+
+    def flush(self) -> List[str]:
+        """Drain the tail (end-of-text or long-idle flush)."""
+        tail = _THINK_BLOCK_RE.sub("", self.buf).strip()
+        self.buf = ""
+        return [tail] if tail else []
+
+
+# ---------------------------------------------------------------------------
+# ABC + registry
+# ---------------------------------------------------------------------------
+
+class StreamingTTSProvider(ABC):
+    """Yields raw int16, little-endian, mono PCM chunks at ``sample_rate``."""
+
+    sample_rate: int = 24000
+    channels: int = 1
+    sample_width: int = 2  # bytes/sample (int16)
+
+    def __init__(self, tts_config: Dict, section: Dict):
+        self.tts_config = tts_config
+        self.section = section
+
+    @staticmethod
+    @abstractmethod
+    def available() -> bool:
+        """True when this provider's credentials/SDK are usable right now."""
+
+    @abstractmethod
+    def stream(self, text: str) -> Iterator[bytes]:
+        """Yield PCM chunks for ``text``. Raise on failure (caller logs)."""
+
+
+_REGISTRY: Dict[str, type[StreamingTTSProvider]] = {}
+
+
+def register(name: str) -> Callable[[type[StreamingTTSProvider]], type[StreamingTTSProvider]]:
+    def _wrap(cls: type[StreamingTTSProvider]) -> type[StreamingTTSProvider]:
+        _REGISTRY[name] = cls
+        return cls
+
+    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()
+    cls = _REGISTRY.get(name)
+    if cls is None or not cls.available():
+        return None
+    try:
+        return cls(tts_config, tts_config.get(name) or {})
+    except Exception as exc:  # pragma: no cover - defensive
+        logger.debug("streaming provider %s init failed: %s", name, exc)
+        return None
+
+
+# ---------------------------------------------------------------------------
+# Providers
+# ---------------------------------------------------------------------------
+
+@register("elevenlabs")
+class ElevenLabsStreamer(StreamingTTSProvider):
+    """ElevenLabs chunked HTTP → pcm_24000 (the original reference path)."""
+
+    sample_rate = 24000
+
+    @staticmethod
+    def available() -> bool:
+        return bool(get_env_value("ELEVENLABS_API_KEY"))
+
+    def stream(self, text: str) -> Iterator[bytes]:
+        from tools.tts_tool import (
+            DEFAULT_ELEVENLABS_STREAMING_MODEL_ID,
+            DEFAULT_ELEVENLABS_VOICE_ID,
+            _import_elevenlabs,
+        )
+
+        client = _import_elevenlabs()(api_key=get_env_value("ELEVENLABS_API_KEY"))
+        voice_id = self.section.get("voice_id", DEFAULT_ELEVENLABS_VOICE_ID)
+        model_id = self.section.get(
+            "streaming_model_id",
+            self.section.get("model_id", DEFAULT_ELEVENLABS_STREAMING_MODEL_ID),
+        )
+        yield from client.text_to_speech.convert(
+            text=text,
+            voice_id=voice_id,
+            model_id=model_id,
+            output_format="pcm_24000",
+        )
+
+
+@register("openai")
+class OpenAIStreamer(StreamingTTSProvider):
+    """OpenAI speech with ``response_format=pcm`` (24 kHz mono int16)."""
+
+    sample_rate = 24000
+
+    @staticmethod
+    def available() -> bool:
+        return bool(get_env_value("OPENAI_API_KEY"))
+
+    def stream(self, text: str) -> Iterator[bytes]:
+        from openai import OpenAI
+
+        client = OpenAI(
+            api_key=get_env_value("OPENAI_API_KEY"),
+            base_url=get_env_value("OPENAI_BASE_URL") or None,
+        )
+        model = self.section.get("model", "gpt-4o-mini-tts")
+        voice = self.section.get("voice", "alloy")
+        with client.audio.speech.with_streaming_response.create(
+            model=model,
+            voice=voice,
+            input=text,
+            response_format="pcm",
+        ) as response:
+            yield from response.iter_bytes()