fix(tts): split long speech by provider and platform limits

Salvage of PR #17973 by @TKCen (Sebastian Hänisch), re-implemented on
current main to preserve speed/instructions/provider params,
prepare_spoken_text normalization, OPUS_VOICE_PLATFORMS, is_write_denied
path security, microsecond timestamps, and the streaming-TTS gate.

- Split long TTS text into provider-safe chunks instead of truncating
- Pack generated audio against platform upload limits (Discord 10MB,
  Telegram 50MB, configurable via tts.delivery_profiles)
- Combine chunks with ffmpeg (OGG/Opus re-encoded, MP3 stream-copied)
- Multi-file delivery when combination fails or would exceed limits
- Remove hard [:4000] truncation from all callers (cli.py, voice.py,
  gateway/run.py, gateway/platforms/base.py)
- Gemini TTS raises ValueError instead of silently truncating when
  composed prompt exceeds the provider limit

Simplify-code fixes: removed dead all_touched_paths set, added
try/finally for scratch file cleanup on exception, clean error response
on chunk failure instead of leaking stale file_path.

cli.py

@@ -12575,14 +12575,14 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin, CLIBillingMixin):
 
             # Strip markdown and non-speech content for cleaner TTS via the
             # shared cleaner (tools/tts_text_normalize): markdown, emoji,
-            # <think> blocks, verifier footer, units, newline flattening.
+            # ⋗ blocks, verifier footer, units, newline flattening.
+            # The TTS tool owns provider request limits and long-form chunking.
             try:
                 from tools.tts_text_normalize import prepare_spoken_text
-                tts_text = prepare_spoken_text(text, max_chars=4000)
+                tts_text = prepare_spoken_text(text, max_chars=None)
             except Exception:
                 # Legacy fallback pipeline — keep voice replies best-effort.
-                tts_text = text[:4000] if len(text) > 4000 else text
-                tts_text = re.sub(r'```[\s\S]*?```', ' ', tts_text)   # fenced code blocks
+                tts_text = re.sub(r'```[\s\S]*?```', ' ', text)   # fenced code blocks
                 tts_text = re.sub(r'\[([^\]]+)\]\([^)]+\)', r'\1', tts_text)  # [text](url) -> text
                 tts_text = re.sub(r'https?://\S+', '', tts_text)      # URLs
                 tts_text = re.sub(r'\*\*(.+?)\*\*', r'\1', tts_text)  # bold
@@ -12611,26 +12611,22 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin, CLIBillingMixin):
             except Exception:
                 tts_result = {}
 
-            # Prefer the requested MP3 when the provider produced it. This
-            # preserves reliable local playback while still supporting
-            # providers that write to and return a different path.
-            audio_path = mp3_path
-            if not os.path.isfile(mp3_path) or os.path.getsize(mp3_path) == 0:
-                audio_path = tts_result.get("file_path") or mp3_path
-
-            if os.path.isfile(audio_path) and os.path.getsize(audio_path) > 0:
-                play_audio_file(audio_path)
-                # Clean up
-                try:
-                    cleanup_paths = {audio_path, mp3_path}
-                    for path in list(cleanup_paths):
-                        ogg_path = path.rsplit(".", 1)[0] + ".ogg"
-                        cleanup_paths.add(ogg_path)
-                    for path in cleanup_paths:
-                        if os.path.isfile(path):
-                            os.unlink(path)
-                except OSError:
-                    pass
+            # The tool result is authoritative — it may return multiple files
+            # for long-form chunked output. Play each in order.
+            play_paths = tts_result.get("file_paths") or [
+                tts_result.get("file_path") or mp3_path
+            ]
+            for play_path in play_paths if tts_result.get("success") else []:
+                if os.path.isfile(play_path) and os.path.getsize(play_path) > 0:
+                    play_audio_file(play_path)
+            # Clean up all generated files (play_paths + mp3_path + ogg variants)
+            cleanup_paths = set(play_paths + [mp3_path, mp3_path.rsplit(".", 1)[0] + ".ogg"])
+            for path in cleanup_paths:
+                if os.path.isfile(path):
+                    try:
+                        os.unlink(path)
+                    except OSError:
+                        pass
         except Exception as e:
             logger.warning("Voice TTS playback failed: %s", e)
             _cprint(f"{_DIM}TTS playback failed: {e}{_RST}")

gateway/platforms/base.py

@@ -4202,19 +4202,22 @@ class BasePlatformAdapter(ABC):
     def prepare_tts_text(self, text: str) -> str:
         """Prepare a spoken script for TTS.
 
-        Auto-TTS should not feed raw chat Markdown, ``<think>`` reasoning
+        Auto-TTS should not feed raw chat Markdown, ``⋗`` reasoning
         blocks, or compact symbols to the speech provider.  It should receive
         a transcript-like script: reasoning blocks removed, headings and
         bullets flattened into sentence pauses, and units like ``°C``
         expanded to words such as ``degrees Celsius``.
+
+        Provider-safe chunking and platform delivery limits are enforced
+        by the TTS tool.
         """
         try:
             from tools.tts_text_normalize import prepare_spoken_text
-            return prepare_spoken_text(text, max_chars=4000)
+            return prepare_spoken_text(text, max_chars=None)
         except Exception:
             # Keep auto-TTS best-effort if the normalizer ever fails.
             text = re.sub(r'<think[\s>].*?</think>', ' ', text, flags=re.DOTALL)
-            return re.sub(r'[*_`#\[\]()]', '', text)[:4000].strip()
+            return re.sub(r'[*_`#\[\]()]', '', text).strip()
 
     async def play_tts(
         self,
@@ -6071,6 +6074,7 @@ class BasePlatformAdapter(ABC):
                 # Skip when streaming TTS already delivered audio for this turn
                 # (#60671) — the gateway streaming-TTS consumer sets the flag.
                 _tts_path = None
+                _tts_paths: List[str] = []
                 _tts_requested_path = None
                 if (self._should_auto_tts_for_chat(event.source.chat_id)
                         and event.message_type == MessageType.VOICE
@@ -6104,14 +6108,21 @@ class BasePlatformAdapter(ABC):
                             )
                             tts_data = _json.loads(tts_result_str)
                             if tts_data.get("success", True):
-                                _tts_path = tts_data.get("file_path") or _tts_requested_path
+                                raw_tts_paths = tts_data.get("file_paths") or [
+                                    tts_data.get("file_path")
+                                ]
+                                _tts_paths = [
+                                    str(path) for path in raw_tts_paths
+                                    if path and Path(path).exists()
+                                ]
+                                _tts_path = _tts_paths[0] if _tts_paths else None
                     except Exception as tts_err:
                         logger.warning("[%s] Auto-TTS failed: %s", self.name, tts_err)
 
                 # Play TTS audio before text (voice-first experience)
                 _tts_caption_delivered = False
-                _tts_cleanup_paths = {_tts_requested_path, _tts_path} - {None}
-                if _tts_path and Path(_tts_path).exists():
+                _tts_cleanup_paths = {_tts_requested_path, *_tts_paths} - {None}
+                for _tts_index, _tts_path in enumerate(_tts_paths):
                     try:
                         # Caption eligibility and payload stay on the ORIGINAL
                         # reply text. The spoken script is for synthesis only:
@@ -6119,9 +6130,11 @@ class BasePlatformAdapter(ABC):
                         # 1024-char caption limit, and captioning that spoken
                         # form would suppress the full formatted reply the
                         # user is meant to receive as a separate message.
+                        # Caption only on the first file.
                         telegram_tts_caption = None
                         if (
-                            self.platform == Platform.TELEGRAM
+                            _tts_index == 0
+                            and self.platform == Platform.TELEGRAM
                             and text_content
                             and text_content[:1024] == text_content
                         ):
@@ -6132,16 +6145,20 @@ class BasePlatformAdapter(ABC):
                             caption=telegram_tts_caption,
                             metadata=_final_thread_metadata,
                         )
