diff --git a/tools/transcription_audio.py b/tools/transcription_audio.py
index 07e45a6734..15413f370a 100644
--- a/tools/transcription_audio.py
+++ b/tools/transcription_audio.py
@@ -2,10 +2,9 @@
 
 Binary discovery, the shared ffmpeg m4a encode (transcode + silence trim),
 source/format validation, WeChat .silk decoding, CAF conversion and the
-best-effort cloud pre-upload silence trim.
-
-Split out of ``tools/transcription_tools.py``, which re-imports every name (patch
-surface) and is imported lazily here so origin patches still intercept.
+best-effort cloud pre-upload silence trim. Every name is re-imported by
+``tools/transcription_tools.py`` (patch surface), which is imported lazily here
+so origin patches still intercept.
 """
 
 from __future__ import annotations
@@ -61,17 +60,11 @@ def _run_quiet(command: list, *, timeout: float, env: Optional[dict] = None) ->
 
 # Shared encode profile for every STT-bound m4a (transcode and silence-trim):
 # 16 kHz mono 32 kbps AAC, faststart. One owner so codec/bitrate never drift.
-_STT_M4A_ENCODE_ARGS = (
-    "-vn", "-ac", "1", "-ar", "16000",
-    "-c:a", "aac", "-b:a", "32k", "-movflags", "+faststart",
-)
+_STT_M4A_ENCODE_ARGS = ("-vn", "-ac", "1", "-ar", "16000", "-c:a", "aac", "-b:a", "32k", "-movflags", "+faststart")
 
 
 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 error semantics (transcode reports, trim swallows).
-    """
+    """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]
@@ -80,11 +73,10 @@ def _run_ffmpeg_stt_encode(ffmpeg: str, input_path: str, output_path: str, *, au
 
 
 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.
+    """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.
-    Returns ``(converted_path, None)`` or ``(None, error)``.
     """
     from tools.transcription_tools import _find_ffmpeg_binary, _run_ffmpeg_stt_encode
     ffmpeg = _find_ffmpeg_binary()
@@ -110,16 +102,13 @@ def _validate_audio_file_size(audio_path: Path, *, enforce_size_limit: bool = Tr
     except OSError as e:
         return _error_result(f"Failed to access file: {e}")
     if enforce_size_limit and file_size > MAX_FILE_SIZE:
-        return _error_result(
-            f"File too large: {file_size / (1024*1024):.1f}MB (max {MAX_FILE_SIZE / (1024*1024):.0f}MB)"
-        )
+        return _error_result(f"File too large: {file_size / (1024*1024):.1f}MB (max {MAX_FILE_SIZE / (1024*1024):.0f}MB)")
     return None
 
 
 def _validate_audio_source_file(file_path: str, *, enforce_size_limit: bool = True) -> Optional[Dict[str, Any]]:
     """Validate source path safety (and optionally size) before any decoder runs."""
     audio_path = Path(file_path)
-
     if os.path.islink(audio_path):
         return _error_result(f"Path is a symbolic link: {file_path}")
     if not audio_path.exists():
@@ -137,9 +126,7 @@ def _validate_audio_file(file_path: str, *, enforce_size_limit: bool = True) ->
 
     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 _error_result(f"Unsupported format: {suffix}. Supported: {', '.join(sorted(SUPPORTED_FORMATS))}")
     return None
 
 
@@ -150,8 +137,7 @@ def _prepare_audio_for_transcription(file_path: str) -> tuple[Optional[str], Opt
     if audio_path.suffix.lower() != ".silk":
         return file_path, None, None
     if not _HAS_PILK:
-        # pilk is a tiny silk-v3 codec binding — lazy-install on first .silk
-        # voice note instead of bloating the base install.
+        # pilk is a tiny silk-v3 codec binding — lazy-installed on first .silk voice note.
         _lazy_ensure_quietly("stt.silk")
         if not _safe_find_spec("pilk"):
             return None, None, _error_result(
@@ -221,23 +207,22 @@ def _convert_caf_to_wav(file_path: str) -> Optional[str]:
 
 # ---- Cloud pre-upload silence trim --------------------------------------
 #
-# Local faster-whisper gets Silero VAD; cloud providers get the raw file, so
-# every second of silence is paid for twice (upload + per-minute billing) and
-# cloud Whisper hallucinates on it. Before uploading to a built-in cloud
-# provider we collapse long pauses with ffmpeg's silenceremove, keeping
-# ``stt.cloud_trim_keep_ms`` of each pause so word boundaries survive.
-# Purely best-effort — ANY of these uploads the original untouched:
-# ``stt.cloud_trim_silence: false``, ffmpeg/ffprobe missing, trim failure or
-# timeout, a ~empty result (the provider, not a dB heuristic, decides "no
-# speech"), or <10% saving. Command-type and plugin providers are NOT trimmed:
-# they may wrap local CLIs that want the original bytes.
+# Local faster-whisper gets Silero VAD; cloud providers get the raw file, so silence
+# is paid for twice (upload + per-minute billing) and cloud Whisper hallucinates on
+# it. Before uploading to a built-in cloud provider we collapse long pauses with
+# ffmpeg's silenceremove, keeping ``stt.cloud_trim_keep_ms`` of each pause so word
+# boundaries survive. Purely best-effort — ANY of these uploads the original:
+# ``stt.cloud_trim_silence: false``, ffmpeg/ffprobe missing, trim failure/timeout, a
+# ~empty result (the provider, not a dB heuristic, decides "no speech"), or <10%
+# saving. Command-type and plugin providers are NOT trimmed: they may wrap local
+# CLIs that want the original bytes.
 
 _CLOUD_TRIM_THRESHOLD_DB_DEFAULT = -40  # audio below this level counts as silence
 _CLOUD_TRIM_KEEP_MS_DEFAULT = 300  # how much of each pause survives the trim
 _CLOUD_TRIM_MIN_SAVING = 0.10  # use the trimmed file only when >=10% shorter
 _CLOUD_TRIM_MIN_RESULT_SECONDS = 0.3  # all-silence guard floor: never upload ~empty audio
-# Below this the trim can't pay for itself (several providers bill a 10s
-# minimum per request) and the encode would sit on the synchronous voice-note path.
+# Below this the trim can't pay for itself (several providers bill a 10s minimum)
+# and the encode would sit on the synchronous voice-note path.
 _CLOUD_TRIM_MIN_INPUT_SECONDS = 12.0
 
 
@@ -252,10 +237,7 @@ def _probe_audio_duration(file_path: str) -> Optional[float]:
     if not ffprobe:
         return None
     command = [
-        ffprobe, "-v", "error",
-        "-show_entries", "format=duration",
-        "-of", "default=noprint_wrappers=1:nokey=1",
-        file_path,
+        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())
@@ -274,11 +256,9 @@ 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.
+    """Return a silence-trimmed copy of *file_path* for cloud upload, or None (= upload the original).
 
-    ``None`` always means "upload the original" (disabled, tools missing, clip
-    too short, trim failed, mostly silence, or not enough saving). 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)
@@ -301,8 +281,7 @@ def _trim_silence_for_cloud_stt(file_path: str, stt_config: Dict[str, Any]) -> O
         return None
 
     keep_seconds = keep_ms / 1000.0
-    # start_periods=1 strips leading silence; stop_periods=-1 collapses every
-    # interior/trailing silence, keeping ``keep_seconds`` of each pause.
+    # 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}:"
@@ -310,8 +289,7 @@ def _trim_silence_for_cloud_stt(file_path: str, stt_config: Dict[str, Any]) -> O
     )
     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: an output consisting solely
-    # of kept pause must never be uploaded as "speech".
+    # Scale the all-silence guard with keep_ms: output that is solely kept pause must never upload as "speech".
     min_result_seconds = max(_CLOUD_TRIM_MIN_RESULT_SECONDS, 2 * keep_seconds)
     keep_result = False
     try:
diff --git a/tools/transcription_cloud.py b/tools/transcription_cloud.py
index c81f008a6d..a1d20f9d51 100644
--- a/tools/transcription_cloud.py
+++ b/tools/transcription_cloud.py
@@ -1,11 +1,10 @@
 """Cloud STT providers.
 
-OpenAI-SDK-shaped backends (groq, openai, deepinfra), Mistral Voxtral, the REST
-multipart backends (xAI, ElevenLabs), and OpenAI audio credential resolution
-(config > keyless local server > env > managed Nous gateway).
-
-Split out of ``tools/transcription_tools.py``, which re-imports every name (patch
-surface) and is imported lazily here so origin patches still intercept.
+OpenAI-SDK-shaped backends (groq, openai, deepinfra), Mistral Voxtral, REST multipart
+backends (xAI, ElevenLabs), and OpenAI audio credential resolution (config > keyless
+local server > env > managed Nous gateway). Every name is re-imported by
+``tools/transcription_tools.py`` (patch surface), imported lazily here so origin
+patches still intercept.
 """
 
 from __future__ import annotations
@@ -14,7 +13,7 @@ import logging
 import re
 import tempfile
 from pathlib import Path
-from typing import Any, Dict, Optional
+from typing import Any, Callable, Dict, Optional
 from urllib.parse import urljoin
 
 from utils import is_truthy_value
@@ -35,25 +34,17 @@ def _has_xai_stt_credentials() -> bool:
     return bool(resolve_xai_http_credentials().get("api_key"))
 
 
-def _close_client(client: Any) -> None:
-    close = getattr(client, "close", None)
-    if callable(close):
-        close()
-
-
 def _with_openai_client(api_key: str, base_url: Optional[str], file_path: str, log_label: str, body):
-    """Run ``body(client)`` against a fresh OpenAI SDK client (30s timeout, no SDK retries).
-
-    Always closes the client; any exception maps to the shared envelope via
-    :func:`_openai_sdk_failure`.
-    """
+    """Run ``body(client)`` on a fresh OpenAI SDK client (30s timeout, no retries); always closed, errors -> envelope."""
     try:
         from openai import OpenAI
         client = OpenAI(api_key=api_key, base_url=base_url, timeout=30, max_retries=0)
         try:
             return body(client)
         finally:
-            _close_client(client)
+            close = getattr(client, "close", None)
+            if callable(close):
+                close()
     except Exception as e:
         return _openai_sdk_failure(e, file_path, log_label)
 
