diff --git a/tools/transcription_audio.py b/tools/transcription_audio.py
index 15413f370a..00d6c25c25 100644
--- a/tools/transcription_audio.py
+++ b/tools/transcription_audio.py
@@ -14,6 +14,7 @@ import os
 import shutil
 import subprocess
 import tempfile
+from functools import partial
 from pathlib import Path
 from typing import Any, Dict, Optional
 
@@ -21,8 +22,7 @@ from hermes_cli._subprocess_compat import windows_hide_flags
 from utils import is_truthy_value
 from tools.transcription_common import (
     COMMON_LOCAL_BIN_DIRS, LOCAL_NATIVE_AUDIO_FORMATS, MAX_FILE_SIZE, SUPPORTED_FORMATS,
-    _config_number, _error_result, _lazy_ensure_quietly, _process_error_detail,
-)
+    _config_number, _error_result, _lazy_ensure_quietly, _process_error_detail)
 
 # Log-record parity with the origin module.
 logger = logging.getLogger("tools.transcription_tools")
@@ -37,16 +37,9 @@ def _find_binary(binary_name: str) -> Optional[str]:
     return shutil.which(binary_name)
 
 
-def _find_ffmpeg_binary() -> Optional[str]:
-    return _find_binary("ffmpeg")
-
-
-def _find_ffprobe_binary() -> Optional[str]:
-    return _find_binary("ffprobe")
-
-
-def _find_whisper_binary() -> Optional[str]:
-    return _find_binary("whisper")
+_find_ffmpeg_binary = partial(_find_binary, "ffmpeg")
+_find_ffprobe_binary = partial(_find_binary, "ffprobe")
+_find_whisper_binary = partial(_find_binary, "whisper")
 
 
 def _run_quiet(command: list, *, timeout: float, env: Optional[dict] = None) -> subprocess.CompletedProcess:
@@ -54,8 +47,7 @@ def _run_quiet(command: list, *, timeout: float, env: Optional[dict] = None) ->
     return subprocess.run(
         command, check=True, capture_output=True, text=True,
         encoding="utf-8", errors="replace", timeout=timeout,
-        stdin=subprocess.DEVNULL, env=env, creationflags=windows_hide_flags(),
-    )
+        stdin=subprocess.DEVNULL, env=env, creationflags=windows_hide_flags())
 
 
 # Shared encode profile for every STT-bound m4a (transcode and silence-trim):
@@ -65,19 +57,15 @@ _STT_M4A_ENCODE_ARGS = ("-vn", "-ac", "1", "-ar", "16000", "-c:a", "aac", "-b:a"
 
 def _run_ffmpeg_stt_encode(ffmpeg: str, input_path: str, output_path: str, *, audio_filter: Optional[str] = None) -> None:
     """Run the shared STT m4a encode, optionally with an ``-af`` filter. Raises on failure; callers own the semantics."""
-    command = [ffmpeg, "-y", "-i", input_path]
-    if audio_filter:
-        command += ["-af", audio_filter]
-    command += [*_STT_M4A_ENCODE_ARGS, output_path]
-    _run_quiet(command, timeout=120)
+    filter_args = ["-af", audio_filter] if audio_filter else []
+    _run_quiet([ffmpeg, "-y", "-i", input_path, *filter_args, *_STT_M4A_ENCODE_ARGS, output_path],
+               timeout=120)
 
 
 def _transcode_audio_for_stt(file_path: str, work_dir: str) -> tuple[Optional[str], Optional[str]]:
     """Transcode to a compact 16 kHz mono AAC/m4a for STT upload; ``(converted_path, None)`` or ``(None, error)``.
-
-    Newer OpenAI models reject containers ``whisper-1`` accepted (notably Ogg/Opus
-    voice notes) and gateway downloads may carry a misleading extension.
-    """
+    Newer OpenAI models reject containers ``whisper-1`` accepted (notably Ogg/Opus voice notes) and
+    gateway downloads may carry a misleading extension."""
     from tools.transcription_tools import _find_ffmpeg_binary, _run_ffmpeg_stt_encode
     ffmpeg = _find_ffmpeg_binary()
     if not ffmpeg:
@@ -121,13 +109,10 @@ def _validate_audio_source_file(file_path: str, *, enforce_size_limit: bool = Tr
 def _validate_audio_file(file_path: str, *, enforce_size_limit: bool = True) -> Optional[Dict[str, Any]]:
     """Validate a supported, decoder-safe audio file."""
     source_error = _validate_audio_source_file(file_path, enforce_size_limit=enforce_size_limit)
-    if source_error:
-        return source_error
-
     suffix = Path(file_path).suffix
-    if suffix.lower() not in SUPPORTED_FORMATS:
-        return _error_result(f"Unsupported format: {suffix}. Supported: {', '.join(sorted(SUPPORTED_FORMATS))}")
-    return None
+    if source_error or suffix.lower() in SUPPORTED_FORMATS:
+        return source_error
+    return _error_result(f"Unsupported format: {suffix}. Supported: {', '.join(sorted(SUPPORTED_FORMATS))}")
 
 
 def _prepare_audio_for_transcription(file_path: str) -> tuple[Optional[str], Optional[str], Optional[Dict[str, Any]]]:
@@ -143,12 +128,10 @@ def _prepare_audio_for_transcription(file_path: str) -> tuple[Optional[str], Opt
             return None, None, _error_result(
                 "Unsupported format: .silk. Install the optional 'pilk' dependency to enable WeChat voice transcription."
             )
-
     temp_dir = tempfile.mkdtemp(prefix="hermes-silk-")
     converted_path = os.path.join(temp_dir, f"{audio_path.stem}.wav")
     try:
         import pilk
-
         pilk.silk_to_wav(file_path, converted_path)
         if not Path(converted_path).is_file() or Path(converted_path).stat().st_size == 0:
             raise RuntimeError("pilk did not produce a readable WAV file")
@@ -165,11 +148,9 @@ def _prepare_local_audio(file_path: str, work_dir: str) -> tuple[Optional[str],
     audio_path = Path(file_path)
     if audio_path.suffix.lower() in LOCAL_NATIVE_AUDIO_FORMATS:
         return file_path, None
-
     ffmpeg = _find_ffmpeg_binary()
     if not ffmpeg:
         return None, "Local STT fallback requires ffmpeg for non-WAV inputs, but ffmpeg was not found"
-
     converted_path = os.path.join(work_dir, f"{audio_path.stem}.wav")
     try:
         _run_quiet([ffmpeg, "-y", "-i", file_path, converted_path], timeout=300)
@@ -194,9 +175,7 @@ def _convert_caf_to_wav(file_path: str) -> Optional[str]:
         ("ffmpeg", [ffmpeg, "-y", "-i", file_path, wav_path] if ffmpeg else None),
         ("afconvert", [afconvert, file_path, wav_path, "-d", "LEI16", "-f", "WAVE"] if afconvert else None),
     )
-    for label, command in candidates:
-        if not command:
-            continue
+    for label, command in ((label, cmd) for label, cmd in candidates if cmd):
         try:
             _run_quiet(command, timeout=300)
             return wav_path
@@ -227,20 +206,17 @@ _CLOUD_TRIM_MIN_INPUT_SECONDS = 12.0
 
 
 def _probe_audio_duration(file_path: str) -> Optional[float]:
-    """Return the audio duration in seconds via ffprobe, or None.
-
-    Canonical sync probe; ``gateway/run.py._probe_audio_duration`` and the
-    Telegram adapter carry local variants — keep the command shape in sync.
-    """
+    """Return the audio duration in seconds via ffprobe, or None. Canonical sync probe;
+    ``gateway/run.py._probe_audio_duration`` and the Telegram adapter carry local variants — keep
+    the command shape in sync."""
     from tools.transcription_tools import _find_ffprobe_binary
     ffprobe = _find_ffprobe_binary()
     if not ffprobe:
         return None
-    command = [
-        ffprobe, "-v", "error", "-show_entries", "format=duration", "-of", "default=noprint_wrappers=1:nokey=1", file_path,
-    ]
     try:
-        return float(_run_quiet(command, timeout=30).stdout.strip())
+        probe = _run_quiet([ffprobe, "-v", "error", "-show_entries", "format=duration", "-of",
+                            "default=noprint_wrappers=1:nokey=1", file_path], timeout=30)
+        return float(probe.stdout.strip())
     except Exception:  # noqa: BLE001 - probe is best-effort
         return None
 
@@ -257,9 +233,7 @@ def _cloud_trim_settings(stt_config: Dict[str, Any]) -> tuple[bool, int, int]:
 
 def _trim_silence_for_cloud_stt(file_path: str, stt_config: Dict[str, Any]) -> Optional[str]:
     """Return a silence-trimmed copy of *file_path* for cloud upload, or None (= upload the original).
-
-    On success the caller owns deleting the returned file's parent directory.
-    """
+    On success the caller owns deleting the returned file's parent directory."""
     from tools.transcription_tools import _find_ffmpeg_binary, _probe_audio_duration, _run_ffmpeg_stt_encode
     enabled, threshold_db, keep_ms = _cloud_trim_settings(stt_config)
     if not enabled:
@@ -274,19 +248,15 @@ def _trim_silence_for_cloud_stt(file_path: str, stt_config: Dict[str, Any]) -> O
         return None
     name = Path(file_path).name
     if original_duration < _CLOUD_TRIM_MIN_INPUT_SECONDS:
-        logger.debug(
-            "Cloud STT silence trim skipped for %s: %.1fs is below the %.0fs gate",
-            name, original_duration, _CLOUD_TRIM_MIN_INPUT_SECONDS,
-        )
+        logger.debug("Cloud STT silence trim skipped for %s: %.1fs is below the %.0fs gate",
+                     name, original_duration, _CLOUD_TRIM_MIN_INPUT_SECONDS)
         return None
-
     keep_seconds = keep_ms / 1000.0
     # start_periods=1 strips leading silence; stop_periods=-1 collapses every interior/trailing silence.
     filter_expr = (
         f"silenceremove="
         f"start_periods=1:start_threshold={threshold_db}dB:start_silence={keep_seconds}:"
-        f"stop_periods=-1:stop_threshold={threshold_db}dB:stop_silence={keep_seconds}"
-    )
+        f"stop_periods=-1:stop_threshold={threshold_db}dB:stop_silence={keep_seconds}")
     work_dir = tempfile.mkdtemp(prefix="hermes-stt-trim-")
     trimmed_path = os.path.join(work_dir, f"{Path(file_path).stem or 'audio'}-trimmed.m4a")
     # Scale the all-silence guard with keep_ms: output that is solely kept pause must never upload as "speech".
@@ -296,20 +266,15 @@ def _trim_silence_for_cloud_stt(file_path: str, stt_config: Dict[str, Any]) -> O
         _run_ffmpeg_stt_encode(ffmpeg, file_path, trimmed_path, audio_filter=filter_expr)
         trimmed_duration = _probe_audio_duration(trimmed_path)
         if not trimmed_duration or trimmed_duration < min_result_seconds:
-            logger.debug(
-                "Cloud STT silence trim discarded for %s: trimmed result ~empty (%.2fs)", name, trimmed_duration or 0.0,
-            )
+            logger.debug("Cloud STT silence trim discarded for %s: trimmed result ~empty (%.2fs)",
+                         name, trimmed_duration or 0.0)
             return None
         if trimmed_duration > original_duration * (1 - _CLOUD_TRIM_MIN_SAVING):
-            logger.debug(
-                "Cloud STT silence trim discarded for %s: saves <%.0f%% (%.1fs -> %.1fs)",
-                name, _CLOUD_TRIM_MIN_SAVING * 100, original_duration, trimmed_duration,
-            )
+            logger.debug("Cloud STT silence trim discarded for %s: saves <%.0f%% (%.1fs -> %.1fs)",
+                         name, _CLOUD_TRIM_MIN_SAVING * 100, original_duration, trimmed_duration)
             return None
-        logger.info(
-            "Trimmed silence from %s before cloud STT upload (%.1fs -> %.1fs, -%d%%)",
-            name, original_duration, trimmed_duration, round((1 - trimmed_duration / original_duration) * 100),
-        )
+        logger.info("Trimmed silence from %s before cloud STT upload (%.1fs -> %.1fs, -%d%%)", name,
+                    original_duration, trimmed_duration, round((1 - trimmed_duration / original_duration) * 100))
         keep_result = True
         return trimmed_path
     except Exception as exc:  # noqa: BLE001 - trim is best-effort
diff --git a/tools/transcription_cloud.py b/tools/transcription_cloud.py
index f3a3414b4b..aca2f9a2de 100644
--- a/tools/transcription_cloud.py
+++ b/tools/transcription_cloud.py
@@ -21,11 +21,14 @@ from tools.transcription_audio import _transcode_audio_for_stt
 from tools.transcription_common import (
     DEFAULT_GROQ_STT_MODEL, DEFAULT_STT_MODEL, ELEVENLABS_STT_BASE_URL, GROQ_BASE_URL, GROQ_MODELS,
     OPENAI_BASE_URL, OPENAI_MODELS, XAI_STT_BASE_URL, _error_result, _get_stt_section,
-    _lazy_ensure_quietly, _log_prompt_unsupported, _ok_result,
-)
+    _lazy_ensure_quietly, _log_prompt_unsupported, _ok_result)
 
 # Log-record parity with the origin module.
 logger = logging.getLogger("tools.transcription_tools")
+# Voxtral-style ``language xx <asr_text> ...`` prefix some SDK responses carry.
+_ASR_TEXT_RE = re.compile(
+    r"\s*language\s+[\w.-]+(?:\s*<audio_language>[^<]*</audio_language>)?\s*<asr_text>\s*(?P<text>.*)",
+    flags=re.IGNORECASE | re.DOTALL)
 
 
 def _has_xai_stt_credentials() -> bool:
@@ -35,10 +38,8 @@ def _has_xai_stt_credentials() -> bool:
 
 def _with_openai_client(api_key: str, base_url: Optional[str], file_path: str, log_label: str, body):
     """Run ``body(client)`` on a fresh OpenAI SDK client (30s timeout, no retries); always closed.
-
-    Errors map to the shared envelope. APIConnectionError is checked before APITimeoutError
-    (its subclass) so timeouts report as connection errors, as they always have.
-    """
+    Errors map to the shared envelope. APIConnectionError is checked before APITimeoutError (its
+    subclass) so timeouts report as connection errors, as they always have."""
     try:
         from openai import OpenAI
         client = OpenAI(api_key=api_key, base_url=base_url, timeout=30, max_retries=0)
@@ -92,27 +93,24 @@ def _transcribe_groq(
     language = language or _resolve_stt_language("groq")
 
     def _run(client):
-        create_kwargs = {"model": model_name, "response_format": "text", **_sdk_prompt_kwargs(language, prompt)}
         with open(file_path, "rb") as audio_file:
-            transcription = client.audio.transcriptions.create(file=audio_file, **create_kwargs)
+            transcription = client.audio.transcriptions.create(file=audio_file, model=model_name,
+                                                               response_format="text",
+                                                               **_sdk_prompt_kwargs(language, prompt))
         transcript_text = str(transcription).strip()
         logger.info("Transcribed %s via Groq API (%s, lang=%s, %d chars)",
                      Path(file_path).name, model_name, language or "auto", len(transcript_text))
         return _ok_result(transcript_text, "groq")
-
     return _with_openai_client(api_key, GROQ_BASE_URL, file_path, "Groq", _run)
 
 
 def _transcribe_openai(
     file_path: str, model_name: str, *, api_key: Optional[str] = None,
     base_url: Optional[str] = None, provider_label: str = "openai", language: Optional[str] = None,
-    prompt: Optional[str] = None,
-) -> Dict[str, Any]:
-    """Transcribe via the OpenAI ``audio.transcriptions.create`` SDK shape.
-
-    Shared by every OpenAI-compatible endpoint (DeepInfra etc.): explicit ``api_key``/
-    ``base_url`` skip the OpenAI-only auth chain; ``provider_label`` names the response's provider.
-    """
+    prompt: Optional[str] = None) -> Dict[str, Any]:
+    """Transcribe via the OpenAI ``audio.transcriptions.create`` SDK shape, shared by every
+    OpenAI-compatible endpoint (DeepInfra etc.): explicit ``api_key``/``base_url`` skip the
+    OpenAI-only auth chain; ``provider_label`` names the response's provider."""
     from tools.transcription_tools import _HAS_OPENAI, _resolve_openai_audio_client_config, _resolve_stt_language
     if api_key is None:
         try:
@@ -137,8 +135,8 @@ def _transcribe_openai(
                 "model": model_name, "response_format": "text" if model_name == "whisper-1" else "json",
             }
             if language:
+                # gpt-transcribe takes a ``languages`` list and rejects the legacy field.
                 if model_name == "gpt-transcribe":
-                    # gpt-transcribe takes a ``languages`` list and rejects the legacy field.
                     create_kwargs["extra_body"] = {"languages": [language]}
                 else:
                     create_kwargs["language"] = language
@@ -147,13 +145,11 @@ def _transcribe_openai(
                 create_kwargs["prompt"] = prompt
             with open(path, "rb") as audio_file:
                 return client.audio.transcriptions.create(file=audio_file, **create_kwargs)
-
         with tempfile.TemporaryDirectory(prefix="hermes-stt-") as work_dir:
             try:
                 transcription = _create_transcription(file_path)
             except BadRequestError as exc:
-                message = str(exc).lower()
-                if not any(k in message for k in ("unsupported", "corrupted", "invalid file")):
+                if not any(k in str(exc).lower() for k in ("unsupported", "corrupted", "invalid file")):
                     raise
                 # Newer models reject containers whisper-1 accepted (Ogg/Opus voice notes): transcode, retry once.
                 converted_path, transcode_error = _transcode_audio_for_stt(file_path, work_dir)
@@ -166,7 +162,6 @@ def _transcribe_openai(
         logger.info("Transcribed %s via %s (%s, %d chars)",
                     Path(file_path).name, provider_label, model_name, len(transcript_text))
         return _ok_result(transcript_text, provider_label)
-
     return _with_openai_client(api_key, base_url, file_path, provider_label, _run)
 
 
@@ -181,24 +176,21 @@ def _transcribe_mistral(
     try:
         _lazy_ensure_quietly("stt.mistral")
         from mistralai.client import Mistral
-        with Mistral(api_key=api_key) as client:
-            with open(file_path, "rb") as audio_file:
-                # Language: hook override > stt.mistral.language > stt.language > env > auto.
-                language = language or _resolve_stt_language("mistral")
-                result = client.audio.transcriptions.complete(
-                    model=model_name, file={"content": audio_file, "file_name": Path(file_path).name},
-                    **_sdk_prompt_kwargs(language, prompt),
-                )
-            transcript_text = _extract_transcript_text(result)
-            logger.info("Transcribed %s via Mistral API (%s, %d chars)",
-                        Path(file_path).name, model_name, len(transcript_text))
-            return _ok_result(transcript_text, "mistral")
+        with Mistral(api_key=api_key) as client, open(file_path, "rb") as audio_file:
+            # Language: hook override > stt.mistral.language > stt.language > env > auto.
+            language = language or _resolve_stt_language("mistral")
+            result = client.audio.transcriptions.complete(
+                model=model_name, file={"content": audio_file, "file_name": Path(file_path).name},
+                **_sdk_prompt_kwargs(language, prompt))
+        transcript_text = _extract_transcript_text(result)
+        logger.info("Transcribed %s via Mistral API (%s, %d chars)",
+                    Path(file_path).name, model_name, len(transcript_text))
+        return _ok_result(transcript_text, "mistral")
     except Exception as e:
         return _cloud_failure(e, file_path, "Mistral transcription", type(e).__name__)
 
 
 # ---- REST multipart backends (xAI, ElevenLabs) ----------------------------
-
 def _post_audio_multipart(url: str, headers: Dict[str, str], file_path: str, data: Dict[str, str]):
     import requests
     with open(file_path, "rb") as audio_file:
@@ -208,14 +200,11 @@ def _post_audio_multipart(url: str, headers: Dict[str, str], file_path: str, dat
 
 def _rest_provider(
     file_path: str, provider: str, label: str, post: Callable[[], Any], extract_detail,
-    extract_text, log: Callable[[str, Dict[str, Any]], None],
-) -> Dict[str, Any]:
-    """Shared multipart REST flow: ``post()`` -> ``log(text, body)`` -> ok envelope.
-
-    Non-200 -> ``"<label> API error (HTTP n): detail"`` (JSON detail via *extract_detail*, else
-    the first 300 body chars); empty text -> the ``no_speech`` envelope (silence is non-fatal);
-    exceptions -> ``_cloud_failure``.
-    """
+    extract_text, log: Callable[[str, Dict[str, Any]], None]) -> Dict[str, Any]:
+    """Shared multipart REST flow: ``post()`` -> ``log(text, body)`` -> ok envelope. Non-200 ->
+    ``"<label> API error (HTTP n): detail"`` (JSON detail via *extract_detail*, else the first 300
+    body chars); empty text -> the ``no_speech`` envelope (silence is non-fatal); exceptions ->
+    ``_cloud_failure``."""
     try:
         response = post()
         if response.status_code != 200:
@@ -245,11 +234,9 @@ def _transcribe_xai(
     # STT is API-billed: prefer the explicit XAI_API_KEY over the xAI OAuth/Grok-subscription
     # credential, which may be valid for Grok yet hit spending-limit errors on /v1/stt.
     direct_api_key = str(get_env_value("XAI_API_KEY") or "").strip()
-    if direct_api_key:
-        creds = {"provider": "xai", "api_key": direct_api_key,
-                 "base_url": str(get_env_value("XAI_BASE_URL") or "https://api.x.ai/v1").strip().rstrip("/")}
-    else:
-        creds = resolve_xai_http_credentials()
+    creds = {"provider": "xai", "api_key": direct_api_key,
+             "base_url": str(get_env_value("XAI_BASE_URL") or "https://api.x.ai/v1").strip().rstrip("/")
+             } if direct_api_key else resolve_xai_http_credentials()
     api_key = str(creds.get("api_key") or "").strip()
     if not api_key:
         return _error_result("No xAI credentials found. Configure xAI OAuth in `hermes model` or set XAI_API_KEY")
@@ -258,10 +245,9 @@ def _transcribe_xai(
 
     def _resolve_base_url(resolved_creds: Dict[str, str]) -> str:
         # OAuth bearers are pinned to the resolver-validated origin; overrides apply to API keys only.
-        if resolved_creds.get("provider") == "xai-oauth":
-            url = resolved_creds.get("base_url")
-        else:
-            url = xai_config.get("base_url") or get_env_value("XAI_STT_BASE_URL") or resolved_creds.get("base_url")
+        url = resolved_creds.get("base_url")
+        if resolved_creds.get("provider") != "xai-oauth":
+            url = xai_config.get("base_url") or get_env_value("XAI_STT_BASE_URL") or url
         return str(url or XAI_STT_BASE_URL).strip().rstrip("/")
 
     # Language: hook override > stt.xai.language > stt.language > env.
@@ -270,9 +256,8 @@ def _transcribe_xai(
     def _post() -> Any:
         from tools.xai_http import hermes_xai_user_agent
         data: Dict[str, str] = {"language": language} if language else {}
-        for flag, default in (("format", True), ("diarize", False)):
-            if is_truthy_value(xai_config.get(flag, default)):
-                data[flag] = "true"
+        data.update({flag: "true" for flag, default in (("format", True), ("diarize", False))
+                     if is_truthy_value(xai_config.get(flag, default))})
 
         def _post_transcription(bearer: str, endpoint_base_url: str):
             headers = {"Authorization": f"Bearer {bearer}", "User-Agent": hermes_xai_user_agent()}
@@ -329,9 +314,7 @@ def _transcribe_elevenlabs(
             "model_id": model_name,
             "tag_audio_events": str(is_truthy_value(elevenlabs_config.get("tag_audio_events", False))).lower(),
             "diarize": str(is_truthy_value(elevenlabs_config.get("diarize", False))).lower(),
-        }
-        if language_code:
-            data["language_code"] = language_code
+            **({"language_code": language_code} if language_code else {})}
         return _post_audio_multipart(f"{base_url}/speech-to-text", {"xi-api-key": api_key}, file_path, data)
 
     def _log(transcript_text: str, _body: Dict[str, Any]) -> None:
@@ -353,20 +336,16 @@ def _transcribe_deepinfra(
     from hermes_cli.models import deepinfra_base_url, deepinfra_model_ids
     # ``stt.deepinfra: null`` in YAML yields None, not {} — coalesce.
     base_url = deepinfra_base_url(_get_stt_section(_load_stt_config(), "deepinfra"))
+    model_name = model_name or next(iter(deepinfra_model_ids("stt")), None)
     if not model_name:
-        candidates = deepinfra_model_ids("stt")
-        if not candidates:
-            return _error_result(
-                "No DeepInfra STT model available. Pin one in config.yaml under stt.deepinfra.model, "
-                "or check connectivity to api.deepinfra.com so the live catalog can be fetched."
-            )
-        model_name = candidates[0]
+        return _error_result(
+            "No DeepInfra STT model available. Pin one in config.yaml under stt.deepinfra.model, "
+            "or check connectivity to api.deepinfra.com so the live catalog can be fetched.")
     return _transcribe_openai(file_path, model_name, api_key=api_key, base_url=base_url,
                               provider_label="deepinfra", language=language, prompt=prompt)
 
 
 # ---- OpenAI audio credential resolution -----------------------------------
-
 def _is_local_or_private_url(url: str) -> bool:
     """True for loopback/RFC-1918/LAN-internal hosts, where an empty ``stt.openai.api_key`` is acceptable
     (local OpenAI-compatible servers ignore the auth header — no sham ``api_key: not-needed`` needed)."""
@@ -374,11 +353,9 @@ def _is_local_or_private_url(url: str) -> bool:
         from urllib.parse import urlparse
         import ipaddress
         host = (urlparse(url).hostname or "").lower()
-        if not host:
-            return False
         if host == "localhost" or host.endswith((".local", ".lan", ".internal")):
             return True
-        addr = ipaddress.ip_address(host)
+        addr = ipaddress.ip_address(host)  # raises for "" and non-IP hostnames
         return addr.is_private or addr.is_loopback
     except Exception:  # unparsable URL or non-IP hostname
         return False
@@ -403,8 +380,7 @@ def _resolve_openai_audio_client_config() -> tuple[str, str]:
     legacy ladder: direct credentials, then the managed gateway. Failures raise ValueError."""
     from tools.transcription_tools import (
         _load_stt_config, managed_nous_tools_enabled, nous_tool_gateway_unavailable_message,
-        resolve_managed_tool_gateway,
-    )
+        resolve_managed_tool_gateway)
     from tools.tool_backend_helpers import NOUS_MANAGED_PROVIDER, read_selection, selection_error
     openai_cfg = _load_stt_config().get("openai") or {}
     selected = read_selection("stt")
@@ -418,10 +394,8 @@ def _resolve_openai_audio_client_config() -> tuple[str, str]:
     if selected == NOUS_MANAGED_PROVIDER:
         managed = _managed()
         if managed is None:
-            raise ValueError(selection_error(
-                "stt", NOUS_MANAGED_PROVIDER,
-                "the Nous Tool Gateway is not available (not entitled or unreachable)",
-            ))
+            raise ValueError(selection_error("stt", NOUS_MANAGED_PROVIDER,
+                                             "the Nous Tool Gateway is not available (not entitled or unreachable)"))
         return managed
     direct = _direct_openai_credentials(openai_cfg.get("api_key", ""), openai_cfg.get("base_url", ""))
     if direct is not None:
@@ -429,8 +403,7 @@ def _resolve_openai_audio_client_config() -> tuple[str, str]:
     if selected is not None:
         raise ValueError(selection_error(
             "stt", selected,
-            "neither stt.openai.api_key in config nor VOICE_TOOLS_OPENAI_KEY/OPENAI_API_KEY is set",
-        ))
+            "neither stt.openai.api_key in config nor VOICE_TOOLS_OPENAI_KEY/OPENAI_API_KEY is set"))
     managed = _managed()
     if managed is None:
         message = "Neither stt.openai.api_key in config nor VOICE_TOOLS_OPENAI_KEY/OPENAI_API_KEY is set"
@@ -446,8 +419,5 @@ def _extract_transcript_text(transcription: Any) -> str:
     if not isinstance(value, str) and isinstance(transcription, dict):
         value = transcription.get("text")
     text = (value if isinstance(value, str) else str(transcription)).strip()
-    match = re.match(
-        r"\s*language\s+[\w.-]+(?:\s*<audio_language>[^<]*</audio_language>)?\s*<asr_text>\s*(?P<text>.*)",
-        text, flags=re.IGNORECASE | re.DOTALL,
-    )
+    match = _ASR_TEXT_RE.match(text)
     return match.group("text").strip() if match else text
diff --git a/tools/transcription_command.py b/tools/transcription_command.py
index 52e00c3dea..c0e63a6fe2 100644
--- a/tools/transcription_command.py
+++ b/tools/transcription_command.py
@@ -12,6 +12,7 @@ from __future__ import annotations
 import logging
 import subprocess
 import tempfile
+from functools import partial
 from pathlib import Path
 from typing import Any, Dict, Optional
 
@@ -19,11 +20,9 @@ from tools.tts_command_provider import (
     _command_output_format, _command_timeout, _is_command_provider_config as _is_command_stt_provider_config,
     _named_provider_config, _resolve_command_config, command_env_passthrough as _command_stt_env_passthrough,
     command_failure_detail, render_command_template as _render_command_stt_template,
-    run_command_provider as _run_command_stt,
-)
+    run_command_provider as _run_command_stt)
 from tools.transcription_common import (
-    BUILTIN_STT_PROVIDERS, _error_result, _log_prompt_unsupported, _ok_result,
-)
+    BUILTIN_STT_PROVIDERS, _error_result, _log_prompt_unsupported, _ok_result)
 
 # Log-record parity with the origin module.
 logger = logging.getLogger("tools.transcription_tools")
@@ -43,81 +42,57 @@ COMMAND_STT_OUTPUT_FORMATS = frozenset({"txt", "json", "srt", "vtt"})
 _NON_COMMAND_STT_NAMES = frozenset(BUILTIN_STT_PROVIDERS | {"none"})
 
 
-def _get_named_stt_provider_config(stt_config: Dict[str, Any], name: str) -> Dict[str, Any]:
-    """``stt.providers.<name>`` (canonical), else ``stt.<name>`` for non-built-in names only."""
-    return _named_provider_config(stt_config, name, BUILTIN_STT_PROVIDERS)
-
-
-def _resolve_command_stt_provider_config(provider: str, stt_config: Dict[str, Any]) -> Optional[Dict[str, Any]]:
-    """The provider config if *provider* is a command type; None for built-ins, ``none``, unknown."""
-    return _resolve_command_config(provider, stt_config, _NON_COMMAND_STT_NAMES)
-
-
-def _get_command_stt_timeout(config: Dict[str, Any]) -> float:
-    return _command_timeout(config, DEFAULT_COMMAND_STT_TIMEOUT_SECONDS)
-
-
-def _get_command_stt_output_format(config: Dict[str, Any]) -> str:
-    return _command_output_format(config, COMMAND_STT_OUTPUT_FORMATS, DEFAULT_COMMAND_STT_OUTPUT_FORMAT)
+# ``stt.providers.<name>`` (canonical), else ``stt.<name>`` for non-built-in names only.
+_get_named_stt_provider_config = partial(_named_provider_config, builtins=BUILTIN_STT_PROVIDERS)
+# The provider config if it is a command type; None for built-ins, ``none``, unknown.
+_resolve_command_stt_provider_config = partial(_resolve_command_config,
+                                               reserved=_NON_COMMAND_STT_NAMES)
+_get_command_stt_timeout = partial(_command_timeout, default=DEFAULT_COMMAND_STT_TIMEOUT_SECONDS)
+_get_command_stt_output_format = partial(_command_output_format, formats=COMMAND_STT_OUTPUT_FORMATS,
+                                         default=DEFAULT_COMMAND_STT_OUTPUT_FORMAT)
 
 
 def _read_command_stt_output(output_path: Path, stdout: str, fmt: str) -> str:
     """Transcript: non-empty output file > non-empty stdout (curl one-liners) > RuntimeError. JSON is returned raw."""
-    if output_path.exists():
-        try:
-            content = output_path.read_text(encoding="utf-8").strip()
-        except UnicodeDecodeError:
-            content = output_path.read_bytes().decode("utf-8", errors="replace").strip()
-        if content:
-            return content
-    if stdout and stdout.strip():
-        return stdout.strip()
+    content = (output_path.read_bytes().decode("utf-8", errors="replace").strip()
+               if output_path.exists() else "")
+    if content or (stdout or "").strip():
+        return content or stdout.strip()
     raise RuntimeError(f"Command STT provider wrote no output file at {output_path} and produced no stdout")
 
 
 def _transcribe_command_stt(
     file_path: str, provider_name: str, config: Dict[str, Any], stt_config: Dict[str, Any],
     model_override: Optional[str] = None, language_override: Optional[str] = None,
-    prompt: Optional[str] = None,
-) -> Dict[str, Any]:
-    """Transcribe via a user-declared ``stt.providers.<name>: type: command``.
-
-    Placeholders (shell-quote-aware; ``{{``/``}}`` stay literal): ``{input_path}``,
-    ``{output_path}`` (transcript file), ``{output_dir}``, ``{format}`` txt/json/srt/vtt,
-    ``{language}`` (default ``en``), ``{model}`` (empty when unset).
-    """
+    prompt: Optional[str] = None) -> Dict[str, Any]:
+    """Transcribe via a user-declared ``stt.providers.<name>: type: command``. Placeholders
+    (shell-quote-aware; ``{{``/``}}`` stay literal): ``{input_path}``, ``{output_path}`` (transcript
+    file), ``{output_dir}``, ``{format}`` txt/json/srt/vtt, ``{language}`` (default ``en``),
+    ``{model}`` (empty when unset)."""
     from tools.transcription_tools import _resolve_stt_language
     if prompt:
         _log_prompt_unsupported(f"Command STT provider '{provider_name}'")
 
     def fail(error: str) -> Dict[str, Any]:
         return _error_result(error, provider=provider_name)
-
     command_template = str(config.get("command") or "").strip()
     if not command_template:
         return fail(f"stt.providers.{provider_name}.command is not configured")
-
     audio = Path(file_path).expanduser()
     if not audio.exists():
         return fail(f"Audio file not found: {file_path}")
-
     timeout = _get_command_stt_timeout(config)
     output_format = _get_command_stt_output_format(config)
-    language = (
-        language_override or config.get("language")
-        or _resolve_stt_language(provider_name, stt_config) or DEFAULT_COMMAND_STT_LANGUAGE
-    )
-    model = model_override or config.get("model") or ""
-
+    language = (language_override or config.get("language")
+                or _resolve_stt_language(provider_name, stt_config) or DEFAULT_COMMAND_STT_LANGUAGE)
     try:
         with tempfile.TemporaryDirectory(prefix=f"hermes-cmd-stt-{provider_name}-") as tmpdir:
             output_path = Path(tmpdir) / f"transcript.{output_format}"
-            placeholders = {
+            command = _render_command_stt_template(command_template, {
                 "input_path": str(audio.resolve()), "output_path": str(output_path),
                 "output_dir": str(output_path.parent), "format": output_format,
-                "language": str(language), "model": str(model),
-            }
-            command = _render_command_stt_template(command_template, placeholders)
+                "language": str(language), "model": str(model_override or config.get("model") or ""),
+            })
             logger.info("Transcribing %s via command STT provider '%s'...", audio.name, provider_name)
             result = _run_command_stt(command, timeout, env_passthrough=_command_stt_env_passthrough(config))
             transcript_text = _read_command_stt_output(output_path, result.stdout or "", output_format)
@@ -131,7 +106,6 @@ def _transcribe_command_stt(
         return fail(str(exc))
     except OSError as exc:
         return fail(f"STT command provider '{provider_name}' failed: {exc}")
-
     logger.info("Transcribed %s via command STT provider '%s' (%d chars)", audio.name, provider_name, len(transcript_text))
     return _ok_result(transcript_text, provider_name)
 
@@ -144,8 +118,7 @@ def _unregistered_stt_provider_error(provider: str) -> Dict[str, Any]:
         "installed STT plugins, or configure a command provider under "
         f"`stt.providers.{key}.command`.",
         provider=key,
-        error_type="provider_not_registered",
-    )
+        error_type="provider_not_registered")
 
 
 def _dispatch_to_plugin_provider(
@@ -153,23 +126,18 @@ def _dispatch_to_plugin_provider(
     model: Optional[str] = None, language: Optional[str] = None, prompt: Optional[str] = None,
 ) -> Optional[Dict[str, Any]]:
     """Route to a plugin-registered transcription provider; None when no plugin claims the name.
-
-    Invariants re-verified here so a caller refactor can't break them: built-in names never
-    reach the registry; a same-name command provider wins over a plugin. A matched plugin with
+    Invariants re-verified here so a caller refactor can't break them: built-in names never reach
+    the registry; a same-name command provider wins over a plugin. A matched plugin with
     ``is_available() == False`` returns an error envelope — not None — because the user
-    explicitly opted in via ``stt.provider``. Provider exceptions become the error envelope.
-    """
-    if not provider:
-        return None
-    key = provider.lower().strip()
-    if key in _NON_COMMAND_STT_NAMES:
+    explicitly opted in via ``stt.provider``. Provider exceptions become the error envelope."""
+    key = (provider or "").lower().strip()
+    if not key or key in _NON_COMMAND_STT_NAMES:
         return None
     if stt_config is not None and _is_command_stt_provider_config(_get_named_stt_provider_config(stt_config, key)):
         return None
     try:
         from agent.transcription_registry import get_provider
         from hermes_cli.plugins import _ensure_plugins_discovered
-
         _ensure_plugins_discovered()
         plugin_provider = get_provider(key)
         if plugin_provider is None:
@@ -182,7 +150,6 @@ def _dispatch_to_plugin_provider(
         return None
     if plugin_provider is None:
         return None
-
     # ``is_available()`` MUST NOT raise per the ABC contract; defend anyway so
     # a buggy plugin can't break dispatch for everyone.
     try:
@@ -196,9 +163,7 @@ def _dispatch_to_plugin_provider(
         logger.info("STT plugin provider '%s' reports not available; returning unavailability envelope.", key)
         return _error_result(
             f"STT plugin '{key}' is not available — check that its required credentials / dependencies are configured.",
-            provider=key,
-        )
-
+            provider=key)
     logger.info("Transcribing with plugin STT provider '%s'...", key)
     # The prompt travels via the ABC's ``**extra`` kwargs and is only sent when
     # set, so pre-prompt providers see byte-identical calls on the no-prompt path.
@@ -208,7 +173,6 @@ def _dispatch_to_plugin_provider(
     except Exception as exc:  # noqa: BLE001
         logger.warning("STT plugin provider '%s' raised: %s", key, exc, exc_info=True)
         return _error_result(f"STT plugin '{key}' raised: {exc}", provider=key)
-
     if not isinstance(result, dict):
         return _error_result(f"STT plugin '{key}' returned a non-dict result", provider=key)
     result.setdefault("provider", key)
@@ -229,16 +193,13 @@ _WHISPER_PROMPT_CAPPED_PROVIDERS = frozenset({"local", "openai", "groq", "deepin
 def _enforce_prompt_length_limit(prompt: Optional[str], provider: str) -> Optional[str]:
     """Truncate *prompt* to the whisper-family token cap, keeping the TAIL (whisper conditions
     on the final context window, so the newest hints survive). Other providers self-validate."""
-    if not prompt or provider not in _WHISPER_PROMPT_CAPPED_PROVIDERS:
-        return prompt
     max_chars = _WHISPER_PROMPT_TOKEN_CAP * _PROMPT_CHARS_PER_TOKEN
-    if len(prompt) <= max_chars:
+    if not prompt or provider not in _WHISPER_PROMPT_CAPPED_PROVIDERS or len(prompt) <= max_chars:
         return prompt
     logger.warning(
         "Transcription prompt is ~%d tokens; whisper-family provider '%s' "
         "only uses the final ~%d — truncating to the last %d characters.",
-        len(prompt) // _PROMPT_CHARS_PER_TOKEN, provider, _WHISPER_PROMPT_TOKEN_CAP, max_chars,
-    )
+        len(prompt) // _PROMPT_CHARS_PER_TOKEN, provider, _WHISPER_PROMPT_TOKEN_CAP, max_chars)
     return prompt[-max_chars:]
 
 
@@ -247,32 +208,24 @@ def _apply_pre_transcription_hook(
     prompt: Optional[str], source: Optional[str],
 ) -> tuple[Optional[str], Optional[str], Optional[str]]:
     """Fire the ``pre_transcription`` plugin hook; returns ``(model, language_override, prompt)``.
-
-    Gated on ``has_hook`` (the no-hook path never builds kwargs) and fail-open: any
-    plumbing error leaves the dispatch untouched. Results apply field-by-field in
-    registration order (last hook wins). ``language_override`` is None unless a hook
-    explicitly set ``language``, so backends keep their own config/env resolution.
-    """
+    Gated on ``has_hook`` (the no-hook path never builds kwargs) and fail-open: any plumbing error
+    leaves the dispatch untouched. Results apply field-by-field in registration order (last hook
+    wins). ``language_override`` is None unless a hook explicitly set ``language``, so backends keep
+    their own config/env resolution."""
     try:
         from hermes_cli.plugins import has_hook, invoke_hook
-
         if not has_hook("pre_transcription"):
             return model, None, prompt
-
         hook_results = invoke_hook(
             "pre_transcription", file_path=file_path, provider=provider,
-            model=model, language=language, prompt=prompt, source=source,
-        )
+            model=model, language=language, prompt=prompt, source=source)
         overrides: Dict[str, Any] = {}
         for hook_result in hook_results:
-            if not isinstance(hook_result, dict):
-                continue
-            for key, value in hook_result.items():
+            for key, value in (hook_result.items() if isinstance(hook_result, dict) else ()):
                 if key == "file_path":
                     logger.warning(
                         "pre_transcription hook attempted to change "
-                        "file_path (read-only) — ignoring the attempt."
-                    )
+                        "file_path (read-only) — ignoring the attempt.")
                 elif key not in _PRE_TRANSCRIPTION_MUTABLE_FIELDS:
                     logger.debug("pre_transcription hook returned unsupported field %r — ignoring.", key)
                 elif not isinstance(value, str):
@@ -281,13 +234,10 @@ def _apply_pre_transcription_hook(
                     )
                 else:
                     overrides[key] = value
-
-        if "model" in overrides:
-            model = overrides["model"]
+        # Hooks win over the static ``stt.prompt`` config; "" clears it.
         if "prompt" in overrides:
-            # Hooks win over the static ``stt.prompt`` config; "" clears it.
             prompt = overrides["prompt"] or None
-        return model, overrides.get("language") or None, prompt
+        return overrides.get("model", model), overrides.get("language") or None, prompt
     except Exception as _hook_err:  # noqa: BLE001 — hook plumbing is fail-open
         logger.debug("pre_transcription hook error: %s", _hook_err)
         return model, None, prompt
diff --git a/tools/transcription_common.py b/tools/transcription_common.py
index 56e4d695a8..94db230c20 100644
--- a/tools/transcription_common.py
+++ b/tools/transcription_common.py
@@ -45,8 +45,7 @@ GROQ_MODELS = {"whisper-large-v3", "whisper-large-v3-turbo", "distil-whisper-lar
 # (a regression test fails on drift); plugins may not register under these names and the
 # dispatcher short-circuits them before command/plugin lookup.
 BUILTIN_STT_PROVIDERS = frozenset({
-    "local", "local_command", "groq", "openai", "mistral", "xai", "elevenlabs", "deepinfra",
-})
+    "local", "local_command", "groq", "openai", "mistral", "xai", "elevenlabs", "deepinfra"})
 # Built-in providers that upload audio to a remote API.
 CLOUD_STT_PROVIDERS = frozenset(BUILTIN_STT_PROVIDERS - {"local", "local_command"})
 
@@ -62,10 +61,8 @@ def _ok_result(transcript: str, provider: str) -> Dict[str, Any]:
 
 def _lazy_ensure_quietly(dep: str) -> None:
     """Best-effort ``tools.lazy_deps.ensure(dep, prompt=False)``; failures are swallowed.
-
-    prompt=False: a bare input() deadlocks under the interactive CLI where
-    prompt_toolkit owns stdin; installs are gated by ``security.allow_lazy_installs``.
-    """
+    prompt=False: a bare input() deadlocks under the interactive CLI where prompt_toolkit owns
+    stdin; installs are gated by ``security.allow_lazy_installs``."""
     try:
         from tools.lazy_deps import ensure
         ensure(dep, prompt=False)
diff --git a/tools/transcription_local.py b/tools/transcription_local.py
index 306ca7d516..515d391f28 100644
--- a/tools/transcription_local.py
+++ b/tools/transcription_local.py
@@ -23,8 +23,7 @@ from tools.transcription_audio import _run_quiet
 from tools.transcription_common import (
     DEFAULT_LOCAL_MODEL, DEFAULT_LOCAL_STT_LANGUAGE, GROQ_MODELS, LOCAL_STT_COMMAND_ENV,
     OPENAI_MODELS, _config_number, _error_result, _log_prompt_unsupported, _ok_result,
-    _process_error_detail,
-)
+    _process_error_detail)
 
 # Log-record parity with the origin module.
 logger = logging.getLogger("tools.transcription_tools")
@@ -36,12 +35,8 @@ def _get_local_command_template() -> Optional[str]:
     if configured:
         return configured
     whisper_binary = _find_whisper_binary()
-    if whisper_binary:
-        return (
-            f"{shlex.quote(whisper_binary)} {{input_path}} --model {{model}} --output_format txt "
-            "--output_dir {output_dir} --language {language}"
-        )
-    return None
+    return (f"{shlex.quote(whisper_binary)} {{input_path}} --model {{model}} --output_format txt "
+            "--output_dir {output_dir} --language {language}") if whisper_binary else None
 
 
 def _has_local_command() -> bool:
@@ -52,13 +47,12 @@ def _normalize_local_model(model_name: Optional[str]) -> str:
     """Return a valid faster-whisper size; cloud-only names (``whisper-1`` …) fall back to the default with a warning."""
     if not model_name:
         return DEFAULT_LOCAL_MODEL
-    if model_name in OPENAI_MODELS or model_name in GROQ_MODELS:
+    if model_name in OPENAI_MODELS | GROQ_MODELS:
         logger.warning(
             "STT model '%s' is a cloud-only name and cannot be used with the local "
             "provider. Falling back to '%s'. Set stt.local.model to a valid "
             "faster-whisper size (tiny, base, small, medium, large-v3).",
-            model_name, DEFAULT_LOCAL_MODEL,
-        )
+            model_name, DEFAULT_LOCAL_MODEL)
         return DEFAULT_LOCAL_MODEL
     return model_name
 
@@ -81,8 +75,7 @@ def _try_lazy_install_stt() -> bool:
             "venv owner: `stat -c '%%u' '$(dirname $(dirname $(which python3)))'` "
             "then `su - <owner> -c 'VIRTUAL_ENV=/opt/hermes/.venv "
             "uv pip install faster-whisper==1.2.1'`",
-            exc,
-        )
+            exc)
     return False
 
 
@@ -93,22 +86,19 @@ def _try_lazy_install_stt() -> bool:
 _CUDA_LIB_ERROR_MARKERS = (
     "libcublas", "libcudnn", "libcudart", "cannot be loaded", "cannot open shared object",
     "no kernel image is available", "CUBLAS_STATUS_NOT_SUPPORTED", "no CUDA-capable device",
-    "CUDA driver version is insufficient",
-)
+    "CUDA driver version is insufficient")
 
 
 def _looks_like_cuda_lib_error(exc: BaseException) -> bool:
     """Heuristic: is this a missing/broken CUDA runtime library (not a legitimate runtime failure)?"""
-    msg = str(exc)
-    return any(marker in msg for marker in _CUDA_LIB_ERROR_MARKERS)
+    return any(marker in str(exc) for marker in _CUDA_LIB_ERROR_MARKERS)
 
 
 def _sysctl_value(name: str) -> str:
     """Return a sysctl value, or an empty string when unavailable."""
     try:
-        return subprocess.check_output(
-            ["/usr/sbin/sysctl", "-n", name], stderr=subprocess.DEVNULL, text=True, timeout=2,
-        ).strip()
+        return subprocess.check_output(["/usr/sbin/sysctl", "-n", name], stderr=subprocess.DEVNULL, text=True,
+                                       timeout=2).strip()
     except Exception:
         return ""
 
@@ -131,37 +121,27 @@ def _get_idle_unload_seconds(local_cfg: Dict[str, Any]) -> int:
 
 
 def _load_local_whisper_model(model_name: str, device: str = "auto", compute_type: str = "auto"):
-    """Load faster-whisper with graceful CUDA → CPU fallback.
-
-    ``device="auto"`` picks CUDA whenever the ctranslate2 wheel ships CUDA libs, even
-    on hosts without the NVIDIA runtime (WSL2, headless servers). Try the requested
-    config first; on a CUDA library load failure fall back to CPU + int8. Pass
-    ``stt.local.device`` / ``compute_type`` to pin.
-    """
+    """Load faster-whisper with graceful CUDA → CPU fallback. ``device="auto"`` picks CUDA
+    whenever the ctranslate2 wheel ships CUDA libs, even on hosts without the NVIDIA runtime (WSL2,
+    headless servers): try the requested config first; on a CUDA library load failure fall back to
+    CPU + int8. Pass ``stt.local.device`` / ``compute_type`` to pin."""
     force_cpu = _should_force_faster_whisper_cpu()
     if force_cpu:
         # Importing ctranslate2 can itself abort on Apple Silicon/Rosetta when
         # multiple Intel OpenMP runtimes are loaded — set before the import.
         os.environ.setdefault("KMP_DUPLICATE_LIB_OK", "TRUE")
-
     from faster_whisper import WhisperModel
     if force_cpu:
-        logger.info(
-            "Apple Silicon/Rosetta detected — loading faster-whisper on CPU "
-            "(int8) to avoid native device autodetection crashes"
-        )
+        logger.info("Apple Silicon/Rosetta detected — loading faster-whisper on CPU "
+                    "(int8) to avoid native device autodetection crashes")
         return WhisperModel(model_name, device="cpu", compute_type="int8")
-
     try:
         return WhisperModel(model_name, device=device, compute_type=compute_type)
     except Exception as exc:
         if not _looks_like_cuda_lib_error(exc):
             raise
-        logger.warning(
-            "faster-whisper CUDA load failed (%s) — falling back to CPU (int8). "
-            "Install the NVIDIA CUDA runtime (libcublas/libcudnn) to use GPU.",
-            exc,
-        )
+        logger.warning("faster-whisper CUDA load failed (%s) — falling back to CPU (int8). "
+                       "Install the NVIDIA CUDA runtime (libcublas/libcudnn) to use GPU.", exc)
         return WhisperModel(model_name, device="cpu", compute_type="int8")
 
 
@@ -180,56 +160,45 @@ def build_local_transcribe_kwargs(stt_config: Optional[Dict[str, Any]] = None) -
     from tools.transcription_tools import _load_stt_config, _resolve_stt_language
     stt_config = stt_config if isinstance(stt_config, dict) else _load_stt_config()
     local_cfg = stt_config.get("local") or {}
-
     # ``vad: null`` in YAML means "default on".
     vad_enabled = local_cfg.get("vad", True)
     kwargs: Dict[str, Any] = {
         "beam_size": 5,
         "condition_on_previous_text": False,
-        "vad_filter": vad_enabled is None or bool(vad_enabled),
-    }
+        "vad_filter": vad_enabled is None or bool(vad_enabled)}
     if kwargs["vad_filter"]:
         kwargs["vad_parameters"] = {
             "min_silence_duration_ms": _config_number(local_cfg, "vad_min_silence_ms", _VAD_MIN_SILENCE_MS_DEFAULT, int)
         }
-
     # Push the confidence gate into faster-whisper itself: its internal defaults drop
     # low-confidence segments BEFORE our post-filter sees them, so the ``stt.local``
     # threshold knobs were dead for that first gate (non-English speech decodes at
     # lower avg_logprob and was silently discarded). Same values feed both gates.
     kwargs["no_speech_threshold"], kwargs["log_prob_threshold"] = _confidence_thresholds(local_cfg)
-
     forced_lang = _resolve_stt_language("local", stt_config)
     if forced_lang:
         kwargs["language"] = forced_lang
-
     initial_prompt = local_cfg.get("initial_prompt")
     if isinstance(initial_prompt, str) and initial_prompt.strip():
         kwargs["initial_prompt"] = initial_prompt
-
     return kwargs
 
 
 def _confidence_thresholds(local_cfg: Dict[str, Any]) -> tuple[float, float]:
     """Resolve (no_speech_prob, avg_logprob) gate thresholds from config."""
-    return (
-        _config_number(local_cfg, "no_speech_prob_threshold", _NO_SPEECH_PROB_THRESHOLD_DEFAULT),
-        _config_number(local_cfg, "logprob_threshold", _LOGPROB_THRESHOLD_DEFAULT),
-    )
+    return (_config_number(local_cfg, "no_speech_prob_threshold", _NO_SPEECH_PROB_THRESHOLD_DEFAULT),
+            _config_number(local_cfg, "logprob_threshold", _LOGPROB_THRESHOLD_DEFAULT))
 
 
 def _is_hallucinated_segment(segment: Any, no_speech_threshold: float, logprob_threshold: float) -> bool:
-    """True when a segment is very likely a silence hallucination.
-
-    Conservative AND gate (openai-whisper's own heuristic): non-speech AND low decode
-    confidence, so quiet-but-real speech survives. Unknown segment shapes are never dropped.
-    """
+    """True when a segment is very likely a silence hallucination. Conservative AND gate
+    (openai-whisper's own heuristic): non-speech AND low decode confidence, so quiet-but-real speech
+    survives. Unknown segment shapes are never dropped."""
     try:
-        no_speech_prob = float(getattr(segment, "no_speech_prob"))
-        avg_logprob = float(getattr(segment, "avg_logprob"))
+        return (float(segment.no_speech_prob) > no_speech_threshold
+                and float(segment.avg_logprob) < logprob_threshold)
     except (AttributeError, TypeError, ValueError):
         return False
-    return no_speech_prob > no_speech_threshold and avg_logprob < logprob_threshold
 
 
 def _join_confident_segments(segments: Any, local_cfg: Dict[str, Any]) -> str:
@@ -238,11 +207,9 @@ def _join_confident_segments(segments: Any, local_cfg: Dict[str, Any]) -> str:
     kept: list[str] = []
     for segment in segments:
         if _is_hallucinated_segment(segment, no_speech_threshold, logprob_threshold):
-            logger.debug(
-                "Dropping probable hallucinated segment %r (no_speech_prob=%.3f, avg_logprob=%.3f)",
-                getattr(segment, "text", ""), getattr(segment, "no_speech_prob", float("nan")),
-                getattr(segment, "avg_logprob", float("nan")),
-            )
+            logger.debug("Dropping probable hallucinated segment %r (no_speech_prob=%.3f, avg_logprob=%.3f)",
+                         getattr(segment, "text", ""), getattr(segment, "no_speech_prob", float("nan")),
+                         getattr(segment, "avg_logprob", float("nan")))
             continue
         kept.append(segment.text.strip())
     return " ".join(kept).strip()
@@ -255,41 +222,30 @@ def _transcribe_local_command(
     from tools.transcription_tools import _prepare_local_audio, _resolve_stt_language
     if prompt:
         _log_prompt_unsupported("STT provider 'local_command'")
-
     command_template = _get_local_command_template()
     if not command_template:
         return _error_result(f"{LOCAL_STT_COMMAND_ENV} not configured and no local whisper binary was found")
-
     # Language: hook override > stt.local.language > stt.language > env > "en".
     language = language or _resolve_stt_language("local") or DEFAULT_LOCAL_STT_LANGUAGE
     normalized_model = _normalize_local_model(model_name)
-
     try:
         with tempfile.TemporaryDirectory(prefix="hermes-local-stt-") as output_dir:
             prepared_input, prep_error = _prepare_local_audio(file_path, output_dir)
             if prep_error:
                 return _error_result(prep_error)
-
             command = command_template.format(
                 input_path=shlex.quote(prepared_input), output_dir=shlex.quote(output_dir),
-                language=shlex.quote(language), model=shlex.quote(normalized_model),
-            )
+                language=shlex.quote(language), model=shlex.quote(normalized_model))
             # Scrub Hermes secrets from the child env (same policy as _run_command_stt).
             from tools.environments.local import hermes_subprocess_env
-
             _run_quiet(shlex.split(command), timeout=300, env=hermes_subprocess_env(inherit_credentials=False))
-
             txt_files = sorted(Path(output_dir).glob("*.txt"))
             if not txt_files:
                 return _error_result("Local STT command completed but did not produce a .txt transcript")
-
             transcript_text = txt_files[0].read_text(encoding="utf-8").strip()
-            logger.info(
-                "Transcribed %s via local STT command (%s, %d chars)",
-                Path(file_path).name, normalized_model, len(transcript_text),
-            )
+            logger.info("Transcribed %s via local STT command (%s, %d chars)",
+                        Path(file_path).name, normalized_model, len(transcript_text))
             return _ok_result(transcript_text, "local_command")
-
     except KeyError as e:
         return _error_result(f"Invalid {LOCAL_STT_COMMAND_ENV} template, missing placeholder: {e}")
     except subprocess.CalledProcessError as e:
diff --git a/tools/transcription_tools.py b/tools/transcription_tools.py
index da1a8ac01b..082fe8389c 100644
--- a/tools/transcription_tools.py
+++ b/tools/transcription_tools.py
@@ -25,42 +25,36 @@ from hermes_cli._subprocess_compat import windows_hide_flags  # noqa: F401  (imp
 from utils import is_truthy_value
 from tools.managed_tool_gateway import resolve_managed_tool_gateway  # noqa: F401  (patched by tests)
 from tools.tool_backend_helpers import (  # noqa: F401  (patched by tests; read lazily by transcription_cloud)
-    managed_nous_tools_enabled, nous_tool_gateway_unavailable_message, resolve_openai_audio_api_key,
-)
+    managed_nous_tools_enabled, nous_tool_gateway_unavailable_message, resolve_openai_audio_api_key)
 from tools.transcription_common import (  # noqa: F401  (re-exported; tests patch tools.transcription_tools.<name>)
     BUILTIN_STT_PROVIDERS, CLOUD_STT_PROVIDERS, DEFAULT_ELEVENLABS_STT_MODEL,
     DEFAULT_GROQ_STT_MODEL, DEFAULT_LOCAL_MODEL, DEFAULT_MISTRAL_STT_MODEL, DEFAULT_PROVIDER,
     DEFAULT_STT_MODEL, ELEVENLABS_STT_BASE_URL, GROQ_MODELS, LOCAL_STT_COMMAND_ENV,
     LOCAL_STT_LANGUAGE_ENV, MAX_FILE_SIZE, OPENAI_MODELS, SUPPORTED_FORMATS, XAI_STT_BASE_URL,
-    _error_result, _get_stt_section, _ok_result,
-)
+    _error_result, _get_stt_section, _ok_result)
 from tools.transcription_audio import (  # noqa: F401  (re-exported; tests patch tools.transcription_tools.<name>)
     _CLOUD_TRIM_KEEP_MS_DEFAULT, _CLOUD_TRIM_MIN_INPUT_SECONDS, _CLOUD_TRIM_THRESHOLD_DB_DEFAULT,
     _cloud_trim_settings, _convert_caf_to_wav, _find_ffmpeg_binary, _find_ffprobe_binary,
     _find_whisper_binary, _prepare_audio_for_transcription, _prepare_local_audio,
     _probe_audio_duration, _run_ffmpeg_stt_encode, _trim_silence_for_cloud_stt,
-    _validate_audio_file, _validate_audio_file_size, _validate_audio_source_file,
-)
+    _validate_audio_file, _validate_audio_file_size, _validate_audio_source_file)
 from tools.transcription_local import (  # noqa: F401  (re-exported; tests patch tools.transcription_tools.<name>)
     _LOGPROB_THRESHOLD_DEFAULT, _NO_SPEECH_PROB_THRESHOLD_DEFAULT, _get_idle_unload_seconds,
     _get_local_command_template, _has_local_command, _is_hallucinated_segment,
     _join_confident_segments, _load_local_whisper_model, _looks_like_cuda_lib_error,
     _normalize_local_model, _transcribe_local_command, _try_lazy_install_stt,
-    build_local_transcribe_kwargs,
-)
+    build_local_transcribe_kwargs)
 from tools.transcription_cloud import (  # noqa: F401  (re-exported; tests patch tools.transcription_tools.<name>)
     _extract_transcript_text, _has_xai_stt_credentials, _is_local_or_private_url,
     _resolve_openai_audio_client_config, _transcribe_deepinfra, _transcribe_elevenlabs,
-    _transcribe_groq, _transcribe_mistral, _transcribe_openai, _transcribe_xai,
-)
+    _transcribe_groq, _transcribe_mistral, _transcribe_openai, _transcribe_xai)
 from tools.transcription_command import (  # noqa: F401  (re-exported; tests patch tools.transcription_tools.<name>)
     COMMAND_STT_OUTPUT_FORMATS, DEFAULT_COMMAND_STT_LANGUAGE, DEFAULT_COMMAND_STT_OUTPUT_FORMAT,
     DEFAULT_COMMAND_STT_TIMEOUT_SECONDS, _PROMPT_CHARS_PER_TOKEN, _WHISPER_PROMPT_TOKEN_CAP,
     _apply_pre_transcription_hook, _dispatch_to_plugin_provider, _enforce_prompt_length_limit,
     _get_command_stt_output_format, _get_command_stt_timeout, _get_named_stt_provider_config,
     _render_command_stt_template, _resolve_command_stt_provider_config, _run_command_stt,
-    _transcribe_command_stt, _unregistered_stt_provider_error,
-)
+    _transcribe_command_stt, _unregistered_stt_provider_error)
 
 logger = logging.getLogger(__name__)
 