+                        _record_delivery(tts_result)
                         _tts_caption_delivered = bool(
-                            telegram_tts_caption and getattr(tts_result, "success", False)
+                            _tts_caption_delivered
+                            or (
+                                telegram_tts_caption
+                                and getattr(tts_result, "success", False)
+                            )
                         )
                     finally:
-                        for _cleanup_path in _tts_cleanup_paths:
-                            try:
-                                os.remove(_cleanup_path)
-                            except OSError:
-                                pass
-                elif _tts_cleanup_paths:
+                        try:
+                            os.remove(_tts_path)
+                        except OSError:
+                            pass
+                if not _tts_paths and _tts_cleanup_paths:
                     for _cleanup_path in _tts_cleanup_paths:
                         try:
                             os.remove(_cleanup_path)

gateway/run.py

@@ -19707,11 +19707,11 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
     async def _send_voice_reply(self, event: MessageEvent, text: str) -> None:
         """Generate TTS audio and send as a voice message before the text reply."""
         audio_path = None
-        actual_path = None
+        actual_paths: List[str] = []
         try:
             from tools.tts_tool import text_to_speech_tool, _strip_markdown_for_tts
 
-            tts_text = _strip_markdown_for_tts(text[:4000])
+            tts_text = _strip_markdown_for_tts(text)
             if not tts_text:
                 return
 
@@ -19731,9 +19731,17 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
                 logger.warning("Auto voice reply TTS returned invalid JSON: %s", result_json[:200] if result_json else result_json)
                 return
 
-            # Use the actual file path from result (may differ after opus conversion)
-            actual_path = result.get("file_path", audio_path)
-            if not result.get("success") or not os.path.isfile(actual_path):
+            # Final delivery may be one combined file or multiple separately
+            # valid files when combination is unavailable or would exceed a
+            # platform limit. Preserve legacy single-file results.
+            actual_paths = result.get("file_paths") or [
+                result.get("file_path", audio_path)
+            ]
+            actual_paths = [
+                str(path) for path in actual_paths
+                if path and os.path.isfile(path)
+            ]
+            if not result.get("success") or not actual_paths:
                 logger.warning("Auto voice reply TTS failed: %s", result.get("error"))
                 return
 
@@ -19741,14 +19749,18 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
 
             # If connected to a voice channel, play there instead of sending a file
             guild_id = self._get_guild_id(event)
-            if (guild_id
-                    and hasattr(adapter, "play_in_voice_channel")
-                    and hasattr(adapter, "is_in_voice_channel")
-                    and adapter.is_in_voice_channel(guild_id)):
-                await adapter.play_in_voice_channel(guild_id, actual_path)
-            elif adapter and hasattr(adapter, "send_voice"):
-                reply_anchor = self._reply_anchor_for_event(event)
-                thread_meta = self._thread_metadata_for_source(event.source, reply_anchor)
+            play_in_voice_channel = getattr(adapter, "play_in_voice_channel", None)
+            is_in_voice_channel = getattr(adapter, "is_in_voice_channel", None)
+            send_voice = getattr(adapter, "send_voice", None)
+            in_voice_channel = bool(
+                guild_id
+                and callable(play_in_voice_channel)
+                and callable(is_in_voice_channel)
+                and is_in_voice_channel(guild_id)
+            )
+            reply_anchor = self._reply_anchor_for_event(event)
+            thread_meta = self._thread_metadata_for_source(event.source, reply_anchor)
+            if not in_voice_channel and callable(send_voice):
                 # Mark the auto voice reply as notify-worthy.  Mirrors the
                 # final-text path in gateway/platforms/base.py which sets
                 # ``notify=True`` so platform adapters that gate push
@@ -19761,17 +19773,23 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
                     thread_meta["notify"] = True
                 else:
                     thread_meta = {"notify": True}
-                send_kwargs: Dict[str, Any] = {
-                    "chat_id": event.source.chat_id,
-                    "audio_path": actual_path,
-                    "reply_to": reply_anchor,
-                    "metadata": thread_meta,
-                }
-                await adapter.send_voice(**send_kwargs)
+            for actual_path in actual_paths:
+                if in_voice_channel:
+                    play_voice = cast(Callable[..., Awaitable[Any]], play_in_voice_channel)
+                    await play_voice(guild_id, actual_path)
+                elif callable(send_voice):
+                    send_voice_call = cast(Callable[..., Awaitable[Any]], send_voice)
+                    send_kwargs: Dict[str, Any] = {
+                        "chat_id": event.source.chat_id,
+                        "audio_path": actual_path,
+                        "reply_to": reply_anchor,
+                        "metadata": thread_meta,
+                    }
+                    await send_voice_call(**send_kwargs)
         except Exception as e:
             logger.warning("Auto voice reply failed: %s", e, exc_info=True)
         finally:
-            for p in {audio_path, actual_path} - {None}:
+            for p in ({audio_path, *actual_paths} - {None}):
                 try:
                     os.unlink(p)
                 except OSError:

hermes_cli/voice.py

@@ -975,14 +975,14 @@ def speak_text(text: str, stop_event: Optional[threading.Event] = None) -> None:
             _debug(f"speak_text: streaming dispatch unavailable ({e}); using sync path")
 
         # Shared cleaner (tools/tts_text_normalize): markdown, emoji,
-        # <think> blocks, verifier footer, units, newline flattening.
+        # ⋗ blocks, verifier footer, units, newline flattening.
+        # The TTS tool owns provider request limits and long-form chunking.
         try:
             from tools.tts_text_normalize import prepare_spoken_text
-            tts_text = prepare_spoken_text(text, max_chars=4000)
+            tts_text = prepare_spoken_text(text, max_chars=None)
         except Exception:
             # Legacy fallback pipeline — keep speak_text best-effort.
-            tts_text = text[:4000] if len(text) > 4000 else text
-            tts_text = re.sub(r'```[\s\S]*?```', ' ', tts_text)             # fenced code blocks
+            tts_text = re.sub(r'```[\s\S]*?```', ' ', text)             # fenced code blocks
             tts_text = re.sub(r'\[([^\]]+)\]\([^)]+\)', r'\1', tts_text)    # [text](url) → text
             tts_text = re.sub(r'https?://\S+', '', tts_text)                # bare URLs
             tts_text = re.sub(r'\*\*(.+?)\*\*', r'\1', tts_text)            # bold
@@ -1013,28 +1013,29 @@ def speak_text(text: str, stop_event: Optional[threading.Event] = None) -> None:
         except Exception:
             tts_result = {}
 