@@ -69,8 +60,8 @@ def _cloud_failure(exc: BaseException, file_path: str, label: str, detail: Optio
 def _openai_sdk_failure(exc: BaseException, file_path: str, log_label: str) -> Dict[str, Any]:
     """Map an OpenAI-SDK-shaped exception to the shared error envelope.
 
-    Order matters: APIConnectionError is checked before APITimeoutError (its
-    subclass) so timeouts report as connection errors, as they always have.
+    APIConnectionError is checked before APITimeoutError (its subclass) so timeouts
+    report as connection errors, as they always have.
     """
     try:
         from openai import APIError, APIConnectionError, APITimeoutError
@@ -78,33 +69,31 @@ def _openai_sdk_failure(exc: BaseException, file_path: str, log_label: str) -> D
         APIError = APIConnectionError = APITimeoutError = ()
     if isinstance(exc, PermissionError):
         return _error_result(f"Permission denied: {file_path}")
-    if isinstance(exc, APIConnectionError):
-        return _error_result(f"Connection error: {exc}")
-    if isinstance(exc, APITimeoutError):
-        return _error_result(f"Request timeout: {exc}")
-    if isinstance(exc, APIError):
-        return _error_result(f"API error: {exc}")
+    for cls, label in ((APIConnectionError, "Connection error"), (APITimeoutError, "Request timeout"), (APIError, "API error")):
+        if isinstance(exc, cls):
+            return _error_result(f"{label}: {exc}")
     logger.error("%s transcription failed: %s", log_label, exc, exc_info=True)
     return _error_result(f"Transcription failed: {exc}")
 
 
-def _transcribe_groq(
-    file_path: str,
-    model_name: str,
-    *,
-    language: Optional[str] = None,
-    prompt: Optional[str] = None,
-) -> Dict[str, Any]:
-    """Transcribe using Groq Whisper API (free tier available).
+def _sdk_prompt_kwargs(language: Optional[str], prompt: Optional[str]) -> Dict[str, Any]:
+    """``language``/``prompt`` create-kwargs, each only when set so the bare request stays byte-identical."""
+    kwargs: Dict[str, Any] = {}
+    if language:
+        kwargs["language"] = language
+    if prompt:
+        kwargs["prompt"] = prompt
+    return kwargs
 
-    Language: hook override > ``stt.groq.language`` > ``stt.language`` > env;
-    otherwise Groq auto-detects.
-    """
+
+def _transcribe_groq(
+    file_path: str, model_name: str, *, language: Optional[str] = None, prompt: Optional[str] = None
+) -> Dict[str, Any]:
+    """Transcribe via the Groq Whisper API; language: hook > ``stt.groq.language`` > ``stt.language`` > env > auto."""
     from tools.transcription_tools import _HAS_OPENAI, _resolve_provider_key, _resolve_stt_language
     api_key = _resolve_provider_key("GROQ_API_KEY", "groq")
     if not api_key:
         return _error_result("GROQ_API_KEY not set")
-
     if not _HAS_OPENAI:
         return _error_result("openai package not installed")
 
@@ -116,45 +105,26 @@ def _transcribe_groq(
     language = language or _resolve_stt_language("groq")
 
     def _run(client):
-        create_kwargs = {
-            "model": model_name,
-            "response_format": "text",
-        }
-        if language:
-            create_kwargs["language"] = language
-        if prompt:
-            # Only sent when set so the no-hook, no-config request stays byte-identical.
-            create_kwargs["prompt"] = prompt
+        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, **create_kwargs)
         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,
+    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 backend for every OpenAI-compatible STT endpoint (DeepInfra etc.):
-    callers pass explicit ``api_key``/``base_url`` to skip the OpenAI-only auth
-    chain and a ``provider_label`` for the response's ``provider``.
+    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:
@@ -170,8 +140,7 @@ def _transcribe_openai(
     if not _HAS_OPENAI:
         return _error_result("openai package not installed")
 
-    # Auto-correct a Groq-only model on the native OpenAI path only —
-    # third-party endpoints may legitimately serve a whisper-large-v3 variant.
+    # Auto-correct a Groq-only model on the native OpenAI path only (third-party endpoints may serve it).
     if provider_label == "openai" and model_name in GROQ_MODELS:
         logger.info("Model %s not available on OpenAI, using %s", model_name, DEFAULT_STT_MODEL)
         model_name = DEFAULT_STT_MODEL
@@ -188,14 +157,12 @@ def _transcribe_openai(
                 }
                 if language:
                     if model_name == "gpt-transcribe":
-                        # gpt-transcribe replaces ``language`` with a ``languages``
-                        # list and rejects requests sending the legacy field.
+                        # gpt-transcribe takes a ``languages`` list and rejects the legacy field.
                         create_kwargs["extra_body"] = {"languages": [language]}
                     else:
                         create_kwargs["language"] = language
                     logger.debug("Using language hint '%s' for OpenAI STT", language)
-                if prompt:
-                    # Only sent when set so the no-hook, no-config request stays byte-identical.
+                if prompt:  # only when set so the bare request stays byte-identical
                     create_kwargs["prompt"] = prompt
                 return client.audio.transcriptions.create(**create_kwargs)
 
@@ -206,8 +173,7 @@ def _transcribe_openai(
                 message = str(exc).lower()
                 if not any(k in message for k in ("unsupported", "corrupted", "invalid file")):
                     raise
-                # Newer models reject some containers whisper-1 accepted
-                # (notably Ogg/Opus voice notes): transcode to m4a, retry once.
+                # 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)
                 if transcode_error:
                     return _error_result(transcode_error)
@@ -222,18 +188,13 @@ def _transcribe_openai(
             "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)
 
 
 def _transcribe_mistral(
-    file_path: str,
-    model_name: str,
-    *,
-    language: Optional[str] = None,
-    prompt: Optional[str] = None,
+    file_path: str, model_name: str, *, language: Optional[str] = None, prompt: Optional[str] = None
 ) -> Dict[str, Any]:
     """Transcribe with the ``mistralai`` SDK (``/v1/audio/transcriptions``); requires ``MISTRAL_API_KEY``."""
     from tools.transcription_tools import _resolve_provider_key, _resolve_stt_language
@@ -247,17 +208,13 @@ def _transcribe_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")
                 complete_kwargs: Dict[str, Any] = {
                     "model": model_name,
                     "file": {"content": audio_file, "file_name": Path(file_path).name},
+                    **_sdk_prompt_kwargs(language, prompt),
                 }
-                # Language: hook override > stt.mistral.language > stt.language > env > auto.
-                language = language or _resolve_stt_language("mistral")
-                if language:
-                    complete_kwargs["language"] = language
-                if prompt:
-                    # Only sent when set so the no-hook, no-config request stays byte-identical.
-                    complete_kwargs["prompt"] = prompt
                 result = client.audio.transcriptions.complete(**complete_kwargs)
 
             transcript_text = _extract_transcript_text(result)
@@ -271,6 +228,9 @@ def _transcribe_mistral(
         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
 
@@ -281,22 +241,17 @@ def _post_audio_multipart(url: str, headers: Dict[str, str], file_path: str, dat
         )
 
 
-def _http_error_detail(response, extract) -> str:
-    """``extract(json_body)`` -> detail string, falling back to the first 300 chars of the body."""
-    try:
-        return extract(response.json()) or response.text[:300]
-    except Exception:
-        return response.text[:300]
-
-
 def _rest_transcript(response, label: str, extract_detail, extract_text):
-    """Turn a multipart STT response into ``(text, body, None)`` or ``(None, None, error_envelope)``.
+    """Multipart STT response -> ``(text, body, None)`` or ``(None, None, error_envelope)``.
 
-    Non-200 -> ``"<label> API error (HTTP n): detail"``; empty text -> the
-    ``no_speech`` envelope so callers treat silence as non-fatal.
+    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).
     """
     if response.status_code != 200:
-        detail = _http_error_detail(response, extract_detail)
+        try:
+            detail = extract_detail(response.json()) or response.text[:300]
+        except Exception:
+            detail = response.text[:300]
         return None, None, _error_result(f"{label} API error (HTTP {response.status_code}): {detail}")
     body = response.json()
     text = extract_text(body)
@@ -305,19 +260,23 @@ def _rest_transcript(response, label: str, extract_detail, extract_text):
     return text, body, None
 
 
-def _elevenlabs_error_detail(err_body: Dict[str, Any]) -> str:
-    error_value = err_body.get("detail") or err_body.get("error")
-    if isinstance(error_value, dict):
-        return str(error_value.get("message") or error_value)
-    return str(error_value) if error_value else ""
+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 REST flow: ``post()`` -> envelope/transcript -> ``log(text, body)`` -> ok; exceptions -> ``_cloud_failure``."""
+    try:
+        transcript_text, body, error = _rest_transcript(post(), label, extract_detail, extract_text)
+        if error:
+            return error
+        log(transcript_text, body)
+        return _ok_result(transcript_text, provider)
+    except Exception as e:
+        return _cloud_failure(e, file_path, f"{label} transcription")
 
 
 def _transcribe_xai(
-    file_path: str,
-    model_name: str,
-    *,
-    language: Optional[str] = None,
-    prompt: Optional[str] = None,
+    file_path: str, model_name: str, *, language: Optional[str] = None, prompt: Optional[str] = None
 ) -> Dict[str, Any]:
     """Transcribe via xAI ``POST /v1/stt`` (multipart). Supports ITN, diarization, word timestamps."""
     from tools.transcription_tools import _load_stt_config, _resolve_stt_language, get_env_value
@@ -326,9 +285,8 @@ def _transcribe_xai(
     if prompt:
         _log_prompt_unsupported("STT provider 'xai'")
 
-    # STT is API-billed: prefer the explicit XAI_API_KEY over the general xAI
-    # OAuth/Grok-subscription credential, which may be valid for Grok yet hit
-    # personal-team spending-limit errors on /v1/stt.
+    # 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 = {
@@ -346,8 +304,7 @@ def _transcribe_xai(
     xai_config = stt_config.get("xai") or {}
 
     def _resolve_base_url(resolved_creds: Dict[str, str]) -> str:
-        # OAuth bearers are pinned to the resolver-validated xAI origin;
-        # config/env base URL overrides only apply to API-key credentials.
+        # 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:
@@ -357,7 +314,7 @@ def _transcribe_xai(
     # Language: hook override > stt.xai.language > stt.language > env.
     language = language or _resolve_stt_language("xai", stt_config) or ""
 
-    try:
+    def _post() -> Any:
         from tools.xai_http import hermes_xai_user_agent
 
         data: Dict[str, str] = {}
@@ -375,7 +332,6 @@ def _transcribe_xai(
             )
 
         response = _post_transcription(api_key, _resolve_base_url(creds))
-
         if response.status_code in {401, 403} and creds.get("provider") == "xai-oauth":
             logger.info("xAI STT got HTTP %d; refreshing OAuth credentials and retrying once", response.status_code)
             try:
@@ -387,30 +343,29 @@ def _transcribe_xai(
                 logger.warning(
                     "xAI STT OAuth refresh-and-retry after HTTP %d failed: %s", response.status_code, retry_exc,
                 )
+        return response
 
-        transcript_text, result, error = _rest_transcript(
-            response, "xAI STT",
-            lambda body: body.get("error", {}).get("message", ""),
-            lambda body: body.get("text", "").strip(),
-        )
-        if error:
-            return error
+    def _log(transcript_text: str, result: Dict[str, Any]) -> None:
         logger.info(
             "Transcribed %s via xAI Grok STT (lang=%s, %.1fs audio, %d chars)",
             Path(file_path).name, result.get("language", language), result.get("duration", 0), len(transcript_text),
         )
-        return _ok_result(transcript_text, "xai")
 
-    except Exception as e:
-        return _cloud_failure(e, file_path, "xAI STT transcription")
+    return _rest_provider(
+        file_path, "xai", "xAI STT", _post, lambda body: body.get("error", {}).get("message", ""),
+        lambda body: body.get("text", "").strip(), _log,
+    )
+
+
+def _elevenlabs_error_detail(err_body: Dict[str, Any]) -> str:
+    error_value = err_body.get("detail") or err_body.get("error")
+    if isinstance(error_value, dict):
+        return str(error_value.get("message") or error_value)
+    return str(error_value) if error_value else ""
 
 
 def _transcribe_elevenlabs(
-    file_path: str,
-    model_name: str,
-    *,
-    language: Optional[str] = None,
-    prompt: Optional[str] = None,
+    file_path: str, model_name: str, *, language: Optional[str] = None, prompt: Optional[str] = None
 ) -> Dict[str, Any]:
     """Transcribe using ElevenLabs Scribe STT API."""
     from tools.transcription_tools import _load_stt_config, _resolve_provider_key, _resolve_stt_language, get_env_value
@@ -428,7 +383,8 @@ def _transcribe_elevenlabs(
     ).strip().rstrip("/")
     # Language: hook override > stt.elevenlabs.language(_code) > stt.language.
     language_code = language or _resolve_stt_language("elevenlabs", stt_config, extra_keys=("language_code",)) or ""
-    try:
+
+    def _post() -> Any:
         data: Dict[str, str] = {
             "model_id": model_name,
             "tag_audio_events": str(is_truthy_value(elevenlabs_config.get("tag_audio_events", False))).lower(),
@@ -436,33 +392,23 @@ def _transcribe_elevenlabs(
         }
         if language_code:
             data["language_code"] = language_code
+        return _post_audio_multipart(f"{base_url}/speech-to-text", {"xi-api-key": api_key}, file_path, data)
 
-        response = _post_audio_multipart(f"{base_url}/speech-to-text", {"xi-api-key": api_key}, file_path, data)
-
-        transcript_text, _body, error = _rest_transcript(
-            response, "ElevenLabs STT", _elevenlabs_error_detail, _extract_transcript_text,
-        )
-        if error:
-            return error
+    def _log(transcript_text: str, _body: Dict[str, Any]) -> None:
         logger.info(
             "Transcribed %s via ElevenLabs Scribe (%s, %d chars)",
             Path(file_path).name, model_name, len(transcript_text),
         )
-        return _ok_result(transcript_text, "elevenlabs")
 
-    except Exception as e:
-        return _cloud_failure(e, file_path, "ElevenLabs STT transcription")
+    return _rest_provider(
+        file_path, "elevenlabs", "ElevenLabs STT", _post, _elevenlabs_error_detail, _extract_transcript_text, _log,
+    )
 
 
 def _transcribe_deepinfra(
-    file_path: str,
-    model_name: str,
-    *,
-    language: Optional[str] = None,
-    prompt: Optional[str] = None,
+    file_path: str, model_name: str, *, language: Optional[str] = None, prompt: Optional[str] = None
 ) -> Dict[str, Any]:
-    """Resolve DeepInfra credentials/model (via the shared ``hermes_cli.models``
-    helpers), then delegate to :func:`_transcribe_openai`."""
+    """Resolve DeepInfra credentials/model (shared ``hermes_cli.models`` helpers), then delegate to :func:`_transcribe_openai`."""
     from tools.transcription_tools import _load_stt_config, _resolve_provider_key, _transcribe_openai
     api_key = _resolve_provider_key("DEEPINFRA_API_KEY", "deepinfra")
     if not api_key:
@@ -488,13 +434,12 @@ def _transcribe_deepinfra(
     )
 
 
-def _is_local_or_private_url(url: str) -> bool:
-    """True for loopback/RFC-1918/LAN-internal hosts.
+# ---- OpenAI audio credential resolution -----------------------------------
 
-    Decides whether an empty ``stt.openai.api_key`` is acceptable: local
-    OpenAI-compatible STT servers ignore the auth header, so users shouldn't
-    need a sham ``api_key: not-needed``.
-    """
+
+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)."""
     try:
         from urllib.parse import urlparse
         import ipaddress
@@ -511,11 +456,8 @@ def _is_local_or_private_url(url: str) -> bool:
 
 
 def _direct_openai_credentials(cfg_api_key: str, cfg_base_url: str) -> Optional[tuple[str, str]]:
-    """Direct-credential ladder: config key > keyless local base_url > env key; None if none apply.
-
-    A local OpenAI-compatible server needs no key — send a placeholder so the
-    SDK doesn't refuse to construct a client.
-    """
+    """Direct-credential ladder: config key > keyless local base_url (placeholder key so the SDK
+    constructs a client) > env key; None if none apply."""
     from tools.transcription_tools import resolve_openai_audio_api_key
     if cfg_api_key:
         return cfg_api_key, (cfg_base_url or OPENAI_BASE_URL)
@@ -528,16 +470,10 @@ def _direct_openai_credentials(cfg_api_key: str, cfg_base_url: str) -> Optional[
 
 
 def _resolve_openai_audio_client_config() -> tuple[str, str]:
-    """Return ``(api_key, base_url)`` for the OpenAI STT client.
-
-    Strict selection semantics on the stored ``stt`` provider string:
-    - ``"nous"`` → managed gateway ONLY; unentitled/unreachable is a
-      selection-naming error (a direct OPENAI_API_KEY must NOT override it).
-    - any other stored provider → direct credentials ONLY; missing credentials
-      is a selection-naming error — no silent managed fallback.
-    - never-configured stt section → legacy ladder: direct credentials, then
-      the managed gateway.
-    """
+    """``(api_key, base_url)`` for the OpenAI STT client, strict on the stored ``stt`` selection:
+    ``"nous"`` -> managed gateway ONLY (a direct OPENAI_API_KEY must NOT override it); any other
+    stored provider -> direct credentials ONLY (no silent managed fallback); never-configured ->
+    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,
diff --git a/tools/transcription_command.py b/tools/transcription_command.py
index 82e8da1d57..52e00c3dea 100644
--- a/tools/transcription_command.py
+++ b/tools/transcription_command.py
@@ -2,10 +2,9 @@
 
 ``stt.providers.<name>: type: command`` registry, plugin-registered
 ``TranscriptionProvider`` dispatch, and the ``pre_transcription`` hook that
-threads prompt/language/model overrides into every backend.
-
-Split out of ``tools/transcription_tools.py``, which re-imports every name (patch
-surface) and is imported lazily here so origin patches still intercept.
+threads prompt/language/model overrides into every backend. Every name is
+re-imported by ``tools/transcription_tools.py`` (patch surface), which is
+imported lazily here so origin patches still intercept.
 """
 
 from __future__ import annotations
@@ -17,12 +16,13 @@ from pathlib import Path
 from typing import Any, Dict, Optional
 
 from tools.tts_command_provider import (
-    command_env_passthrough as _command_stt_env_passthrough,
-    render_command_template as _render_command_stt_template,
+    _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,
 )
 from tools.transcription_common import (
-    BUILTIN_STT_PROVIDERS, _error_result, _get_stt_section, _log_prompt_unsupported, _ok_result,
+    BUILTIN_STT_PROVIDERS, _error_result, _log_prompt_unsupported, _ok_result,
 )
 
 # Log-record parity with the origin module.
@@ -31,86 +31,38 @@ logger = logging.getLogger("tools.transcription_tools")
 
 # ---- Command-provider registry (``stt.providers.<name>: type: command``) ---
 #
-# Mirrors the TTS command-provider registry: same placeholder grammar,
-# shell-quote-aware rendering and process-tree termination on timeout.
-# Resolution order: built-in name (always wins) > stt.providers.<name> command
-# > plugin-registered TranscriptionProvider > "No STT provider available".
-# The single-env-var HERMES_LOCAL_STT_COMMAND escape hatch stays untouched via
-# the built-in ``local_command`` path.
+# Mirrors the TTS command-provider registry (same placeholder grammar, quote-aware
+# rendering, process-tree termination on timeout). Resolution order: built-in name
+# (always wins) > stt.providers.<name> command > plugin TranscriptionProvider >
+# "No STT provider available". The single-env-var HERMES_LOCAL_STT_COMMAND escape
+# hatch stays untouched via the built-in ``local_command`` path.
 DEFAULT_COMMAND_STT_TIMEOUT_SECONDS = 300
 DEFAULT_COMMAND_STT_LANGUAGE = "en"
 DEFAULT_COMMAND_STT_OUTPUT_FORMAT = "txt"
 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]:
-    """Return the config for a user-declared STT provider, or {}.
-
-    ``stt.providers.<name>`` is canonical; ``stt.<name>`` is accepted for
-    back-compat only when *name* is not a built-in, so a user's ``stt.openai``
-    block still means the OpenAI provider. Built-in sections can't be mistaken
-    for command providers anyway: ``_is_command_stt_provider_config`` requires
-    an explicit ``command:``.
-    """
-    providers = _get_stt_section(stt_config, "providers")
-    section = providers.get(name)
-    if isinstance(section, dict):
-        return section
-    if name.lower() not in BUILTIN_STT_PROVIDERS:
-        return _get_stt_section(stt_config, name)
-    return {}
+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 _is_command_stt_provider_config(config: Dict[str, Any]) -> bool:
-    """Return True when *config* declares a command-type STT provider."""
-    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())
-
-
-def _resolve_command_stt_provider_config(
-    provider: str,
-    stt_config: Dict[str, Any],
-) -> Optional[Dict[str, Any]]:
-    """Return the provider config if *provider* is a command type; None for built-ins, ``none``, unknown."""
-    if not provider:
-        return None
-    key = provider.lower().strip()
-    if key in BUILTIN_STT_PROVIDERS or key == "none":
-        return None
-    config = _get_named_stt_provider_config(stt_config, key)
-    return config if _is_command_stt_provider_config(config) else None
+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 timeout in seconds (``timeout`` > ``timeout_seconds``), falling back when invalid or <= 0."""
-    raw = config.get("timeout", config.get("timeout_seconds", DEFAULT_COMMAND_STT_TIMEOUT_SECONDS))
-    try:
-        value = float(raw)
-    except (TypeError, ValueError):
-        value = 0.0
-    return value if value > 0 else float(DEFAULT_COMMAND_STT_TIMEOUT_SECONDS)
+    return _command_timeout(config, DEFAULT_COMMAND_STT_TIMEOUT_SECONDS)
 
 
 def _get_command_stt_output_format(config: Dict[str, Any]) -> str:
-    """Return the validated output format (txt/json/srt/vtt)."""
-    raw = config.get("format") or config.get("output_format") or DEFAULT_COMMAND_STT_OUTPUT_FORMAT
-    fmt = str(raw).lower().strip().lstrip(".")
-    return fmt if fmt in COMMAND_STT_OUTPUT_FORMATS else DEFAULT_COMMAND_STT_OUTPUT_FORMAT
+    return _command_output_format(config, COMMAND_STT_OUTPUT_FORMATS, DEFAULT_COMMAND_STT_OUTPUT_FORMAT)
 
 
 def _read_command_stt_output(output_path: Path, stdout: str, fmt: str) -> str:
-    """Return the transcript: non-empty output file > non-empty stdout (curl one-liners) > RuntimeError.
-
-    JSON output is returned raw — users configure ``format: txt`` or post-process.
-    """
+    """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()
@@ -120,26 +72,18 @@ def _read_command_stt_output(output_path: Path, stdout: str, fmt: str) -> str:
             return content
     if stdout and stdout.strip():
         return stdout.strip()
-    raise RuntimeError(
-        f"Command STT provider wrote no output file at {output_path} "
-        f"and produced no stdout"
-    )
+    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,
+    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 (all shell-quote-aware; ``{{``/``}}`` stay literal):
-    ``{input_path}`` original audio path, ``{output_path}`` file to write the
-    transcript to, ``{output_dir}`` its parent, ``{format}`` txt/json/srt/vtt,
+    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
@@ -160,10 +104,8 @@ def _transcribe_command_stt(
     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
+        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 ""
 
@@ -171,12 +113,9 @@ def _transcribe_command_stt(
         with tempfile.TemporaryDirectory(prefix=f"hermes-cmd-stt-{provider_name}-") as tmpdir:
             output_path = Path(tmpdir) / f"transcript.{output_format}"
             placeholders = {
-                "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),
+                "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)
             logger.info("Transcribing %s via command STT provider '%s'...", audio.name, provider_name)
@@ -185,11 +124,9 @@ def _transcribe_command_stt(
     except subprocess.TimeoutExpired:
         return fail(f"STT command provider '{provider_name}' timed out after {timeout:g}s")
     except subprocess.CalledProcessError as exc:
-        detail_parts = [
-            f"{stream}: {text.strip()}" for stream, text in (("stderr", exc.stderr), ("stdout", exc.stdout)) if text
-        ]
-        detail = "; ".join(detail_parts) or "no command output"
-        return fail(f"STT command provider '{provider_name}' exited with code {exc.returncode}: {detail}")
+        return fail(
+            f"STT command provider '{provider_name}' exited with code {exc.returncode}: {command_failure_detail(exc)}"
+        )
     except RuntimeError as exc:
         return fail(str(exc))
     except OSError as exc:
@@ -212,32 +149,22 @@ def _unregistered_stt_provider_error(provider: str) -> Dict[str, Any]:
 
 
 def _dispatch_to_plugin_provider(
-    file_path: str,
-    provider: str,
-    stt_config: Optional[Dict[str, Any]] = None,
-    *,
-    model: Optional[str] = None,
-    language: Optional[str] = None,
-    prompt: Optional[str] = None,
+    file_path: str, provider: str, stt_config: Optional[Dict[str, Any]] = None, *,
+    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 even though the caller short-circuits first,
-    so a caller refactor can't silently break them): built-in names never reach
-    the registry; a same-name ``stt.providers.<name>: type: command`` wins over
-    a plugin. A matched plugin reporting ``is_available() == False`` returns an
-    error envelope — not None — because the user explicitly opted in via
-    ``stt.provider`` and the generic fall-through message would mislead.
-    Provider exceptions become the standard error envelope.
+    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 BUILTIN_STT_PROVIDERS or key == "none":
+    if 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)
-    ):
+    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
@@ -288,24 +215,20 @@ def _dispatch_to_plugin_provider(
     return result
 
 
-# Fields a pre_transcription hook may mutate. ``file_path`` is read-only —
-# attempts to change it are logged and dropped.
+# Fields a pre_transcription hook may mutate; ``file_path`` is read-only (logged and dropped).
 _PRE_TRANSCRIPTION_MUTABLE_FIELDS = ("prompt", "language", "model")
 
-# Whisper-family models only use the final ~224 tokens of the prompt; longer
-# values waste upload bytes and can trip stricter OpenAI-compatible servers.
-# Enforced client-side (truncate with a warning, never error), ~4 chars/token.
+# Whisper-family models only use the final ~224 tokens of the prompt; longer values
+# waste upload bytes and can trip stricter OpenAI-compatible servers. Enforced
+# client-side (truncate with a warning, never error), ~4 chars/token.
 _WHISPER_PROMPT_TOKEN_CAP = 224
 _PROMPT_CHARS_PER_TOKEN = 4
 _WHISPER_PROMPT_CAPPED_PROVIDERS = frozenset({"local", "openai", "groq", "deepinfra"})
 
 
 def _enforce_prompt_length_limit(prompt: Optional[str], provider: str) -> Optional[str]:
-    """Truncate *prompt* to the whisper-family token cap, keeping the TAIL (fail-open).
-
-    Whisper conditions on the final context window, so the most recently
-    appended hints survive. Other providers own their own validation.
-    """
+    """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
@@ -314,35 +237,21 @@ def _enforce_prompt_length_limit(prompt: Optional[str], provider: str) -> Option
     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:]
 
 
 def _apply_pre_transcription_hook(
-    *,
-    file_path: str,
-    provider: str,
-    model: Optional[str],
-    language: Optional[str],
-    prompt: Optional[str],
-    source: Optional[str],
+    *, file_path: str, provider: str, model: Optional[str], language: Optional[str],
+    prompt: Optional[str], source: Optional[str],
 ) -> tuple[Optional[str], Optional[str], Optional[str]]:
-    """Fire the ``pre_transcription`` plugin hook and merge its results.
+    """Fire the ``pre_transcription`` plugin hook; returns ``(model, language_override, prompt)``.
 
-    Gated on ``has_hook`` so the no-hook path never builds hook kwargs, and
-    fail-open: any hook-plumbing error leaves the dispatch untouched. Results
-    arrive in registration order (plugins discovered in sorted order) and are
-    applied field-by-field, so the last hook to write a field wins. Model
-    values are accepted as-is and flow through the same per-backend
-    normalization a caller-supplied model would.
-
-    Returns ``(model, language_override, prompt)``; ``language_override`` is
-    None unless a hook explicitly set ``language``, so backends keep their own
-    config/env language 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
@@ -351,13 +260,8 @@ def _apply_pre_transcription_hook(
             return model, None, prompt
 
         hook_results = invoke_hook(
-            "pre_transcription",
-            file_path=file_path,
-            provider=provider,
-            model=model,
-            language=language,
-            prompt=prompt,
-            source=source,
+            "pre_transcription", file_path=file_path, provider=provider,
+            model=model, language=language, prompt=prompt, source=source,
         )
         overrides: Dict[str, Any] = {}
         for hook_result in hook_results:
diff --git a/tools/transcription_common.py b/tools/transcription_common.py
index 8fe1dd8eb7..56e4d695a8 100644
--- a/tools/transcription_common.py
+++ b/tools/transcription_common.py
@@ -1,7 +1,7 @@
 """Constants, result envelopes and tiny config readers shared by every STT module.
 
-Split out of ``tools/transcription_tools.py``; every name is re-imported there, so
-``tools.transcription_tools.<name>`` keeps resolving (and monkeypatching) as before.
+Every name is re-imported by ``tools/transcription_tools.py``, so
+``tools.transcription_tools.<name>`` keeps resolving (and monkeypatching).
 """
 
 from __future__ import annotations
@@ -11,6 +11,8 @@ import os
 import subprocess  # noqa: F401  (type annotation only)
 from typing import Any, Dict
 
+from tools.tts_command_provider import _get_provider_section as _get_stt_section  # noqa: F401  (re-exported)
+
 # Log-record parity with the origin module.
 logger = logging.getLogger("tools.transcription_tools")
 
@@ -39,10 +41,9 @@ MAX_FILE_SIZE = 25 * 1024 * 1024  # 25 MB
 OPENAI_MODELS = {"whisper-1", "gpt-4o-mini-transcribe", "gpt-4o-transcribe", "gpt-transcribe"}
 GROQ_MODELS = {"whisper-large-v3", "whisper-large-v3-turbo", "distil-whisper-large-v3-en"}
 
-# Providers with native handlers. Kept in sync with
-# ``agent.transcription_registry._BUILTIN_NAMES`` (a regression test fails on
-# drift); plugins may not register under these names and the dispatcher
-# short-circuits them before command/plugin lookup.
+# Providers with native handlers. Kept in sync with ``agent.transcription_registry._BUILTIN_NAMES``
+# (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",
 })
@@ -59,14 +60,6 @@ def _ok_result(transcript: str, provider: str) -> Dict[str, Any]:
     return {"success": True, "transcript": transcript, "provider": provider}
 
 
-def _get_stt_section(stt_config: Dict[str, Any], name: str) -> Dict[str, Any]:
-    """Return an stt sub-section if it's a dict, else an empty dict."""
-    if not isinstance(stt_config, dict):
-        return {}
-    section = stt_config.get(name)
-    return section if isinstance(section, dict) else {}
-
-
 def _lazy_ensure_quietly(dep: str) -> None:
     """Best-effort ``tools.lazy_deps.ensure(dep, prompt=False)``; failures are swallowed.
 
diff --git a/tools/transcription_local.py b/tools/transcription_local.py
index 976fc128da..306ca7d516 100644
--- a/tools/transcription_local.py
+++ b/tools/transcription_local.py
@@ -2,11 +2,9 @@
 
 faster-whisper loading (CUDA->CPU fallback, Apple Silicon pinning), the
 anti-hallucination transcribe kwargs and segment gate, and the local whisper CLI
-(``local_command``) provider. The cached-model singleton and its idle-unload
-watcher stay in ``transcription_tools`` (they own the module state).
-
-Split out of ``tools/transcription_tools.py``, which re-imports every name (patch
-surface) and is imported lazily here so origin patches still intercept.
+(``local_command``) provider. The cached-model singleton and idle-unload watcher
+stay in ``transcription_tools`` (module state), which re-imports every name here
+(patch surface) and is imported lazily so origin patches still intercept.
 """
 
 from __future__ import annotations
@@ -69,9 +67,8 @@ def _try_lazy_install_stt() -> bool:
     """Lazy-install faster-whisper and re-check dynamically so it's usable without a restart."""
     try:
         from tools.lazy_deps import ensure
-        # prompt=False: a bare input() deadlocks under the interactive CLI where
-        # prompt_toolkit owns stdin; the install is already gated by
-        # security.allow_lazy_installs, so reaching here is opt-in.
+        # prompt=False: a bare input() deadlocks under the interactive CLI where prompt_toolkit
+        # owns stdin; the install is already gated by security.allow_lazy_installs.
         ensure("stt.faster_whisper", prompt=False)
         if _ilu.find_spec("faster_whisper"):
             return True
@@ -89,10 +86,9 @@ def _try_lazy_install_stt() -> bool:
     return False
 
 
-# Substrings identifying a missing/unloadable CUDA runtime library: when
-# ctranslate2 can't dlopen one of these the "auto" device picker has already
-# committed to CUDA, so we fall back to CPU and reload. Deliberately narrow
-# (library names + dlopen phrasing) so legitimate runtime failures like "CUDA
+# Substrings identifying a missing/unloadable CUDA runtime library: the "auto" device
+# picker has already committed to CUDA, so we fall back to CPU and reload. Deliberately
+# narrow (library names + dlopen phrasing) so legitimate runtime failures like "CUDA
 # out of memory" surface to the user instead of silently running on CPU.
 _CUDA_LIB_ERROR_MARKERS = (
     "libcublas", "libcudnn", "libcudart", "cannot be loaded", "cannot open shared object",
@@ -111,10 +107,7 @@ 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,
+            ["/usr/sbin/sysctl", "-n", name], stderr=subprocess.DEVNULL, text=True, timeout=2,
         ).strip()
     except Exception:
         return ""
@@ -140,10 +133,10 @@ 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, CPU-only
-    dev boxes). Try the requested config first; on a CUDA library load failure
-    fall back to CPU + int8. Pass ``stt.local.device`` / ``compute_type`` to pin.
+    ``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:
@@ -172,23 +165,18 @@ def _load_local_whisper_model(model_name: str, device: str = "auto", compute_typ
         return WhisperModel(model_name, device="cpu", compute_type="int8")
 
 
-# Silence-hallucination hardening for local faster-whisper (whisper decodes
-# junk like "You"/"Thank you." from pure silence). Three layers, all tunable
-# under ``stt.local``: Silero VAD so silence never reaches the model
-# (``vad: false`` restores raw behaviour for music/ambient audio);
-# condition_on_previous_text=False so one hallucinated token can't seed a run;
-# and the segment confidence gate in _is_hallucinated_segment.
+# Silence-hallucination hardening for local faster-whisper (whisper decodes junk like
+# "You"/"Thank you." from pure silence). Three layers, all tunable under ``stt.local``:
+# Silero VAD so silence never reaches the model (``vad: false`` restores raw behaviour
+# for music/ambient audio); condition_on_previous_text=False so one hallucinated token
+# can't seed a run; and the segment confidence gate in _is_hallucinated_segment.
 _VAD_MIN_SILENCE_MS_DEFAULT = 500
 _NO_SPEECH_PROB_THRESHOLD_DEFAULT = 0.6
 _LOGPROB_THRESHOLD_DEFAULT = -1.0
 
 
 def build_local_transcribe_kwargs(stt_config: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
-    """Build the kwargs for EVERY local faster-whisper ``model.transcribe`` call.
-
-    Single owner for the anti-hallucination hardening — new local-whisper call
-    sites must go through here instead of hand-rolling kwargs.
-    """
+    """Kwargs for EVERY local faster-whisper ``model.transcribe`` call — single owner of the anti-hallucination hardening."""
     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 {}
@@ -205,11 +193,10 @@ def build_local_transcribe_kwargs(stt_config: Optional[Dict[str, Any]] = None) -
             "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 without this 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; defaults unchanged.
+    # 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)
@@ -234,9 +221,8 @@ def _confidence_thresholds(local_cfg: Dict[str, Any]) -> tuple[float, float]:
 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): the model must BOTH
-    think the window is non-speech AND have decoded it with low confidence, so
-    quiet-but-real speech survives. Unknown segment shapes are never dropped.
+    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"))
@@ -254,8 +240,7 @@ def _join_confident_segments(segments: Any, local_cfg: Dict[str, Any]) -> str:
         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, "text", ""), getattr(segment, "no_speech_prob", float("nan")),
                 getattr(segment, "avg_logprob", float("nan")),
             )
             continue
@@ -264,11 +249,7 @@ def _join_confident_segments(segments: Any, local_cfg: Dict[str, Any]) -> str:
 
 
 def _transcribe_local_command(
-    file_path: str,
-    model_name: str,
-    *,
-    language: Optional[str] = None,
-    prompt: Optional[str] = None,
+    file_path: str, model_name: str, *, language: Optional[str] = None, prompt: Optional[str] = None
 ) -> Dict[str, Any]:
     """Run the configured local STT command template and read back a .txt transcript."""
     from tools.transcription_tools import _prepare_local_audio, _resolve_stt_language
@@ -290,10 +271,8 @@ def _transcribe_local_command(
                 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),
+                input_path=shlex.quote(prepared_input), output_dir=shlex.quote(output_dir),
+                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
diff --git a/tools/transcription_tools.py b/tools/transcription_tools.py
index 94f585bea2..a93d28bdc1 100644
--- a/tools/transcription_tools.py
+++ b/tools/transcription_tools.py
@@ -1,16 +1,13 @@
 #!/usr/bin/env python3
 """Speech-to-text transcription used by the gateway for voice messages.
 
-Built-in providers: local (faster-whisper, default/free), local_command, groq,
-openai (also serves the managed ``nous`` selection), mistral, xai, elevenlabs,
-deepinfra; plus user-declared command providers and plugin providers.
-
-    result = transcribe_audio("/path/to/audio.ogg")   # {"success", "transcript", "error"?, "provider"?}
-
-This module owns provider resolution, the dispatcher, and the cached local
-model + idle-unload state. Backends live in sibling modules
-(``transcription_{common,audio,local,cloud,command}``) and are re-imported
-here so ``tools.transcription_tools.<name>`` stays the patch/import surface.
+Built-in providers: local (faster-whisper, default/free), local_command, groq, openai
+(also serves the managed ``nous`` selection), mistral, xai, elevenlabs, deepinfra; plus
+user-declared command providers and plugin providers. ``transcribe_audio(path)`` returns
+``{"success", "transcript", "error"?, "provider"?}``. This module owns provider resolution,
+the dispatcher and the cached local model + idle-unload state; backends live in
+``transcription_{common,audio,local,cloud,command}`` and are re-imported here so
+``tools.transcription_tools.<name>`` stays the patch surface.
 """
 
 import logging
@@ -69,11 +66,7 @@ logger = logging.getLogger(__name__)
 
 
 def get_env_value(name, default=None):
-    """Read env values through the live config module.
-
-    Resolved at call time: tests monkeypatch/restore ``hermes_cli.config.get_env_value``
-    around this module's import, so a cached import would go stale.
-    """
+    """Read env values through the live config module (resolved per call: tests monkeypatch it around import)."""
     try:
         from hermes_cli.config import get_env_value as _get_env_value
     except ImportError:
@@ -83,9 +76,9 @@ def get_env_value(name, default=None):
 
 
 def _resolve_provider_key(env_var: str, provider_id: str) -> str:
-    """Resolve an STT API key via the shared voice-key resolver (config > env/.env > credential pool).
+    """STT API key via the shared voice-key resolver (config > env/.env > credential pool).
 
-    Resolved at call time so tests that reload the helpers module see the live function.
+    Resolved per call so tests that reload the helpers module see the live function.
     """
     try:
         from tools.tool_backend_helpers import resolve_provider_secret
@@ -107,18 +100,14 @@ _HAS_MISTRAL = _safe_find_spec("mistralai")
 _HAS_PILK = _safe_find_spec("pilk")
 
 
-# Singleton for the local model — loaded once, reused across calls. The lock
-# guards the check-then-load so two concurrent voice messages can't both
-# download/load the model.
+# Local model singleton; the lock guards check-then-load against concurrent voice messages.
 _local_model: Optional[object] = None
 _local_model_name: Optional[str] = None
 _local_model_lock = threading.Lock()
 
-# Idle unload: a single daemon thread checks _last_transcription_time and
-# releases the model (hundreds of MB of RAM/VRAM) after a configurable idle
-# period, then exits; the next voice message reloads and restarts it.
-# _idle_unload_mgmt_lock serializes the start check so two concurrent
-# transcriptions can't both observe "no watcher alive" and spawn duplicates.
+# Idle unload: one daemon thread releases the model (hundreds of MB of RAM/VRAM) after a
+# configurable idle period, then exits; the next voice message reloads and restarts it.
+# _idle_unload_mgmt_lock serializes the start check so no duplicate watchers spawn.
 _last_transcription_time: float = 0.0
 _idle_unload_thread: Optional[threading.Thread] = None
 _idle_unload_stop = threading.Event()
@@ -147,16 +136,12 @@ def is_stt_enabled(stt_config: Optional[dict] = None) -> bool:
 
 
 def _resolve_stt_language(
-    provider_key: str,
-    stt_config: Optional[Dict[str, Any]] = None,
-    *,
-    extra_keys: tuple = (),
+    provider_key: str, stt_config: Optional[Dict[str, Any]] = None, *, extra_keys: tuple = ()
 ) -> Optional[str]:
-    """Resolve the language hint for an STT provider; first non-empty wins.
+    """Language hint for an STT provider, first non-empty wins (never "").
 
-    Order: ``stt.<provider>.language`` (plus *extra_keys* aliases, e.g. ElevenLabs'
-    ``language_code``) > ``stt.language`` > ``HERMES_LOCAL_STT_LANGUAGE`` env >
-    None (provider auto-detects). Never returns "".
+    ``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()
@@ -192,6 +177,15 @@ 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))
+    return probe
+
+
 def _has_xai_stt_credentials_quietly() -> bool:
     try:
         return _has_xai_stt_credentials()
@@ -199,30 +193,11 @@ def _has_xai_stt_credentials_quietly() -> bool:
         return False
 
 
-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:
-            return False
-        if needs_mistral and not _HAS_MISTRAL:
-            return False
-        return bool(_resolve_provider_key(env_var, provider))
-    return probe
-
-
-_has_groq_key = _has_key("GROQ_API_KEY", "groq", needs_openai=True)
-_has_mistral_key = _has_key("MISTRAL_API_KEY", "mistral", needs_mistral=True)
-_has_elevenlabs_key = _has_key("ELEVENLABS_API_KEY", "elevenlabs")
-_has_deepinfra_key = _has_key("DEEPINFRA_API_KEY", "deepinfra", needs_openai=True)
-
-
 def _resolve_explicit_openai() -> str:
     if not _HAS_OPENAI:
         logger.warning("STT provider 'openai' configured but no API key available")
         return "none"
-    # Resolve directly rather than via the boolean probe so a managed
-    # openai-audio gateway outage is logged with its real reason, not a
-    # generic "no API key" hint.
+    # Resolved directly so a managed openai-audio gateway outage is logged with its real reason.
     reason = _openai_audio_unavailable_reason()
     if reason is None:
         return "openai"
@@ -262,14 +237,18 @@ def _resolve_explicit_local_command() -> str:
     return "none"
 
 
+_has_groq_key = _has_key("GROQ_API_KEY", "groq", needs_openai=True)
+_has_mistral_key = _has_key("MISTRAL_API_KEY", "mistral", needs_mistral=True)
+_has_elevenlabs_key = _has_key("ELEVENLABS_API_KEY", "elevenlabs")
+_has_deepinfra_key = _has_key("DEEPINFRA_API_KEY", "deepinfra", needs_openai=True)
+
 # Cloud providers in AUTO-DETECT priority order:
 #   name -> (explicit-selection probe, auto-detect probe, explicit warning, auto-detect log)
-# The two probes differ only for openai (auto-detect additionally requires the
-# SDK) and xai (auto-detect must never raise). DeepInfra is LAST so a
-# DEEPINFRA_API_KEY set for the chat surface never displaces an existing
-# xAI/ElevenLabs auto-selection. Mistral only auto-selects when the SDK is
-# already present — no lazy-install during passive auto-detection (explicit
-# ``provider: mistral`` installs on first use).
+# The probes differ only for openai (explicit has its own resolver in _EXPLICIT_RESOLVERS;
+# auto-detect also requires the SDK) and xai (auto-detect must never raise). DeepInfra is
+# LAST so a DEEPINFRA_API_KEY set for chat never displaces an xAI/ElevenLabs auto-selection.
+# Mistral only auto-selects when the SDK is present — no lazy-install during passive
+# auto-detection (explicit ``provider: mistral`` installs on first use).
 _CLOUD_PROVIDER_SPECS = {
     "groq": (
         _has_groq_key, _has_groq_key,
@@ -278,13 +257,12 @@ _CLOUD_PROVIDER_SPECS = {
     ),
     "openai": (
         None, lambda: _HAS_OPENAI and _has_openai_audio_backend(),
-        None,  # explicit openai has its own resolver (logs the real gateway reason)
+        None,
         "No local STT available, using OpenAI Whisper API",
     ),
     "mistral": (
         _has_mistral_key, _has_mistral_key,
-        "STT provider 'mistral' configured but mistralai package "
-        "not installed or MISTRAL_API_KEY not set",
+        "STT provider 'mistral' configured but mistralai package not installed or MISTRAL_API_KEY not set",
         "No local STT available, using Mistral Voxtral Transcribe API",
     ),
     "xai": (
@@ -299,25 +277,25 @@ _CLOUD_PROVIDER_SPECS = {
     ),
     "deepinfra": (
         _has_deepinfra_key, _has_deepinfra_key,
-        "STT provider 'deepinfra' configured but DEEPINFRA_API_KEY not set "
-        "(or openai package missing)",
+        "STT provider 'deepinfra' configured but DEEPINFRA_API_KEY not set (or openai package missing)",
         "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,
+}
+
 
 def _resolve_explicit_provider(provider: str) -> str:
-    """Resolve an explicit ``stt.provider`` to a usable provider name or ``"none"``.
-
-    Unknown names pass through untouched so the dispatcher can fail with the
-    provider-not-registered message.
-    """
-    if provider == "local":
-        return _resolve_explicit_local()
-    if provider == "local_command":
-        return _resolve_explicit_local_command()
-    if provider == "openai":
-        return _resolve_explicit_openai()
+    """Explicit ``stt.provider`` -> usable name or ``"none"``; unknown names pass through untouched
+    so the dispatcher fails with the provider-not-registered message."""
+    resolver = _EXPLICIT_RESOLVERS.get(provider)
+    if resolver is not None:
+        return resolver()
     spec = _CLOUD_PROVIDER_SPECS.get(provider)
     if spec is None:
         return provider
@@ -329,29 +307,21 @@ def _resolve_explicit_provider(provider: str) -> str:
 
 
 def _get_provider(stt_config: dict) -> str:
-    """Determine which STT provider to use.
-
-    An explicit ``stt.provider`` is honoured — no silent cloud fallback. With no
-    provider configured, auto-detect tries local > groq > openai > mistral > xai
-    > elevenlabs > deepinfra.
-    """
+    """Which STT provider to use: an explicit ``stt.provider`` is honoured (no silent cloud
+    fallback); otherwise auto-detect local > groq > openai > mistral > xai > elevenlabs > deepinfra."""
     if not is_stt_enabled(stt_config):
         return "none"
 
     explicit = "provider" in stt_config
     provider = stt_config.get("provider", DEFAULT_PROVIDER)
 
-    # The managed "Nous Subscription" selection is serviced by the OpenAI
-    # implementation, routed through the managed gateway by
-    # _resolve_openai_audio_client_config.
+    # The managed "Nous Subscription" selection is the OpenAI backend routed via the managed gateway.
     if isinstance(provider, str) and provider.strip().lower() == "nous":
         provider = "openai"
 
     if explicit and provider == "local":
-        # Legacy DEFAULT_CONFIG seeded ``stt.provider: local`` on every install,
-        # so a merged-config "local" is not proof of a user pick. Only a raw
-        # config.yaml selection counts as explicit; otherwise autodetect (which
-        # prefers local first anyway).
+        # Legacy DEFAULT_CONFIG seeded ``stt.provider: local`` on every install, so only a
+        # raw config.yaml selection counts as explicit; otherwise autodetect (local-first anyway).
         try:
             from tools.tool_backend_helpers import read_selection
 
@@ -381,10 +351,7 @@ def _unload_local_model() -> None:
     global _local_model, _local_model_name
     with _local_model_lock:
         if _local_model is not None:
-            logger.info(
-                "Unloading local whisper model '%s' after idle timeout",
-                _local_model_name or "unknown",
-            )
+            logger.info("Unloading local whisper model '%s' after idle timeout", _local_model_name or "unknown")
             _local_model = None
             _local_model_name = None
 
@@ -392,12 +359,10 @@ def _unload_local_model() -> None:
 def _start_idle_unload_watcher(timeout_seconds: int) -> None:
     """Ensure the single idle-unload watcher thread is running.
 
-    Started only when none is alive (one lock + one ``is_alive()`` per
-    transcription). 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. After unloading, when the timeout becomes 0, or when
-    the model is already gone, the thread exits; the next transcription restarts it.
+    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.
     """
     global _idle_unload_thread
     with _idle_unload_mgmt_lock:
@@ -412,9 +377,7 @@ def _start_idle_unload_watcher(timeout_seconds: int) -> None:
                 if _local_model is None:
                     break
                 try:
-                    timeout = _get_idle_unload_seconds(
-                        _load_stt_config().get("local") or {}
-                    )
+                    timeout = _get_idle_unload_seconds(_load_stt_config().get("local") or {})
                 except Exception:  # noqa: BLE001 - keep the seed value
                     timeout = initial_timeout
                 if timeout <= 0:
@@ -424,9 +387,7 @@ def _start_idle_unload_watcher(timeout_seconds: int) -> None:
                     break
 
         _idle_unload_stop.clear()
-        _idle_unload_thread = threading.Thread(
-            target=_watch, name="hermes-stt-idle-unload", daemon=True
-        )
+        _idle_unload_thread = threading.Thread(target=_watch, name="hermes-stt-idle-unload", daemon=True)
         _idle_unload_thread.start()
 
 
@@ -437,11 +398,9 @@ def _touch_transcription_time() -> None:
 
 
 def _get_or_load_local_model(model_name: str, local_cfg: Dict[str, Any]):
-    """Return the cached faster-whisper model, (re)loading under the lock when needed.
+    """Cached faster-whisper model, (re)loaded under a double-checked lock when needed.
 
-    Double-checked lock: concurrent voice messages must not both download/load.
-    The returned strong reference stays valid even if the idle watcher nulls the
-    module global mid-transcription.
+    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
@@ -449,11 +408,9 @@ def _get_or_load_local_model(model_name: str, local_cfg: Dict[str, Any]):
         with _local_model_lock:
             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 let users pin a configuration
-                # where ``auto`` mis-detects; the loader keeps the CUDA→CPU fallback.
+                # 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"),
+                    model_name, device=local_cfg.get("device", "auto"),
                     compute_type=local_cfg.get("compute_type", "auto"),
                 )
                 _local_model_name = model_name
@@ -473,11 +430,7 @@ def _replace_cached_model_on_cpu(model_name: str):
 
 
 def _transcribe_local(
-    file_path: str,
-    model_name: str,
-    *,
-    language: Optional[str] = None,
-    prompt: Optional[str] = None,
+    file_path: str, model_name: str, *, language: Optional[str] = None, prompt: Optional[str] = None
 ) -> Dict[str, Any]:
     """Transcribe using faster-whisper (local, free)."""
     if not _HAS_FASTER_WHISPER and not _try_lazy_install_stt():
@@ -486,8 +439,7 @@ def _transcribe_local(
     try:
         stt_config = _load_stt_config()
         local_cfg = stt_config.get("local") or {}
-        # Reset the idle timer BEFORE loading/transcribing so the watcher can't
-        # count a long in-flight transcription as idle time and unload mid-use.
+        # Reset the idle timer BEFORE loading so a long in-flight transcription isn't counted as idle.
         _touch_transcription_time()
         model = _get_or_load_local_model(model_name, local_cfg)
         if model is None:  # defensive: load failed without raising
@@ -503,9 +455,8 @@ def _transcribe_local(
         try:
             segments, info = model.transcribe(file_path, **transcribe_kwargs)
         except Exception as exc:
-            # CUDA libs sometimes only fail at dlopen-on-first-use, AFTER the
-            # model loaded. Evict the poisoned cached model, reload on CPU and
-            # retry once — otherwise every later voice message fails until restart.
+            # CUDA libs can fail at dlopen-on-first-use, AFTER loading: evict the poisoned
+            # cached model, reload on CPU and retry once, else every later message fails.
             if not _looks_like_cuda_lib_error(exc):
                 raise
             logger.warning(
@@ -538,8 +489,7 @@ def _transcribe_local(
 
 
 def _read_block_error(file_path: str) -> Optional[Dict[str, Any]]:
-    """Refuse to feed a credential / secret store (auth.json, .env, OAuth tokens, ...)
-    to an STT provider, which would ship its plaintext to a third-party API.
+    """Refuse to ship a credential store (auth.json, .env, OAuth tokens) to an STT provider.
     Mirrors the image-gen / video-gen read guards."""
     from agent.file_safety import get_read_block_error
     blocked = get_read_block_error(file_path)
@@ -547,23 +497,19 @@ 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,
+    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/provider default; ``source`` is a caller-surface
-    label (``"gateway"``, ``"voice_mode"``) forwarded to the ``pre_transcription``
-    hook for observability only. Returns the standard result envelope.
+    ``model`` overrides the config default; ``source`` is a caller-surface label
+    (``"gateway"``, ``"voice_mode"``) forwarded to the ``pre_transcription`` hook only.
     """
     blocked = _read_block_error(file_path)
     if blocked:
         return blocked
 
-    # Validate before provider resolution so invalid files cannot trigger
-    # provider setup or lazy installation. The remote-upload size cap is
-    # enforced below, only for non-local providers.
+    # 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 = _validate_audio_file(file_path, enforce_size_limit=False)
     if error:
         return error
@@ -599,9 +545,8 @@ def _transcribe_prepared_audio(
             shutil.rmtree(trim_cleanup_dir, ignore_errors=True)
 
 
-# Built-in provider -> (stt section, config key, default, treat-empty-as-missing).
-# "local_command" shares the ``stt.local`` section; xAI takes no model parameter
-# (the name is logging-only); deepinfra resolves from the live catalog when empty.
+# Built-in provider -> (stt section, config key, default, treat-empty-as-missing). "local_command"
+# shares ``stt.local``; xAI takes no model (logging-only); deepinfra uses the live catalog when empty.
 _BUILTIN_MODEL_KEYS = {
     "local": ("local", "model", DEFAULT_LOCAL_MODEL, False),
     "local_command": ("local", "model", DEFAULT_LOCAL_MODEL, False),
@@ -626,29 +571,17 @@ def _builtin_model_name(provider: str, stt_config: Dict[str, Any], model: Option
     return cfg.get(key, default)
 
 
-def _builtin_handler(provider: str):
-    """Handler for a built-in provider, looked up in this module at call time so tests may patch ``_transcribe_*``."""
-    if provider not in BUILTIN_STT_PROVIDERS:
-        return None
-    return globals()[f"_transcribe_{provider}"]
-
-
 def _dispatch_stt_provider(
-    file_path: str,
-    provider: str,
-    stt_config: Dict[str, Any],
-    model: Optional[str] = None,
+    file_path: str, provider: str, stt_config: Dict[str, Any], model: Optional[str] = None,
     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; pre_transcription hook results mutate
-    # on top in registration order (last hook to set a field wins).
+    # 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
 
-    # The hook fires after provider resolution and BEFORE any backend is
-    # invoked; ``language`` stays None unless a hook overrides it.
+    # 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,
         language=_get_stt_section(stt_config, provider).get("language"),
@@ -656,16 +589,16 @@ def _dispatch_stt_provider(
     )
     prompt = _enforce_prompt_length_limit(prompt, provider)
 
-    handler = _builtin_handler(provider)
-    if handler is not None:
+    if provider in BUILTIN_STT_PROVIDERS:
+        # Looked up in this module at call time so tests may patch ``_transcribe_*``.
+        handler = globals()[f"_transcribe_{provider}"]
         model_name = _builtin_model_name(provider, stt_config, model)
         if provider in ("local", "local_command"):
             model_name = _normalize_local_model(model_name)
         return handler(file_path, model_name, language=language, prompt=prompt)
 
-    # User-declared command provider: after built-ins (so ``stt.providers.openai
-    # .command`` can't override the real handler) and BEFORE plugins, because
-    # config is more local than a plugin install (same precedence as TTS).
+    # Command providers: after built-ins (``stt.providers.openai.command`` can't override the
+    # real handler) and BEFORE plugins, since config is more local than a plugin install.
     command_provider_config = _resolve_command_stt_provider_config(provider, stt_config)
     if command_provider_config is not None:
         return _transcribe_command_stt(
@@ -673,14 +606,11 @@ def _dispatch_stt_provider(
             model_override=model, language_override=language, prompt=prompt,
         )
 
-    # Plugin-registered backend. Plugins read per-provider config under
-    # ``stt.<provider>`` like built-ins; the ``model`` argument overrides it.
+    # Plugin backend: reads ``stt.<provider>`` like built-ins; the ``model`` argument overrides it.
     plugin_cfg = _get_stt_section(stt_config, provider)
     plugin_result = _dispatch_to_plugin_provider(
-        file_path, provider, stt_config,
-        model=model or plugin_cfg.get("model"),
-        language=language or _resolve_stt_language(provider, stt_config),
-        prompt=prompt,
+        file_path, provider, stt_config, model=model or plugin_cfg.get("model"),
+        language=language or _resolve_stt_language(provider, stt_config), prompt=prompt,
     )
     if plugin_result is not None:
         return plugin_result
@@ -693,9 +623,7 @@ def _no_provider_error(provider: str, stt_config: Dict[str, Any]) -> Dict[str, A
     if "provider" in stt_config and provider_key and provider_key not in BUILTIN_STT_PROVIDERS and provider_key != "none":
         return _unregistered_stt_provider_error(provider_key)
 
-    # An explicit openai selection flattened to "none" carries a
-    # selection-specific reason (e.g. managed openai-audio gateway down);
-    # surface it with its remediation instead of the all-provider hint.
+    # An explicit openai selection flattened to "none" has a specific reason (e.g. managed gateway down).
     if provider_key == "none" and str(stt_config.get("provider") or "") == "openai" and _HAS_OPENAI:
         reason = _openai_audio_unavailable_reason()
         if reason is not None:
@@ -712,24 +640,20 @@ def _no_provider_error(provider: str, stt_config: Dict[str, Any]) -> Dict[str, A
 
 
 def transcribe_audio(
-    file_path: str,
-    model: Optional[str] = None,
-    source: Optional[str] = None,
+    file_path: str, model: Optional[str] = None, source: Optional[str] = None
 ) -> Dict[str, Any]:
-    """Safely validate, preprocess supported inputs, and dispatch transcription.
+    """Validate, preprocess supported inputs, and dispatch transcription.
 
-    ``source`` is an optional caller-surface label (``"gateway"``, ``"voice_mode"``)
-    forwarded to the ``pre_transcription`` hook for observability only.
+    ``source`` is a caller-surface label (``"gateway"``, ``"voice_mode"``) forwarded to
+    the ``pre_transcription`` hook for observability only.
     """
-    # Secret-store refusal runs before ANY validation so the error names the
-    # real reason rather than a format error.
+    # Secret-store refusal runs before ANY validation so the error names the real reason.
     blocked = _read_block_error(file_path)
     if blocked:
         return blocked
 
-    # Cap .silk sources before the decoder runs (decoder safety); for all other
-    # inputs the upload cap is provider-scoped in _transcribe_prepared_audio,
-    # so local whisper can handle big files.
+    # Cap .silk sources before the decoder runs; for other inputs the upload cap is
+    # provider-scoped in _transcribe_prepared_audio so local whisper can take big files.
     is_silk = Path(file_path).suffix.lower() == ".silk"
     source_error = _validate_audio_source_file(file_path, enforce_size_limit=is_silk)
     if source_error:
@@ -750,13 +674,12 @@ def transcribe_audio(
 
 
 def transcribe_audio_local_fallback(
-    file_path: str,
-    model: Optional[str] = None,
+    file_path: str, model: Optional[str] = None
 ) -> Dict[str, Any]:
     """Try an already-installed local STT backend without changing config.
 
-    For passive inbound-media recovery after the configured provider failed:
-    never lazy-installs or falls through to a cloud provider.
+    Passive inbound-media recovery after the configured provider failed: never
+    lazy-installs or falls through to a cloud provider.
     """
     error = _validate_audio_file(file_path)
     if error:
@@ -770,5 +693,3 @@ def transcribe_audio_local_fallback(
     if _has_local_command():
         return _transcribe_local_command(file_path, _normalize_local_model(local_model))
     return _error_result("No installed local STT backend is available.", provider="local")
-
-
diff --git a/tools/tts_command_provider.py b/tools/tts_command_provider.py
index c0ddf24650..852d1c48b1 100644
--- a/tools/tts_command_provider.py
+++ b/tools/tts_command_provider.py
@@ -1,25 +1,20 @@
 """Shared runner for user-configured shell ("command") TTS/STT providers.
 
-Both ``tools.tts_tool`` and ``tools.transcription_tools`` let users declare a
-provider as a shell command template with ``{placeholders}``. This module owns
-the shell-quote-aware template rendering and the idle-timeout process runner
-they share, plus the TTS side's ``tts.providers.<name>`` config layer. Each
-origin module re-imports these under its historical private names.
-
-TTS config shape::
+``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 -m ~/model.onnx -f {output_path} < {input_path}"
-          output_format: wav
+        piper-en: {type: command, command: "piper -f {output_path} < {input_path}", output_format: wav}
 
-Placeholders: ``{input_path}``, ``{text_path}`` (alias), ``{output_path}``,
-``{format}``, ``{voice}``, ``{model}``, ``{speed}``; ``{{``/``}}`` for literal
-braces. Values are shell-quoted for their surrounding quote context. Built-in
-provider names always win over a same-named entry under ``tts.providers``.
+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``.
 """
 
 from __future__ import annotations
@@ -33,7 +28,7 @@ import tempfile
 import threading
 import time
 from pathlib import Path
-from typing import Any, Dict, Optional
+from typing import Any, Dict, FrozenSet, Optional
 
 
 def shell_quote_context(command_template: str, position: int) -> Optional[str]:
@@ -68,43 +63,26 @@ def quote_command_placeholder(value: str, quote_context: Optional[str]) -> str:
     if quote_context == "'":
         return value.replace("'", r"'\''")
     if quote_context == '"':
-        return (
-            value
-            .replace("\\", "\\\\")
-            .replace('"', r'\"')
-            .replace("$", r"\$")
-            .replace("`", r"\`")
-        )
+        return value.replace("\\", "\\\\").replace('"', r'\"').replace("$", r"\$").replace("`", r"\`")
     if os.name == "nt":
         return subprocess.list2cmdline([value])
     return shlex.quote(value)
 
 
-def render_command_template(
-    command_template: str,
-    placeholders: Dict[str, str],
-) -> str:
+def render_command_template(command_template: str, placeholders: Dict[str, str]) -> str:
     """Replace ``{name}`` placeholders (quote-aware) while preserving ``{{``/``}}``."""
     names = "|".join(re.escape(name) for name in placeholders)
-    pattern = re.compile(
-        rf"(?<!\$)(?:\{{\{{(?P<double>{names})\}}\}}|\{{(?P<single>{names})\}})"
-    )
+    pattern = re.compile(rf"(?<!\$)(?:\{{\{{(?P<double>{names})\}}\}}|\{{(?P<single>{names})\}})")
     replacements: list[tuple[str, str]] = []
 
     def replace_match(match: re.Match[str]) -> str:
         name = match.group("double") or match.group("single")
         token = f"__HERMES_CMD_PLACEHOLDER_{len(replacements)}__"
-        replacements.append((
-            token,
-            quote_command_placeholder(
-                placeholders[name],
-                shell_quote_context(command_template, match.start()),
-            ),
-        ))
+        quoted = quote_command_placeholder(placeholders[name], shell_quote_context(command_template, match.start()))
+        replacements.append((token, quoted))
         return token
 
-    rendered = pattern.sub(replace_match, command_template)
-    rendered = rendered.replace("{{", "{").replace("}}", "}")
+    rendered = pattern.sub(replace_match, command_template).replace("{{", "{").replace("}}", "}")
     for token, value in replacements:
         rendered = rendered.replace(token, value)
     return rendered
@@ -130,20 +108,15 @@ def terminate_command_process_tree(proc: subprocess.Popen) -> None:
     """Best-effort termination of a shell process and all of its children."""
     if proc.poll() is not 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,
+                stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, timeout=5, stdin=subprocess.DEVNULL,
             )
         except Exception:
             proc.kill()
         return
-
     try:
         import psutil  # type: ignore
     except ImportError:
@@ -153,40 +126,35 @@ def terminate_command_process_tree(proc: subprocess.Popen) -> None:
         except subprocess.TimeoutExpired:
             proc.kill()
         return
-
     _signal_process_tree(psutil, proc, "terminate")
     try:
         proc.wait(timeout=2)
-        return
     except subprocess.TimeoutExpired:
-        pass
-    _signal_process_tree(psutil, proc, "kill")
+        _signal_process_tree(psutil, proc, "kill")
 
 
 def command_env_passthrough(config: Dict[str, Any]) -> list:
-    """Return the provider's ``env_passthrough`` allowlist.
-
-    The child env is scrubbed of Hermes secrets by default; this list names
-    variables copied back from the parent env so a trusted template (e.g. a
-    curl one-liner using its own API key) keeps working.
-    """
+    """``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()]
 
 
+def command_failure_detail(exc: subprocess.CalledProcessError) -> str:
+    """``stderr: ...; stdout: ...`` for a failed command provider, or ``no command output``."""
+    parts = [f"{stream}: {text.strip()}" for stream, text in (("stderr", exc.stderr), ("stdout", exc.stdout)) if text]
+    return "; ".join(parts) or "no command output"
+
+
 def run_command_provider(
-    command: str,
-    timeout: float,
-    env_passthrough: Optional[list] = None,
+    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
@@ -197,15 +165,9 @@ def run_command_provider(
         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 on non-UTF-8 Windows.
-        "encoding": "utf-8",
-        "errors": "replace",
-        "env": delegated_child_subprocess_env(scrubbed),
+        "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)
@@ -222,10 +184,7 @@ def run_command_provider(
         read1 = getattr(getattr(stream, "buffer", None), "read1", None)
         try:
             while True:
-                if read1 is None:
-                    chunk = stream.read(65536)
-                else:
-                    chunk = read1(65536).decode(encoding, errors="replace")
+                chunk = stream.read(65536) if read1 is None else read1(65536).decode(encoding, errors="replace")
                 if not chunk:
                     break
                 output_queue.put((name, chunk))
@@ -273,63 +232,42 @@ def run_command_provider(
                 break
             if chunk:
                 chunks[name].append(chunk)
-        stdout = "".join(chunks["stdout"])
-        stderr = "".join(chunks["stderr"])
+        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 = "".join(chunks["stdout"])
-    stderr = "".join(chunks["stderr"])
+            raise subprocess.TimeoutExpired(command, timeout, output=stdout, stderr=stderr) from exc
 
+    stdout, stderr = "".join(chunks["stdout"]), "".join(chunks["stderr"])
     if proc.returncode:
-        raise subprocess.CalledProcessError(
-            proc.returncode, command, output=stdout, stderr=stderr,
-        )
+        raise subprocess.CalledProcessError(proc.returncode, command, output=stdout, stderr=stderr)
     return subprocess.CompletedProcess(command, proc.returncode, stdout, stderr)
 
 
 # ===========================================================================
-# TTS ``tts.providers.<name>`` config layer
+# Generic ``<section>.providers.<name>`` config layer (TTS and STT share it)
 # ===========================================================================
 
-# 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",
-})
 
-DEFAULT_COMMAND_TTS_TIMEOUT_SECONDS = 120
-DEFAULT_COMMAND_TTS_OUTPUT_FORMAT = "mp3"
-COMMAND_TTS_OUTPUT_FORMATS = frozenset(
-    {"mp3", "wav", "ogg", "flac", "m4a", "aac", "amr", "opus"}
-)
-DEFAULT_COMMAND_TTS_MAX_TEXT_LENGTH = 5000
-
-
-def _get_provider_section(tts_config: Dict[str, Any], name: str) -> Dict[str, Any]:
-    """Return a provider config block if it's a dict, else an empty dict."""
-    if not isinstance(tts_config, dict):
+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 = tts_config.get(name)
+    section = config.get(name)
     return section if isinstance(section, dict) else {}
 
 
-def _get_named_provider_config(tts_config: Dict[str, Any], name: str) -> Dict[str, Any]:
-    """Config dict for a user-declared provider, or {}.
+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.
 
-    ``tts.providers.<name>`` is canonical; ``tts.<name>`` is accepted as
-    back-compat only for non-built-in names (so a user's ``tts.openai`` block
-    still means the OpenAI provider, not a custom command).
+    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 = _get_provider_section(tts_config, "providers").get(name)
+    section = _get_provider_section(config, "providers").get(name)
     if isinstance(section, dict):
         return section
-    if name.lower() not in BUILTIN_TTS_PROVIDERS:
-        return _get_provider_section(tts_config, name)
+    if name.lower() not in builtins:
+        return _get_provider_section(config, name)
     return {}
 
 
@@ -344,59 +282,76 @@ def _is_command_provider_config(config: Dict[str, Any]) -> bool:
     return isinstance(command, str) and bool(command.strip())
 
 
-def _resolve_command_provider_config(
-    provider: str,
-    tts_config: Dict[str, Any],
+def _resolve_command_config(
+    provider: str, config: Dict[str, Any], reserved: FrozenSet[str],
 ) -> Optional[Dict[str, Any]]:
-    """The provider config when *provider* is a user-declared command provider.
-
-    None for built-in names (native handlers win), unknown names, or
-    non-command types.
-    """
+    """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 BUILTIN_TTS_PROVIDERS:
+    if key in reserved:
         return None
-    config = _get_named_provider_config(tts_config, key)
-    return config if _is_command_provider_config(config) else None
+    named = _named_provider_config(config, key, reserved)
+    return named if _is_command_provider_config(named) else None
+
+
+def _command_timeout(config: Dict[str, Any], default: float) -> float:
+    """Timeout in seconds (``timeout`` > ``timeout_seconds``); invalid or non-positive -> *default*."""
+    raw = config.get("timeout", config.get("timeout_seconds", default))
+    try:
+        value = float(raw)
+    except (TypeError, ValueError):
+        return float(default)
+    return value if value > 0 else float(default)
+
+
+def _command_output_format(config: Dict[str, Any], formats: FrozenSet[str], default: str) -> str:
+    """Validated ``format``/``output_format`` from *config*, else *default*."""
+    raw = config.get("format") or config.get("output_format") or default
+    fmt = str(raw).lower().strip().lstrip(".")
+    return fmt if fmt in formats else default
+
+
+# ---- TTS ``tts.providers.<name>`` layer -----------------------------------
+
+# 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",
+})
+
+DEFAULT_COMMAND_TTS_TIMEOUT_SECONDS = 120
+DEFAULT_COMMAND_TTS_OUTPUT_FORMAT = "mp3"
+COMMAND_TTS_OUTPUT_FORMATS = frozenset({"mp3", "wav", "ogg", "flac", "m4a", "aac", "amr", "opus"})
+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)
 
 
 def _iter_command_providers(tts_config: Dict[str, Any]):
     """Yield (name, config) pairs for every declared command-type provider."""
     for name, cfg in _get_provider_section(tts_config, "providers").items():
-        if (
-            isinstance(name, str)
-            and name.lower() not in BUILTIN_TTS_PROVIDERS
-            and _is_command_provider_config(cfg)
-        ):
+        if isinstance(name, str) and name.lower() not in BUILTIN_TTS_PROVIDERS and _is_command_provider_config(cfg):
             yield name, cfg
 
 
 def _get_command_tts_timeout(config: Dict[str, Any]) -> float:
-    """Timeout in seconds; invalid or non-positive values fall back to the default."""
-    raw = config.get("timeout", config.get("timeout_seconds", DEFAULT_COMMAND_TTS_TIMEOUT_SECONDS))
-    try:
-        value = float(raw)
-    except (TypeError, ValueError):
-        return float(DEFAULT_COMMAND_TTS_TIMEOUT_SECONDS)
-    if value <= 0:
-        return float(DEFAULT_COMMAND_TTS_TIMEOUT_SECONDS)
-    return value
+    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:
+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
-    raw = config.get("format") or config.get("output_format") or DEFAULT_COMMAND_TTS_OUTPUT_FORMAT
-    fmt = str(raw).lower().strip().lstrip(".")
-    return fmt if fmt in COMMAND_TTS_OUTPUT_FORMATS else DEFAULT_COMMAND_TTS_OUTPUT_FORMAT
+    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:
@@ -413,23 +368,16 @@ def _configured_command_tts_output_path(path: Path, config: Dict[str, Any]) -> P
 
 
 def _generate_command_tts(
-    text: str,
-    output_path: str,
-    provider_name: str,
-    config: Dict[str, Any],
-    tts_config: Dict[str, Any],
+    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.
+    """Generate speech by running a user-configured shell command; returns the audio path it wrote.
 
-    Returns the absolute path of the audio file the command wrote. Raises
-    ``ValueError`` for invalid provider config and ``RuntimeError`` for
+    Raises ``ValueError`` for invalid provider config and ``RuntimeError`` for
     timeouts / non-zero exits / empty output.
     """
     command_template = str(config.get("command") or "").strip()
     if not command_template:
-        raise ValueError(
-            f"tts.providers.{provider_name}.command is not configured"
-        )
+        raise ValueError(f"tts.providers.{provider_name}.command is not configured")
 
     output = Path(output_path).expanduser()
     output.parent.mkdir(parents=True, exist_ok=True)
@@ -443,46 +391,24 @@ def _generate_command_tts(
     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),
+            "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),
         }
         command = render_command_template(command_template, placeholders)
-
         try:
-            # Resolved through the origin so tests patching
-            # ``tools.tts_tool._run_command_tts`` still intercept.
+            # 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),
-            )
+            _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
+            raise RuntimeError(f"TTS provider '{provider_name}' timed out after {timeout:g}s") from exc
         except subprocess.CalledProcessError as exc:
-            detail_parts = []
-            if exc.stderr:
-                detail_parts.append(f"stderr: {exc.stderr.strip()}")
-            if exc.stdout:
-                detail_parts.append(f"stdout: {exc.stdout.strip()}")
-            detail = "; ".join(detail_parts) or "no command output"
             raise RuntimeError(
-                f"TTS provider '{provider_name}' exited with code "
-                f"{exc.returncode}: {detail}"
+                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}"
-        )
+        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 f80ac11025..823e48aec1 100644
--- a/tools/tts_streaming.py
+++ b/tools/tts_streaming.py
@@ -1,16 +1,11 @@
-"""Provider-agnostic streaming TTS: sentence text → int16 PCM chunk iterator.
+"""Provider-agnostic streaming TTS: sentence text → int16 mono PCM chunk iterator.
 
