From 3969d10e40a43bddfdcc224c56d514a648d2b64a Mon Sep 17 00:00:00 2001
From: Teknium <127238744+teknium1@users.noreply.github.com>
Date: Wed, 2 Sep 2026 22:17:23 -0700
Subject: [PATCH] refactor(tools): compact
 wake_word/write_approval/voice_mode_transcript/working_diff (-12% LOC,
 behavior-neutral)

---
 tools/voice_mode_transcript.py |  95 ++++-----
 tools/wake_word.py             | 353 +++++++++++----------------------
 tools/wake_word_engines.py     | 130 +++++-------
 tools/working_diff.py          |  53 ++---
 tools/write_approval.py        | 172 ++++++----------
 5 files changed, 279 insertions(+), 524 deletions(-)

diff --git a/tools/voice_mode_transcript.py b/tools/voice_mode_transcript.py
index b0e22b8dc3..b5706f2a9a 100644
--- a/tools/voice_mode_transcript.py
+++ b/tools/voice_mode_transcript.py
@@ -7,8 +7,8 @@ from typing import Optional
 
 
 def _voice_config() -> dict:
-    """``voice`` section of config.yaml, or ``{}`` when missing, malformed,
-    or the config system can't be imported (broken config mid-install)."""
+    """``voice`` section of config.yaml, or ``{}`` when missing, malformed, or the
+    config system can't be imported (broken config mid-install)."""
     try:
         from hermes_cli.config import load_config
         voice_cfg = load_config().get("voice", {})
@@ -28,12 +28,9 @@ WHISPER_HALLUCINATIONS = {
     "amara.org", "www.mooji.org", "ご視聴ありがとうございました",
 }
 
-
 # Repetitive hallucinations (e.g. "Thank you. Thank you. Thank you.")
-_HALLUCINATION_REPEAT_RE = re.compile(
-    r'^(?:thank you|thanks|bye|you|ok|okay|the end|\.|\s|,|!)+$',
-    flags=re.IGNORECASE,
-)
+_HALLUCINATION_REPEAT_RE = re.compile(r'^(?:thank you|thanks|bye|you|ok|okay|the end|\.|\s|,|!)+$',
+                                      flags=re.IGNORECASE)
 
 
 def is_whisper_hallucination(transcript: str) -> bool:
@@ -41,28 +38,23 @@ def is_whisper_hallucination(transcript: str) -> bool:
     cleaned = transcript.strip().lower()
     if not cleaned:
         return True
-    return (
-        cleaned.rstrip('.!') in WHISPER_HALLUCINATIONS
-        or bool(_HALLUCINATION_REPEAT_RE.match(cleaned))
-    )
+    return cleaned.rstrip('.!') in WHISPER_HALLUCINATIONS or bool(_HALLUCINATION_REPEAT_RE.match(cleaned))
 
 
 DEFAULT_VOICE_STOP_PHRASES = ("stop",)
 
 
 def _load_voice_stop_phrases() -> tuple:
-    """Configured ``voice.stop_phrases`` (default ``("stop",)``); an empty tuple
-    disables the feature. Malformed config (dict, list of non-strings) falls
-    back to the default rather than crashing the voice loop."""
+    """Configured ``voice.stop_phrases`` (default ``("stop",)``); an empty tuple disables
+    the feature. Malformed config (dict, list of non-strings) falls back to the default
+    rather than crashing the voice loop."""
     try:
         raw = _voice_config().get("stop_phrases", DEFAULT_VOICE_STOP_PHRASES)
         if isinstance(raw, str):
             raw = [raw]
         if isinstance(raw, (list, tuple)):
-            return tuple(
-                str(p).strip().lower() for p in raw
-                if isinstance(p, (str, int, float)) and str(p).strip()
-            )
+            return tuple(str(p).strip().lower() for p in raw
+                         if isinstance(p, (str, int, float)) and str(p).strip())
     except Exception:
         pass
     return DEFAULT_VOICE_STOP_PHRASES
@@ -78,31 +70,24 @@ def _configured_stop_phrases() -> tuple:
 def is_voice_stop_phrase(transcript: str, stop_phrases: Optional[tuple] = None) -> bool:
     """True when *transcript* is EXACTLY a configured stop phrase.
 
-    Deliberately strict: the whole utterance — lowercased, surrounding
-    punctuation stripped — must equal a phrase, so "stop doing that and try
-    again" still reaches the agent. ``voice.stop_phrases: []`` disables.
+    Deliberately strict: the whole utterance — lowercased, surrounding punctuation
+    stripped — must equal a phrase, so "stop doing that and try again" still reaches
+    the agent. ``voice.stop_phrases: []`` disables.
     """
-    if not transcript:
-        return False
-    cleaned = transcript.strip().lower().strip(".,!?;: \t\n\"'")
+    cleaned = transcript.strip().lower().strip(".,!?;: \t\n\"'") if transcript else ""
     if not cleaned:
         return False
-    if stop_phrases is None:
-        stop_phrases = _configured_stop_phrases()
-    return cleaned in stop_phrases
+    return cleaned in (_configured_stop_phrases() if stop_phrases is None else stop_phrases)
 
 
-# Similarity ratio (difflib.SequenceMatcher) above which a playback-phase barge
-# transcript is treated as a self-capture of Hermes' own TTS: the full-duplex
-# listener has no echo cancellation, so speaker bleed can trip the barge
-# trigger and get transcribed near-verbatim (a TTS -> STT -> TTS loop).
+# Similarity ratio (difflib.SequenceMatcher) above which a playback-phase barge transcript
+# is treated as a self-capture of Hermes' own TTS: the full-duplex listener has no echo
+# cancellation, so speaker bleed can be transcribed near-verbatim (TTS -> STT -> TTS loop).
 DEFAULT_TTS_ECHO_SIMILARITY_THRESHOLD = 0.6
 
-
-# Minimum normalized-transcript length before the sliding-window fallback
-# runs. Below this a genuine one-word barge-in ("yes") landing verbatim inside
-# a longer reply would score a trivial 1.0 and be misread as self-capture; a
-# real self-capture spans pre-roll plus time-to-silence, so it is longer.
+# Minimum normalized-transcript length before the sliding-window fallback runs. Below
+# this a genuine one-word barge-in ("yes") landing verbatim inside a longer reply would
+# score a trivial 1.0; a real self-capture spans pre-roll plus time-to-silence, so it is longer.
 MIN_FRAGMENT_LENGTH_FOR_ECHO = 10
 
 
@@ -110,25 +95,18 @@ def _normalize_for_echo_compare(text: str) -> str:
     return re.sub(r"\s+", " ", text).strip().lower()
 
 
-def is_tts_echo(
-    transcript: str,
-    spoken_text: str,
-    threshold: float = DEFAULT_TTS_ECHO_SIMILARITY_THRESHOLD,
-) -> bool:
+def is_tts_echo(transcript: str, spoken_text: str,
+                threshold: float = DEFAULT_TTS_ECHO_SIMILARITY_THRESHOLD) -> bool:
     """True when *transcript* looks like a self-capture of *spoken_text*.
 
-    Character-level similarity (language-agnostic, no word tokenization): a
-    genuine user interjection is very unlikely to closely match Hermes' own
-    words, so a high ratio signals speaker-bleed self-capture (fail-closed
-    guard for the playback-phase listener, which has no echo cancellation).
-
-    The playback-phase capture is cut when the trigger fires and only spans
-    pre-roll plus time-to-silence, so for replies longer than a clause the
-    transcript is a short FRAGMENT of `spoken_text` and the whole-string
-    ratio dilutes toward 0. When it misses, a window sized to the transcript
-    slides across `spoken_text` (character-based, so it works without word
-    boundaries). Transcripts shorter than `MIN_FRAGMENT_LENGTH_FOR_ECHO` skip
-    this fallback: a short interjection trivially matches a short window.
+    Character-level similarity (language-agnostic, no word tokenization): a genuine
+    user interjection is very unlikely to closely match Hermes' own words, so a high
+    ratio signals speaker-bleed (fail-closed guard for the playback-phase listener).
+    The playback-phase capture only spans pre-roll plus time-to-silence, so for long
+    replies the transcript is a short FRAGMENT and the whole-string ratio dilutes toward
+    0; when it misses, a transcript-sized window slides across `spoken_text`. Transcripts
+    shorter than `MIN_FRAGMENT_LENGTH_FOR_ECHO` skip the fallback (a short interjection
+    trivially matches a short window).
     """
     if not transcript or not spoken_text:
         return False
