hermes-admin/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions 32 Packages Projects Releases Wiki Activity

refactor(voice): WSL detection through hermes_platform.host.runtime.is_wsl

Browse Source 
One runtime predicate keeps WSL detection from diverging between call sites.
The shared result is cached per process.
This commit is contained in:
alt-glitch
2026-09-20 19:09:40 +05:30
committed by GitHub
parent a6b2364862
commit 62aa4b7d0d
3 changed files with 22 additions and 76 deletions
Download Patch File Download Diff File Expand all files Collapse all files

62
tests/tools/test_voice_mode.py
View File

@@ -10,18 +10,6 @@ from unittest.mock import MagicMock, patch
import pytest
def _non_wsl_proc_version(real_open):
"""Return an open() shim that makes host WSL detection deterministic."""
def _fake_open(file, *args, **kwargs):
if file == "/proc/version":
from io import StringIO
return StringIO("Linux test-kernel")
return real_open(file, *args, **kwargs)
return _fake_open
# ============================================================================
# Fixtures
# ============================================================================
@@ -162,7 +150,7 @@ class TestDetectAudioEnvironment:
monkeypatch.setattr("hermes_constants.is_container", lambda: False)
monkeypatch.setattr("tools.voice_mode._import_audio",
lambda: (MagicMock(), MagicMock()))
monkeypatch.setattr("builtins.open", _non_wsl_proc_version(open))
monkeypatch.setattr("tools.voice_mode.is_wsl", lambda: False)
from tools.voice_mode import detect_audio_environment
result = detect_audio_environment()
@@ -190,7 +178,7 @@ class TestDetectAudioEnvironment:
monkeypatch.delenv("PIPEWIRE_REMOTE", raising=False)
monkeypatch.setattr("tools.voice_mode._import_audio",
lambda: (MagicMock(), MagicMock()))
monkeypatch.setattr("builtins.open", _non_wsl_proc_version(open))
monkeypatch.setattr("tools.voice_mode.is_wsl", lambda: False)
from tools.voice_mode import detect_audio_environment
result = detect_audio_environment()
@@ -198,7 +186,7 @@ class TestDetectAudioEnvironment:
assert result["warnings"] == []
assert any("SSH" in n for n in result.get("notices", []))
def test_wsl_without_pulse_blocks_voice(self, monkeypatch, tmp_path):
def test_wsl_without_pulse_blocks_voice(self, monkeypatch):
"""WSL without PULSE_SERVER should block voice mode."""
monkeypatch.delenv("SSH_CLIENT", raising=False)
monkeypatch.delenv("SSH_TTY", raising=False)
@@ -208,18 +196,10 @@ class TestDetectAudioEnvironment:
monkeypatch.setattr("tools.voice_mode._import_audio",
lambda: (MagicMock(), MagicMock()))
proc_version = tmp_path / "proc_version"
proc_version.write_text("Linux 5.15.0-microsoft-standard-WSL2")
monkeypatch.setattr("tools.voice_mode.is_wsl", lambda: True)
_real_open = open
def _fake_open(f, *a, **kw):
if f == "/proc/version":
return _real_open(str(proc_version), *a, **kw)
return _real_open(f, *a, **kw)
with patch("builtins.open", side_effect=_fake_open):
from tools.voice_mode import detect_audio_environment
result = detect_audio_environment()
from tools.voice_mode import detect_audio_environment
result = detect_audio_environment()
assert result["available"] is False
assert any("WSL" in w for w in result["warnings"])
@@ -1445,7 +1425,7 @@ class TestWSL2PowerShellFallback:
m.wait = MagicMock(return_value=m.returncode)
return m
with patch("tools.voice_mode._is_wsl2_env", return_value=True), \
with patch("tools.voice_mode.is_wsl", return_value=True), \
patch("tools.voice_mode._import_audio", side_effect=ImportError), \
patch("tools.voice_mode.shutil.which",
side_effect=lambda x: f"/bin/{x}" if x in ("powershell.exe", "ffmpeg", "ffplay", "sh") else (x if x.startswith("/") else None)), \
@@ -1491,13 +1471,7 @@ class TestWSL2PowerShellFallback:
return f"C:\\Temp\\{wsl_path.split('/')[-1]}\n".encode()
return b""
def _fake_open(path, *args, **kwargs):
if str(path) == "/proc/version":
import io
return io.StringIO("Linux Microsoft WSL2")
return open(path, *args, **kwargs)
with patch("builtins.open", side_effect=_fake_open), \
with patch("tools.voice_mode.is_wsl", return_value=True), \
patch("shutil.which", side_effect=lambda x: f"/bin/{x}" if x in ("powershell.exe", "ffmpeg", "ffplay") else None), \
patch("subprocess.check_output", side_effect=_capture_check_output), \
patch("subprocess.Popen", return_value=MagicMock(returncode=0, wait=lambda **k: 0)), \
@@ -1531,13 +1505,7 @@ class TestWSL2PowerShellFallback:
m.wait.return_value = 0
return m
def _fake_open(path, *args, **kwargs):
if str(path) == "/proc/version":
import io
return io.StringIO("Linux version 5.15.0-generic #72-Ubuntu")
return open(path, *args, **kwargs)
with patch("builtins.open", side_effect=_fake_open), \
with patch("tools.voice_mode.is_wsl", return_value=False), \
patch("tools.voice_mode._import_audio", side_effect=ImportError), \
patch("shutil.which", side_effect=lambda x: f"/bin/{x}" if x in ("ffplay", "aplay") else None), \
patch("subprocess.Popen", side_effect=_capture_popen), \
@@ -1558,12 +1526,6 @@ class TestWSLAudioEnvironmentGate:
not be hard-blocked, but the recording/STT PulseAudio-bridge guidance
must still be surfaced (as a non-blocking notice)."""
def _fake_open_wsl(self, path, *args, **kwargs):
if str(path) == "/proc/version":
import io
return io.StringIO("Linux version 5.15 Microsoft Standard WSL2")
return open(path, *args, **kwargs)
def test_wsl_no_pulse_but_powershell_available_not_hard_blocked(self, monkeypatch):
from unittest.mock import patch
from tools import voice_mode as vm
@@ -1574,7 +1536,7 @@ class TestWSLAudioEnvironmentGate:
monkeypatch.delenv(_ssh_var, raising=False)
monkeypatch.setattr("tools.voice_mode._import_audio",
lambda: (MagicMock(), MagicMock()))
with patch("builtins.open", side_effect=self._fake_open_wsl), \
with patch("tools.voice_mode.is_wsl", return_value=True), \
patch("tools.voice_mode._wsl_powershell_tts_available", return_value=True), \
patch("tools.voice_mode._pulse_socket_reachable", return_value=False), \
patch("hermes_constants.is_container", return_value=False):
@@ -1601,7 +1563,7 @@ class TestWSLAudioEnvironmentGate:
monkeypatch.delenv(_ssh_var, raising=False)
monkeypatch.setattr("tools.voice_mode._import_audio",
lambda: (MagicMock(), MagicMock()))
with patch("builtins.open", side_effect=self._fake_open_wsl), \
with patch("tools.voice_mode.is_wsl", return_value=True), \
patch("tools.voice_mode._wsl_powershell_tts_available", return_value=False), \
patch("tools.voice_mode._pulse_socket_reachable", return_value=False), \
patch("hermes_constants.is_container", return_value=False):
@@ -1622,7 +1584,7 @@ class TestWSLAudioEnvironmentGate:
monkeypatch.delenv(_ssh_var, raising=False)
monkeypatch.setattr("tools.voice_mode._import_audio",
lambda: (MagicMock(), MagicMock()))
with patch("builtins.open", side_effect=self._fake_open_wsl), \
with patch("tools.voice_mode.is_wsl", return_value=True), \
patch("hermes_constants.is_container", return_value=False):
result = vm.detect_audio_environment()

15
tests/tools/test_voice_wsl_pipewire.py
View File

@@ -4,22 +4,15 @@ detect_audio_environment() honors forwarded audio (has_forwarded_audio =
PULSE_SERVER or PIPEWIRE_REMOTE or a reachable socket) in the SSH and container
blocks, but the WSL block previously checked only PULSE_SERVER — so a WSL user
with PipeWire forwarding (PIPEWIRE_REMOTE) was wrongly blocked from voice mode.
These tests mock /proc/version so they reproduce the WSL path on any host.
These tests patch the voice module's WSL predicate so they reproduce the WSL path on any host.
"""
import builtins
import io
from unittest.mock import MagicMock
WSL = "Linux version 5.15.0-microsoft-standard-WSL2 (oe-user@oe-host)"
def _force_wsl(monkeypatch):
import tools.voice_mode as voice_mode
def _force_wsl(monkeypatch, content=WSL):
real_open = builtins.open
def fake_open(file, *a, **k):
if str(file) == "/proc/version":
return io.StringIO(content)
return real_open(file, *a, **k)
monkeypatch.setattr(builtins, "open", fake_open)
monkeypatch.setattr(voice_mode, "is_wsl", lambda: True)
def _base(monkeypatch):