-``stream_tts_to_speaker`` (``tools.tts_tool``) owns the sentence buffer,
-sounddevice output and stop/queue protocol; this module owns the *provider*
-half — turning one sentence into audio the moment it's ready so playback starts
-on sentence one instead of after the whole reply.
-
-One contract (int16 mono PCM at ``sample_rate``): **true streamers**
-(`StreamingTTSProvider.stream`) wrap chunked APIs (ElevenLabs pcm_24000, OpenAI
-pcm, …); providers with no chunked API (edge, the default) still get per-
-*sentence* playback via the sync ``text_to_speech_tool`` path in the dispatcher.
-Adding a streamer is ``@register("name")`` on a subclass; the dispatcher, config
-gate (``tts.<name>.streaming``) and resolver come free.
+``stream_tts_to_speaker`` (``tools.tts_tool``) owns the sentence buffer, sounddevice
+output and stop/queue protocol; this module owns the *provider* half so playback
+starts on sentence one. True streamers (``StreamingTTSProvider.stream``) wrap chunked
+APIs; providers with no chunked API (edge, the default) get per-sentence playback via
+the sync ``text_to_speech_tool`` path. Adding a streamer is ``@register("name")`` on
+a subclass; the dispatcher, config gate (``tts.<name>.streaming``) and resolver come free.
 """
 
 from __future__ import annotations
@@ -26,16 +21,14 @@ from tools.tts_tool import _get_provider, _load_tts_config, get_env_value
 
 logger = logging.getLogger(__name__)
 
-# Per-sentence PCM byte cap, mirroring the 16 MiB bounded-body invariant of the
-# sync providers: a buggy or hostile endpoint must not feed unbounded audio.
+# Per-sentence PCM byte cap, mirroring the sync providers' 16 MiB bounded-body invariant.
 _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).
+    """Provider secret lookup (config > env/.env > credential pool); seam over ``tts_tool._resolve_provider_key``.
 