-        # Prefer the requested MP3 when the provider produced it. This
-        # preserves reliable local playback while still supporting providers
-        # that write to and return a different path.
-        audio_path = mp3_path
-        if not os.path.isfile(mp3_path) or os.path.getsize(mp3_path) == 0:
-            audio_path = tts_result.get("file_path") or mp3_path
-
-        if os.path.isfile(audio_path) and os.path.getsize(audio_path) > 0:
-            _debug(f"speak_text: playing {audio_path} ({os.path.getsize(audio_path)} bytes)")
-            play_audio_file(audio_path)
-            try:
-                cleanup_paths = {audio_path, mp3_path}
-                for path in list(cleanup_paths):
-                    ogg_path = path.rsplit(".", 1)[0] + ".ogg"
-                    cleanup_paths.add(ogg_path)
-                for path in cleanup_paths:
-                    if os.path.isfile(path):
-                        os.unlink(path)
-            except OSError:
-                pass
-        else:
-            _debug(f"speak_text: TTS tool produced no audio at {audio_path}")
+        # The tool result is authoritative — it may return multiple files
+        # for long-form chunked output. Play each in order.
+        play_paths = tts_result.get("file_paths") or [
+            tts_result.get("file_path") or mp3_path
+        ]
+        played_any = False
+        for play_path in play_paths if tts_result.get("success") else []:
+            if os.path.isfile(play_path) and os.path.getsize(play_path) > 0:
+                _debug(
+                    f"speak_text: playing {play_path} "
+                    f"({os.path.getsize(play_path)} bytes)"
+                )
+                play_audio_file(play_path)
+                played_any = True
+        cleanup_paths = set(play_paths + [mp3_path, mp3_path.rsplit(".", 1)[0] + ".ogg"])
+        for path in cleanup_paths:
+            if os.path.isfile(path):
+                try:
+                    os.unlink(path)
+                except OSError:
+                    pass
+        if not played_any:
+            _debug(f"speak_text: TTS tool produced no audio at {mp3_path}")
     except Exception as e:
         logger.warning("Voice TTS playback failed: %s", e)
         _debug(f"speak_text raised {type(e).__name__}: {e}")

tests/hermes_cli/test_voice_wrapper.py

@@ -10,6 +10,8 @@ stack.
 """
 
 
+import json
+
 import pytest
 
 
@@ -186,18 +188,22 @@ class TestSpeakTextGuards:
         assert voice.speak_text("Hello world") is None
         assert played == [returned_path]
 
-    def test_speak_text_prefers_requested_mp3_over_returned_ogg(self, monkeypatch):
+    def test_speak_text_plays_returned_file_paths(self, monkeypatch):
         import hermes_cli.voice as voice
         from tools import tts_tool
 
         played = []
-        requested_paths = []
 
         def fake_tts(**kwargs):
             requested_path = kwargs["output_path"]
-            requested_paths.append(requested_path)
             ogg_path = requested_path.rsplit(".", 1)[0] + ".ogg"
-            return f'{{"success": true, "file_path": "{ogg_path}"}}'
+            # The tool may return a different path than the requested MP3;
+            # the result's file_paths is authoritative for playback.
+            return json.dumps({
+                "success": True,
+                "file_path": ogg_path,
+                "file_paths": [ogg_path],
+            })
 
         monkeypatch.setattr(tts_tool, "text_to_speech_tool", fake_tts)
         monkeypatch.setattr(voice.os, "makedirs", lambda *_args, **_kwargs: None)
@@ -207,7 +213,9 @@ class TestSpeakTextGuards:
         monkeypatch.setattr(voice, "play_audio_file", lambda path: played.append(path))
 
         assert voice.speak_text("Hello world") is None
-        assert played == requested_paths
+        # Should play the path from the result, not the requested MP3 path
+        assert len(played) == 1
+        assert played[0].endswith(".ogg")
 
 
 class TestContinuousAPI:

tests/tools/test_tts_long_form_chunking.py

@@ -0,0 +1,150 @@
+"""Tests for the long-form TTS chunking and delivery packing pipeline.
+
+Verifies that text exceeding a provider's per-request cap is split without
+content loss, that chunks are synthesized in order, and that the delivery
+packing respects platform upload limits.
+"""
+
+import json
+import os
+from pathlib import Path
+from unittest.mock import patch
+
+import pytest
+
+from tools.tts_tool import (
+    AudioDeliveryProfile,
+    _build_audio_delivery_files,
+    _concat_audio_files,
+    _pack_audio_files_for_delivery,
+    _split_oversized_sentence,
+    _split_text_for_tts,
+)
+
+
+class TestSplitTextForTts:
+    def test_short_text_returns_single_chunk(self):
+        result = _split_text_for_tts("Hello world.", 4096)
+        assert result == ["Hello world."]
+
+    def test_empty_text_returns_empty_list(self):
+        assert _split_text_for_tts("", 4096) == []
+        assert _split_text_for_tts("   ", 4096) == []
+
+    def test_long_text_is_split_without_loss(self):
+        text = "A" * 5000
+        chunks = _split_text_for_tts(text, 4096)
+        assert len(chunks) == 2
+        assert chunks[0] == "A" * 4096
+        assert chunks[1] == "A" * 904
+        assert "".join(chunks) == text
+
+    def test_splits_on_sentence_boundaries(self):
+        text = "First sentence. Second sentence. Third sentence."
+        chunks = _split_text_for_tts(text, 30)
+        assert len(chunks) >= 2
+        # No content lost
+        assert "".join(chunks).replace(" ", "") == text.replace(" ", "")
+
+    def test_handles_very_long_word(self):
+        text = "A" * 100
+        chunks = _split_text_for_tts(text, 30)
+        assert all(len(c) <= 30 for c in chunks)
+        assert "".join(chunks) == text
+
+
+class TestSplitOversizedSentence:
+    def test_short_sentence_returns_as_is(self):
+        assert _split_oversized_sentence("Hello world.", 100) == ["Hello world."]
+
+    def test_long_word_is_hard_split(self):
+        word = "A" * 100
+        chunks = _split_oversized_sentence(word, 30)
+        assert all(len(c) <= 30 for c in chunks)
+        assert "".join(chunks) == word
+
+    def test_word_boundary_split(self):
+        words = " ".join(["word"] * 50)
+        chunks = _split_oversized_sentence(words, 30)
+        assert all(len(c) <= 30 for c in chunks)
+
+
+class TestAudioDeliveryProfile:
+    def test_default_profile(self):
+        profile = AudioDeliveryProfile(platform="default", max_file_bytes=10 * 1024 * 1024)
+        assert profile.target_file_bytes > 0
+        assert profile.target_file_bytes < profile.max_file_bytes
+
+    def test_custom_safety_ratio(self):
+        profile = AudioDeliveryProfile(
+            platform="custom", max_file_bytes=1000, safety_ratio=0.5
+        )
+        assert profile.target_file_bytes == 500
+
+
+class TestPackAudioFilesForDelivery:
+    def test_single_file_returns_one_group(self, tmp_path):
+        f = tmp_path / "a.mp3"
+        f.write_bytes(b"x" * 100)
+        profile = AudioDeliveryProfile(platform="default", max_file_bytes=10000)
+        groups = _pack_audio_files_for_delivery([str(f)], profile)
+        assert len(groups) == 1
+        assert groups[0] == [str(f)]
+
+    def test_splits_on_size_limit(self, tmp_path):
+        files = []
+        for i in range(5):
+            f = tmp_path / f"chunk{i:02d}.mp3"
+            f.write_bytes(b"x" * 300)
+            files.append(str(f))
+        # Target is 500 bytes, each file is 300 → at most 1 file per group
+        profile = AudioDeliveryProfile(platform="default", max_file_bytes=1000, safety_ratio=0.5)
+        groups = _pack_audio_files_for_delivery(files, profile)
+        assert len(groups) == 5
+        for group in groups:
+            assert len(group) == 1
+
+    def test_splits_on_suffix_mismatch(self, tmp_path):
+        f1 = tmp_path / "a.mp3"
+        f1.write_bytes(b"x" * 100)
+        f2 = tmp_path / "b.ogg"
+        f2.write_bytes(b"x" * 100)
+        profile = AudioDeliveryProfile(platform="default", max_file_bytes=10000)
+        groups = _pack_audio_files_for_delivery([str(f1), str(f2)], profile)
+        assert len(groups) == 2
+
+
+class TestBuildAudioDeliveryFiles:
+    def test_single_file_passes_through(self, tmp_path):
+        f = tmp_path / "chunk.mp3"
+        f.write_bytes(b"x" * 100)
+        out = str(tmp_path / "output.mp3")
+        profile = AudioDeliveryProfile(platform="default", max_file_bytes=10000)
+        paths, combined = _build_audio_delivery_files([str(f)], out, profile)
+        assert len(paths) == 1
+        assert combined is False
+
+    def test_oversized_chunk_raises(self, tmp_path):
+        f = tmp_path / "chunk.mp3"
+        f.write_bytes(b"x" * 100)
+        out = str(tmp_path / "output.mp3")
+        profile = AudioDeliveryProfile(platform="default", max_file_bytes=50)
+        with pytest.raises(ValueError, match="exceeds"):
+            _build_audio_delivery_files([str(f)], out, profile)
+
+    def test_combines_multiple_files(self, tmp_path):
+        files = []
+        for i in range(3):
+            f = tmp_path / f"chunk{i:02d}.mp3"
+            f.write_bytes(b"\x00" * 100)
+            files.append(str(f))
+        out = str(tmp_path / "output.mp3")
+        profile = AudioDeliveryProfile(platform="default", max_file_bytes=10000)
+
+        with patch("tools.tts_tool._concat_audio_files") as mock_concat:
+            mock_concat.return_value = out
+            # Copy the first file to output so the size check passes
+            Path(out).write_bytes(b"\x00" * 300)
+            paths, combined = _build_audio_delivery_files(files, out, profile)
+            assert len(paths) == 1
+            assert combined is True

tests/tools/test_tts_max_text_length.py

@@ -1,8 +1,8 @@
 """Tests for per-provider TTS input-character limits.
 