21
tools/voice_mode.py
View File

@@ -23,8 +23,9 @@ from typing import Any, Callable, Dict, List, Optional
logger = logging.getLogger(__name__)
from tools.voice_mode_transcript import _voice_config, is_voice_stop_phrase, is_whisper_hallucination
from hermes_constants import is_termux as _is_termux_environment
from hermes_platform.host.runtime import is_wsl
from tools.voice_mode_transcript import _voice_config, is_voice_stop_phrase, is_whisper_hallucination
# ── Recording parameters ──
SAMPLE_RATE = 16000 # Whisper native rate
@@ -310,7 +311,7 @@ def detect_audio_environment() -> dict:
# WSL: the PowerShell/Media.SoundPlayer fallback only covers OUTPUT, so when
# it is all that's available downgrade to a notice (recording guidance stays
# visible, TTS-only usage isn't blocked).
if _is_wsl2_env():
if is_wsl():
if has_forwarded_audio:
notices.append("Running in WSL with a reachable PulseAudio/PipeWire sound server")
elif _wsl_powershell_tts_available():
@@ -961,20 +962,10 @@ def stop_playback() -> None:
sd.stop()
def _is_wsl2_env() -> bool:
"""True inside WSL (Microsoft kernel signature in /proc/version); False on any error.
Module-level so tests can patch it instead of ``builtins.open``."""
try:
with open("/proc/version", encoding="utf-8", errors="replace") as _fv:
return "microsoft" in _fv.read().lower()
except OSError:
return False
def _wsl_powershell_tts_available() -> bool:
"""WSL2 PowerShell TTS fallback usable. OUTPUT only (Media.SoundPlayer on the host) —
recording still needs a PulseAudio bridge, so callers keep surfacing that guidance."""
return bool(_is_wsl2_env() and shutil.which("powershell.exe") and shutil.which("ffmpeg"))
return bool(is_wsl() and shutil.which("powershell.exe") and shutil.which("ffmpeg"))
def play_audio_file(file_path: str) -> bool:
@@ -1000,7 +991,7 @@ def _play_wav_via_sounddevice(file_path: str) -> bool:
# ~100 ms to stabilise and the small default blocksize worsens
# clock-adjustment jitter (microsoft/wslg#1257).
blocksize = 0 # default (auto)
if _is_wsl2_env():
if is_wsl():
fade_samples = int(0.1 * sample_rate)
audio_float = audio_data.astype(np.float64)
audio_float[:fade_samples] *= np.linspace(0.0, 1.0, fade_samples, dtype=np.float64)
@@ -1023,7 +1014,7 @@ def _wsl_powershell_player_cmd(file_path: str) -> Optional[List[str]]:
ffplay/aplay have no device, but Media.SoundPlayer on the host does: convert to a
uniquely-named WAV in Windows %TEMP% (concurrent TTS must not collide), play, always
delete, and re-raise the ORIGINAL exit status past the cleanup (rm -f exits 0)."""
if not (shutil.which("powershell.exe") and shutil.which("ffmpeg") and _is_wsl2_env()):
if not (shutil.which("powershell.exe") and shutil.which("ffmpeg") and is_wsl()):
return None
try:
import uuid