-    Monkeypatchable seam over ``tools.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
@@ -49,17 +42,11 @@ def _gemini_key() -> str:
     return _resolve_key("GEMINI_API_KEY", "gemini") or _resolve_key("GOOGLE_API_KEY", "gemini")
 
 
-# ---------------------------------------------------------------------------
-# Interruption latch — lets the model know it was cut off mid-speech
-# ---------------------------------------------------------------------------
-# When the user barges in on a spoken reply, the surface marks the latch; 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
+# 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.
-
-SPEECH_INTERRUPTED_NOTE = (
-    "[Note: the user interrupted your previous spoken reply before it finished.]"
-)
+SPEECH_INTERRUPTED_NOTE = "[Note: the user interrupted your previous spoken reply before it finished.]"
 _INTERRUPT_TTL_S = 120.0
 _interrupted_at: Optional[float] = None
 
@@ -83,10 +70,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, so "Ha!" rides along instead of stalling as a tiny clip.
+    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):
@@ -117,10 +103,6 @@ class SentenceChunker:
         return [tail] if tail else []
 
 
-# ---------------------------------------------------------------------------
-# ABC + registry
-# ---------------------------------------------------------------------------
-
 class StreamingTTSProvider(ABC):
     """Yields raw int16, little-endian, mono PCM chunks at ``sample_rate``."""
 