@@ -91,10 +85,8 @@ def _safe_find_spec(module_name: str) -> bool:
         return module_name in globals() or module_name in os.sys.modules
 
 
-_HAS_FASTER_WHISPER = _safe_find_spec("faster_whisper")
-_HAS_OPENAI = _safe_find_spec("openai")
-_HAS_MISTRAL = _safe_find_spec("mistralai")
-_HAS_PILK = _safe_find_spec("pilk")
+_HAS_FASTER_WHISPER, _HAS_OPENAI, _HAS_MISTRAL, _HAS_PILK = map(
+    _safe_find_spec, ("faster_whisper", "openai", "mistralai", "pilk"))
 
 # Local model singleton; the lock guards check-then-load against concurrent voice messages.
 _local_model: Optional[object] = None
@@ -112,7 +104,6 @@ _IDLE_UNLOAD_CHECK_INTERVAL = 30  # seconds between idle checks
 
 
 # ---- Config helpers -----------------------------------------------------
-
 def _load_stt_config() -> dict:
     """Load the ``stt`` section from user config, falling back to defaults."""
     try:
@@ -123,19 +114,16 @@ def _load_stt_config() -> dict:
 
 
 def is_stt_enabled(stt_config: Optional[dict] = None) -> bool:
-    if stt_config is None:
-        stt_config = _load_stt_config()
-    return is_truthy_value(stt_config.get("enabled", True), default=True)
+    cfg = _load_stt_config() if stt_config is None else stt_config
+    return is_truthy_value(cfg.get("enabled", True), default=True)
 
 
 def _resolve_stt_language(
     provider_key: str, stt_config: Optional[Dict[str, Any]] = None, *, extra_keys: tuple = ()
 ) -> Optional[str]:
-    """Language hint for an STT provider, first non-empty wins (never "").
-
-    ``stt.<provider>.language`` (plus *extra_keys* aliases, e.g. ``language_code``) >
-    ``stt.language`` > ``HERMES_LOCAL_STT_LANGUAGE`` env > None (provider auto-detects).
-    """
+    """Language hint for an STT provider, first non-empty wins (never ""): ``stt.<provider>.language``
+    (plus *extra_keys* aliases, e.g. ``language_code``) > ``stt.language`` > ``HERMES_LOCAL_STT_LANGUAGE``
+    env > None (provider auto-detects)."""
     if stt_config is None:
         stt_config = _load_stt_config()
     provider_cfg = _get_stt_section(stt_config, provider_key)
@@ -143,10 +131,7 @@ def _resolve_stt_language(
     if isinstance(stt_config, dict):
         candidates.append(stt_config.get("language"))
     candidates.append(os.getenv(LOCAL_STT_LANGUAGE_ENV))
-    for candidate in candidates:
-        if isinstance(candidate, str) and candidate.strip():
-            return candidate.strip()
-    return None
+    return next((c.strip() for c in candidates if isinstance(c, str) and c.strip()), None)
 
 
 def _openai_audio_unavailable_reason() -> Optional[str]:
@@ -168,13 +153,11 @@ def _is_local_stt_provider(provider: str, stt_config: Dict[str, Any]) -> bool:
 
 
 # ---- Provider resolution ------------------------------------------------
-
 def _has_key(env_var: str, provider: str, *, needs_openai: bool = False, needs_mistral: bool = False):
     """Availability probe factory: optional SDK flag AND a resolvable API key."""
     def probe() -> bool:
-        if (needs_openai and not _HAS_OPENAI) or (needs_mistral and not _HAS_MISTRAL):
-            return False
-        return bool(_resolve_provider_key(env_var, provider))
+        sdk_ok = (not needs_openai or _HAS_OPENAI) and (not needs_mistral or _HAS_MISTRAL)
+        return sdk_ok and bool(_resolve_provider_key(env_var, provider))
     return probe
 
 
@@ -201,18 +184,15 @@ def _detect_local_backend() -> Optional[str]:
     """faster-whisper > local whisper CLI > lazy-installed faster-whisper; None when nothing local works."""
     if _HAS_FASTER_WHISPER:
         return "local"
-    if _has_local_command():
-        return "local_command"
-    return "local" if _try_lazy_install_stt() else None
+    return "local_command" if _has_local_command() else ("local" if _try_lazy_install_stt() else None)
 
 
 def _resolve_explicit_local() -> str:
     backend = _detect_local_backend()
-    if backend:
-        return backend
-    logger.warning("STT provider 'local' configured but unavailable "
-                   "(install faster-whisper or set HERMES_LOCAL_STT_COMMAND)")
-    return "none"
+    if not backend:
+        logger.warning("STT provider 'local' configured but unavailable "
+                       "(install faster-whisper or set HERMES_LOCAL_STT_COMMAND)")
+    return backend or "none"
 
 
 def _resolve_explicit_local_command() -> str:
@@ -255,15 +235,13 @@ _CLOUD_PROVIDER_SPECS = {
                    "No local STT available, using ElevenLabs Scribe STT API"),
     "deepinfra": (_has_deepinfra_key, _has_deepinfra_key,
                   "STT provider 'deepinfra' configured but DEEPINFRA_API_KEY not set (or openai package missing)",
-                  "No local STT available, using DeepInfra Whisper API"),
-}
+                  "No local STT available, using DeepInfra Whisper API")}
 
 # Explicit selections whose resolution is more than a probe + warning.
 _EXPLICIT_RESOLVERS = {
     "local": _resolve_explicit_local,
     "local_command": _resolve_explicit_local_command,
-    "openai": _resolve_explicit_openai,
-}
+    "openai": _resolve_explicit_openai}
 
 
 def _resolve_explicit_provider(provider: str) -> str:
@@ -273,12 +251,9 @@ def _resolve_explicit_provider(provider: str) -> str:
     if resolver is not None:
         return resolver()
     spec = _CLOUD_PROVIDER_SPECS.get(provider)
-    if spec is None:
+    if spec is None or spec[0]():
         return provider
-    probe, _auto, warning, _log = spec
-    if probe():
-        return provider
-    logger.warning(warning)
+    logger.warning(spec[2])
     return "none"
 
 
@@ -314,7 +289,6 @@ def _get_provider(stt_config: dict) -> str:
 
 
 # ---- Provider: local (faster-whisper) -----------------------------------
-
 def _unload_local_model() -> None:
     """Release the cached local whisper model. Thread-safe via the model lock."""
     global _local_model, _local_model_name
@@ -326,22 +300,17 @@ def _unload_local_model() -> None:
 
 
 def _start_idle_unload_watcher(timeout_seconds: int) -> None:
-    """Ensure the single idle-unload watcher thread is running.
-
-    The loop re-reads ``stt.local.unload_after_idle_seconds`` every cycle so config edits
-    apply within one interval; ``timeout_seconds`` seeds the first cycle so a just-written
-    config is honored even if a concurrent read races. The thread exits after unloading,
-    when the timeout becomes 0, or when the model is already gone.
-    """
+    """Ensure the single idle-unload watcher thread is running. The loop re-reads
+    ``stt.local.unload_after_idle_seconds`` every cycle so config edits apply within one interval;
+    ``timeout_seconds`` seeds the first cycle so a just-written config is honored even if a
+    concurrent read races. Exits after unloading, when the timeout becomes 0, or when the model is gone."""
     global _idle_unload_thread
     with _idle_unload_mgmt_lock:
         if _idle_unload_thread is not None and _idle_unload_thread.is_alive():
             return
 
         def _watch(initial_timeout=timeout_seconds):
-            while not _idle_unload_stop.is_set():
-                if _idle_unload_stop.wait(_IDLE_UNLOAD_CHECK_INTERVAL) or _local_model is None:
-                    break
+            while not _idle_unload_stop.wait(_IDLE_UNLOAD_CHECK_INTERVAL) and _local_model is not None:
                 try:
                     timeout = _get_idle_unload_seconds(_load_stt_config().get("local") or {})
                 except Exception:  # noqa: BLE001 - keep the seed value
@@ -351,7 +320,6 @@ def _start_idle_unload_watcher(timeout_seconds: int) -> None:
                 if time.monotonic() - _last_transcription_time >= timeout:
                     _unload_local_model()
                     break
-
         _idle_unload_stop.clear()
         _idle_unload_thread = threading.Thread(target=_watch, name="hermes-stt-idle-unload", daemon=True)
         _idle_unload_thread.start()
@@ -364,10 +332,8 @@ def _touch_transcription_time() -> None:
 
 
 def _get_or_load_local_model(model_name: str, local_cfg: Dict[str, Any]):
-    """Cached faster-whisper model, (re)loaded under a double-checked lock when needed.
-
-    The returned strong reference stays valid even if the idle watcher nulls the global mid-transcription.
-    """
+    """Cached faster-whisper model, (re)loaded under a double-checked lock when needed. The returned
+    strong reference stays valid even if the idle watcher nulls the global mid-transcription."""
     global _local_model, _local_model_name
     model = _local_model
     if model is None or _local_model_name != model_name:
@@ -375,10 +341,8 @@ def _get_or_load_local_model(model_name: str, local_cfg: Dict[str, Any]):
             if _local_model is None or _local_model_name != model_name:
                 logger.info("Loading faster-whisper model '%s' (first load downloads the model)...", model_name)
                 # stt.local.device / compute_type pin a configuration where ``auto`` mis-detects.
-                _local_model = _load_local_whisper_model(
-                    model_name, device=local_cfg.get("device", "auto"),
-                    compute_type=local_cfg.get("compute_type", "auto"),
-                )
+                _local_model = _load_local_whisper_model(model_name, device=local_cfg.get("device", "auto"),
+                                                         compute_type=local_cfg.get("compute_type", "auto"))
                 _local_model_name = model_name
             model = _local_model
     return model
@@ -390,8 +354,7 @@ def _replace_cached_model_on_cpu(model_name: str):
     from faster_whisper import WhisperModel
     model = WhisperModel(model_name, device="cpu", compute_type="int8")
     with _local_model_lock:
-        _local_model = model
-        _local_model_name = model_name
+        _local_model, _local_model_name = model, model_name
     return model
 
 
@@ -409,12 +372,10 @@ def _transcribe_local(
         model = _get_or_load_local_model(model_name, local_cfg)
         if model is None:  # defensive: load failed without raising
             return _error_result("Local whisper model failed to load")
-        transcribe_kwargs = build_local_transcribe_kwargs(stt_config)
         # pre_transcription hook overrides win over config-resolved values.
-        if language:
-            transcribe_kwargs["language"] = language
-        if prompt:
-            transcribe_kwargs["initial_prompt"] = prompt
+        transcribe_kwargs = build_local_transcribe_kwargs(stt_config)
+        transcribe_kwargs.update({k: v for k, v in (("language", language), ("initial_prompt", prompt))
+                                  if v})
         try:
             segments, info = model.transcribe(file_path, **transcribe_kwargs)
         except Exception as exc:
@@ -440,7 +401,6 @@ def _transcribe_local(
 
 
 # ---- Public API ---------------------------------------------------------
-
 def _read_block_error(file_path: str) -> Optional[Dict[str, Any]]:
     """Refuse to ship a credential store (auth.json, .env, OAuth tokens) to an STT provider.
     Mirrors the image-gen / video-gen read guards."""
@@ -450,13 +410,10 @@ def _read_block_error(file_path: str) -> Optional[Dict[str, Any]]:
 
 
 def _transcribe_prepared_audio(
-    file_path: str, model: Optional[str] = None, source: Optional[str] = None
-) -> Dict[str, Any]:
-    """Transcribe a validated audio file with the configured STT provider.
-
-    ``model`` overrides the config default; ``source`` is a caller-surface label
-    (``"gateway"``, ``"voice_mode"``) forwarded to the ``pre_transcription`` hook only.
-    """
+    file_path: str, model: Optional[str] = None, source: Optional[str] = None) -> Dict[str, Any]:
+    """Transcribe a validated audio file with the configured STT provider. ``model`` overrides the
+    config default; ``source`` is a caller-surface label (``"gateway"``, ``"voice_mode"``) forwarded
+    to the ``pre_transcription`` hook only."""
     # Validate before provider resolution so invalid files can't trigger provider setup
     # or lazy installation; the remote-upload size cap applies to non-local only.
     error = _read_block_error(file_path) or _validate_audio_file(file_path, enforce_size_limit=False)
@@ -470,11 +427,11 @@ def _transcribe_prepared_audio(
         error = _validate_audio_file_size(Path(file_path))
         if error:
             return error
-    # Convert CAF (iMessage voice notes) to WAV for cloud STT providers.
-    if Path(file_path).suffix.lower() == ".caf" and provider not in ("local", "local_command"):
-        file_path = _convert_caf_to_wav(file_path)
-        if not file_path:
-            return _error_result("CAF audio could not be converted to WAV.")
+        # Convert CAF (iMessage voice notes) to WAV for cloud STT providers.
+        if Path(file_path).suffix.lower() == ".caf":
+            file_path = _convert_caf_to_wav(file_path)
+            if not file_path:
+                return _error_result("CAF audio could not be converted to WAV.")
     # Best-effort pre-upload silence trim for built-in cloud providers.
     trim_cleanup_dir: Optional[str] = None
     if provider in CLOUD_STT_PROVIDERS:
@@ -498,8 +455,7 @@ _BUILTIN_MODEL_KEYS = {
     "openai": ("openai", "model", DEFAULT_STT_MODEL, False),
     "mistral": ("mistral", "model", DEFAULT_MISTRAL_STT_MODEL, False),
     "elevenlabs": ("elevenlabs", "model_id", DEFAULT_ELEVENLABS_STT_MODEL, False),
-    "deepinfra": ("deepinfra", "model", "", True),
-}
+    "deepinfra": ("deepinfra", "model", "", True)}
 
 
 def _builtin_model_name(provider: str, stt_config: Dict[str, Any], model: Optional[str]) -> str:
@@ -515,13 +471,11 @@ def _builtin_model_name(provider: str, stt_config: Dict[str, Any], model: Option
 
 def _dispatch_stt_provider(
     file_path: str, provider: str, stt_config: Dict[str, Any], model: Optional[str] = None,
-    source: Optional[str] = None,
-) -> Dict[str, Any]:
+    source: Optional[str] = None) -> Dict[str, Any]:
     """Route *file_path* to the handler for *provider* (built-in > command > plugin)."""
     # Static ``stt.prompt`` is the base; hook results mutate on top (last hook to set a field wins).
     prompt = stt_config.get("prompt")
-    if not isinstance(prompt, str) or not prompt.strip():
-        prompt = None
+    prompt = prompt if isinstance(prompt, str) and prompt.strip() else None
     # Fires after provider resolution and BEFORE any backend; ``language`` stays None unless a hook sets it.
     model, language, prompt = _apply_pre_transcription_hook(
         file_path=file_path, provider=provider, model=model,
@@ -544,8 +498,7 @@ def _dispatch_stt_provider(
     # Plugin backend: reads ``stt.<provider>`` like built-ins; the ``model`` argument overrides it.
     plugin_result = _dispatch_to_plugin_provider(
         file_path, provider, stt_config, model=model or _get_stt_section(stt_config, provider).get("model"),
-        language=language or _resolve_stt_language(provider, stt_config), prompt=prompt,
-    )
+        language=language or _resolve_stt_language(provider, stt_config), prompt=prompt)
     return plugin_result if plugin_result is not None else _no_provider_error(provider, stt_config)
 
 
@@ -565,18 +518,13 @@ def _no_provider_error(provider: str, stt_config: Dict[str, Any]) -> Dict[str, A
         "set GROQ_API_KEY for free Groq Whisper, set MISTRAL_API_KEY for Mistral "
         "Voxtral Transcribe, configure xAI OAuth or set XAI_API_KEY for xAI Grok STT, "
         "set ELEVENLABS_API_KEY for ElevenLabs Scribe, or set VOICE_TOOLS_OPENAI_KEY "
-        "or OPENAI_API_KEY for the OpenAI Whisper API."
-    )
+        "or OPENAI_API_KEY for the OpenAI Whisper API.")
 
 
 def transcribe_audio(
-    file_path: str, model: Optional[str] = None, source: Optional[str] = None
-) -> Dict[str, Any]:
-    """Validate, preprocess supported inputs, and dispatch transcription.
-
-    ``source`` is a caller-surface label (``"gateway"``, ``"voice_mode"``) forwarded to
-    the ``pre_transcription`` hook for observability only.
-    """
+    file_path: str, model: Optional[str] = None, source: Optional[str] = None) -> Dict[str, Any]:
+    """Validate, preprocess supported inputs, and dispatch transcription. ``source`` is a caller-surface
+    label (``"gateway"``, ``"voice_mode"``) forwarded to the ``pre_transcription`` hook only."""
     # Secret-store refusal runs before ANY validation so the error names the real reason.
     blocked = _read_block_error(file_path)
     if blocked:
@@ -599,16 +547,12 @@ def transcribe_audio(
 
 
 def transcribe_audio_local_fallback(file_path: str, model: Optional[str] = None) -> Dict[str, Any]:
-    """Try an already-installed local STT backend without changing config.
-
-    Passive inbound-media recovery after the configured provider failed: never
-    lazy-installs or falls through to a cloud provider.
-    """
+    """Try an already-installed local STT backend without changing config: passive inbound-media
+    recovery after the configured provider failed — never lazy-installs or falls through to cloud."""
     error = _validate_audio_file(file_path)
     if error:
         return error
-    local_cfg = _load_stt_config().get("local") or {}
-    local_model = model or local_cfg.get("model", DEFAULT_LOCAL_MODEL)
+    local_model = model or (_load_stt_config().get("local") or {}).get("model", DEFAULT_LOCAL_MODEL)
     if _HAS_FASTER_WHISPER:
         return _transcribe_local(file_path, _normalize_local_model(local_model))
     if _has_local_command():
diff --git a/tools/tts_command_provider.py b/tools/tts_command_provider.py
index 852d1c48b1..c5c6f412d3 100644
--- a/tools/tts_command_provider.py
+++ b/tools/tts_command_provider.py
@@ -1,20 +1,12 @@
 """Shared runner for user-configured shell ("command") TTS/STT providers.
 
-``tools.tts_tool`` and ``tools.transcription_tools`` both let users declare a
-provider as a shell command template with ``{placeholders}`` (``{{``/``}}`` stay
-literal; values are shell-quoted for their surrounding quote context). This
-module owns the quote-aware rendering, the idle-timeout process runner and the
-generic ``<section>.providers.<name>`` config readers; each origin module
-re-imports them under its historical private names. TTS config shape::
-
-    tts:
-      provider: piper-en
-      providers:
-        piper-en: {type: command, command: "piper -f {output_path} < {input_path}", output_format: wav}
-
-TTS placeholders: ``{input_path}``/``{text_path}``, ``{output_path}``, ``{format}``,
-``{voice}``, ``{model}``, ``{speed}``. Built-in provider names always win over a
-same-named entry under ``providers``.
+``tts.providers.<name>: {type: command, command: "piper -f {output_path} < {input_path}"}``
+(and the ``stt.`` twin): ``{placeholders}`` are shell-quoted for their surrounding quote
+context, ``{{``/``}}`` stay literal. Owns the quote-aware rendering, the idle-timeout
+process runner and the generic ``<section>.providers.<name>`` readers, re-imported by
+``tts_tool``/``transcription_tools`` under their historical private names. TTS placeholders:
+``{input_path}``/``{text_path}``, ``{output_path}``, ``{format}``, ``{voice}``, ``{model}``,
+``{speed}``. Built-in provider names always win over a same-named ``providers`` entry.
 """
 
 from __future__ import annotations
@@ -27,9 +19,12 @@ import subprocess
 import tempfile
 import threading
 import time
+from functools import partial
 from pathlib import Path
 from typing import Any, Dict, FrozenSet, Optional
 
+from utils import is_truthy_value
+
 
 def shell_quote_context(command_template: str, position: int) -> Optional[str]:
     """Return the shell quote char (``'``/``"``) active right before *position*, or None."""
@@ -48,10 +43,8 @@ def shell_quote_context(command_template: str, position: int) -> Optional[str]:
                 escaped = True
             elif char == '"':
                 quote = None
-        elif char == "'":
-            quote = "'"
-        elif char == '"':
-            quote = '"'
+        elif char in ("'", '"'):
+            quote = char
         elif char == "\\":
             i += 1
         i += 1
@@ -64,9 +57,7 @@ def quote_command_placeholder(value: str, quote_context: Optional[str]) -> str:
         return value.replace("'", r"'\''")
     if quote_context == '"':
         return value.replace("\\", "\\\\").replace('"', r'\"').replace("$", r"\$").replace("`", r"\`")
-    if os.name == "nt":
-        return subprocess.list2cmdline([value])
-    return shlex.quote(value)
+    return subprocess.list2cmdline([value]) if os.name == "nt" else shlex.quote(value)
 
 
 def render_command_template(command_template: str, placeholders: Dict[str, str]) -> str:
@@ -110,35 +101,29 @@ def terminate_command_process_tree(proc: subprocess.Popen) -> None:
         return
     if os.name == "nt":
         try:
-            subprocess.run(
-                ["taskkill", "/F", "/T", "/PID", str(proc.pid)],
-                stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, timeout=5, stdin=subprocess.DEVNULL,
-            )
+            subprocess.run(["taskkill", "/F", "/T", "/PID", str(proc.pid)], stdout=subprocess.DEVNULL,
+                           stderr=subprocess.DEVNULL, timeout=5, stdin=subprocess.DEVNULL)
         except Exception:
             proc.kill()
         return
     try:
         import psutil  # type: ignore
     except ImportError:
-        proc.terminate()
-        try:
-            proc.wait(timeout=2)
-        except subprocess.TimeoutExpired:
-            proc.kill()
-        return
-    _signal_process_tree(psutil, proc, "terminate")
+        psutil = None
+    # Without psutil only the shell itself is signalled (children may survive).
+    signal = ((lambda m: getattr(proc, m)()) if psutil is None
+              else (lambda m: _signal_process_tree(psutil, proc, m)))
+    signal("terminate")
     try:
         proc.wait(timeout=2)
     except subprocess.TimeoutExpired:
-        _signal_process_tree(psutil, proc, "kill")
+        signal("kill")
 
 
 def command_env_passthrough(config: Dict[str, Any]) -> list:
     """``env_passthrough`` allowlist: parent env vars copied back into the secret-scrubbed child env."""
     raw = config.get("env_passthrough")
-    if not isinstance(raw, (list, tuple)):
-        return []
-    return [str(item).strip() for item in raw if str(item).strip()]
+    return [str(x).strip() for x in raw if str(x).strip()] if isinstance(raw, (list, tuple)) else []
 
 
 def command_failure_detail(exc: subprocess.CalledProcessError) -> str:
@@ -151,30 +136,23 @@ def run_command_provider(
     command: str, timeout: float, env_passthrough: Optional[list] = None,
 ) -> subprocess.CompletedProcess:
     """Run a command-provider shell command with process-tree idle cleanup.
-
-    ``timeout`` is an IDLE timeout, reset whenever the command emits output — a
-    slow-but-alive provider survives, a silently stalled one is killed. Child env
-    is scrubbed of Hermes secrets while propagating delegated-child lineage markers.
-    """
+    ``timeout`` is an IDLE timeout, reset whenever the command emits output — a slow-but-alive
+    provider survives, a silently stalled one is killed. Child env is scrubbed of Hermes secrets
+    while propagating delegated-child lineage markers."""
     from agent.delegation_context import delegated_child_subprocess_env
     from tools.environments.local import hermes_subprocess_env
-
     scrubbed = hermes_subprocess_env(inherit_credentials=False)
     for key in env_passthrough or []:
         value = os.environ.get(key)
         if value is not None:
             scrubbed[key] = value
-    popen_kwargs: Dict[str, Any] = {
-        "shell": True, "stdout": subprocess.PIPE, "stderr": subprocess.PIPE, "text": True,
-        # Lossy UTF-8 decode: locale-mismatched bytes must not raise in the reader threads.
-        "encoding": "utf-8", "errors": "replace", "env": delegated_child_subprocess_env(scrubbed),
-    }
-    if os.name == "nt":
-        popen_kwargs["creationflags"] = getattr(subprocess, "CREATE_NEW_PROCESS_GROUP", 0)
-    else:
-        popen_kwargs["start_new_session"] = True
-
-    proc = subprocess.Popen(command, **popen_kwargs, stdin=subprocess.DEVNULL)
+    # Own process group so the whole tree can be signalled on idle timeout. Lossy UTF-8 decode:
+    # locale-mismatched bytes must not raise in the reader threads.
+    group = ({"creationflags": getattr(subprocess, "CREATE_NEW_PROCESS_GROUP", 0)} if os.name == "nt"
+             else {"start_new_session": True})
+    proc = subprocess.Popen(command, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True,
+                            encoding="utf-8", errors="replace", env=delegated_child_subprocess_env(scrubbed),
+                            stdin=subprocess.DEVNULL, **group)
     output_queue: "queue.Queue[tuple[str, Optional[str]]]" = queue.Queue()
     chunks: Dict[str, list[str]] = {"stdout": [], "stderr": []}
     open_streams = {"stdout", "stderr"}
@@ -191,13 +169,10 @@ def run_command_provider(
         finally:
             output_queue.put((name, None))
 
-    readers = [
-        threading.Thread(target=read_stream, args=(name, stream), daemon=True)
-        for name, stream in (("stdout", proc.stdout), ("stderr", proc.stderr))
-    ]
+    readers = [threading.Thread(target=read_stream, args=(name, stream), daemon=True)
+               for name, stream in (("stdout", proc.stdout), ("stderr", proc.stderr))]
     for reader in readers:
         reader.start()
-
     deadline = time.monotonic() + timeout
     timed_out = False
     while open_streams:
@@ -214,61 +189,42 @@ def run_command_provider(
             continue
         chunks[name].append(chunk)
         deadline = time.monotonic() + timeout
-
     if not timed_out:
         try:
             proc.wait(timeout=max(0.0, deadline - time.monotonic()))
         except subprocess.TimeoutExpired:
             timed_out = True
-
     if timed_out:
         terminate_command_process_tree(proc)
         for reader in readers:
             reader.join(timeout=0.5)
-        while True:
-            try:
-                name, chunk = output_queue.get_nowait()
-            except queue.Empty:
-                break
+        while not output_queue.empty():
+            name, chunk = output_queue.get_nowait()
             if chunk:
                 chunks[name].append(chunk)
-        stdout, stderr = "".join(chunks["stdout"]), "".join(chunks["stderr"])
-        try:
-            raise subprocess.TimeoutExpired(command, timeout)
-        except subprocess.TimeoutExpired as exc:
-            raise subprocess.TimeoutExpired(command, timeout, output=stdout, stderr=stderr) from exc
-
     stdout, stderr = "".join(chunks["stdout"]), "".join(chunks["stderr"])
+    if timed_out:
+        raise subprocess.TimeoutExpired(command, timeout, output=stdout, stderr=stderr) from (
+            subprocess.TimeoutExpired(command, timeout))
     if proc.returncode:
         raise subprocess.CalledProcessError(proc.returncode, command, output=stdout, stderr=stderr)
     return subprocess.CompletedProcess(command, proc.returncode, stdout, stderr)
 
 
-# ===========================================================================
-# Generic ``<section>.providers.<name>`` config layer (TTS and STT share it)
-# ===========================================================================
-
-
+# ---- Generic ``<section>.providers.<name>`` config layer (TTS and STT share it) ----
 def _get_provider_section(config: Dict[str, Any], name: str) -> Dict[str, Any]:
     """Return ``config[name]`` if it's a dict, else an empty dict."""
-    if not isinstance(config, dict):
-        return {}
-    section = config.get(name)
+    section = config.get(name) if isinstance(config, dict) else None
     return section if isinstance(section, dict) else {}
 
 
 def _named_provider_config(config: Dict[str, Any], name: str, builtins: FrozenSet[str]) -> Dict[str, Any]:
-    """``<section>.providers.<name>`` (canonical), else ``<section>.<name>`` for non-built-in names only.
-
-    The back-compat form is refused for built-ins so a user's ``openai:`` block
-    still means the OpenAI provider, not a custom command.
-    """
+    """``<section>.providers.<name>`` (canonical), else ``<section>.<name>`` for non-built-in names
+    only — refused for built-ins so a user's ``openai:`` block still means OpenAI, not a command."""
     section = _get_provider_section(config, "providers").get(name)
     if isinstance(section, dict):
         return section
-    if name.lower() not in builtins:
-        return _get_provider_section(config, name)
-    return {}
+    return _get_provider_section(config, name) if name.lower() not in builtins else {}
 
 
 def _is_command_provider_config(config: Dict[str, Any]) -> bool:
@@ -276,20 +232,15 @@ def _is_command_provider_config(config: Dict[str, Any]) -> bool:
     if not isinstance(config, dict):
         return False
     ptype = str(config.get("type") or "").strip().lower()
-    if ptype and ptype != "command":
-        return False
     command = config.get("command")
-    return isinstance(command, str) and bool(command.strip())
+    return ptype in ("", "command") and isinstance(command, str) and bool(command.strip())
 
 
 def _resolve_command_config(
-    provider: str, config: Dict[str, Any], reserved: FrozenSet[str],
-) -> Optional[Dict[str, Any]]:
+    provider: str, config: Dict[str, Any], reserved: FrozenSet[str]) -> Optional[Dict[str, Any]]:
     """Config of a user-declared command provider; None for *reserved* names, unknown or non-command."""