-Replaces the old global ``MAX_TEXT_LENGTH = 4000`` cap that truncated every
-provider at 4000 chars even though OpenAI allows 4096, xAI allows 15000,
-MiniMax allows 10000, and ElevenLabs allows 5000-40000 depending on model.
+With long-form chunking, text exceeding the provider cap is split into
+ordered chunks instead of silently truncated. Each chunk is synthesized
+separately and the results are combined or delivered as multiple files.
 """
 
 import json
@@ -56,25 +56,30 @@ class TestResolveMaxTextLength:
         assert expected.issubset(PROVIDER_MAX_TEXT_LENGTH.keys())
 
 
-class TestTextToSpeechToolTruncation:
-    """End-to-end: verify the resolver actually drives the text_to_speech_tool
-    truncation path rather than the old 4000-char global."""
-
-    def test_openai_truncates_at_4096_not_4000(self, tmp_path, monkeypatch, caplog):
-        import logging
-        caplog.set_level(logging.WARNING, logger="tools.tts_tool")
+class TestTextToSpeechToolChunking:
+    """End-to-end: verify the resolver drives text_to_speech_tool to split
+    per-request chunks rather than the old 4000-char global truncation."""
 
+    def test_openai_chunks_at_4096_without_dropping_text(self, tmp_path, monkeypatch):
         # 5000 chars -- over OpenAI's 4096 limit but under xAI's 15k
         text = "A" * 5000
-        captured_text = {}
+        captured_text = []
 
         def fake_openai(t, out, cfg, **_kw):
-            captured_text["text"] = t
+            captured_text.append(t)
             with open(out, "wb") as f:
                 f.write(b"\x00")
             return out
 
+        def fake_combine(paths, output_path, *, voice_compatible=False):
+            with open(output_path, "wb") as destination:
+                for path in paths:
+                    with open(path, "rb") as source:
+                        destination.write(source.read())
+            return output_path
+
         monkeypatch.setattr("tools.tts_tool._generate_openai_tts", fake_openai)
+        monkeypatch.setattr("tools.tts_tool._concat_audio_files", fake_combine)
         monkeypatch.setattr("tools.tts_tool._load_tts_config",
                             lambda: {"provider": "openai"})
 
@@ -83,10 +88,9 @@ class TestTextToSpeechToolTruncation:
         result = json.loads(text_to_speech_tool(text=text, output_path=out))
 
         assert result["success"] is True
-        # Should be truncated to 4096, not the old 4000
-        assert len(captured_text["text"]) == 4096
-        # And the warning should mention the provider
-        assert any("openai" in rec.message.lower() for rec in caplog.records)
+        assert [len(chunk) for chunk in captured_text] == [4096, 904]
+        assert "".join(captured_text) == text
+        assert result["chunk_count"] == 2
 
     def test_xai_accepts_much_longer_input(self, tmp_path, monkeypatch):
         # 12000 chars -- over old global 4000, under xAI's 15000
@@ -108,21 +112,29 @@ class TestTextToSpeechToolTruncation:
         result = json.loads(text_to_speech_tool(text=text, output_path=out))
 
         assert result["success"] is True
-        # xAI should accept the full 12000 chars
+        # xAI should accept the full 12000 chars in a single chunk
         assert len(captured_text["text"]) == 12000
 
     def test_user_override_is_respected(self, tmp_path, monkeypatch):
         # User says "cap openai at 100 chars" -- we must honor it
         text = "C" * 500
-        captured_text = {}
+        captured_text = []
 
         def fake_openai(t, out, cfg, **_kw):
-            captured_text["text"] = t
+            captured_text.append(t)
             with open(out, "wb") as f:
                 f.write(b"\x00")
             return out
 
+        def fake_combine(paths, output_path, *, voice_compatible=False):
+            with open(output_path, "wb") as destination:
+                for path in paths:
+                    with open(path, "rb") as source:
+                        destination.write(source.read())
+            return output_path
+
         monkeypatch.setattr("tools.tts_tool._generate_openai_tts", fake_openai)
+        monkeypatch.setattr("tools.tts_tool._concat_audio_files", fake_combine)
         monkeypatch.setattr("tools.tts_tool._load_tts_config",
                             lambda: {"provider": "openai",
                                      "openai": {"max_text_length": 100}})
@@ -132,4 +144,5 @@ class TestTextToSpeechToolTruncation:
         result = json.loads(text_to_speech_tool(text=text, output_path=out))
 
         assert result["success"] is True
-        assert len(captured_text["text"]) == 100
+        assert all(len(chunk) <= 100 for chunk in captured_text)
+        assert "".join(captured_text) == text

tests/tools/test_voice_cli_integration.py

@@ -1,6 +1,7 @@
 """Tests for CLI voice mode integration -- markdown stripping, voice state
 management, TTS/STT wiring, barge-in and the full-duplex listener."""
 
+import json
 import queue
 import threading
 from types import SimpleNamespace
@@ -287,21 +288,28 @@ class TestVoiceSpeakResponseReal:
     @patch("cli.os.makedirs")
     @patch("tools.voice_mode.play_audio_file")
     @patch("tools.tts_tool.text_to_speech_tool")
-    def test_play_audio_prefers_requested_mp3_over_returned_ogg(
+    def test_play_audio_uses_returned_file_paths(
         self, mock_tts, mock_play, _mkd, _isf, _gsz, _unl, _cp
     ):
         def fake_tts(**kwargs):
             mp3_path = kwargs["output_path"]
             ogg_path = mp3_path.rsplit(".", 1)[0] + ".ogg"
-            return f'{{"success": true, "file_path": "{ogg_path}"}}'
+            # The tool result is authoritative — file_paths drives playback
+            return json.dumps({
+                "success": True,
+                "file_path": ogg_path,
+                "file_paths": [ogg_path],
+            })
 
         mock_tts.side_effect = fake_tts
 
         cli = _make_voice_cli(_voice_tts=True)
         cli._voice_speak_response("Hello world")
 
-        requested_path = mock_tts.call_args.kwargs["output_path"]
-        mock_play.assert_called_once_with(requested_path)
+        # Should play the returned OGG path, not the requested MP3 path
+        mock_play.assert_called_once_with(
+            mock_tts.call_args.kwargs["output_path"].rsplit(".", 1)[0] + ".ogg"
+        )
 
 
 class TestVoiceStopAndTranscribeReal:

tools/tts_tool.py

@@ -53,7 +53,7 @@ import uuid
 from concurrent.futures import Future, ThreadPoolExecutor
 from dataclasses import dataclass, field
 from pathlib import Path
-from typing import Callable, Dict, Any, Iterator, Optional
+from typing import Callable, Dict, Any, Iterator, List, Optional, Tuple
 from urllib.parse import urljoin, urlparse
 
 from hermes_cli._subprocess_compat import windows_hide_flags
@@ -463,6 +463,165 @@ def _resolve_max_text_length(
     return FALLBACK_MAX_TEXT_LENGTH
 
 
+# ===========================================================================
+# Long-form chunking and delivery packing
+# ===========================================================================
+
+@dataclass(frozen=True)
+class AudioDeliveryProfile:
+    """Destination-platform constraints for generated TTS audio."""
+
+    platform: str
+    max_file_bytes: int
+    safety_ratio: float = 0.85
+
+    @property
+    def target_file_bytes(self) -> int:
+        """Conservative packing target below the platform hard limit."""
+        return max(1, int(self.max_file_bytes * self.safety_ratio))
+
+
+_PLATFORM_AUDIO_DEFAULTS: Dict[str, Dict[str, Any]] = {
+    "discord": {
+        "max_file_bytes": 10 * 1024 * 1024,
+        "safety_ratio": 0.85,
+    },
+    "telegram": {
+        "max_file_bytes": 50 * 1024 * 1024,
+        "safety_ratio": 0.85,
+    },
+    "default": {
+        "max_file_bytes": 10 * 1024 * 1024,
+        "safety_ratio": 0.85,
+    },
+}
+
+
+def _resolve_audio_delivery_profile(
+    platform: Optional[str],
+    tts_config: Optional[Dict[str, Any]] = None,
+) -> AudioDeliveryProfile:
+    """Resolve upload constraints, including optional per-platform overrides."""
+    key = (platform or "default").lower().strip() or "default"
+    defaults = dict(
+        _PLATFORM_AUDIO_DEFAULTS.get(key) or _PLATFORM_AUDIO_DEFAULTS["default"]
+    )
+    cfg = tts_config or {}
+    profiles = cfg.get("delivery_profiles")
+    overrides = profiles.get(key, {}) if isinstance(profiles, dict) else {}
+    if isinstance(overrides, dict):
+        defaults.update({k: v for k, v in overrides.items() if v is not None})
+
+    max_file_bytes = defaults.get("max_file_bytes")
+    if (
+        isinstance(max_file_bytes, bool)
+        or not isinstance(max_file_bytes, int)
+        or max_file_bytes <= 0
+    ):
+        max_file_bytes = _PLATFORM_AUDIO_DEFAULTS["default"]["max_file_bytes"]
+
+    safety_ratio = defaults.get("safety_ratio", 0.85)
+    if (
+        isinstance(safety_ratio, bool)
+        or not isinstance(safety_ratio, (int, float))
+        or not 0 < safety_ratio <= 1
+    ):
+        safety_ratio = 0.85
+
+    return AudioDeliveryProfile(
+        platform=key,
+        max_file_bytes=max_file_bytes,
+        safety_ratio=float(safety_ratio),
+    )
+
+
+def _split_oversized_sentence(sentence: str, max_chars: int) -> List[str]:
+    """Split one over-limit sentence on word boundaries, then hard boundaries."""
+    words = sentence.split()
+    chunks: List[str] = []
+    current = ""
+    for word in words:
+        if len(word) > max_chars:
+            if current:
+                chunks.append(current)
+                current = ""
+            chunks.extend(word[i:i + max_chars] for i in range(0, len(word), max_chars))
+            continue
+        candidate = f"{current} {word}".strip()
+        if current and len(candidate) > max_chars:
+            chunks.append(current)
+            current = word
+        else:
+            current = candidate
+    if current:
+        chunks.append(current)
+    return chunks
+
+
+def _split_text_for_tts(text: str, max_chars: int) -> List[str]:
+    """Split text under a provider cap without dropping normalized content."""
+    if max_chars <= 0:
+        max_chars = FALLBACK_MAX_TEXT_LENGTH
+    normalized = " ".join((text or "").split())
+    if not normalized:
+        return []
+    if len(normalized) <= max_chars:
+        return [normalized]
+
+    sentences = [
+        sentence.strip()
+        for sentence in re.split(r"(?<=[.!?;:,])\s+", normalized)
+        if sentence.strip()
+    ]
+    expanded: List[str] = []
+    for sentence in sentences:
+        if len(sentence) <= max_chars:
+            expanded.append(sentence)
+        else:
+            expanded.extend(_split_oversized_sentence(sentence, max_chars))
+
+    chunks: List[str] = []
+    current = ""
+    for sentence in expanded:
+        candidate = f"{current} {sentence}".strip()
+        if current and len(candidate) > max_chars:
+            chunks.append(current)
+            current = sentence
+        else:
+            current = candidate
+    if current:
+        chunks.append(current)
+    return chunks
+
+
+def _pack_audio_files_for_delivery(
+    audio_paths: List[str],
+    profile: AudioDeliveryProfile,
+) -> List[List[str]]:
+    """Group already-final-encoded chunks under the conservative size target."""
+    groups: List[List[str]] = []
+    current: List[str] = []
+    current_size = 0
+    current_suffix = ""
+    for path in audio_paths:
+        size = Path(path).stat().st_size
+        suffix = Path(path).suffix.lower()
+        if current and (
+            current_size + size > profile.target_file_bytes
+            or suffix != current_suffix
+        ):
+            groups.append(current)
+            current = []
+            current_size = 0
+            current_suffix = ""
+        current.append(path)
+        current_size += size
+        current_suffix = suffix
+    if current:
+        groups.append(current)
+    return groups
+
+
 # ===========================================================================
 # Config loader -- reads tts: section from ~/.hermes/config.yaml
 # ===========================================================================
@@ -1365,6 +1524,193 @@ def _repair_ogg_container(file_str: str) -> str:
         return file_str
 
 
+# ===========================================================================
+# Long-form audio combination and delivery packing
+# ===========================================================================
+
+def _concat_audio_files(
+    audio_paths: List[str],
+    output_path: str,
+    *,
+    voice_compatible: bool = False,
+) -> Optional[str]:
+    """Combine independently encoded chunks with ffmpeg.
+
+    OGG/Opus is always decoded and re-encoded, even when a custom provider did
+    not opt in to voice-message presentation. Matching MP3 chunks preserve their
+    encoded frames. A failed or unavailable combine returns ``None`` so callers
+    can preserve the original, individually valid files. Structured audio
+    containers are never byte-joined.
+    """
+    if not audio_paths:
+        raise ValueError("No audio chunks to combine")
+    if len(audio_paths) == 1:
+        source = audio_paths[0]
+        if os.path.abspath(source) != os.path.abspath(output_path):
+            shutil.copyfile(source, output_path)
+        return output_path
+
+    ffmpeg = shutil.which("ffmpeg")
+    if not ffmpeg:
+        return None
+
+    destination = Path(output_path)
+    destination.parent.mkdir(parents=True, exist_ok=True)
+    concat_path = destination.with_name(f".{destination.name}.{uuid.uuid4().hex}.concat.txt")
+    temp_output = destination.with_name(
+        f".{destination.stem}.{uuid.uuid4().hex}.combining{destination.suffix}"
+    )
+    try:
+        with concat_path.open("w", encoding="utf-8") as concat_file:
+            for path in audio_paths:
+                concat_file.write(f"file {shlex.quote(os.path.abspath(path))}\n")
+
+        command = [
+            ffmpeg,
+            "-y",
+            "-loglevel",
+            "error",
+            "-f",
+            "concat",
+            "-safe",
+            "0",
+            "-i",
+            str(concat_path),
+            "-vn",
+        ]
+        suffix = destination.suffix.lower()
+        if voice_compatible or suffix in {".ogg", ".opus"}:
+            command.extend([
+                "-c:a", "libopus", "-ac", "1", "-b:a", "64k", "-vbr", "off",
+            ])
+        elif suffix == ".mp3" and all(
+            Path(path).suffix.lower() == ".mp3" for path in audio_paths
+        ):
+            # Matching MP3 provider chunks already share one output codec/config.
+            # Preserve those encoded frames instead of imposing a second lossy pass.
+            command.extend(["-c:a", "copy"])
+        command.append(str(temp_output))
+
+        result = subprocess.run(
+            command,
+            capture_output=True,
+            timeout=120,
+            stdin=subprocess.DEVNULL,
+            creationflags=windows_hide_flags(),
+        )
+        if (
+            result.returncode == 0
+            and temp_output.exists()
+            and temp_output.stat().st_size > 0
+        ):
+            os.replace(temp_output, destination)
+            return str(destination)
+        logger.warning(
+            "ffmpeg audio combine failed: %s",
+            result.stderr.decode("utf-8", errors="ignore")[:500],
+        )
+    except (OSError, subprocess.TimeoutExpired) as exc:
+        logger.warning("ffmpeg audio combine failed: %s", exc)
+    finally:
+        for path in (concat_path, temp_output):
+            try:
+                path.unlink()
+            except OSError:
+                pass
+    return None
+
+
+def _build_audio_delivery_files(
+    audio_paths: List[str],
+    output_path: str,
+    profile: AudioDeliveryProfile,
+    *,
+    voice_compatible: bool = False,
+) -> Tuple[List[str], bool]:
+    """Pack final-encoded chunks and enforce the hard upload limit.
+
+    Packing uses the conservative target. Every combined artifact is then
+    checked at its actual post-encoding size; an over-limit group is split and
+    retried. If combining fails, the valid constituent files are returned
+    separately. A single final-encoded chunk above the hard limit fails closed
+    rather than returning an upload that the destination will reject.
+    """
+    if not audio_paths:
+        raise ValueError("No final-encoded TTS audio chunks")
+    for path in audio_paths:
+        size = Path(path).stat().st_size
+        if size > profile.max_file_bytes:
+            raise ValueError(
+                f"Final-encoded TTS chunk exceeds {profile.platform} delivery "
+                f"limit ({size} > {profile.max_file_bytes} bytes): {path}"
+            )
+
+    base = Path(output_path)
+    scratch_outputs: List[str] = []
+    combined_any = False
+    combine_index = 0
+
+    def emit(group: List[str]) -> List[str]:
+        nonlocal combined_any, combine_index
+        if len(group) == 1:
+            return list(group)
+
+        combine_index += 1
+        scratch = base.with_name(
+            f".{base.stem}.delivery{combine_index:03d}.{uuid.uuid4().hex}{base.suffix}"
+        )
+        combined = _concat_audio_files(
+            group, str(scratch), voice_compatible=voice_compatible,
+        )
+        if not combined:
+            return list(group)
+        scratch_outputs.append(combined)
+        combined_size = Path(combined).stat().st_size
+        if combined_size <= profile.max_file_bytes:
+            combined_any = True
+            return [combined]
+
+        try:
+            Path(combined).unlink()
+        except OSError:
+            pass
+        midpoint = max(1, len(group) // 2)
+        return emit(group[:midpoint]) + emit(group[midpoint:])
+
+    packed: List[str] = []
+    for group in _pack_audio_files_for_delivery(audio_paths, profile):
+        packed.extend(emit(group))
+
+    final_paths: List[str] = []
+    for index, source in enumerate(packed, start=1):
+        if len(packed) == 1:
+            destination = base
+        else:
+            source_suffix = Path(source).suffix or base.suffix
+            destination = base.with_name(
+                f"{base.stem}.part{index:02d}{source_suffix}"
+            )
+        if os.path.abspath(source) != os.path.abspath(destination):
+            destination.parent.mkdir(parents=True, exist_ok=True)
+            os.replace(source, destination)
+        if destination.stat().st_size > profile.max_file_bytes:
+            raise ValueError(
+                f"Final TTS deliverable exceeds {profile.platform} delivery limit: "
+                f"{destination}"
+            )
+        final_paths.append(str(destination))
+
+    try:
+        return final_paths, combined_any
+    finally:
+        for scratch in scratch_outputs:
+            if scratch not in final_paths:
+                try:
+                    Path(scratch).unlink()
+                except OSError:
+                    pass
+
+
 # ===========================================================================
 # Provider: Edge TTS (free)
 # ===========================================================================
@@ -2294,11 +2640,12 @@ def _generate_gemini_tts(text: str, output_path: str, tts_config: Dict[str, Any]
     )
     max_len = _resolve_max_text_length("gemini", tts_config)
     if len(prompt_text) > max_len:
-        logger.warning(
-            "Gemini TTS composed prompt too long (%d chars), truncating to %d",
-            len(prompt_text), max_len,
+        raise ValueError(
+            "Gemini TTS composed prompt exceeds the provider request limit "
+            f"({len(prompt_text)} > {max_len} chars). Reduce the persona/audio-tag "
+            "prompt or lower tts.gemini.max_text_length so long-form text is "
+            "split with enough prompt headroom."
         )
-        prompt_text = prompt_text[:max_len]
 
     payload: Dict[str, Any] = {
         "contents": [{"parts": [{"text": prompt_text}]}],
@@ -2779,55 +3126,30 @@ def _generate_kittentts(text: str, output_path: str, tts_config: Dict[str, Any])
 # ===========================================================================
 # Main tool function
 # ===========================================================================
-def text_to_speech_tool(
+def _text_to_speech_single(
     text: str,
     output_path: Optional[str] = None,
+    *,
     speed: Optional[float] = None,
     instructions: Optional[str] = None,
     provider: Optional[str] = None,
+    tts_config_override: Optional[Dict[str, Any]] = None,
 ) -> str:
-    """
-    Convert text to speech audio.
+    """Synthesize one provider-safe text chunk and return one final-encoded file.
 
-    Reads provider/voice config from ~/.hermes/config.yaml (tts: section).
-    The model sends text; the user configures voice and provider.
-
-    On messaging platforms, the returned MEDIA:<path> tag is intercepted
-    by the send pipeline and delivered as a native voice message.
-    In CLI mode, the file is saved to ~/voice-memos/.
-
-    Args:
-        text: The text to convert to speech.
-        output_path: Optional custom save path. Defaults to ~/voice-memos/<timestamp>.mp3
-        speed: Optional playback speed multiplier (0.25-4.0). Overrides config.yaml.
-        instructions: Optional voice-design guidance (tone, emotion, pacing,
-            accent, whispering). Forwarded to the OpenAI backend
-            (gpt-4o-mini-tts and OpenAI-compatible servers). Silently
-            ignored by backends that don't support it.
-        provider: Optional TTS provider override. When set, bypasses the
-            configured ``tts.provider`` and uses this provider instead.
-            Accepts built-in names (``edge``, ``openai``, ``elevenlabs``,
-            ``minimax``, ``xai``, ``mistral``, ``gemini``, ``neutts``,
-            ``kittentts``, ``piper``), user-declared command provider names
-            from ``tts.providers.<name>``, or plugin-registered provider
-            names.  When ``None`` (the default), the configured provider
-            from ``tts.provider`` in config.yaml is used.
-
-    Returns:
-        str: JSON result with success, file_path, and optionally MEDIA tag.
+    The public :func:`text_to_speech_tool` wrapper owns long-form splitting,
+    delivery packing, and post-encoding size enforcement.
     """
     if not text or not text.strip():
         return tool_error("Text is required", success=False)
 
-    try:
-        from tools.tts_text_normalize import prepare_spoken_text
-        text = prepare_spoken_text(text, max_chars=None)
-    except Exception:
-        text = text.strip()
-    if not text:
-        return tool_error("Text is empty after TTS cleanup", success=False)
-
-    tts_config = _load_tts_config()
+    # The wrapper already normalizes text via prepare_spoken_text; the inner
+    # function should not re-normalize or truncate.
+    tts_config = (
+        tts_config_override
+        if tts_config_override is not None
+        else _load_tts_config()
+    )
 
     # When the model supplies a speed parameter, inject it into the config
     # so all downstream provider functions pick it up uniformly.
@@ -2848,15 +3170,16 @@ def text_to_speech_tool(
     # OpenAI handler.
     command_provider_config = _resolve_command_provider_config(provider, tts_config)
 
-    # Truncate very long text with a warning. The cap is per-provider
-    # (OpenAI 4096, xAI 15k, MiniMax 10k, ElevenLabs model-aware, etc.).
+    # The wrapper splits text into provider-safe chunks before calling this
+    # function. If text exceeds the cap here, it means the caller bypassed
+    # the wrapper — log a warning but don't silently truncate.
     max_len = _resolve_max_text_length(provider, tts_config)
     if len(text) > max_len:
         logger.warning(
-            "TTS text too long for provider %s (%d chars), truncating to %d",
+            "TTS text exceeds provider %s cap (%d > %d chars) — "
+            "use text_to_speech_tool() for automatic chunking",
             provider, len(text), max_len,
         )
-        text = text[:max_len]
 
     # Detect platform from gateway env var to choose the best output format.
     # Several platforms deliver native voice bubbles only for Ogg/Opus
@@ -3157,6 +3480,224 @@ def text_to_speech_tool(
         return tool_error(error_msg, success=False)
 
 
+def text_to_speech_tool(
+    text: str,
+    output_path: Optional[str] = None,
+    speed: Optional[float] = None,
+    instructions: Optional[str] = None,
+    provider: Optional[str] = None,
+) -> str:
+    """Convert text to speech audio with long-form chunking.
+
+    Long text is normalized, split into provider-safe chunks, synthesized
+    sequentially, and packed against destination platform upload limits.
+    Each provider request is encoded to its final format before files are
+    packed. Multi-chunk voice output is re-encoded when combined; failed
+    combines preserve separate valid files, and no over-limit final artifact
+    is returned.
+
+    On messaging platforms, the returned MEDIA:<path> tag is intercepted
+    by the send pipeline and delivered as a native voice message.
+    In CLI mode, the file is saved to ~/voice-memos/.
+
+    Args:
+        text: The text to convert to speech. Provider-specific per-request
+            character caps apply automatically (OpenAI 4096, xAI 15000,
+            MiniMax 10000, ElevenLabs 5k-40k depending on model); longer
+            input is split into ordered chunks without silent truncation.
+        output_path: Optional custom save path.
+        speed: Optional playback speed multiplier (0.25-4.0).
+        instructions: Optional voice-design guidance (tone, emotion, pacing).
+        provider: Optional TTS provider override.
+
+    Returns:
+        str: JSON result with success, file_path, file_paths, and MEDIA tag.
+    """
+    if not text or not text.strip():
+        return tool_error("Text is required", success=False)
+
+    # Normalize text via the shared cleaner: markdown, emoji, think blocks,
+    # verifier footer, units, newline flattening.
+    try:
+        from tools.tts_text_normalize import prepare_spoken_text
+        text = prepare_spoken_text(text, max_chars=None)
+    except Exception:
+        text = text.strip()
+    if not text:
+        return tool_error("Text is empty after TTS cleanup", success=False)
+
+    tts_config = _load_tts_config()
+
+    # When the model supplies a speed parameter, inject it into the config
+    # so all downstream provider functions pick it up uniformly.
+    if speed is not None:
+        clamped = max(0.25, min(4.0, float(speed)))
+        tts_config = dict(tts_config)  # shallow copy to avoid mutating the cache
+        tts_config["speed"] = clamped
+
+    # Allow per-call provider override; fall back to the configured default.
+    if provider:
+        provider = provider.lower().strip()
+    else:
+        provider = _get_provider(tts_config)
+
+    command_provider_config = _resolve_command_provider_config(provider, tts_config)
+    max_len = _resolve_max_text_length(provider, tts_config)
+    chunks = _split_text_for_tts(text, max_len)
+    if not chunks:
+        return tool_error("Text is required", success=False)
+    if len(chunks) > 1:
+        logger.info(
+            "TTS text for provider %s split into %d chunks (input=%d chars, cap=%d)",
+            provider,
+            len(chunks),
+            len(text),
+            max_len,
+        )
+
+    from gateway.session_context import get_session_env
+    platform = get_session_env("HERMES_SESSION_PLATFORM", "").lower()
+    want_opus = platform in OPUS_VOICE_PLATFORMS
+    delivery_profile = _resolve_audio_delivery_profile(platform, tts_config)
+
+    # Determine output path (single-chunk short-circuit uses the final path).
+    if output_path:
+        from tools.path_security import has_traversal_component
+        if has_traversal_component(output_path):
+            return json.dumps({
+                "success": False,
+                "error": (
+                    f"output_path contains '..' traversal component: {output_path}. "
+                    "Use an absolute path or one relative to the current directory "
+                    "without '..'."
+                ),
+            }, ensure_ascii=False)
+        base_path = Path(output_path).expanduser()
+        if command_provider_config is not None:
+            base_path = _configured_command_tts_output_path(
+                base_path, command_provider_config,
+            )
+        from agent.file_safety import is_write_denied
+        if is_write_denied(str(base_path)):
+            return json.dumps({
+                "success": False,
+                "error": (
+                    f"output_path targets a protected credential or system path: "
+                    f"{base_path}. Choose a normal audio output location."
+                ),
+            }, ensure_ascii=False)
+    else:
+        timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S_%f")
+        out_dir = Path(DEFAULT_OUTPUT_DIR)
+        out_dir.mkdir(parents=True, exist_ok=True)
+        if command_provider_config is not None:
+            fmt = _get_command_tts_output_format(command_provider_config)
+            base_path = out_dir / f"tts_{timestamp}.{fmt}"
+        elif want_opus and provider in {"openai", "elevenlabs", "mistral", "gemini"}:
+            base_path = out_dir / f"tts_{timestamp}.ogg"
+        else:
+            base_path = out_dir / f"tts_{timestamp}.mp3"
+    base_path.parent.mkdir(parents=True, exist_ok=True)
+
+    generated_artifacts: set[str] = set()
+    final_paths: List[str] = []
+    chunk_results: List[Dict[str, Any]] = []
+    try:
+        encoded_paths: List[str] = []
+        for index, chunk in enumerate(chunks, start=1):
+            if len(chunks) == 1:
+                chunk_path = base_path
+            else:
+                chunk_path = base_path.with_name(
+                    f"{base_path.stem}.chunk{index:03d}{base_path.suffix}"
+                )
+            generated_artifacts.add(str(chunk_path))
+            raw_result = _text_to_speech_single(
+                text=chunk,
+                output_path=str(chunk_path),
+                speed=speed,
+                instructions=instructions,
+                provider=provider,
+                tts_config_override=tts_config,
+            )
+            try:
+                chunk_result = json.loads(raw_result)
+            except (json.JSONDecodeError, TypeError):
+                raise RuntimeError(
+                    f"TTS chunk {index} returned invalid JSON: {str(raw_result)[:200]}"
+                )
+            if not chunk_result.get("success"):
+                error_msg = chunk_result.get("error", "unknown error")
+                return tool_error(
+                    f"TTS chunk {index} failed ({provider}): {error_msg}",
+                    success=False,
+                )
+            actual_path = str(chunk_result.get("file_path") or chunk_path)
+            if not os.path.isfile(actual_path) or os.path.getsize(actual_path) <= 0:
+                raise RuntimeError(
+                    f"TTS chunk {index} produced no final audio: {actual_path}"
+                )
+            generated_artifacts.add(actual_path)
+            encoded_paths.append(actual_path)
+            chunk_results.append(chunk_result)
+
+        voice_compatible = bool(chunk_results) and all(
+            bool(result.get("voice_compatible")) for result in chunk_results
+        )
+        delivery_base = base_path.with_suffix(Path(encoded_paths[0]).suffix)
+        final_paths, combined_chunks = _build_audio_delivery_files(
+            encoded_paths,
+            str(delivery_base),
+            delivery_profile,
+            voice_compatible=voice_compatible,
+        )
+
+        for path in final_paths:
+            logger.info(
+                "TTS audio saved: %s (%s bytes, provider: %s)",
+                path,
+                f"{os.path.getsize(path):,}",
+                provider,
+            )
+        media_tag = "\n".join(f"MEDIA:{path}" for path in final_paths)
+        if voice_compatible:
+            media_tag = f"[[audio_as_voice]]\n{media_tag}"
+
+        return json.dumps({
+            "success": True,
+            "file_path": final_paths[0],
+            "file_paths": final_paths,
+            "media_tag": media_tag,
+            "provider": chunk_results[0].get("provider", provider),
+            "voice_compatible": voice_compatible,
+            "chunk_count": len(chunks),
+            "delivery_file_count": len(final_paths),
+            "combined_chunks": bool(combined_chunks),
+            "delivery_profile": {
+                "platform": delivery_profile.platform,
+                "max_file_bytes": delivery_profile.max_file_bytes,
+                "target_file_bytes": delivery_profile.target_file_bytes,
+            },
+        }, ensure_ascii=False)
+    except ValueError as exc:
+        error_msg = f"TTS delivery error ({provider}): {exc}"
+        logger.error("%s", error_msg)
+        return tool_error(error_msg, success=False)
+    except Exception as exc:
+        error_msg = f"TTS long-form generation failed ({provider}): {exc}"
+        logger.error("%s", error_msg, exc_info=True)
+        return tool_error(error_msg, success=False)
+    finally:
+        final_absolute = {os.path.abspath(path) for path in final_paths}
+        for artifact in generated_artifacts:
+            if os.path.abspath(artifact) in final_absolute:
+                continue
+            try:
+                os.unlink(artifact)
+            except OSError:
+                pass
+
+
 # ===========================================================================
 # Requirements check
 # ===========================================================================
@@ -3915,7 +4456,7 @@ TTS_SCHEMA = {
         "properties": {
             "text": {
                 "type": "string",
-                "description": "The text to convert to speech. Provider-specific character caps apply and are enforced automatically (OpenAI 4096, xAI 15000, MiniMax 10000, ElevenLabs 5k-40k depending on model); over-long input is truncated."
+                "description": "The text to convert to speech. Provider-specific per-request character caps apply automatically (OpenAI 4096, xAI 15000, MiniMax 10000, ElevenLabs 5k-40k depending on model); longer input is split into ordered chunks without silent truncation."
             },
             "output_path": {
                 "type": "string",

website/docs/user-guide/features/tts.md

@@ -149,7 +149,7 @@ The rewrite uses `auxiliary.tts_audio_tags` and defaults to your main chat model
 
 ### Input length limits
 
-Each provider has a documented per-request input-character cap. Hermes truncates text before calling the provider so requests never fail with a length error:
+Each provider has a documented per-request input-character cap. Hermes splits longer replies into ordered, sentence-aware chunks before calling the provider, so the full normalized text is preserved instead of silently truncated:
 
 | Provider | Default cap (chars) |
 |----------|---------------------|
@@ -182,7 +182,7 @@ tts:
     max_text_length: 8192   # raise or lower the provider cap
 ```
 
-Only positive integers are honored. Zero, negative, non-numeric, or boolean values fall through to the provider default, so a broken config can't accidentally disable truncation.
+Only positive integers are honored. Zero, negative, non-numeric, or boolean values fall through to the provider default, so a broken config can't accidentally bypass the provider request limit.
 
 ### Telegram Voice Bubbles & ffmpeg
 
@@ -346,7 +346,7 @@ Use `{{` and `}}` for literal braces.
 | `timeout`          | `120`   | Idle seconds; stdout or stderr output resets the deadline. The process tree is killed after inactivity (Unix `killpg`, Windows `taskkill /T`). |
 | `output_format`    | `mp3`   | One of `mp3` / `wav` / `ogg` / `flac`. Auto-inferred from the output extension if Hermes picks a path.      |
 | `voice_compatible` | `false` | When `true`, Hermes converts MP3/WAV output to Opus/OGG via ffmpeg so Telegram renders a voice bubble.      |
-| `max_text_length`  | `5000`  | Input is truncated to this length before rendering the command.                                             |
+| `max_text_length`  | `5000`  | Maximum input characters per command invocation; longer text is split into ordered chunks.                  |
 | `voice` / `model`  | empty   | Passed to the command as placeholder values only.                                                           |
 
 #### Behavior notes