@@ -149,7 +131,6 @@ def register(name: str) -> Callable[[type[StreamingTTSProvider]], type[Streaming
     def _wrap(cls: type[StreamingTTSProvider]) -> type[StreamingTTSProvider]:
         _REGISTRY[name] = cls
         return cls
-
     return _wrap
 
 
@@ -165,39 +146,30 @@ def _try_instantiate(name: str, tts_config: Dict) -> Optional[StreamingTTSProvid
         return None
 
 
-# Fallback priority for ``tts.streaming.provider: auto`` — best chunked
-# latency/quality first. Deliberately hard-coded (a UX decision, not a config
-# knob); edge is absent because it has no chunked-PCM API.
+# Fallback priority for ``tts.streaming.provider: auto`` — best chunked latency/quality
+# first. Deliberately hard-coded (a UX decision); edge is absent (no chunked-PCM API).
 _PROVIDER_PRIORITY: List[str] = ["elevenlabs", "gemini", "openai", "xai"]
 
 
 def resolve_streaming_provider(
-    tts_config: Dict,
-    preferred: Optional[str] = None,
+    tts_config: Dict, preferred: Optional[str] = None
 ) -> Optional[StreamingTTSProvider]:
     """Return a ready streamer for the *configured* provider, else ``None``.
 
-    1. ``tts.streaming.provider`` when set: a name pins that exact streamer
-       (or ``None`` if unusable); ``auto`` walks ``_PROVIDER_PRIORITY`` and
-       returns the first usable one.
-    2. 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.
     """
-    streaming_cfg = tts_config.get("streaming") or {}
-    pinned = str(streaming_cfg.get("provider") or "").lower().strip()
+    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
-    if pinned:
-        return _try_instantiate(pinned, tts_config)
-
-    name = (preferred or _get_provider(tts_config)).lower().strip()
-    return _try_instantiate(name, tts_config)
+    return _try_instantiate(pinned or (preferred or _get_provider(tts_config)).lower().strip(), tts_config)
 
 
 def _capped(chunks: Iterator[bytes], label: str) -> Iterator[bytes]:
@@ -206,16 +178,11 @@ def _capped(chunks: Iterator[bytes], label: str) -> Iterator[bytes]:
     for chunk in chunks:
         total += len(chunk)
         if total > _STREAM_SENTENCE_BYTE_CAP:
-            logger.warning("%s exceeded %d bytes for one sentence; truncating",
-                           label, _STREAM_SENTENCE_BYTE_CAP)
+            logger.warning("%s exceeded %d bytes for one sentence; truncating", label, _STREAM_SENTENCE_BYTE_CAP)
             return
         yield chunk
 
 
-# ---------------------------------------------------------------------------
-# Providers
-# ---------------------------------------------------------------------------
-
 @register("elevenlabs")
 class ElevenLabsStreamer(StreamingTTSProvider):
     """ElevenLabs chunked HTTP → pcm_24000 (the original reference path)."""
@@ -229,25 +196,18 @@ class ElevenLabsStreamer(StreamingTTSProvider):
     def stream(self, text: str) -> Iterator[bytes]:
         from tools.tts_tool import _import_elevenlabs
         from tools.tts_tool_providers import (
-            DEFAULT_ELEVENLABS_STREAMING_MODEL_ID,
-            DEFAULT_ELEVENLABS_VOICE_ID,
-            _elevenlabs_environment_kwargs,
+            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),
+            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),
+            "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=voice_id, model_id=model_id, output_format="pcm_24000",
         )
 
 
@@ -275,30 +235,18 @@ class OpenAIStreamer(StreamingTTSProvider):
 
         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),
         )