-    if not provider:
-        return None
-    key = provider.lower().strip()
-    if key in reserved:
+    key = (provider or "").lower().strip()
+    if not key or key in reserved:
         return None
     named = _named_provider_config(config, key, reserved)
     return named if _is_command_provider_config(named) else None
@@ -317,8 +268,7 @@ def _command_output_format(config: Dict[str, Any], formats: FrozenSet[str], defa
 # Any ``tts.provider`` value NOT in this set refers to ``tts.providers.<name>``.
 BUILTIN_TTS_PROVIDERS = frozenset({
     "edge", "elevenlabs", "openai", "minimax", "xai", "mistral", "gemini",
-    "neutts", "kittentts", "piper", "deepinfra",
-})
+    "neutts", "kittentts", "piper", "deepinfra"})
 
 DEFAULT_COMMAND_TTS_TIMEOUT_SECONDS = 120
 DEFAULT_COMMAND_TTS_OUTPUT_FORMAT = "mp3"
@@ -326,12 +276,9 @@ COMMAND_TTS_OUTPUT_FORMATS = frozenset({"mp3", "wav", "ogg", "flac", "m4a", "aac
 DEFAULT_COMMAND_TTS_MAX_TEXT_LENGTH = 5000
 
 
-def _get_named_provider_config(tts_config: Dict[str, Any], name: str) -> Dict[str, Any]:
-    return _named_provider_config(tts_config, name, BUILTIN_TTS_PROVIDERS)
-
-
-def _resolve_command_provider_config(provider: str, tts_config: Dict[str, Any]) -> Optional[Dict[str, Any]]:
-    return _resolve_command_config(provider, tts_config, BUILTIN_TTS_PROVIDERS)
+_get_named_provider_config = partial(_named_provider_config, builtins=BUILTIN_TTS_PROVIDERS)
+_resolve_command_provider_config = partial(_resolve_command_config, reserved=BUILTIN_TTS_PROVIDERS)
+_get_command_tts_timeout = partial(_command_timeout, default=DEFAULT_COMMAND_TTS_TIMEOUT_SECONDS)
 
 
 def _iter_command_providers(tts_config: Dict[str, Any]):
@@ -341,25 +288,17 @@ def _iter_command_providers(tts_config: Dict[str, Any]):
             yield name, cfg
 
 
-def _get_command_tts_timeout(config: Dict[str, Any]) -> float:
-    return _command_timeout(config, DEFAULT_COMMAND_TTS_TIMEOUT_SECONDS)
-
-
 def _get_command_tts_output_format(config: Dict[str, Any], output_path: Optional[str] = None) -> str:
     """Validated output format: the output path's suffix wins, then ``format``/``output_format``."""
-    if output_path:
-        suffix = Path(output_path).suffix.lower().strip().lstrip(".")
-        if suffix in COMMAND_TTS_OUTPUT_FORMATS:
-            return suffix
+    suffix = Path(output_path).suffix.lower().strip().lstrip(".") if output_path else ""
+    if suffix in COMMAND_TTS_OUTPUT_FORMATS:
+        return suffix
     return _command_output_format(config, COMMAND_TTS_OUTPUT_FORMATS, DEFAULT_COMMAND_TTS_OUTPUT_FORMAT)
 
 
 def _is_command_tts_voice_compatible(config: Dict[str, Any]) -> bool:
     """True only when the user explicitly opted in to voice delivery."""
-    value = config.get("voice_compatible", False)
-    if isinstance(value, str):
-        return value.strip().lower() in {"1", "true", "yes", "on"}
-    return bool(value)
+    return is_truthy_value(config.get("voice_compatible", False))
 
 
 def _configured_command_tts_output_path(path: Path, config: Dict[str, Any]) -> Path:
@@ -371,36 +310,28 @@ def _generate_command_tts(
     text: str, output_path: str, provider_name: str, config: Dict[str, Any], tts_config: Dict[str, Any],
 ) -> str:
     """Generate speech by running a user-configured shell command; returns the audio path it wrote.
-
-    Raises ``ValueError`` for invalid provider config and ``RuntimeError`` for
-    timeouts / non-zero exits / empty output.
-    """
+    Raises ``ValueError`` for bad provider config, ``RuntimeError`` for timeouts / bad exits / no output."""
     command_template = str(config.get("command") or "").strip()
     if not command_template:
         raise ValueError(f"tts.providers.{provider_name}.command is not configured")
-
     output = Path(output_path).expanduser()
     output.parent.mkdir(parents=True, exist_ok=True)
     if output.exists():
         output.unlink()
-
     timeout = _get_command_tts_timeout(config)
-    output_format = _get_command_tts_output_format(config, str(output))
-    speed = config.get("speed", tts_config.get("speed", ""))
-
     with tempfile.TemporaryDirectory() as tmpdir:
         text_path = Path(tmpdir) / "input.txt"
         text_path.write_text(text, encoding="utf-8")
         placeholders = {
             "input_path": str(text_path), "text_path": str(text_path), "output_path": str(output),
-            "format": output_format, "voice": str(config.get("voice", "")),
-            "model": str(config.get("model", "")), "speed": str(speed),
+            "format": _get_command_tts_output_format(config, str(output)),
+            "voice": str(config.get("voice", "")), "model": str(config.get("model", "")),
+            "speed": str(config.get("speed", tts_config.get("speed", ""))),
         }
         command = render_command_template(command_template, placeholders)
         try:
             # Resolved through the origin so tests patching ``tools.tts_tool._run_command_tts`` still intercept.
             from tools.tts_tool import _run_command_tts
-
             _run_command_tts(command, timeout, env_passthrough=command_env_passthrough(config))
         except subprocess.TimeoutExpired as exc:
             raise RuntimeError(f"TTS provider '{provider_name}' timed out after {timeout:g}s") from exc
@@ -408,7 +339,6 @@ def _generate_command_tts(
             raise RuntimeError(
                 f"TTS provider '{provider_name}' exited with code {exc.returncode}: {command_failure_detail(exc)}"
             ) from exc
-
     if not output.exists() or output.stat().st_size <= 0:
         raise RuntimeError(f"TTS provider '{provider_name}' produced no output at {output}")
     return str(output)
diff --git a/tools/tts_streaming.py b/tools/tts_streaming.py
index a84625048f..5447cf272d 100644
--- a/tools/tts_streaming.py
+++ b/tools/tts_streaming.py
@@ -27,12 +27,9 @@ _STREAM_SENTENCE_BYTE_CAP = 16 * 1024 * 1024
 
 def _resolve_key(env_var: str, provider_id: str) -> str:
     """Provider secret lookup (config > env/.env > credential pool); seam over ``tts_tool._resolve_provider_key``.
-
-    ALL streaming-provider key lookups go through here — never bare ``get_env_value``.
-    """
+    ALL streaming-provider key lookups go through here — never bare ``get_env_value``."""
     try:
         from tools.tts_tool import _resolve_provider_key
-
         return _resolve_provider_key(env_var, provider_id) or ""
     except Exception:
         return get_env_value(env_var) or ""
@@ -42,10 +39,9 @@ def _gemini_key() -> str:
     return _resolve_key("GEMINI_API_KEY", "gemini") or _resolve_key("GOOGLE_API_KEY", "gemini")
 
 
-# Interruption latch: when the user barges in on a spoken reply, the surface marks
-# it; the next turn's submit path takes it and prepends SPEECH_INTERRUPTED_NOTE to
-# the model-bound message (API-call local, never persisted). The TTL keeps a stale
-# barge from annotating an unrelated message minutes later.
+# Interruption latch: a barge-in on a spoken reply marks it; the next turn's submit path takes it
+# and prepends SPEECH_INTERRUPTED_NOTE to the model-bound message (API-call local, never
+# persisted). The TTL keeps a stale barge from annotating an unrelated message minutes later.
 SPEECH_INTERRUPTED_NOTE = "[Note: the user interrupted your previous spoken reply before it finished.]"
 _INTERRUPT_TTL_S = 120.0
 _interrupted_at: Optional[float] = None
@@ -68,12 +64,9 @@ _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 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.
-    """
+    """Incremental sentence cutter for LLM token deltas, shared by the speaker pipeline 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."""
 
     def __init__(self, min_len: int = 20):
         self.min_len = min_len
@@ -98,13 +91,12 @@ class SentenceChunker:
 
     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 = ""
+        tail, self.buf = _THINK_BLOCK_RE.sub("", self.buf).strip(), ""
         return [tail] if tail else []
 
 
 class StreamingTTSProvider(ABC):
-    """Yields raw int16, little-endian, mono PCM chunks at ``sample_rate``."""
+    """Yields raw int16, little-endian, mono PCM chunks at ``sample_rate`` (built-ins: 24 kHz)."""
 
     sample_rate: int = 24000
     channels: int = 1
@@ -152,23 +144,17 @@ _PROVIDER_PRIORITY: List[str] = ["elevenlabs", "gemini", "openai", "xai"]
 
 
 def resolve_streaming_provider(
-    tts_config: Dict, preferred: Optional[str] = None
-) -> Optional[StreamingTTSProvider]:
+    tts_config: Dict, preferred: Optional[str] = None) -> Optional[StreamingTTSProvider]:
     """Return a ready streamer for the *configured* provider, else ``None``.
-
-    ``tts.streaming.provider`` when set: a name pins that exact streamer (``None``
-    if unusable); ``auto`` returns the first usable in ``_PROVIDER_PRIORITY``.
-    Otherwise the configured TTS provider (or ``preferred``): ``None`` means "no
-    chunked API" — the dispatcher speaks per-sentence via the sync path, preserving
-    the user's chosen voice. We never silently swap providers just to get streaming.
-    """
+    ``tts.streaming.provider`` when set: a name pins that exact streamer (``None`` if unusable);
+    ``auto`` returns the first usable in ``_PROVIDER_PRIORITY``. Otherwise the configured TTS
+    provider (or ``preferred``): ``None`` means "no chunked API" — the dispatcher speaks
+    per-sentence via the sync path, preserving the user's chosen voice. We never silently swap
+    providers just to get streaming."""
     pinned = str((tts_config.get("streaming") or {}).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
+        return next((inst for name in _PROVIDER_PRIORITY
+                     if (inst := _try_instantiate(name, tts_config))), None)
     return _try_instantiate(pinned or (preferred or _get_provider(tts_config)).lower().strip(), tts_config)
 
 
@@ -187,8 +173,6 @@ def _capped(chunks: Iterator[bytes], label: str) -> Iterator[bytes]:
 class ElevenLabsStreamer(StreamingTTSProvider):
     """ElevenLabs chunked HTTP → pcm_24000 (the original reference path)."""
 
-    sample_rate = 24000
-
     @staticmethod
     def available() -> bool:
         return bool(_resolve_key("ELEVENLABS_API_KEY", "elevenlabs"))
@@ -198,45 +182,37 @@ class ElevenLabsStreamer(StreamingTTSProvider):
         from tools.tts_tool_providers import (
             DEFAULT_ELEVENLABS_STREAMING_MODEL_ID, DEFAULT_ELEVENLABS_VOICE_ID, _elevenlabs_environment_kwargs,
         )
-
         client = _import_elevenlabs()(
             api_key=_resolve_key("ELEVENLABS_API_KEY", "elevenlabs"), **_elevenlabs_environment_kwargs(self.section),
         )
-        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",
-        )
+            text=text, 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)),
+            output_format="pcm_24000")
 
 
 def _openai_config_api_key() -> str:
     """Return ``tts.openai.api_key`` from config.yaml, or empty string."""
     try:
-        openai_cfg = (_load_tts_config().get("openai") or {})
+        return (_load_tts_config().get("openai") or {}).get("api_key") or ""
     except Exception:
         return ""
-    return openai_cfg.get("api_key") or ""
 
 
 @register("openai")
 class OpenAIStreamer(StreamingTTSProvider):
     """OpenAI speech with ``response_format=pcm`` (24 kHz mono int16)."""
 
-    sample_rate = 24000
-
     @staticmethod
     def available() -> bool:
         return bool(_openai_config_api_key() or resolve_openai_audio_api_key())
 
     def stream(self, text: str) -> Iterator[bytes]:
         from openai import OpenAI
-
         client = OpenAI(
             api_key=(self.section.get("api_key") or resolve_openai_audio_api_key()),
-            base_url=(self.section.get("base_url") or get_env_value("OPENAI_BASE_URL") or None),
-        )
+            base_url=(self.section.get("base_url") or get_env_value("OPENAI_BASE_URL") or None))
         with client.audio.speech.with_streaming_response.create(
             model=self.section.get("model", "gpt-4o-mini-tts"), voice=self.section.get("voice", "alloy"),
             input=text, response_format="pcm",
@@ -248,8 +224,6 @@ class OpenAIStreamer(StreamingTTSProvider):
 class GeminiStreamer(StreamingTTSProvider):
     """Gemini ``streamGenerateContent?alt=sse`` → SSE feed of base64 PCM chunks (24 kHz), bounded streamed body."""
 
-    sample_rate = 24000
-
     @staticmethod
     def available() -> bool:
         return bool(_gemini_key())
@@ -257,13 +231,9 @@ class GeminiStreamer(StreamingTTSProvider):
     def stream(self, text: str) -> Iterator[bytes]:
         import base64
         import json as _json
-
         import requests
-
         from tools.tts_tool_providers import (
-            DEFAULT_GEMINI_TTS_BASE_URL, DEFAULT_GEMINI_TTS_MODEL, DEFAULT_GEMINI_TTS_VOICE,
-        )
-
+            DEFAULT_GEMINI_TTS_BASE_URL, DEFAULT_GEMINI_TTS_MODEL, DEFAULT_GEMINI_TTS_VOICE)
         api_key = _gemini_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
@@ -274,9 +244,7 @@ class GeminiStreamer(StreamingTTSProvider):
             "contents": [{"parts": [{"text": text}]}],
             "generationConfig": {
                 "responseModalities": ["AUDIO"],
-                "speechConfig": {"voiceConfig": {"prebuiltVoiceConfig": {"voiceName": voice}}},
-            },
-        }
+                "speechConfig": {"voiceConfig": {"prebuiltVoiceConfig": {"voiceName": voice}}}}}
         url = f"{base_url}/models/{model}:streamGenerateContent"
 
         def _sse_chunks() -> Iterator[bytes]:
@@ -288,13 +256,11 @@ class GeminiStreamer(StreamingTTSProvider):
                     if not line or not line.startswith("data: "):
                         continue
                     try:
-                        event = _json.loads(line[len("data: "):])
-                        parts = event["candidates"][0]["content"]["parts"]
+                        parts = _json.loads(line[len("data: "):])["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", "")
+                        b64 = (part.get("inlineData") or part.get("inline_data") or {}).get("data", "")
                         if not b64:
                             continue
                         try:
@@ -308,21 +274,15 @@ class GeminiStreamer(StreamingTTSProvider):
 @register("xai")
 class XAIStreamer(StreamingTTSProvider):
     """xAI WebSocket TTS (``wss://api.x.ai/v1/tts``) → binary PCM frames (24 kHz mono int16).
-
-    Credentials route through ``resolve_xai_http_credentials`` (OAuth or XAI_API_KEY),
-    same as the sync path. ``_collect_async`` bridges the async WS loop to the sync
-    iterator contract — the seam unit tests patch.
-    """
-
-    sample_rate = 24000
+    Credentials route through ``resolve_xai_http_credentials`` (OAuth or XAI_API_KEY), same as the
+    sync path. ``_collect_async`` bridges the async WS loop to the sync iterator contract (test
+    seam)."""
 
     @staticmethod
     def available() -> bool:
         try:
             from tools.xai_http import resolve_xai_http_credentials
-
-            creds = resolve_xai_http_credentials()
-            return bool(str(creds.get("api_key") or "").strip())
+            return bool(str(resolve_xai_http_credentials().get("api_key") or "").strip())
         except Exception:
             return False
 
@@ -332,26 +292,20 @@ class XAIStreamer(StreamingTTSProvider):
     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]:
-        return [frame async for frame in self._async_frames(text)]
+        async def _drain() -> List[bytes]:
+            return [frame async for frame in self._async_frames(text)]
+        return asyncio.run(_drain())
 
     async def _async_frames(self, text: str):
         import json as _json
-
         import websockets
-
         from tools.tts_tool_providers import DEFAULT_XAI_VOICE_ID
         from tools.xai_http import resolve_xai_http_credentials
-
-        creds = resolve_xai_http_credentials()
-        api_key = str(creds.get("api_key") or "").strip()
+        api_key = str(resolve_xai_http_credentials().get("api_key") or "").strip()
         if not api_key:
             raise RuntimeError("No xAI credentials for 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:
@@ -367,15 +321,13 @@ class XAIStreamer(StreamingTTSProvider):
                             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,
                         )
+                    if etype in ("done", "error"):
                         return
             except Exception as exc:
-                if exc.__class__.__name__ == "ConnectionClosed":
-                    return
-                logger.warning("xAI WS receive failed: %s", exc)
+                if exc.__class__.__name__ != "ConnectionClosed":
+                    logger.warning("xAI WS receive failed: %s", exc)
                 return
diff --git a/tools/tts_text_normalize.py b/tools/tts_text_normalize.py
index 45213cd9ce..dccb01f088 100644
--- a/tools/tts_text_normalize.py
+++ b/tools/tts_text_normalize.py
@@ -31,11 +31,11 @@ _MD_HR_RE = re.compile(r"^\s*[-*_]{3,}\s*$", flags=re.MULTILINE)
 _MD_TABLE_PIPE_RE = re.compile(r"\s*\|\s*")
 _URL_RE = re.compile(r"https?://\S+")
 
+_DEGREE_UNITS = (("C", "Celsius"), ("F", "Fahrenheit"))
 # Unit suffix (regex, after a digit) -> spoken word; km/h variants before the bare "m".
 _UNIT_WORDS = (
     (r"km\s*/\s*h", "kilometres per hour"), (r"km/h", "kilometres per hour"),
-    (r"mm", "millimetres"), (r"cm", "centimetres"), (r"m", "metres"),
-)
+    (r"mm", "millimetres"), (r"cm", "centimetres"), (r"m", "metres"))
 # Currency prefix (regex) -> spoken word; order matters (NZ$/A$/US$ before bare $).
 _CURRENCY_WORDS = (
     (r"NZ\$", "New Zealand dollars", re.IGNORECASE), (r"A\$", "Australian dollars", re.IGNORECASE),
@@ -47,8 +47,7 @@ _EMOJI_RE = re.compile(
     "[\U0001F1E6-\U0001F1FF\U0001F300-\U0001F5FF\U0001F600-\U0001F64F\U0001F680-\U0001F6FF"
     "\U0001F700-\U0001F77F\U0001F780-\U0001F7FF\U0001F800-\U0001F8FF\U0001F900-\U0001F9FF"
     "\U0001FA00-\U0001FAFF☀-➿]+",
-    flags=re.UNICODE,
-)
+    flags=re.UNICODE)
 _VARIATION_SELECTOR_RE = re.compile("[︎️]")
 
 
@@ -56,7 +55,6 @@ def strip_markdown_for_tts(text: str) -> str:
     """Strip Markdown/Telegram formatting while preserving readable words."""
     if not text:
         return ""
-
     text = html.unescape(str(text))
     text = _MD_CODE_BLOCK_RE.sub(" ", text)
     text = _MD_IMAGE_RE.sub(lambda m: f" {m.group(1)} " if m.group(1) else " ", text)
@@ -73,24 +71,20 @@ def strip_markdown_for_tts(text: str) -> str:
     text = _MD_BLOCKQUOTE_RE.sub("", text)
     text = _MD_LIST_ITEM_RE.sub("", text)
     text = _MD_HR_RE.sub("", text)
-
     # Leftover table pipes become pauses instead of a spoken "vertical bar".
-    text = _MD_TABLE_PIPE_RE.sub("; ", text)
-    return text
+    return _MD_TABLE_PIPE_RE.sub("; ", text)
 
 
 def _normalize_temperature_ranges(text: str) -> str:
     """``11-17°C`` -> ``11 to 17 degrees Celsius`` (en/em dash or hyphen; unicode minus normalized)."""
     number = r"([-+\u2212]?\d+(?:\.\d+)?)"
-    for unit, word in (("C", "Celsius"), ("F", "Fahrenheit")):
+    for unit, word in _DEGREE_UNITS:
         text = re.sub(
             r"(?<!\w)" + number + r"\s*[\u2013\u2014-]\s*" + number + r"\s*°\s*" + unit + r"\b",
             lambda m, w=word: (
                 f"{m.group(1).replace(chr(0x2212), '-')} to {m.group(2).replace(chr(0x2212), '-')} degrees {w}"
             ),
-            text,
-            flags=re.IGNORECASE,
-        )
+            text, flags=re.IGNORECASE)
     return text
 
 
@@ -98,59 +92,43 @@ def normalize_symbols_for_tts(text: str) -> str:
     """Expand common symbols/shorthand into words a TTS engine reads well."""
     if not text:
         return ""
-
-    text = str(text)
-    text = re.sub("[   ]", " ", text)  # non-breaking / thin spaces
-    text = text.replace("\u2212", "-")  # minus sign
-    text = text.replace("…", "...")  # ellipsis
+    text = re.sub("[   ]", " ", str(text))  # non-breaking / thin spaces
+    text = text.replace("\u2212", "-").replace("…", "...")  # minus sign, ellipsis
     text = _normalize_temperature_ranges(text)
-
     # Temperatures with a number first, then bare units ("measured in degrees C"),
     # then any remaining degree symbol (angles, stray cases).
-    for unit, word in (("C", "Celsius"), ("F", "Fahrenheit")):
+    for unit, word in _DEGREE_UNITS:
         text = re.sub(
             r"(?<!\w)([-+]?\d+(?:\.\d+)?)\s*°\s*" + unit + r"\b", r"\1 degrees " + word, text, flags=re.IGNORECASE,
         )
-    for unit, word in (("C", "Celsius"), ("F", "Fahrenheit")):
+    for unit, word in _DEGREE_UNITS:
         text = re.sub(r"°\s*" + unit + r"\b", "degrees " + word, text, flags=re.IGNORECASE)
-    text = re.sub(r"(?<!\w)([-+]?\d+(?:\.\d+)?)\s*°", r"\1 degrees", text)
-    text = text.replace("°", " degrees")
-
+    text = re.sub(r"(?<!\w)([-+]?\d+(?:\.\d+)?)\s*°", r"\1 degrees", text).replace("°", " degrees")
     # Common weather/travel units.
     for pattern, word in _UNIT_WORDS:
         text = re.sub(r"(?<=\d)\s*" + pattern + r"\b", " " + word, text, flags=re.IGNORECASE)
-
     # Numeric rates only ("5/month" -> "5 per month").  Requiring digit-then-letter
     # keeps "and/or", "N/A", "TCP/IP" and dates like "2026/06" intact.
     text = re.sub(r"(?<=\d)\s*/\s*(?=[A-Za-z])", " per ", text)
-
     # Money and percentages. The integer part must END in a digit so a trailing
     # comma ("A$50, ...") is not swallowed into the spoken amount. Prefixed
     # currencies run first so "$" doesn't eat "NZ$".
     for symbol, word, flags in _CURRENCY_WORDS:
         text = re.sub(symbol + r"\s*([\d,]*\d(?:\.\d+)?)", r"\1 " + word, text, flags=flags)
     text = re.sub(r"(?<=\d)\s*%", " percent", text)
-
     # Operators and separators that commonly leak from formatted answers.
-    text = text.replace("&", " and ")
-    text = re.sub("[•◦▪▫]", " ", text)  # bullet glyphs
+    text = re.sub("[•◦▪▫]", " ", text.replace("&", " and "))  # bullet glyphs
     for symbol, word in (("→", " to "), ("⇒", " to "), ("≈", " about "), ("~", " about ")):
         text = text.replace(symbol, word)
-
-    text = _VARIATION_SELECTOR_RE.sub("", text)
-    text = _EMOJI_RE.sub("", text)
-    return text
+    return _EMOJI_RE.sub("", _VARIATION_SELECTOR_RE.sub("", text))
 
 
 def smooth_whitespace_for_tts(text: str) -> str:
-    """Collapse visual formatting into calm spoken paragraphs.
-
-    A _HEAD-marked heading folds into the next content line as a lead-in ("Weather,
-    It will be sunny."); a heading with no content after it becomes its own sentence.
-    """
+    """Collapse visual formatting into calm spoken paragraphs. A _HEAD-marked heading folds into
+    the next content line as a lead-in ("Weather, It will be sunny."); a heading with no content
+    after it becomes its own sentence."""
     if not text:
         return ""
-
     raw_lines = text.splitlines()
     add_sentence_pauses = sum(1 for raw_line in raw_lines if raw_line.replace(_HEAD, "").strip()) > 1
     lines: list[str] = []
@@ -161,7 +139,6 @@ def smooth_whitespace_for_tts(text: str) -> str:
         if pending_heading is not None:
             lines.append(pending_heading.rstrip(".:;,") + ".")
             pending_heading = None
-
     for raw_line in raw_lines:
         is_heading = raw_line.rstrip().endswith(_HEAD)
         line = raw_line.replace(_HEAD, "").strip()
@@ -180,15 +157,11 @@ def smooth_whitespace_for_tts(text: str) -> str:
         if add_sentence_pauses and line[-1] not in ".!?;:":
             line += "."
         lines.append(line)
-
     flush_pending()
-
     text = "\n".join(lines)
-    text = re.sub(r"\n{3,}", "\n\n", text)
-    text = re.sub(r"[ \t]{2,}", " ", text)
-    text = re.sub(r"\s+([,.;:!?])", r"\1", text)
-    text = re.sub(r"([,.;:!?])([A-Za-z])", r"\1 \2", text)
-    text = re.sub(r"\.{4,}", "...", text)
+    for pattern, repl in ((r"\n{3,}", "\n\n"), (r"[ \t]{2,}", " "), (r"\s+([,.;:!?])", r"\1"),
+                          (r"([,.;:!?])([A-Za-z])", r"\1 \2"), (r"\.{4,}", "...")):
+        text = re.sub(pattern, repl, text)
     return text.strip()
 
 
@@ -212,27 +185,21 @@ def strip_nonspoken_blocks(text: str) -> str:
 
 
 def flatten_newlines_for_payload(text: str) -> str:
-    """Collapse newlines into sentence breaks for single-line TTS payloads.
-
-    Some OpenAI-compatible backends (e.g. Kokoro) truncate at the first newline; the
-    smoothing pass already terminates each line with punctuation, so this is safe.
-    """
+    """Collapse newlines into sentence breaks for single-line TTS payloads: some OpenAI-compatible
+    backends (e.g. Kokoro) truncate at the first newline; smoothing already ends each line with
+    punctuation, so this is safe."""
     if not text:
         return ""
-    text = re.sub(r"\n{2,}", ". ", text)
-    text = re.sub(r"(?<=[.!?;:,])\n", " ", text)
-    text = text.replace("\n", ". ")
-    text = re.sub(r"\.\s*\.", ".", text)
-    text = re.sub(r"[ \t]{2,}", " ", text)
+    for pattern, repl in ((r"\n{2,}", ". "), (r"(?<=[.!?;:,])\n", " "), (r"\n", ". "), (r"\.\s*\.", "."),
+                          (r"[ \t]{2,}", " ")):
+        text = re.sub(pattern, repl, text)
     return text.strip()
 
 
 def prepare_spoken_text(text: str, max_chars: int | None = 4000) -> str:
     """Return a TTS-friendly script from assistant text (deterministic cleanup, not a rewrite).
-
-    Pipeline: non-spoken blocks > Markdown > symbols/units > line formatting into
-    sentence pauses > single line (for newline-sensitive providers), then ``max_chars``.
-    """
+    Pipeline: non-spoken blocks > Markdown > symbols/units > line formatting into sentence
+    pauses > single line (for newline-sensitive providers), then ``max_chars``."""
     spoken = text
     for step in (strip_nonspoken_blocks, strip_markdown_for_tts, normalize_symbols_for_tts,
                  smooth_whitespace_for_tts, flatten_newlines_for_payload):
@@ -242,30 +209,6 @@ def prepare_spoken_text(text: str, max_chars: int | None = 4000) -> str:
     return spoken
 
 
-# Legacy regex fallback, only used if the shared normalizer raises.
-_LEGACY_TTS_STRIP_STEPS = (
-    (re.compile(r'<think[\s>].*?</think>', flags=re.DOTALL), ' '),
-    (re.compile(r'```[\s\S]*?```'), ' '),
-    (re.compile(r'\[([^\]]+)\]\([^)]+\)'), r'\1'),
-    (re.compile(r'https?://\S+'), ''),
-    (re.compile(r'\*\*(.+?)\*\*'), r'\1'),
-    (re.compile(r'\*(.+?)\*'), r'\1'),
-    (re.compile(r'`(.+?)`'), r'\1'),
-    (re.compile(r'^#+\s*', flags=re.MULTILINE), ''),
-    (re.compile(r'^\s*[-*]\s+', flags=re.MULTILINE), ''),
-    (re.compile(r'---+'), ''),
-    # Emoji + variation selectors/ZWJ: providers speak them as awkward labels.
-    (re.compile('[\U0001F000-\U0001FAFF\u2600-\u27BF\uFE0F\u200D\U000E0020-\U000E007F]+'), ' '),
-    (re.compile(r'\n{3,}'), '\n\n'),
-)
-
-
 def _strip_markdown_for_tts(text: str) -> str:
-    """``prepare_spoken_text`` without a length cap; falls back to the legacy regex pipeline if it raises."""
-    try:
-        return prepare_spoken_text(text, max_chars=None)
-    except Exception:
-        pass
-    for pattern, repl in _LEGACY_TTS_STRIP_STEPS:
-        text = pattern.sub(repl, text)
-    return text.strip()
+    """``prepare_spoken_text`` without a length cap (``tts_tool`` compatibility name)."""
+    return prepare_spoken_text(text, max_chars=None)