@@ -150,12 +128,9 @@ def is_tts_echo(
 def voice_stop_hint() -> str:
     """One-line 'Say "stop" to end the voice chat.' hint for voice-mode start.
 
-    Uses the first ``voice.stop_phrases`` entry so a custom phrase renders
-    correctly; returns "" when stop phrases are disabled so surfaces show no
-    hint. Every surface announcing voice-mode start (CLI, TUI, desktop) uses
-    this one owner instead of hardcoding the wording.
+    Uses the first ``voice.stop_phrases`` entry so a custom phrase renders correctly;
+    "" when stop phrases are disabled. Every surface announcing voice-mode start (CLI,
+    TUI, desktop) uses this one owner instead of hardcoding the wording.
     """
     phrases = _configured_stop_phrases()
-    if not phrases:
-        return ""
-    return f'Say "{phrases[0]}" to end the voice chat.'
+    return f'Say "{phrases[0]}" to end the voice chat.' if phrases else ""
diff --git a/tools/wake_word.py b/tools/wake_word.py
index ac59bab889..10f1e4a44e 100644
--- a/tools/wake_word.py
+++ b/tools/wake_word.py
@@ -1,17 +1,11 @@
 """Wake-word ("Hey Hermes") detection — hands-free session trigger.
 
-An always-on hotword listener shared by CLI, TUI and desktop GUI (one owns it,
-gated by ``wake_surface_enabled``): on wake Hermes opens a fresh session and
-captures voice via the existing pipeline. Engines (openwakeword default,
-sherpa open-vocabulary, porcupine premium) are all on-device and live in
-:mod:`tools.wake_word_engines`; this module owns config, the capture loop and
-the process-wide listener singleton.
-
+One always-on hotword listener shared by CLI, TUI and desktop GUI (a single owner,
+gated by ``wake_surface_enabled``). Engines live in :mod:`tools.wake_word_engines`;
+this module owns config, the capture loop and the process-wide listener singleton.
 Capture reuses voice mode's 16 kHz mono int16 ``sounddevice`` path on a daemon
-thread; callers ``pause()`` while a voice turn holds the mic and ``resume()``
-once idle (two input streams on one device is unreliable cross-platform).
-Nothing here touches agent context or the prompt cache — on wake the caller
-gets a plain string, like a transcript.
+thread; callers ``pause()`` while a voice turn holds the mic and ``resume()`` once
+idle (two input streams on one device is unreliable cross-platform).
 """
 
 from __future__ import annotations
@@ -32,22 +26,20 @@ from tools.wake_word_engines import (  # noqa: F401  (re-exported for callers/te
 
 logger = logging.getLogger(__name__)
 
-# 16 kHz mono int16 — Whisper-native and what every engine expects.
-SAMPLE_RATE = 16000
+SAMPLE_RATE = 16000  # 16 kHz mono int16 — Whisper-native and what every engine expects.
 
-# Minimum gap between two wake fires, so one "hey hermes" can't retrigger
-# across several frames while the caller is still reacting.
+# Minimum gap between two wake fires, so one "hey hermes" can't retrigger across
+# several frames while the caller is still reacting.
 _FIRE_COOLDOWN_SECONDS = 2.0
 _START_TIMEOUT_SECONDS = 5.0
 
-# Ambient-speech rejection: require N consecutive over-threshold frames before
-# firing (a stray phoneme spikes one frame; a real phrase holds several).
+# Ambient-speech rejection: N consecutive over-threshold frames before firing
+# (a stray phoneme spikes one frame; a real phrase holds several).
 _DEFAULT_CONFIRMATION_FRAMES = 3
 
-# Dead-mic detection: an int16 stream whose peak stays at/below _SILENCE_PEAK
-# for this many consecutive seconds is flagged silent (desktop push-to-talk and
-# the backend listener use different capture paths, so one can work while the
-# backend-selected stream is all zeros).
+# Dead-mic detection: an int16 stream whose peak stays at/below _SILENCE_PEAK for
+# this many consecutive seconds is flagged silent (desktop push-to-talk and the
+# backend listener capture differently, so one can work while the other is all zeros).
 _SILENCE_PEAK = 10
 _SILENCE_ALERT_SECONDS = 10
 
@@ -70,9 +62,8 @@ _DEFAULTS: Dict[str, Any] = {
     "enabled": False,
     "surface": "auto",
     "input_device": None,
-    # Where PCM is captured: "local" (PortAudio on the backend host),
-    # "client" (desktop/TUI streams int16 frames via wake.feed), or
-    # "auto" (local when a device exists, else client capture).
+    # capture: "local" (PortAudio on the backend host), "client" (desktop/TUI streams
+    # int16 frames via wake.feed), or "auto" (local when a device exists, else client).
     "capture": "auto",
     "provider": "openwakeword",
     "phrase": "hey hermes",
@@ -81,8 +72,8 @@ _DEFAULTS: Dict[str, Any] = {
     "start_new_session": True,
 }
 
-# Bundled "hey hermes" model (tools/wakewords/) — the default. Config names in
-# _ALIASES resolve to it, not to an openWakeWord built-in.
+# Bundled "hey hermes" model (tools/wakewords/) — the default; alias names resolve
+# to it, not to an openWakeWord built-in.
 _BUNDLED_MODEL_NAME = "hey_hermes"
 _BUNDLED_MODEL_ALIASES = frozenset({"", "hey_hermes", "hey hermes", "hermes"})
 
@@ -100,9 +91,8 @@ def _is_macos_arm64() -> bool:
 
 
 def default_inference_framework() -> str:
-    """tflite on macOS ARM64, onnx elsewhere: openWakeWord's ONNX *embedding*
-    model scores near-zero on Apple Silicon (upstream #336) — the detector arms
-    but no phrase ever crosses threshold."""
+    """tflite on macOS ARM64, onnx elsewhere: openWakeWord's ONNX embedding model
+    scores near-zero on Apple Silicon — the detector arms but never fires."""
     return "tflite" if _is_macos_arm64() else "onnx"
 
 
@@ -110,19 +100,15 @@ _warned_onnx_coerced = False
 
 
 def resolve_inference_framework(cfg: Dict[str, Any]) -> str:
-    """Effective openWakeWord backend: explicit ``openwakeword.inference_framework``
-    or the platform default. The one provably dead combination — explicit
-    ``onnx`` on macOS ARM64 (upstream #336) — is coerced to tflite with a
-    one-time warning so a pre-fix pin doesn't keep a wake word that never fires.
-    """
+    """Effective openWakeWord backend: explicit ``openwakeword.inference_framework`` or
+    the platform default. Explicit ``onnx`` on macOS ARM64 is provably dead, so it is
+    coerced to tflite with a one-time warning (a pre-fix pin must not stay deaf)."""
     global _warned_onnx_coerced
 
     sub = cfg.get("openwakeword") if isinstance(cfg.get("openwakeword"), dict) else {}
     framework = str(sub.get("inference_framework") or "").strip().lower()
-
     if not framework:
         return default_inference_framework()
-
     if framework == "onnx" and _is_macos_arm64():
         if not _warned_onnx_coerced:
             _warned_onnx_coerced = True
@@ -133,16 +119,14 @@ def resolve_inference_framework(cfg: Dict[str, Any]) -> str:
                 "'tflite' in config.yaml to silence this."
             )
         return "tflite"
-
     return framework
 
 
 def ensure_tflite_runtime() -> bool:
     """Make ``import tflite_runtime.interpreter`` resolve, returning success.
 
-    openWakeWord hardcodes that import but only declares ``tflite-runtime`` on
-    Linux; on macOS the equivalent wheel is ``ai-edge-litert``. Alias the
-    module in-process (nothing is written to site-packages).
+    openWakeWord hardcodes that import but only declares ``tflite-runtime`` on Linux;
+    on macOS the wheel is ``ai-edge-litert``. Alias it in-process (site-packages untouched).
     """
     try:
         import tflite_runtime.interpreter  # noqa: F401
@@ -150,12 +134,10 @@ def ensure_tflite_runtime() -> bool:
         return True
     except ImportError:
         pass
-
     try:
         from ai_edge_litert import interpreter as _litert  # type: ignore[import-not-found]
     except ImportError:
         return False
-
     import types
 
     pkg = types.ModuleType("tflite_runtime")
@@ -178,7 +160,7 @@ def load_wake_word_config() -> Dict[str, Any]:
 
 
 def _get(cfg: Dict[str, Any], key: str) -> Any:
-    val = cfg.get(key, _DEFAULTS.get(key))
+    val = cfg.get(key)
     return _DEFAULTS.get(key) if val is None else val
 
 
@@ -200,9 +182,7 @@ def _input_device(cfg: Dict[str, Any]) -> int | str | None:
     raw = _get(cfg, "input_device")
     if raw is None or isinstance(raw, bool):
         return None
-    if isinstance(raw, int):
-        return raw
-    return str(raw).strip() or None
+    return raw if isinstance(raw, int) else (str(raw).strip() or None)
 
 
 def _sensitivity(cfg: Dict[str, Any]) -> float:
@@ -210,11 +190,7 @@ def _sensitivity(cfg: Dict[str, Any]) -> float:
 
 
 def _confirmation_frames(cfg: Dict[str, Any]) -> int:
-    """Consecutive over-threshold frames required to fire, clamped 1..10.
-
-    ``1`` restores single-frame behaviour; higher rejects ambient blips at the
-    cost of a few tens of ms of latency.
-    """
+    """Consecutive over-threshold frames required to fire, clamped 1..10 (1 = single-frame)."""
     return _clamped(cfg, "confirmation_frames", int, 1, 10)
 
 
@@ -224,19 +200,14 @@ def wake_phrase(cfg: Optional[Dict[str, Any]] = None) -> str:
     return str(_get(cfg, "phrase")) or "hey hermes"
 
 
-def resolve_capture_mode(
-    cfg: Optional[Dict[str, Any]] = None,
-    *,
-    prefer_client: bool = False,
-    force_local: bool = False,
-) -> str:
+def resolve_capture_mode(cfg: Optional[Dict[str, Any]] = None, *, prefer_client: bool = False,
+                         force_local: bool = False) -> str:
     """Return ``local`` or ``client`` capture mode for this arm.
 
-    ``prefer_client`` is set by remote desktop; ``force_local`` keeps CLI/TUI on
-    the process mic. Under ``auto`` a working backend input always wins (local
-    desktops keep PortAudio + ``input_device``); client is the fallback only for
-    a preferring surface with no usable backend mic — CLI/TUI stay local so
-    status reports the real requirement rather than a path nothing will feed.
+    ``prefer_client`` is set by remote desktop; ``force_local`` keeps CLI/TUI on the
+    process mic. Under ``auto`` a working backend input always wins; client is the
+    fallback only for a preferring surface with no usable backend mic — CLI/TUI stay
+    local so status reports the real requirement rather than a path nothing will feed.
     """
     cfg = cfg if cfg is not None else load_wake_word_config()
     if force_local:
@@ -244,9 +215,7 @@ def resolve_capture_mode(
     raw = str(_get(cfg, "capture") or "auto").strip().lower()
     if raw in ("client", "remote", "external"):
         return "client"
-    if raw == "local":
-        return "local"
-    if prefer_client and not _local_input_device_ready():
+    if raw != "local" and prefer_client and not _local_input_device_ready():
         return "client"
     return "local"
 
@@ -262,9 +231,6 @@ def _local_input_device_ready() -> bool:
     """True when PortAudio is importable and at least one input device exists."""
     try:
         sd, _ = _import_audio()
-    except (ImportError, OSError):
-        return False
-    try:
         devices = sd.query_devices()
         if isinstance(devices, dict):
             return _input_channels(devices) > 0
@@ -279,9 +245,8 @@ def _local_input_device_ready() -> bool:
 def wake_surface_enabled(surface: str, cfg: Optional[Dict[str, Any]] = None) -> bool:
     """Should ``surface`` (``cli`` / ``tui`` / ``gui``) host the listener?
 
-    True when enabled and the configured ``surface`` is ``auto`` or this exact
-    surface. ``auto`` only makes a surface eligible; the process/machine
-    ownership lock still permits a single claimant.
+    True when enabled and the configured ``surface`` is ``auto`` or this exact surface.
+    ``auto`` only makes a surface eligible; the ownership lock still admits one claimant.
     """
     cfg = cfg if cfg is not None else load_wake_word_config()
     if not cfg.get("enabled"):
@@ -304,10 +269,9 @@ def _active_profile_name() -> str:
 def enrolled_profile_phrases() -> Dict[str, str]:
     """Map ``profile name -> wake phrase`` for every wake-enabled profile.
 
-    Reads each profile's own ``config.yaml`` raw (``load_config()`` targets only
-    the ACTIVE profile). Enrolled = ``wake_word.enabled`` truthy; phrase defaults
-    to ``"hey <profile>"``. The sherpa engine listens for all of them at once and
-    routes the wake to the matching profile. Best-effort: unreadable skipped.
+    Reads each profile's own ``config.yaml`` raw (``load_config()`` targets only the
+    ACTIVE profile). Phrase defaults to ``"hey <profile>"``; the sherpa engine listens
+    for all of them and routes the wake to the matching profile. Unreadable → skipped.
     """
     phrases: Dict[str, str] = {}
     try:
@@ -317,8 +281,7 @@ def enrolled_profile_phrases() -> Dict[str, str]:
         for info in list_profiles():
             name = getattr(info, "name", None) or str(info)
             try:
-                raw = read_user_config_raw(Path(get_profile_dir(name)) / "config.yaml")
-                wc = raw.get("wake_word") or {}
+                wc = read_user_config_raw(Path(get_profile_dir(name)) / "config.yaml").get("wake_word") or {}
                 if not isinstance(wc, dict) or not wc.get("enabled"):
                     continue
                 phrase = str(wc.get("phrase") or f"hey {name}").strip()
@@ -351,8 +314,7 @@ def _audio_available() -> bool:
 def _describe_input_device(sd, selector: int | str | None) -> Dict[str, Any]:
     """Resolve a PortAudio selector into JSON-safe diagnostics.
 
-    Diagnostic only: ``InputStream`` remains the authority on whether the
-    device can actually open at the requested format.
+    Diagnostic only: ``InputStream`` stays the authority on whether the device opens.
     """
     details: Dict[str, Any] = {"selector": selector}
     try:
@@ -362,14 +324,11 @@ def _describe_input_device(sd, selector: int | str | None) -> Dict[str, Any]:
         return details
     if not isinstance(info, dict):
         return details
-
     if info.get("name"):
         details["name"] = str(info["name"])
-    for key, out_key, cast in (
-        ("max_input_channels", "max_input_channels", int),
-        ("default_samplerate", "default_samplerate", float),
-        ("hostapi", "hostapi_index", int),
-    ):
+    for key, out_key, cast in (("max_input_channels", "max_input_channels", int),
+                               ("default_samplerate", "default_samplerate", float),
+                               ("hostapi", "hostapi_index", int)):
         if isinstance(info.get(key), (int, float)):
             details[out_key] = cast(info[key])
     if "hostapi_index" in details:
@@ -409,18 +368,15 @@ def _resample_audio_frame(np, frame, output_length: int):
         return np.asarray(frame, dtype=np.int16).reshape(-1)
     if source.size == 0:
         return np.zeros(output_length, dtype=np.int16)
-
     if source.size > output_length:
-        # Average each source window when reducing (matches the desktop wake
-        # capture path) so speech energy is retained instead of decimated.
+        # Average each source window when reducing (matches the desktop wake capture
+        # path) so speech energy is retained instead of decimated.
         edges = np.linspace(0, source.size, output_length + 1, dtype=np.int64)
         values = np.add.reduceat(source, edges[:-1]) / np.diff(edges)
     else:
         # Unusual low-rate devices: interpolate up to the 16 kHz frame size.
         source_positions = np.arange(source.size, dtype=np.float64)
-        target_positions = np.linspace(0, source.size - 1, output_length)
-        values = np.interp(target_positions, source_positions, source)
-
+        values = np.interp(np.linspace(0, source.size - 1, output_length), source_positions, source)
     return np.rint(values).clip(-32768, 32767).astype(np.int16)
 
 
@@ -432,16 +388,10 @@ def silent_audio_hint(details: Dict[str, Any]) -> str:
             "microphone access in System Settings > Privacy & Security > "
             "Microphone, then toggle the wake word."
         )
-    if sys.platform == "win32":
-        return (
-            f"Microphone delivers only silence from {_device_label(details)}. "
-            "Set wake_word.input_device to a different PortAudio input device, "
-            "then toggle the wake word."
-        )
-    return (
-        f"Microphone delivers only silence from {_device_label(details)}. "
-        "Check the selected input device, then toggle the wake word."
-    )
+    fix = ("Set wake_word.input_device to a different PortAudio input device"
+           if sys.platform == "win32" else "Check the selected input device")
+    return (f"Microphone delivers only silence from {_device_label(details)}. "
+            f"{fix}, then toggle the wake word.")
 
 
 # ── Engines (implementations live in tools.wake_word_engines) ──
@@ -456,11 +406,8 @@ def _build_engine(cfg: Dict[str, Any]) -> _Engine:
 # ── Requirements probe (for /wake status + enable path) ──
 
 def _stt_ready() -> bool:
-    """Is a speech-to-text provider configured and enabled?
-
-    A wake without STT arms the mic but every utterance dies at transcription.
-    Same standard as voice mode's ``check_voice_requirements``.
-    """
+    """Is a speech-to-text provider configured and enabled? (A wake without STT arms the
+    mic but every utterance dies at transcription — same bar as ``check_voice_requirements``.)"""
     try:
         from tools.transcription_tools import _get_provider, _load_stt_config, is_stt_enabled
 
@@ -476,10 +423,9 @@ _LAZY_TTS_FEATURES = {"edge": "tts.edge", "elevenlabs": "tts.elevenlabs", "mistr
 def _tts_ready() -> bool:
     """Can the configured TTS provider run (or install at first use)?
 
-    PROBE, not an installer: ``check_tts_requirements`` lazily pip-installs the
-    provider SDK, which froze wake.status polls for a whole pip run (a failed
-    install unmounted the desktop ear). Uninstalled deps count as ready iff
-    lazy installs are allowed; pip is never touched from here.
+    PROBE, not an installer: ``check_tts_requirements`` lazily pip-installs the provider
+    SDK, which froze wake.status polls for a whole pip run. Uninstalled deps count as
+    ready iff lazy installs are allowed; pip is never touched from here.
     """
     try:
         from tools.tts_tool import _get_provider, _load_tts_config
@@ -487,7 +433,6 @@ def _tts_ready() -> bool:
         provider = _get_provider(_load_tts_config())
     except Exception:
         return False
-
     feature = _LAZY_TTS_FEATURES.get(provider)
     if feature is not None:
         try:
@@ -497,7 +442,6 @@ def _tts_ready() -> bool:
                 return lazy_deps._allow_lazy_installs()
         except Exception:
             return False
-
     try:
         from tools.tts_tool import check_tts_requirements
 
@@ -515,41 +459,35 @@ def check_wake_word_requirements(cfg: Optional[Dict[str, Any]] = None) -> Dict[s
     feature = _PROVIDERS.get(provider, ("", "wake.openwakeword"))[1]
     deps_ok = lazy_deps.is_available(feature)
     lazy_ok = lazy_deps._allow_lazy_installs()
-    # The audio probe imports sounddevice + numpy — packages the lazy installer
-    # would fetch — so only trust it once deps are installed; on a fresh install
-    # the engine constructors' ``lazy_deps.ensure()`` + stream-open surface any
-    # real audio problem (gating on the probe made lazy install unreachable).
+    # The audio probe imports sounddevice + numpy — packages the lazy installer would
+    # fetch — so only trust it once deps are installed; on a fresh install the engine
+    # constructors' ``lazy_deps.ensure()`` + stream-open surface any real audio problem.
     audio_ok = _audio_available() if deps_ok else False
-    key_ok = True
-    # Loop is wake → record → STT → agent → TTS; without either end the mic
-    # hears you and nothing perceptible happens — refuse with a hint.
-    stt_ok = _stt_ready()
-    tts_ok = _tts_ready()
-    hint = ""
-
-    # tflite needs a runtime openWakeWord doesn't declare off Linux; report it
-    # as a remediation instead of arming a detector that can't fire.
+    # Loop is wake → record → STT → agent → TTS; without either end the mic hears you
+    # and nothing perceptible happens — refuse with a hint.
+    stt_ok, tts_ok = _stt_ready(), _tts_ready()
+    # tflite needs a runtime openWakeWord doesn't declare off Linux; report it as a
+    # remediation instead of arming a detector that can't fire.
     tflite_ok = True
     if feature == "wake.openwakeword" and resolve_inference_framework(cfg) == "tflite":
         tflite_ok = ensure_tflite_runtime() or lazy_deps.is_available("wake.openwakeword.tflite") or lazy_ok
-
-    if provider == "porcupine" and not (os.getenv("PORCUPINE_ACCESS_KEY") or "").strip():
-        key_ok = False
-        hint = "Set PORCUPINE_ACCESS_KEY (free key at https://console.picovoice.ai)."
-    elif not deps_ok and not lazy_ok:
-        hint = lazy_deps.feature_install_command(feature) or ""
-    elif not tflite_ok:
-        hint = "The wake word needs the tflite runtime on this Mac: pip install ai-edge-litert"
-    elif deps_ok and not audio_ok and resolve_capture_mode(cfg) == "local":
-        hint = "Microphone capture needs sounddevice + numpy and a working audio device."
-    elif not stt_ok or not tts_ok:
-        missing = " and ".join(
-            name for name, ok in (("speech-to-text", stt_ok), ("text-to-speech", tts_ok)) if not ok
-        )
-        hint = (f"Wake word needs {missing} configured — run `hermes tools` "
-                f"(Voice section) or see the voice-mode docs.")
-
+    key_ok = provider != "porcupine" or bool((os.getenv("PORCUPINE_ACCESS_KEY") or "").strip())
     capture_mode = resolve_capture_mode(cfg)
+    missing = " and ".join(n for n, ok in (("speech-to-text", stt_ok), ("text-to-speech", tts_ok)) if not ok)
+
+    # Ordered remediation ladder: first true predicate wins.
+    ladder = (
+        (not key_ok, lambda: "Set PORCUPINE_ACCESS_KEY (free key at https://console.picovoice.ai)."),
+        (not deps_ok and not lazy_ok, lambda: lazy_deps.feature_install_command(feature) or ""),
+        (not tflite_ok,
+         lambda: "The wake word needs the tflite runtime on this Mac: pip install ai-edge-litert"),
+        (deps_ok and not audio_ok and capture_mode == "local",
+         lambda: "Microphone capture needs sounddevice + numpy and a working audio device."),
+        (bool(missing), lambda: (f"Wake word needs {missing} configured — run `hermes tools` "
+                                 f"(Voice section) or see the voice-mode docs.")),
+    )
+    hint = next((make() for cond, make in ladder if cond), "")
+
     # Client capture needs deps (engine) but not a server-side PortAudio device.
     if capture_mode == "client":
         mic_ok = deps_ok or lazy_ok
@@ -590,8 +528,7 @@ class _Capture:
     frame_length: int = 1280  # samples per read at ``rate``
 
     def read(self):
-        """One raw block; None when no client frame arrived within 250 ms.
-        Stream errors propagate."""
+        """One raw block; None when no client frame arrived within 250 ms. Stream errors propagate."""
         if self.stream is not None:
             return self.stream.read(self.frame_length)[0]
         try:
@@ -611,8 +548,8 @@ class _Capture:
 class WakeWordDetector:
     """Background hotword listener. Fires ``on_wake()`` when the phrase is heard.
 
-    The engine is built once and kept alive across pause/resume; only the audio
-    stream + reader thread cycle, so toggling the mic for a voice turn is cheap.
+    The engine is built once and kept alive across pause/resume; only the audio stream
+    + reader thread cycle, so toggling the mic for a voice turn is cheap.
     """
 
     def __init__(self, engine: _Engine, on_wake: Callable[[], None],
@@ -620,6 +557,8 @@ class WakeWordDetector:
                  on_failure: Optional[Callable[["WakeWordDetector"], None]] = None,
                  input_device: int | str | None = None,
                  external_audio: bool = False):
+        import queue as _queue
+
         self.engine = engine
         self.on_wake = on_wake
         self.cooldown = cooldown
@@ -628,8 +567,7 @@ class WakeWordDetector:
         self.external_audio = bool(external_audio)
         self.input_device_details: Dict[str, Any] = (
             {"selector": "client", "name": "client capture", "hostapi": "remote"}
-            if self.external_audio
-            else {"selector": input_device}
+            if self.external_audio else {"selector": input_device}
         )
         self._thread: Optional[threading.Thread] = None
         self._stop = threading.Event()
@@ -637,11 +575,9 @@ class WakeWordDetector:
         self._last_fire = 0.0
         self._lock = threading.Lock()
         # Client-capture PCM queue (int16 mono frames). Local mode ignores this.
-        import queue as _queue
-
         self._audio_q: "_queue.Queue[Any]" = _queue.Queue(maxsize=64)
-        # True when the stream is open but every frame is (near-)silence, so
-        # status surfaces can tell "armed" from "deaf".
+        # True when the stream is open but every frame is (near-)silence, so status
+        # surfaces can tell "armed" from "deaf".
         self.audio_silent = False
         self._silent_frames = 0
 
@@ -653,8 +589,8 @@ class WakeWordDetector:
     def feed(self, pcm_int16) -> None:
         """Enqueue one int16 mono frame (or raw bytes) for client capture.
 
-        Short frames are zero-padded to ``engine.frame_length``; long frames are
-        split. On queue overflow the oldest frame is dropped to stay real-time.
+        Short frames are zero-padded to ``engine.frame_length``; long frames are split.
+        On queue overflow the oldest frame is dropped to stay real-time.
         """
         if not self.external_audio:
             return
@@ -692,12 +628,8 @@ class WakeWordDetector:
             self._stop.clear()
             ready = threading.Event()
             startup_errors: list[BaseException] = []
-            self._thread = threading.Thread(
-                target=self._run,
-                args=(ready, startup_errors),
-                daemon=True,
-                name="wake-word",
-            )
+            self._thread = threading.Thread(target=self._run, args=(ready, startup_errors),
+                                            daemon=True, name="wake-word")
             self._thread.start()
         if not ready.wait(_START_TIMEOUT_SECONDS):
             self._halt_thread()
@@ -737,16 +669,13 @@ class WakeWordDetector:
     def _open_capture(self, frame_length: int) -> _Capture:
         """Open the audio source; raises on any local-mic failure."""
         if self.external_audio:
-            # Drain any stale frames from a previous arm.
-            try:
+            try:  # drain stale frames from a previous arm
                 while True:
                     self._audio_q.get_nowait()
             except Exception:
                 pass
-            logger.info(
-                "wake word: client-capture mode (frame=%d, rate=%d) — waiting for wake.feed",
-                frame_length, SAMPLE_RATE,
-            )
+            logger.info("wake word: client-capture mode (frame=%d, rate=%d) — waiting for wake.feed",
+                        frame_length, SAMPLE_RATE)
             return _Capture(queue=self._audio_q, frame_length=frame_length)
 
         try:
@@ -754,28 +683,18 @@ class WakeWordDetector:
         except (ImportError, OSError) as e:
             logger.error("wake word: audio libraries unavailable: %s", e)
             raise
-
-        self.input_device_details = _describe_input_device(sd, self.input_device)
-        cap = _Capture(np=np, rate=_capture_sample_rate(self.input_device_details))
+        details = self.input_device_details = _describe_input_device(sd, self.input_device)
+        cap = _Capture(np=np, rate=_capture_sample_rate(details))
         cap.frame_length = max(1, int(round(frame_length * cap.rate / SAMPLE_RATE)))
         logger.info(
             "wake word: opening microphone device=%s selector=%r hostapi=%s "
             "default_rate=%s capture_rate=%d engine_rate=%d",
-            self.input_device_details.get("name") or "system default",
-            self.input_device,
-            self.input_device_details.get("hostapi") or "unknown",
-            self.input_device_details.get("default_samplerate") or "unknown",
-            cap.rate,
-            SAMPLE_RATE,
+            details.get("name") or "system default", self.input_device, details.get("hostapi") or "unknown",
+            details.get("default_samplerate") or "unknown", cap.rate, SAMPLE_RATE,
         )
         try:
-            cap.stream = sd.InputStream(
-                device=self.input_device,
-                samplerate=cap.rate,
-                channels=1,
-                dtype="int16",
-                blocksize=cap.frame_length,
-            )
+            cap.stream = sd.InputStream(device=self.input_device, samplerate=cap.rate, channels=1,
+                                        dtype="int16", blocksize=cap.frame_length)
             cap.stream.start()
         except Exception as e:
             logger.error("wake word: failed to open microphone: %s", e)
@@ -792,11 +711,8 @@ class WakeWordDetector:
             self._silent_frames += 1
             if self._silent_frames == silent_alert_frames:
                 self.audio_silent = True
-                logger.warning(
-                    "wake word: mic delivers only silence (peak<=%d for %ds); %s",
-                    _SILENCE_PEAK, _SILENCE_ALERT_SECONDS,
-                    silent_audio_hint(self.input_device_details),
-                )
+                logger.warning("wake word: mic delivers only silence (peak<=%d for %ds); %s", _SILENCE_PEAK,
+                               _SILENCE_ALERT_SECONDS, silent_audio_hint(self.input_device_details))
         elif self._silent_frames:
             if self.audio_silent:
                 logger.info("wake word: mic audio detected — stream healthy")
@@ -815,8 +731,7 @@ class WakeWordDetector:
             self._callback_inflight.set()
             threading.Thread(target=self._dispatch_wake, daemon=True, name="wake-word-callback").start()
 
-    def _run(self, ready: threading.Event,
-             startup_errors: list[BaseException]) -> None:
+    def _run(self, ready: threading.Event, startup_errors: list[BaseException]) -> None:
         frame_length = self.engine.frame_length
         try:
             cap = self._open_capture(frame_length)
@@ -824,15 +739,12 @@ class WakeWordDetector:
             startup_errors.append(e)
             ready.set()
             return
-
-        # Drop buffered audio/feature state so a resume right after a voice turn
-        # can't re-fire on audio captured before the pause (the wake → voice →
-        # resume → wake runaway loop).
+        # Drop buffered audio/feature state so a resume right after a voice turn can't
+        # re-fire on audio captured before the pause (wake → voice → resume → wake loop).
         try:
             self.engine.reset()
         except Exception:
             pass
-
         logger.info("wake word: listening (frame=%d, rate=%d, external=%s)",
                     frame_length, SAMPLE_RATE, self.external_audio)
         ready.set()
@@ -846,8 +758,7 @@ class WakeWordDetector:
                     logger.warning("wake word: stream read error: %s", e)
                     failed = not self._stop.is_set()
                     break
-                if data is None:
-                    # No client frames yet — count as silence for status.
+                if data is None:  # no client frames yet — counts as silence for status
                     self._silent_frames += 1
                     if self._silent_frames == silent_alert_frames:
                         self.audio_silent = True
@@ -930,9 +841,7 @@ def _clear_singleton_locked() -> tuple[Optional[WakeWordDetector], Any]:
     """Forget the armed detector (caller holds ``_detector_lock``); returns (detector, lock handle)."""
     global _detector, _detector_owner, _detector_file_lock
     det, handle = _detector, _detector_file_lock
-    _detector = None
-    _detector_owner = None
-    _detector_file_lock = None
+    _detector = _detector_owner = _detector_file_lock = None
     return det, handle
 
 
@@ -953,18 +862,13 @@ def _detector_failed(detector: WakeWordDetector) -> None:
             _release_machine_lock(lock_handle)
 
 
-def start_listening(
-    on_wake: Callable[[], None],
-    *,
-    owner: object,
-    config: Optional[Dict[str, Any]] = None,
-    external_audio: bool = False,
-) -> WakeWordDetector:
+def start_listening(on_wake: Callable[[], None], *, owner: object, config: Optional[Dict[str, Any]] = None,
+                    external_audio: bool = False) -> WakeWordDetector:
     """Claim, build, and start the detector. Idempotent for the same owner.
 
-    Raises if engine construction fails (missing deps / access key / model);
-    callers should probe :func:`check_wake_word_requirements` first. A different
-    owner, including another process, receives :class:`WakeWordInUse`.
+    Raises if engine construction fails (missing deps / access key / model); callers
+    should probe :func:`check_wake_word_requirements` first. A different owner,
+    including another process, receives :class:`WakeWordInUse`.
     """
     if owner is None:
         raise ValueError("wake-word owner must not be None")
@@ -980,17 +884,9 @@ def start_listening(
         lock_handle = _acquire_machine_lock()
         try:
             cfg = config if config is not None else load_wake_word_config()
-            engine = _build_engine(cfg)
-            detector = WakeWordDetector(
-                engine,
-                on_wake,
-                on_failure=_detector_failed,
-                input_device=_input_device(cfg),
-                external_audio=external_audio,
-            )
-            _detector = detector
-            _detector_owner = owner
-            _detector_file_lock = lock_handle
+            detector = WakeWordDetector(_build_engine(cfg), on_wake, on_failure=_detector_failed,
+                                        input_device=_input_device(cfg), external_audio=external_audio)
+            _detector, _detector_owner, _detector_file_lock = detector, owner, lock_handle
             detector.start()
             return detector
         except Exception:
@@ -1052,11 +948,8 @@ def is_listening() -> bool:
 
 
 def audio_is_silent() -> bool:
-    """True when the armed stream has delivered only silence (dead mic).
-
-    The stream opens fine but every frame is zeros, so detection can never
-    fire; status surfaces show "listening but the microphone appears silent".
-    """
+    """True when the armed stream opens fine but delivers only silence (dead mic), so
+    detection can never fire; status shows "listening but the microphone appears silent"."""
     det = _current_detector()
     return det is not None and det.audio_silent
 
@@ -1066,7 +959,6 @@ def get_input_device_status(cfg: Optional[Dict[str, Any]] = None) -> Dict[str, A
     det = _current_detector()
     if det is not None:
         return dict(det.input_device_details)
-
     cfg = cfg if cfg is not None else load_wake_word_config()
     selector = _input_device(cfg)
     try:
@@ -1077,17 +969,14 @@ def get_input_device_status(cfg: Optional[Dict[str, Any]] = None) -> Dict[str, A
 
 
 def get_last_match() -> Optional[tuple[str, str]]:
-    """(matched phrase, profile) of the most recent wake fire, if the engine
-    reports per-phrase matches (sherpa multi-profile routing). None otherwise."""
+    """(matched phrase, profile) of the most recent wake fire when the engine reports
+    per-phrase matches (sherpa multi-profile routing); None otherwise."""
     det = _current_detector()
     return None if det is None else getattr(det.engine, "last_match", None)
 
 
 def feed_audio(*, owner: object, pcm_int16) -> bool:
-    """Push client-captured PCM into the armed detector (client capture mode).
-
-    Returns True when the frame was accepted for ``owner``'s armed detector.
-    """
+    """Push client-captured PCM into ``owner``'s armed detector; True when accepted."""
     with _detector_lock:
         det = _owned_detector(owner)
         if det is None or not det.external_audio:
diff --git a/tools/wake_word_engines.py b/tools/wake_word_engines.py
index adca319077..3a3afb4825 100644
--- a/tools/wake_word_engines.py
+++ b/tools/wake_word_engines.py
@@ -1,8 +1,8 @@
 """Wake-word hotword engines (openWakeWord / sherpa-onnx KWS / Porcupine).
 
-All three run fully on-device. Config, platform probes and sensitivity
-accessors live in :mod:`tools.wake_word`; engines read them lazily through
-that module so test seams (``patch("tools.wake_word.<name>")``) keep working.
+All three run fully on-device. Config, platform probes and sensitivity accessors
+live in :mod:`tools.wake_word`; engines read them lazily through that module so
+test seams (``patch("tools.wake_word.<name>")``) keep working.
 """
 
 from __future__ import annotations
@@ -21,14 +21,19 @@ def _ww():
     return wake_word
 
 
+def _ensure_dep(feature: str) -> None:
+    from tools import lazy_deps
+
+    lazy_deps.ensure(feature, prompt=False)
+
+
 class _Engine:
     """Minimal hotword-engine contract: feed int16 frames, get a bool."""
 
     frame_length: int = 1280  # 80 ms at 16 kHz
 
-    #: (matched phrase, profile name) of the most recent fire. Multi-phrase
-    #: engines (sherpa) set this for profile routing; single-phrase engines
-    #: leave it None (callers fall back to configured phrase / active profile).
+    #: (matched phrase, profile name) of the most recent fire. Multi-phrase engines
+    #: (sherpa) set this for profile routing; single-phrase engines leave it None.
     last_match: Optional[tuple[str, str]] = None
 
     def process(self, frame) -> bool:  # frame: 1-D int16 ndarray
@@ -53,19 +58,15 @@ def _sub(cfg: Dict[str, Any], key: str) -> Dict[str, Any]:
 class _OpenWakeWordEngine(_Engine):
     """openWakeWord — free, local ONNX/tflite hotword detection.
 
-    Scores one ~80 ms frame at a time; ``sensitivity`` IS the raw 0..1
-    threshold (higher = stricter). A real utterance holds the score high
-    across frames while a stray ambient phoneme spikes one, so we require
-    ``confirmation_frames`` consecutive over-threshold frames before firing.
+    Scores one ~80 ms frame at a time; ``sensitivity`` IS the raw 0..1 threshold
+    (higher = stricter). A real utterance holds the score high across frames while a
+    stray phoneme spikes one, so ``confirmation_frames`` consecutive hits are required.
     """
 
     frame_length = 1280  # openWakeWord recommends 80 ms frames.
 
     def __init__(self, cfg: Dict[str, Any]):
-        from tools import lazy_deps
-
-        lazy_deps.ensure("wake.openwakeword", prompt=False)
-
+        _ensure_dep("wake.openwakeword")
         import openwakeword
         from openwakeword.model import Model
 
@@ -75,20 +76,15 @@ class _OpenWakeWordEngine(_Engine):
         self._threshold = ww._sensitivity(cfg)
         self._confirm_needed = ww._confirmation_frames(cfg)
         self._confirm_streak = 0
-
-        # Default (or explicit "hey_hermes") → the bundled model; a built-in
-        # name or custom path is used as-is.
+        # Default (or explicit "hey_hermes") → the bundled model; built-in names / paths as-is.
         if model_ref.lower() in ww._BUNDLED_MODEL_ALIASES:
             model_ref = ww._bundled_wakeword_path(framework)
-
-        # download_models() also fetches the shared feature models (melspectrogram
-        # + embedding) needed for ANY model, so a custom path must call it too or a
-        # fresh install crashes on a missing melspectrogram.onnx.
+        # download_models() also fetches the shared feature models (melspectrogram +
+        # embedding) needed for ANY model, so a custom path must call it too.
         try:
             openwakeword.utils.download_models([model_ref])
         except Exception as e:  # pragma: no cover - network/path dependent
             logger.debug("openwakeword model download skipped: %s", e)
-
         self._model = Model(wakeword_models=[model_ref], inference_framework=framework)
         self._labels = list(self._model.models.keys())
 
@@ -96,19 +92,16 @@ class _OpenWakeWordEngine(_Engine):
     def _usable_framework(framework: str) -> str:
         """Refuse openWakeWord's silent tflite→onnx downgrade.
 
-        Without a tflite runtime openWakeWord falls back to onnx, which on
-        macOS ARM64 is the backend whose embedding model never fires — the
-        listener would arm and stay deaf. Install + bridge the runtime first
-        (the platform gate lives here because dep specs can't carry PEP 508
+        Without a tflite runtime openWakeWord falls back to onnx, which on macOS ARM64
+        never fires — the listener would arm and stay deaf. Install + bridge the runtime
+        first (the platform gate lives here because dep specs can't carry PEP 508
         markers); on that Mac raise instead of downgrading.
         """
         ww = _ww()
         if framework != "tflite" or ww.ensure_tflite_runtime():
             return framework
         try:
-            from tools import lazy_deps
-
-            lazy_deps.ensure("wake.openwakeword.tflite", prompt=False)
+            _ensure_dep("wake.openwakeword.tflite")
         except Exception as e:
             logger.debug("wake word: tflite runtime install failed: %s", e)
         if ww.ensure_tflite_runtime():
@@ -133,8 +126,8 @@ class _OpenWakeWordEngine(_Engine):
         return True
 
     def reset(self) -> None:
-        # Clears openWakeWord's rolling feature/prediction buffer so stale audio
-        # captured before a pause can't re-fire the moment we resume.
+        # Clears openWakeWord's rolling feature buffer so stale audio captured before a
+        # pause can't re-fire the moment we resume.
         self._confirm_streak = 0
         try:
             self._model.reset()
@@ -145,9 +138,8 @@ class _OpenWakeWordEngine(_Engine):
         self.reset()
 
 
-# sherpa-onnx open-vocabulary KWS model: a small streaming zipformer transducer
-# (English, GigaSpeech); one-time download cached under HERMES_HOME. Keywords
-# are typed phrases tokenized at RUNTIME — no training step.
+# sherpa-onnx open-vocabulary KWS model: small streaming zipformer transducer (English,
+# GigaSpeech), downloaded once under HERMES_HOME. Keywords are tokenized at RUNTIME.
 _SHERPA_KWS_MODEL_URL = (
     "https://github.com/k2-fsa/sherpa-onnx/releases/download/kws-models/"
     "sherpa-onnx-kws-zipformer-gigaspeech-3.3M-2024-01-01.tar.bz2"
@@ -185,18 +177,16 @@ def _ensure_sherpa_model(root: Optional[Path] = None) -> Path:
 class _SherpaKwsEngine(_Engine):
     """sherpa-onnx open-vocabulary keyword spotting — any typed phrase, zero training.
 
-    ``wake_word.phrase`` is BPE-tokenized at runtime against the model's
-    vocabulary, so here ``phrase`` is DETECTION config, not a cosmetic label.
+    ``wake_word.phrase`` is BPE-tokenized at runtime against the model's vocabulary,
+    so here ``phrase`` is DETECTION config, not a cosmetic label.
     """
 
     frame_length = 1280  # streaming zipformer accepts any chunk; match capture path.
 
     def __init__(self, cfg: Dict[str, Any]):
-        from tools import lazy_deps
-
-        lazy_deps.ensure("wake.sherpa", prompt=False)
-
+        _ensure_dep("wake.sherpa")
         import sherpa_onnx
+        import tempfile
         from sherpa_onnx import text2token
 
         ww = _ww()
@@ -205,30 +195,21 @@ class _SherpaKwsEngine(_Engine):
         if not (d / "tokens.txt").exists():
             raise RuntimeError(f"sherpa KWS model not found at {d}")
 
-        # Phrase set: this profile's own phrase plus — when profile routing is
-        # on — every other wake-enabled profile's phrase, so ONE listener can
-        # wake any profile. phrase → profile is kept for routing the match back.
+        # Phrase set: this profile's phrase plus — with profile routing on — every other
+        # wake-enabled profile's phrase, so ONE listener can wake any profile.
         phrase = str(ww._get(cfg, "phrase") or "hey hermes").strip()
         phrase_map: Dict[str, str] = {phrase: ww._active_profile_name()}
         if bool(cfg.get("profile_routing", True)):
             for prof, p in ww.enrolled_profile_phrases().items():
                 phrase_map.setdefault(p.strip(), prof)
-
         phrases = list(phrase_map)
-        tokens = text2token(
-            [p.upper() for p in phrases],
-            tokens=str(d / "tokens.txt"),
-            tokens_type="bpe",
-            bpe_model=str(d / "bpe.model"),
-        )
-        import tempfile
-
-        # sherpa keyword entries reject spaces in the @display-name; underscore
-        # them and map display → profile for match routing.
+        tokens = text2token([p.upper() for p in phrases], tokens=str(d / "tokens.txt"),
+                            tokens_type="bpe", bpe_model=str(d / "bpe.model"))
+        # sherpa keyword entries reject spaces in the @display-name; underscore them and
+        # map display → profile for match routing.
         self._display_to_profile: Dict[str, str] = {}
-        kw = tempfile.NamedTemporaryFile(
-            mode="w", suffix=".txt", prefix="hermes-kws-", delete=False, encoding="utf-8"
-        )
+        kw = tempfile.NamedTemporaryFile(mode="w", suffix=".txt", prefix="hermes-kws-", delete=False,
+                                         encoding="utf-8")
         for p, toks in zip(phrases, tokens):
             display = p.upper().replace(" ", "_")
             self._display_to_profile[display] = phrase_map[p]
@@ -237,9 +218,9 @@ class _SherpaKwsEngine(_Engine):
         self._keywords_file = kw.name
         self.last_match: Optional[tuple[str, str]] = None
 
-        # Shared 0..1 sensitivity → sherpa keywords_threshold. 0.5 lands on
-        # sherpa's recommended 0.25; a stricter 0.35 missed ~12% of true
-        # positives in live TTS matrix tests while 0.25 held zero false fires.
+        # Shared 0..1 sensitivity → sherpa keywords_threshold. 0.5 lands on sherpa's
+        # recommended 0.25; a stricter 0.35 missed ~12% of true positives in live TTS
+        # matrix tests while 0.25 held zero false fires.
         threshold = 0.05 + 0.4 * ww._sensitivity(cfg)
 
         def _model_file(pattern: str) -> str:
@@ -262,8 +243,7 @@ class _SherpaKwsEngine(_Engine):
     def process(self, frame) -> bool:
         import numpy as np
 
-        samples = np.asarray(frame, dtype=np.float32) / 32768.0
-        self._stream.accept_waveform(_ww().SAMPLE_RATE, samples)
+        self._stream.accept_waveform(_ww().SAMPLE_RATE, np.asarray(frame, dtype=np.float32) / 32768.0)
         fired = False
         while self._spotter.is_ready(self._stream):
             self._spotter.decode_stream(self._stream)
@@ -271,17 +251,13 @@ class _SherpaKwsEngine(_Engine):
             if result:
                 fired = True
                 display = str(result)
-                self.last_match = (
-                    display.replace("_", " ").lower(),
-                    self._display_to_profile.get(display, ""),
-                )
-                # Reset decoder state so one utterance can't fire repeatedly.
-                self._spotter.reset_stream(self._stream)
+                self.last_match = (display.replace("_", " ").lower(),
+                                   self._display_to_profile.get(display, ""))
+                self._spotter.reset_stream(self._stream)  # one utterance must not fire repeatedly
         return fired
 
     def reset(self) -> None:
-        # Fresh stream drops buffered audio/decoder state (pause → resume must
-        # not re-fire on stale audio).
+        # Fresh stream drops buffered audio/decoder state (pause → resume must not re-fire).
         try:
             self._stream = self._spotter.create_stream()
         except Exception:
@@ -298,10 +274,7 @@ class _PorcupineEngine(_Engine):
     """Picovoice Porcupine — premium, on-device, needs an access key."""
 
     def __init__(self, cfg: Dict[str, Any]):
-        from tools import lazy_deps
-
-        lazy_deps.ensure("wake.porcupine", prompt=False)
-
+        _ensure_dep("wake.porcupine")
         import pvporcupine
 
         access_key = (os.getenv("PORCUPINE_ACCESS_KEY") or "").strip()
@@ -310,19 +283,16 @@ class _PorcupineEngine(_Engine):
                 "Porcupine wake word requires PORCUPINE_ACCESS_KEY "
                 "(get a free key at https://console.picovoice.ai)."
             )
-
         keyword = str(_sub(cfg, "porcupine").get("keyword") or "jarvis").strip()
-        # Porcupine's `sensitivities` runs the OPPOSITE way to our shared knob
-        # (higher = looser); invert so "higher = stricter" holds for every engine.
+        # Porcupine's `sensitivities` runs the OPPOSITE way to our shared knob (higher =
+        # looser); invert so "higher = stricter" holds for every engine.
         kwargs: Dict[str, Any] = {"access_key": access_key, "sensitivities": [1.0 - _ww()._sensitivity(cfg)]}
         kwargs["keyword_paths" if _looks_like_path(keyword) else "keywords"] = [keyword]
-
         self._porcupine = pvporcupine.create(**kwargs)
         self.frame_length = self._porcupine.frame_length
 
     def process(self, frame) -> bool:
-        # pvporcupine wants a plain list/sequence of int16 samples.
-        return self._porcupine.process(frame) >= 0
+        return self._porcupine.process(frame) >= 0  # pvporcupine wants a plain sequence of int16
 
     def close(self) -> None:
         try:
diff --git a/tools/working_diff.py b/tools/working_diff.py
index 4ab4205683..2b6f74d92d 100644
--- a/tools/working_diff.py
+++ b/tools/working_diff.py
@@ -1,13 +1,10 @@
 """Working-tree git diff collection shared by the CLI and gateway ``/diff``.
 
 Surface-agnostic so the CLI (colored terminal) and gateway (fenced, truncated
-messages) render the same data.
-
-Modes: ``working`` (unstaged + untracked — what ``git checkout . && git clean
--fd`` would lose), ``staged`` (``git diff --cached``), ``all`` (everything since
-HEAD plus untracked). Untracked files are folded in via ``git diff --no-index
-/dev/null <file>`` so brand-new files show as additions instead of being
-invisible (mirrors Codex CLI's ``/diff``).
+messages) render the same data. Modes: ``working`` (unstaged + untracked),
+``staged`` (``git diff --cached``), ``all`` (everything since HEAD plus untracked).
+Untracked files are folded in via ``git diff --no-index /dev/null <file>`` so
+brand-new files show as additions instead of being invisible.
 """
 
 from __future__ import annotations
@@ -34,15 +31,13 @@ def _run(args: List[str], cwd: str, timeout: int = _GIT_TIMEOUT):
     """Run git, returning (returncode, stdout). Never raises on git failure.
 
     Hardened against a malicious repo's ``.git/config`` (GHSA-7x36-8jrh-v4pw):
-    ``noninteractive_git_env`` disables fsmonitor/hooks/pager/editor/credential
-    sinks, and ``harden_git_argv`` appends ``--no-ext-diff --no-textconv`` to
-    the diff-rendering subcommands so attribute-scoped diff/textconv drivers
-    can't execute either.
+    ``noninteractive_git_env`` disables fsmonitor/hooks/pager/editor/credential sinks,
+    and ``harden_git_argv`` appends ``--no-ext-diff --no-textconv`` to diff-rendering
+    subcommands so attribute-scoped diff/textconv drivers can't execute either.
     """
     proc = subprocess.run(
         ["git", "-c", "core.quotePath=false", *harden_git_argv(args)],
-        cwd=cwd, capture_output=True, text=True, timeout=timeout,
-        encoding="utf-8", errors="replace",
+        cwd=cwd, capture_output=True, text=True, timeout=timeout, encoding="utf-8", errors="replace",
         stdin=subprocess.DEVNULL, env=noninteractive_git_env(),
     )
     return proc.returncode, proc.stdout
@@ -50,9 +45,7 @@ def _run(args: List[str], cwd: str, timeout: int = _GIT_TIMEOUT):
 
 def _untracked_files(cwd: str) -> List[str]:
     code, out = _run(["ls-files", "--others", "--exclude-standard"], cwd)
-    if code != 0:
-        return []
-    return [line for line in out.splitlines() if line.strip()]
+    return [line for line in out.splitlines() if line.strip()] if code == 0 else []
 
 
 def _untracked_diff(cwd: str, files: List[str]) -> str:
@@ -60,35 +53,28 @@ def _untracked_diff(cwd: str, files: List[str]) -> str:
     chunks: List[str] = []
     for rel in files[:_MAX_UNTRACKED_FILES]:
         try:
-            # --no-index exits 1 when files differ — that's the success path,
-            # so the return code is ignored.
+            # --no-index exits 1 when files differ — the success path, so the code is ignored.
             _, out = _run(["diff", "--no-index", "--", os.devnull, rel], cwd)
             if out.strip():
                 chunks.append(out.rstrip("\n"))
         except (subprocess.TimeoutExpired, OSError):
             continue
     if len(files) > _MAX_UNTRACKED_FILES:
-        chunks.append(
-            f"... ({len(files) - _MAX_UNTRACKED_FILES} more untracked files not shown)"
-        )
+        chunks.append(f"... ({len(files) - _MAX_UNTRACKED_FILES} more untracked files not shown)")
     return "\n".join(chunks)
 
 
-def collect_working_diff(cwd: str, mode: str = "working",
-                         paths: List[str] | None = None) -> Dict:
+def collect_working_diff(cwd: str, mode: str = "working", paths: List[str] | None = None) -> Dict:
     """Collect a git diff of the working directory.
 
     Returns ``{"success", "stat", "diff", "untracked", "empty"}`` on success or
-    ``{"success": False, "error": ...}`` when git is unavailable / not a repo.
-    ``paths`` optionally restricts the diff to pathspecs (passed to git
-    verbatim); when given, untracked files are not collected.
+    ``{"success": False, "error": ...}`` when git is unavailable / not a repo. ``paths``
+    restricts the diff to pathspecs (passed verbatim); untracked files are then skipped.
     """
     if mode not in _MODE_ARGS:
-        return {"success": False,
-                "error": f"Unknown mode '{mode}'. Use: {', '.join(VALID_MODES)}"}
+        return {"success": False, "error": f"Unknown mode '{mode}'. Use: {', '.join(VALID_MODES)}"}
     if not shutil.which("git"):
         return {"success": False, "error": "git is not installed or not on PATH."}
-
     try:
         code, _ = _run(["rev-parse", "--is-inside-work-tree"], cwd, timeout=5)
     except (subprocess.TimeoutExpired, OSError) as e:
@@ -101,12 +87,8 @@ def collect_working_diff(cwd: str, mode: str = "working",
     try:
         _, stat_out = _run([*base_args, "--stat", *pathspec], cwd)
         _, diff_out = _run([*base_args, *pathspec], cwd, timeout=_GIT_TIMEOUT * 2)
-        untracked: List[str] = []
-        untracked_diff = ""
-        if mode in ("working", "all") and not paths:
-            untracked = _untracked_files(cwd)
-            if untracked:
-                untracked_diff = _untracked_diff(cwd, untracked)
+        untracked = _untracked_files(cwd) if mode in ("working", "all") and not paths else []
+        untracked_diff = _untracked_diff(cwd, untracked) if untracked else ""
     except subprocess.TimeoutExpired:
         return {"success": False, "error": "git diff timed out."}
     except OSError as e:
@@ -116,7 +98,6 @@ def collect_working_diff(cwd: str, mode: str = "working",
     diff = diff_out.strip()
     if untracked_diff:
         diff = f"{diff}\n{untracked_diff}".strip()
-
     result = {"success": True, "stat": stat, "diff": diff, "untracked": untracked}
     if not stat and not diff and not untracked:
         result["empty"] = True
diff --git a/tools/write_approval.py b/tools/write_approval.py
index a47f4719cb..5adc70493b 100644
--- a/tools/write_approval.py
+++ b/tools/write_approval.py
@@ -1,21 +1,12 @@
 #!/usr/bin/env python3
 """Write-approval gate + pending store for memory and skill writes.
 
-The agent writes to two cross-session stores — **memory** (MEMORY.md / USER.md,
-small entries) and **skills** (SKILL.md + files, potentially 10-100 KB) — from
-two origins: **foreground** (a normal turn) and **background_review** (the
-autonomous self-improvement fork). A per-subsystem boolean ``write_approval``
-gates those writes: ``false`` (default) writes freely; ``true`` never commits
-directly — it prompts inline (memory, interactive CLI only) or **stages** the
-write to a pending store for out-of-band review.
-
-Staging is mandatory for background writes (a daemon thread cannot block on a
-prompt), gateway sessions (no inline channel — review via ``/memory pending``),
-and all skill writes (too big to eyeball mid-loop). Memory shows full content;
-skills show metadata + a gist + a ``diff`` escape hatch.
-
-Pending records live under ``<HERMES_HOME>/pending/{memory,skills}/<id>.json``
-so they survive restarts and can be reviewed from CLI, gateway, or dashboard.
+A per-subsystem boolean ``write_approval`` gates the agent's cross-session writes —
+**memory** (MEMORY.md / USER.md) and **skills** (SKILL.md + files) — from either
+origin (**foreground** turn or **background_review** fork). ``false`` (default)
+writes freely; ``true`` never commits directly: it prompts inline (memory,
+interactive CLI only) or **stages** the write under
+``<HERMES_HOME>/pending/{memory,skills}/<id>.json`` for out-of-band review.
 """
 
 from __future__ import annotations
@@ -40,9 +31,8 @@ MEMORY = "memory"
 SKILLS = "skills"
 _SUBSYSTEMS = (MEMORY, SKILLS)
 
-# Per-subsystem config key. Intentionally a single boolean with no "block all
-# writes" state — to disable a subsystem use its own enable flag
-# (e.g. ``memory.memory_enabled: false``).
+# Per-subsystem config key. Intentionally a single boolean with no "block all writes"
+# state — to disable a subsystem use its own enable flag (e.g. ``memory.memory_enabled``).
 CONFIG_KEY = "write_approval"
 
 
@@ -61,11 +51,8 @@ def write_approval_enabled(subsystem: str) -> bool:
 
 
 def _normalize_enabled(value: Any) -> bool:
-    """Coerce a config value to bool; unknown → False (gate off).
-
-    YAML already parses bare on/off/yes/no as bools; the string branch covers
-    hand-edited configs.
-    """
+    """Coerce a config value to bool; unknown → False (gate off). The string branch
+    covers hand-edited configs (YAML already parses bare on/off/yes/no)."""
     if isinstance(value, bool):
         return value
     if isinstance(value, str):
@@ -87,15 +74,13 @@ def _read_record(path: Path) -> Dict[str, Any]:
     return json.loads(path.read_text(encoding="utf-8"))
 
 
-def stage_write(subsystem: str, payload: Dict[str, Any],
-                *, summary: str, origin: str) -> Dict[str, Any]:
+def stage_write(subsystem: str, payload: Dict[str, Any], *, summary: str, origin: str) -> Dict[str, Any]:
     """Persist a pending write and return its record (``id`` + metadata).
 
-    ``payload`` is the exact kwargs to replay the write on approval; ``summary``
-    is the one-line description shown in pending lists; ``origin`` is
-    ``foreground`` or ``background_review`` (audit). Best-effort: on disk
-    failure it logs and still returns a record — the write is lost, which is
-    the safe failure for an approval gate (nothing silently committed).
+    ``payload`` is the exact kwargs to replay the write on approval; ``origin`` is
+    ``foreground`` or ``background_review`` (audit). Best-effort: on disk failure it
+    logs and still returns a record — the write is lost, which is the safe failure
+    for an approval gate (nothing silently committed).
     """
     pid = uuid.uuid4().hex[:8]
     record = {
@@ -136,10 +121,8 @@ def list_pending(subsystem: str) -> List[Dict[str, Any]]:
 def get_pending(subsystem: str, pending_id: str) -> Optional[Dict[str, Any]]:
     """Return a single pending record by id, or None."""
     path = _pending_path(subsystem, pending_id)
-    if not path.exists():
-        return None
     try:
-        return _read_record(path)
+        return _read_record(path) if path.exists() else None
     except Exception:
         return None
 
@@ -159,10 +142,8 @@ def discard_pending(subsystem: str, pending_id: str) -> bool:
 def pending_count(subsystem: str) -> int:
     """Cheap count of pending records (for notification badges)."""
     d = _pending_dir(subsystem)
-    if not d.exists():
-        return 0
     try:
-        return sum(1 for _ in d.glob("*.json"))
+        return sum(1 for _ in d.glob("*.json")) if d.exists() else 0
     except Exception:
         return 0
 
@@ -170,11 +151,8 @@ def pending_count(subsystem: str) -> int:
 # --- Write origin ---
 
 def current_origin() -> str:
-    """Return ``foreground`` or ``background_review``.
-
-    Reuses the skill-provenance ContextVar the background review fork sets;
-    foreground turns leave it at the default.
-    """
+    """``foreground`` or ``background_review`` — reuses the skill-provenance ContextVar
+    the background review fork sets; foreground turns leave it at the default."""
     try:
         from tools.skill_provenance import get_current_write_origin
         return get_current_write_origin()
@@ -188,9 +166,9 @@ def current_origin() -> str:
 class GateDecision:
     """Result of evaluating the write gate. Exactly one flag is True.
 
-    ``allow`` proceed with the real write; ``blocked`` the user denied an inline
-    prompt (``message`` explains why); ``stage`` the caller must ``stage_write``
-    the payload (``message`` is the user-facing "staged for approval" note).
+    ``allow`` proceed with the real write; ``blocked`` the user denied an inline prompt
+    (``message`` explains why); ``stage`` the caller must ``stage_write`` the payload
+    (``message`` is the user-facing "staged for approval" note).
     """
 
     allow: bool = False
@@ -201,59 +179,39 @@ class GateDecision:
 
 def _staged(subsystem: str) -> GateDecision:
     where = "/skills pending" if subsystem == SKILLS else "/memory pending"
-    return GateDecision(
-        stage=True,
-        message=(
-            f"Staged for approval ({subsystem}.write_approval is on). "
-            f"Not yet saved — review with {where}."
-        ),
-    )
+    return GateDecision(stage=True, message=(f"Staged for approval ({subsystem}.write_approval is on). "
+                                             f"Not yet saved — review with {where}."))
 
 
-def evaluate_gate(subsystem: str, *, inline_summary: str = "",
-                  inline_detail: str = "") -> GateDecision:
+def evaluate_gate(subsystem: str, *, inline_summary: str = "", inline_detail: str = "") -> GateDecision:
     """Decide what to do with a pending write for ``subsystem``.
 
-    Decision matrix:
-        gate off (default)                    → allow
-        gate on, memory + interactive CLI     → inline approve/deny prompt
-        gate on, memory + gateway/script/bg   → stage
-        gate on, skills (any origin)          → stage (too big to review inline)
-
-    The gate only ever delays a write, never silently refuses it; ``blocked``
-    is produced only when the user actively denies the inline prompt.
-    ``inline_summary``/``inline_detail`` feed the memory inline prompt.
+    gate off → allow; gate on + skills (any origin) or background → stage; gate on +
+    memory + foreground → inline prompt when an interactive channel exists, else stage.
+    The gate only ever delays a write, never silently refuses it; ``blocked`` is
+    produced only when the user actively denies the inline prompt.
     """
     if not write_approval_enabled(subsystem):
         return GateDecision(allow=True)
-
-    # Skills always stage; a background write runs in a daemon thread with no user.
+    # Skills are too big to review inline; a background write runs in a daemon thread with no user.
     if subsystem == SKILLS or current_origin() == "background_review":
         return _staged(subsystem)
-
-    # Memory + foreground: prompt inline if an interactive channel exists;
-    # otherwise (gateway, script, prompt failure) stage instead of blind-denying.
     granted = _prompt_inline_memory_approval(inline_summary, inline_detail)
     if granted is True:
         return GateDecision(allow=True)
     if granted is False:
-        return GateDecision(
-            blocked=True,
-            message="Memory write denied by user. The change was not saved.",
-        )
+        return GateDecision(blocked=True, message="Memory write denied by user. The change was not saved.")
     return _staged(MEMORY)
 
 
 def _prompt_inline_memory_approval(summary: str, detail: str) -> Optional[bool]:
     """Prompt inline for a memory write: True approved, False denied, None → stage.
 
-    Uses the per-thread CLI approval callback registered for dangerous
-    commands (``tools.terminal_tool.set_approval_callback``), invoked directly
-    rather than via ``prompt_dangerous_approval``: that wrapper falls back to
-    ``input()`` (deadlock-prone under prompt_toolkit; silent deny in gateway
-    sessions, whose ``/approve`` round-trip lives in the pending-approval
-    queue) and converts callback errors into a deny. Here a missing channel or
-    failed prompt must stage instead.
+    Uses the per-thread CLI approval callback (``tools.terminal_tool.set_approval_callback``)
+    directly rather than ``prompt_dangerous_approval``: that wrapper falls back to
+    ``input()`` (deadlock-prone under prompt_toolkit; silent deny in gateway sessions)
+    and turns callback errors into a deny, whereas here a missing channel or failed
+    prompt must stage instead.
     """
     try:
         from tools.terminal_tool import _get_approval_callback
@@ -262,32 +220,30 @@ def _prompt_inline_memory_approval(summary: str, detail: str) -> Optional[bool]:
     callback = _get_approval_callback()
     if callback is None:
         return None
-
     header = summary.strip() or "Save to memory?"
-    body = detail.strip()
     try:
-        choice = callback(body or header, f"Save to memory: {header}", allow_permanent=False)
+        choice = callback(detail.strip() or header, f"Save to memory: {header}", allow_permanent=False)
     except Exception as e:
         logger.error("Inline memory approval prompt failed: %s", e)
         return None
-
     if choice in {"once", "session"}:
         return True
-    if choice == "deny":
-        return False
-    return None  # unknown outcome → no decision, stage rather than drop
+    return False if choice == "deny" else None  # unknown outcome → stage rather than drop
 
 
 # --- Skill-specific helpers (gist + diff for the review affordances) ---
 
-def skill_gist(action: str, name: str, *, content: str = "",
-               file_path: str = "", old_string: str = "",
-               new_string: str = "") -> str:
-    """Build a one-line heuristic gist (no model call) for a pending skill write.
+_GIST_TEMPLATES = {
+    "write_file": "write {file_path} in '{name}'",
+    "remove_file": "remove {file_path} from '{name}'",
+    "delete": "delete skill '{name}'",
+}
 
-    create/edit use the frontmatter ``description:``; patch/write_file describe
-    the size of the change. The full diff stays behind /skills diff.
-    """
+
+def skill_gist(action: str, name: str, *, content: str = "", file_path: str = "",
+               old_string: str = "", new_string: str = "") -> str:
+    """One-line heuristic gist (no model call) for a pending skill write: create/edit use
+    the frontmatter ``description:``; patch/write_file describe the size of the change."""
     if action in {"create", "edit"} and content:
         desc = _frontmatter_description(content)
         size = f"{len(content) // 1024 + 1} KB" if len(content) >= 1024 else f"{len(content)} chars"
@@ -297,13 +253,8 @@ def skill_gist(action: str, name: str, *, content: str = "",
         removed = old_string.count("\n") + 1 if old_string else 0
         added = new_string.count("\n") + 1 if new_string else 0
         return f"patch '{name}' {file_path or 'SKILL.md'} (+{added}/-{removed} lines)"
-    if action == "write_file":
-        return f"write {file_path} in '{name}'"
-    if action == "remove_file":
-        return f"remove {file_path} from '{name}'"
-    if action == "delete":
-        return f"delete skill '{name}'"
-    return f"{action} '{name}'"
+    template = _GIST_TEMPLATES.get(action, "{action} '{name}'")
+    return template.format(action=action, name=name, file_path=file_path)
 
 
 def _frontmatter_description(content: str) -> str:
@@ -323,15 +274,11 @@ def _find_skill_path(name: str) -> Optional[Path]:
 
 
 def skill_pending_diff(record: Dict[str, Any]) -> str:
-    """Full content (create) or unified diff vs. the on-disk skill (edit/patch/write_file).
-
-    Rendered by /skills diff <id> on surfaces that can show it (CLI pager,
-    dashboard, pending JSON file).
-    """
+    """Full content (create) or unified diff vs. the on-disk skill (edit/patch/write_file),
+    rendered by /skills diff <id> on surfaces that can show it."""
     payload = record.get("payload", {})
     action = payload.get("action", "")
     name = payload.get("name", "")
-
     if action == "create":
         return payload.get("content") or ""
     if action == "remove_file":
@@ -350,24 +297,17 @@ def skill_pending_diff(record: Dict[str, Any]) -> str:
             target_label = payload.get("file_path") or "SKILL.md"
         try:
             p = skill_dir / target_label
-            if p.exists():
-                current = p.read_text(encoding="utf-8")
+            current = p.read_text(encoding="utf-8") if p.exists() else ""
         except Exception:
             current = ""
 
     if action == "edit":
         new = payload.get("content") or ""
     elif action == "patch":
-        old_s = payload.get("old_string") or ""
-        new_s = payload.get("new_string") or ""
+        old_s, new_s = payload.get("old_string") or "", payload.get("new_string") or ""
         new = current.replace(old_s, new_s) if current else f"(patch {old_s!r} → {new_s!r})"
     else:
         new = payload.get("file_content") or ""
-
-    diff = difflib.unified_diff(
-        current.splitlines(keepends=True),
-        new.splitlines(keepends=True),
-        fromfile=f"a/{target_label}",
-        tofile=f"b/{target_label}",
-    )
+    diff = difflib.unified_diff(current.splitlines(keepends=True), new.splitlines(keepends=True),
+                                fromfile=f"a/{target_label}", tofile=f"b/{target_label}")
     return "".join(diff) or "(no textual change)"