-        model = self.section.get("model", "gpt-4o-mini-tts")
-        voice = self.section.get("voice", "alloy")
         with client.audio.speech.with_streaming_response.create(
-            model=model,
-            voice=voice,
-            input=text,
-            response_format="pcm",
+            model=self.section.get("model", "gpt-4o-mini-tts"), voice=self.section.get("voice", "alloy"),
+            input=text, response_format="pcm",
         ) as response:
             yield from _capped(response.iter_bytes(), "OpenAI streaming TTS")
 
 
 @register("gemini")
 class GeminiStreamer(StreamingTTSProvider):
-    """Gemini ``streamGenerateContent?alt=sse`` → base64 PCM chunks (24 kHz).
-
-    ``?alt=sse`` flips the response from one JSON blob to an SSE feed of
-    base64 PCM chunks. Uses requests with a bounded streamed body.
-    """
+    """Gemini ``streamGenerateContent?alt=sse`` → SSE feed of base64 PCM chunks (24 kHz), bounded streamed body."""
 
     sample_rate = 24000
 
@@ -313,40 +261,27 @@ class GeminiStreamer(StreamingTTSProvider):
         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
         base_url = str(
-            self.section.get("base_url")
-            or get_env_value("GEMINI_BASE_URL")
-            or DEFAULT_GEMINI_TTS_BASE_URL
+            self.section.get("base_url") or get_env_value("GEMINI_BASE_URL") or DEFAULT_GEMINI_TTS_BASE_URL
         ).strip().rstrip("/")
-
         payload = {
             "contents": [{"parts": [{"text": text}]}],
             "generationConfig": {
                 "responseModalities": ["AUDIO"],
-                "speechConfig": {
-                    "voiceConfig": {
-                        "prebuiltVoiceConfig": {"voiceName": voice},
-                    },
-                },
+                "speechConfig": {"voiceConfig": {"prebuiltVoiceConfig": {"voiceName": voice}}},
             },
         }
         url = f"{base_url}/models/{model}:streamGenerateContent"
 
         def _sse_chunks() -> Iterator[bytes]:
             with requests.post(
-                url,
-                params={"alt": "sse", "key": api_key},
-                json=payload,
-                timeout=60,
-                stream=True,
+                url, params={"alt": "sse", "key": api_key}, json=payload, timeout=60, stream=True,
             ) as response:
                 response.raise_for_status()
                 for line in response.iter_lines(decode_unicode=True):
@@ -374,9 +309,9 @@ class GeminiStreamer(StreamingTTSProvider):
 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. The async WS loop is bridged to the
-    sync iterator contract via ``_collect_async`` — the seam unit tests patch.
+    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
@@ -394,8 +329,6 @@ class XAIStreamer(StreamingTTSProvider):
     def stream(self, text: str) -> Iterator[bytes]:
         yield from _capped(iter(self._collect_async(text)), "xAI streaming TTS")
 
-    # -- async→sync bridge (test seam) ------------------------------------
-
     def _collect_async(self, text: str) -> List[bytes]:
         import asyncio
 
@@ -420,18 +353,10 @@ class XAIStreamer(StreamingTTSProvider):
         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()
+        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",
-            }))
+        async with websockets.connect(ws_url, extra_headers={"Authorization": f"Bearer {api_key}"}) as ws:
+            await ws.send(_json.dumps({"text": text, "voice_id": voice, "response_format": "pcm"}))
             try:
                 while True:
                     message = await ws.recv()
@@ -448,8 +373,9 @@ class XAIStreamer(StreamingTTSProvider):
                     if etype == "done":
                         return
                     if etype == "error":
-                        logger.warning("xAI WS error envelope: %s",
-                                       envelope.get("error") or envelope.get("message") or envelope)
+                        logger.warning(
+                            "xAI WS error envelope: %s", envelope.get("error") or envelope.get("message") or envelope,
+                        )
                         return
             except Exception as exc:
                 if exc.__class__.__name__ == "ConnectionClosed":
diff --git a/tools/tts_text_normalize.py b/tools/tts_text_normalize.py
index f96e79022c..45213cd9ce 100644
--- a/tools/tts_text_normalize.py
+++ b/tools/tts_text_normalize.py
@@ -1,11 +1,8 @@
-"""Utilities for preparing assistant text for speech synthesis.
+"""Deterministic cleanup turning assistant Markdown into a spoken script.
 
-The TTS provider should receive a spoken script, not raw chat Markdown.  This
-module centralises the lightweight, deterministic cleanup used by explicit TTS
-calls and gateway auto-TTS replies.
-
-Non-ASCII characters are written as escapes on purpose so the file stays free of
-invisible/look-alike glyphs.
+Shared by explicit TTS calls, gateway auto-TTS, voice-mode streaming and the web
+dashboard. Non-ASCII characters are written as escapes on purpose so the file
+stays free of invisible/look-alike glyphs.
 """
 
 from __future__ import annotations
@@ -13,9 +10,9 @@ from __future__ import annotations
 import html
 import re
 
-# Sentinel appended to former heading lines so smooth_whitespace_for_tts can
-# fold a heading into the sentence that follows it ("Weather, it will be sunny")
-# rather than leaving a bare "Weather." label that reads abruptly aloud.
+# Sentinel appended to former heading lines so smooth_whitespace_for_tts folds the
+# heading into the sentence after it ("Weather, it will be sunny") instead of a bare
+# "Weather." label.
 _HEAD = "\x00"
 
 _MD_CODE_BLOCK_RE = re.compile(r"```[\s\S]*?```")
@@ -34,21 +31,22 @@ _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+")
 
-# Broad emoji / pictograph cleanup.  Voice providers vary a lot here; most read
-# emojis as awkward labels, so keep the speech script calm and literal.
+# 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"),
+)
+# 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),
+    (r"US\$", "US dollars", re.IGNORECASE), ("€", "euros", 0), ("£", "pounds", 0), (r"\$", "dollars", 0),
+)
+
+# Broad emoji / pictograph cleanup: most voice providers read emojis as awkward labels.
 _EMOJI_RE = re.compile(
-    "["
-    "\U0001F1E6-\U0001F1FF"
-    "\U0001F300-\U0001F5FF"
-    "\U0001F600-\U0001F64F"
-    "\U0001F680-\U0001F6FF"
-    "\U0001F700-\U0001F77F"
-    "\U0001F780-\U0001F7FF"
-    "\U0001F800-\U0001F8FF"
-    "\U0001F900-\U0001F9FF"
-    "\U0001FA00-\U0001FAFF"
-    "☀-➿"
-    "]+",
+    "[\U0001F1E6-\U0001F1FF\U0001F300-\U0001F5FF\U0001F600-\U0001F64F\U0001F680-\U0001F6FF"
+    "\U0001F700-\U0001F77F\U0001F780-\U0001F7FF\U0001F800-\U0001F8FF\U0001F900-\U0001F9FF"
+    "\U0001FA00-\U0001FAFF☀-➿]+",
     flags=re.UNICODE,
 )
 _VARIATION_SELECTOR_RE = re.compile("[︎️]")
@@ -70,34 +68,29 @@ def strip_markdown_for_tts(text: str) -> str:
     text = _MD_ITALIC_RE.sub(r"\1", text)
     text = _MD_UNDERSCORE_ITALIC_RE.sub(r"\1", text)
     text = _MD_STRIKE_RE.sub(r"\1", text)
-    # Mark headings (do not just delete the marker): the whitespace pass folds a
-    # heading into the sentence after it so speech says "Weather, it will be
-    # sunny" instead of a clipped "Weather." then a separate sentence.
+    # Mark headings (do not just delete the marker): see _HEAD.
     text = _MD_HEADING_LINE_RE.sub(lambda m: m.group(1).rstrip() + _HEAD, text)
     text = _MD_BLOCKQUOTE_RE.sub("", text)
     text = _MD_LIST_ITEM_RE.sub("", text)
     text = _MD_HR_RE.sub("", text)
 
-    # Pipe tables are terrible read aloud.  Turn any leftover pipes into pauses
-    # instead of letting a provider speak "vertical bar".
+    # Leftover table pipes become pauses instead of a spoken "vertical bar".
     text = _MD_TABLE_PIPE_RE.sub("; ", text)
     return text
 
 
 def _normalize_temperature_ranges(text: str) -> str:
-    # 11-17 degrees C -> "11 to 17 degrees Celsius" (en/em dash or hyphen).
-    text = re.sub(
-        r"(?<!\w)([-+\u2212]?\d+(?:\.\d+)?)\s*[\u2013\u2014-]\s*([-+\u2212]?\d+(?:\.\d+)?)\s*°\s*C\b",
-        lambda m: f"{m.group(1).replace(chr(0x2212), '-')} to {m.group(2).replace(chr(0x2212), '-')} degrees Celsius",
-        text,
-        flags=re.IGNORECASE,
-    )
-    text = re.sub(
-        r"(?<!\w)([-+\u2212]?\d+(?:\.\d+)?)\s*[\u2013\u2014-]\s*([-+\u2212]?\d+(?:\.\d+)?)\s*°\s*F\b",
-        lambda m: f"{m.group(1).replace(chr(0x2212), '-')} to {m.group(2).replace(chr(0x2212), '-')} degrees Fahrenheit",
-        text,
-        flags=re.IGNORECASE,
-    )
+    """``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")):
+        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,
+        )
     return text
 
 
@@ -112,44 +105,37 @@ def normalize_symbols_for_tts(text: str) -> str:
     text = text.replace("…", "...")  # ellipsis
     text = _normalize_temperature_ranges(text)
 
-    # Temperatures with a number.  Do this before generic degree handling.
-    text = re.sub(r"(?<!\w)([-+]?\d+(?:\.\d+)?)\s*°\s*C\b", r"\1 degrees Celsius", text, flags=re.IGNORECASE)
-    text = re.sub(r"(?<!\w)([-+]?\d+(?:\.\d+)?)\s*°\s*F\b", r"\1 degrees Fahrenheit", text, flags=re.IGNORECASE)
-    # Bare units with no leading number ("measured in degrees C").
-    text = re.sub(r"°\s*C\b", "degrees Celsius", text, flags=re.IGNORECASE)
-    text = re.sub(r"°\s*F\b", "degrees Fahrenheit", text, flags=re.IGNORECASE)
-    # Any remaining degree symbol (angles, stray cases).
+    # 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")):
+        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")):
+        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")
 
     # Common weather/travel units.
-    text = re.sub(r"(?<=\d)\s*km\s*/\s*h\b", " kilometres per hour", text, flags=re.IGNORECASE)
-    text = re.sub(r"(?<=\d)\s*km/h\b", " kilometres per hour", text, flags=re.IGNORECASE)
-    text = re.sub(r"(?<=\d)\s*mm\b", " millimetres", text, flags=re.IGNORECASE)
-    text = re.sub(r"(?<=\d)\s*cm\b", " centimetres", text, flags=re.IGNORECASE)
-    text = re.sub(r"(?<=\d)\s*m\b", " metres", text, flags=re.IGNORECASE)
+    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.
-    text = re.sub(r"NZ\$\s*([\d,]*\d(?:\.\d+)?)", r"\1 New Zealand dollars", text, flags=re.IGNORECASE)
-    text = re.sub(r"A\$\s*([\d,]*\d(?:\.\d+)?)", r"\1 Australian dollars", text, flags=re.IGNORECASE)
-    text = re.sub(r"US\$\s*([\d,]*\d(?:\.\d+)?)", r"\1 US dollars", text, flags=re.IGNORECASE)
-    text = re.sub(r"€\s*([\d,]*\d(?:\.\d+)?)", r"\1 euros", text)
-    text = re.sub(r"£\s*([\d,]*\d(?:\.\d+)?)", r"\1 pounds", text)
-    text = re.sub(r"\$\s*([\d,]*\d(?:\.\d+)?)", r"\1 dollars", 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 = text.replace("→", " to ")  # ->
-    text = text.replace("⇒", " to ")  # =>
-    text = text.replace("≈", " about ")  # almost equal
-    text = text.replace("~", " about ")
+    for symbol, word in (("→", " to "), ("⇒", " to "), ("≈", " about "), ("~", " about ")):
+        text = text.replace(symbol, word)
 
     text = _VARIATION_SELECTOR_RE.sub("", text)
     text = _EMOJI_RE.sub("", text)
@@ -159,10 +145,8 @@ def normalize_symbols_for_tts(text: str) -> str:
 def smooth_whitespace_for_tts(text: str) -> str:
     """Collapse visual formatting into calm spoken paragraphs.
 
-    A former heading line (marked with the _HEAD sentinel) folds into the next
-    content line as a spoken lead-in: "Weather" + "It will be sunny" becomes
-    "Weather, It will be sunny."  A heading with no content after it becomes its
-    own short sentence.
+    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 ""
@@ -182,8 +166,7 @@ def smooth_whitespace_for_tts(text: str) -> str:
         is_heading = raw_line.rstrip().endswith(_HEAD)
         line = raw_line.replace(_HEAD, "").strip()
         if not line:
-            # Hold a pending heading across blank lines so it still folds into
-            # the next real content line; otherwise just collapse the blank.
+            # Hold a pending heading across blank lines so it still folds into the next content line.
             if pending_heading is None and lines and lines[-1] != "":
                 lines.append("")
             continue
@@ -209,44 +192,30 @@ def smooth_whitespace_for_tts(text: str) -> str:
     return text.strip()
 
 
-# Reasoning blocks: models with ``/reasoning show`` enabled emit
-# ``<think>...</think>`` blocks in the final assistant message.  Users want to
-# SEE reasoning, not hear it read aloud (#34213).
+# ``/reasoning show`` emits ``<think>...</think>`` in the final message: users want to
+# SEE reasoning, not hear it. An unterminated block (streaming cut-off) is also silenced.
 _THINK_BLOCK_RE = re.compile(r"<think[\s>].*?</think>", flags=re.DOTALL | re.IGNORECASE)
-# An unterminated block (streaming cut-off) should still not be spoken.
 _THINK_BLOCK_OPEN_RE = re.compile(r"<think[\s>].*\Z", flags=re.DOTALL | re.IGNORECASE)
 
-# Turn-end file-mutation verifier footer appended by run_agent.py
-# (``_format_file_mutation_failure_footer``).  It's a UI affordance — reading
-# "warning file mutation verifier, 2 files were NOT modified..." aloud is
-# noise (#40772).  The footer is a ``⚠️ File-mutation verifier:`` header line
-# followed by indented ``•`` bullet lines; strip the whole block.
-_VERIFIER_FOOTER_RE = re.compile(
-    r"^\s*⚠️?\s*File-mutation verifier:.*(?:\n[ \t]+•.*)*",
-    flags=re.MULTILINE,
-)
+# run_agent.py's turn-end file-mutation verifier footer (a ``⚠️ File-mutation verifier:``
+# header line plus indented ``•`` bullets) is a UI affordance, not speech.
+_VERIFIER_FOOTER_RE = re.compile(r"^\s*⚠️?\s*File-mutation verifier:.*(?:\n[ \t]+•.*)*", flags=re.MULTILINE)
 
 
 def strip_nonspoken_blocks(text: str) -> str:
-    """Remove blocks that must never reach a speech provider.
-
-    Currently: ``<think>`` reasoning blocks and the end-of-turn
-    file-mutation verifier footer.
-    """
+    """Remove ``<think>`` reasoning blocks and the file-mutation verifier footer."""
     if not text:
         return ""
-    text = _THINK_BLOCK_RE.sub(" ", text)
-    text = _THINK_BLOCK_OPEN_RE.sub(" ", text)
-    text = _VERIFIER_FOOTER_RE.sub(" ", text)
+    for pattern in (_THINK_BLOCK_RE, _THINK_BLOCK_OPEN_RE, _VERIFIER_FOOTER_RE):
+        text = pattern.sub(" ", text)
     return text
 
 
 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 synthesis at the
-    first newline (#9004).  The smoothing pass already terminates each line
-    with punctuation, so newlines can safely become plain spaces.
+    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.
     """
     if not text:
         return ""
@@ -259,28 +228,20 @@ def flatten_newlines_for_payload(text: str) -> str:
 
 
 def prepare_spoken_text(text: str, max_chars: int | None = 4000) -> str:
-    """Return a TTS-friendly script from assistant text.
+    """Return a TTS-friendly script from assistant text (deterministic cleanup, not a rewrite).
 
-    Deterministic cleanup, not a semantic rewrite: it removes ``<think>``
-    reasoning blocks and the file-mutation verifier footer, removes Markdown,
-    expands common symbols such as a degree-Celsius sign to "degrees Celsius",
-    turns visual line formatting into speakable sentence pauses, and flattens
-    the result to a single line so newline-sensitive providers (Kokoro) speak
-    the whole script.
+    Pipeline: non-spoken blocks > Markdown > symbols/units > line formatting into
+    sentence pauses > single line (for newline-sensitive providers), then ``max_chars``.
     """
-    spoken = strip_nonspoken_blocks(text)
-    spoken = strip_markdown_for_tts(spoken)
-    spoken = normalize_symbols_for_tts(spoken)
-    spoken = smooth_whitespace_for_tts(spoken)
-    spoken = flatten_newlines_for_payload(spoken)
+    spoken = text
+    for step in (strip_nonspoken_blocks, strip_markdown_for_tts, normalize_symbols_for_tts,
+                 smooth_whitespace_for_tts, flatten_newlines_for_payload):
+        spoken = step(spoken)
     if max_chars is not None and max_chars > 0 and len(spoken) > max_chars:
         spoken = spoken[:max_chars].rstrip()
     return spoken
 
 
-# ===========================================================================
-# Speech text cleanup (shared by voice-mode streaming and gateway auto-TTS)
-# ===========================================================================
 # Legacy regex fallback, only used if the shared normalizer raises.
 _LEGACY_TTS_STRIP_STEPS = (
     (re.compile(r'<think[\s>].*?</think>', flags=re.DOTALL), ' '),
@@ -300,14 +261,7 @@ _LEGACY_TTS_STRIP_STEPS = (
 
 
 def _strip_markdown_for_tts(text: str) -> str:
-    """Prepare text for speech via the shared cleaner in tts_text_normalize.
-
-    One cleaner for every TTS path (tool, gateway auto-TTS, voice-mode
-    streaming, web dashboard): strips <think> blocks, the verifier footer,
-    markdown and emoji; expands units; flattens newlines so newline-sensitive
-    providers (Kokoro) speak the whole script. Falls back to the legacy regex
-    pipeline if the normalizer ever fails.
-    """
+    """``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: