fix(voice): add WSL2 PowerShell audio fallback for TTS playback

Ports #63768 forward onto current main per teknium1's review.

On WSL2 without a PulseAudio bridge, ffplay and aplay have no audio
device and TTS playback silently fails (issue #17608). When
powershell.exe and ffmpeg are available, convert the audio to a
uniquely-named WAV in the Windows %TEMP% directory and play it via
Media.SoundPlayer.

Per review, this fixes two gaps in the original port:

1. Exit-status masking: the cleanup subshell was
   '( ffmpeg && powershell ); rm -f wav' -- the shell's exit status is
   the LAST command's (rm -f, which is always 0), so a real
   ffmpeg/PowerShell failure could never be detected by the rc-checking
   fallback logic added to the player loop. Now captures the real
   status before cleanup and re-exits with it:
   '( ffmpeg && powershell ); rc=0; rm -f wav; exit '.

2. The no-Pulse WSL gate in detect_audio_environment() still hard-blocked
   voice mode entirely (input AND output) even when the PowerShell
   fallback made TTS output viable. Added _wsl_powershell_tts_available()
   and use it to downgrade the WSL-without-Pulse case from a hard
   'warnings' block to a non-blocking 'notices' entry when the fallback
   is available -- the same PulseAudio-bridge recording guidance is still
   surfaced (mic capture genuinely still needs it), it just no longer
   blocks /voice on for TTS-only usage. cli.py's existing
   env_check['available'] gate needed no changes since it already
   respects this flag.

Also fixed the flaky uniqueness test (the original asserted
len(filenames) >= 2, which passed trivially on zero captured
filenames) and added a real fallback-triggering regression test for
the exit-status fix.

10 new/fixed tests pass in TestWSL2PowerShellFallback and the new
TestWSLAudioEnvironmentGate; 80/80 in the full tests/tools/test_voice_mode.py file.
This commit is contained in:
ygd58 2026-07-21 12:47:41 +00:00 committed by Teknium
parent ec7a46a6fe
commit 0560f52047
2 changed files with 361 additions and 3 deletions

View file

@ -2101,3 +2101,255 @@ class TestDefaultInputSamplerate:
assert wav_path is not None
with wave.open(wav_path, "rb") as wf:
assert wf.getframerate() == 48000
class TestWSL2PowerShellFallback:
"""Regression tests for WSL2 PowerShell TTS fallback (issue #17608).
On WSL2 without a PulseAudio bridge, ffplay/aplay have no audio device.
play_audio_file() should insert a PowerShell-based player at the front
of the player list when powershell.exe and ffmpeg are available.
"""
def _fake_check_output(self, responses):
"""Build a subprocess.check_output side_effect from a list of responses."""
it = iter(responses)
def _side_effect(cmd, **kwargs):
return next(it)
return _side_effect
def test_wsl2_powershell_player_inserted_first(self, monkeypatch, sample_wav):
"""When WSL2 is detected and powershell.exe + ffmpeg are available,
a sh -c pipeline must be inserted before ffplay/aplay in the player list."""
from unittest.mock import patch, MagicMock
from tools import voice_mode as vm
captured_players = []
def _capture_popen(cmd, **kw):
captured_players.append(list(cmd))
m = MagicMock()
m.returncode = 0
m.wait = MagicMock(return_value=0)
return m
with patch("tools.voice_mode._is_wsl2_env", 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)), \
patch("tools.voice_mode.subprocess.check_output",
side_effect=self._fake_check_output([
b"C:/Temp\r\n",
b"/mnt/c/Temp\n",
b"C:/Temp/hermes.wav\n",
])), \
patch("tools.voice_mode.subprocess.Popen", side_effect=_capture_popen):
vm.play_audio_file(str(sample_wav))
assert captured_players, "No players were tried"
first_cmd = captured_players[0]
assert first_cmd[0] in ("/bin/sh", "sh") and first_cmd[1] == "-c", (
f"Expected sh -c as first player, got {first_cmd}"
)
assert "powershell.exe" in first_cmd[2]
assert "PlaySync" in first_cmd[2]
def test_powershell_pipeline_preserves_real_exit_status(self, sample_wav):
"""Regression (review of #63768): the shell pipeline must preserve
the (ffmpeg && powershell) exit status past the unconditional
cleanup, so a real conversion/playback failure falls through to the
next player instead of being masked by rm -f's always-zero exit."""
from unittest.mock import patch, MagicMock
from tools import voice_mode as vm
captured_cmds = []
def _capture_popen(cmd, **kw):
captured_cmds.append(list(cmd))
m = MagicMock()
# Simulate the PowerShell pipeline failing (nonzero rc), and
# the fallback ffplay succeeding.
if cmd[0] in ("/bin/sh", "sh"):
m.returncode = 1
else:
m.returncode = 0
m.wait = MagicMock(return_value=m.returncode)
return m
with patch("tools.voice_mode._is_wsl2_env", 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)), \
patch("tools.voice_mode.subprocess.check_output",
side_effect=self._fake_check_output([
b"C:/Temp\r\n",
b"/mnt/c/Temp\n",
b"C:/Temp/hermes.wav\n",
])), \
patch("tools.voice_mode.subprocess.Popen", side_effect=_capture_popen):
result = vm.play_audio_file(str(sample_wav))
assert result is True, "Must fall through to ffplay and succeed"
assert len(captured_cmds) == 2, (
f"Expected sh pipeline to be tried and fail, then ffplay to be "
f"tried: {captured_cmds}"
)
assert captured_cmds[0][0] in ("/bin/sh", "sh")
assert captured_cmds[1][0] == "ffplay"
# The subshell command must capture and re-exit with $rc, not rely
# on rm -f's exit status.
sh_script = captured_cmds[0][2]
assert "rc=$?" in sh_script and "exit $rc" in sh_script, (
"Shell pipeline must preserve the real exit status past cleanup: " + sh_script
)
def test_wsl2_unique_temp_filename(self, monkeypatch, tmp_path, sample_wav):
"""Two concurrent calls must use different temp WAV filenames."""
from unittest.mock import patch, MagicMock
from tools import voice_mode as vm
filenames = []
def _capture_check_output(cmd, **kwargs):
cmd_str = " ".join(str(c) for c in cmd)
if "TEMP" in cmd_str:
return b"C:\\Temp\r\n"
if "wslpath" in cmd_str and "-u" in cmd_str:
return b"/mnt/c/Temp\n"
if "wslpath" in cmd_str and "-w" in cmd_str:
wsl_path = cmd[-1] if isinstance(cmd[-1], str) else cmd[-1].decode()
filenames.append(wsl_path.split("/")[-1])
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), \
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)), \
patch("tools.voice_mode._playback_lock"), \
patch("tools.voice_mode._active_playback", None):
vm.play_audio_file(str(sample_wav))
vm.play_audio_file(str(sample_wav))
# Regression (review of #63768): the original test made this
# assertion conditional on len(filenames) >= 2, so a broken
# (zero-captured) run passed trivially. Require exactly two.
assert len(filenames) == 2, (
f"Expected exactly 2 captured temp filenames from 2 calls, got "
f"{len(filenames)}: {filenames}"
)
assert filenames[0] != filenames[1], (
"Concurrent TTS calls must use unique temp WAV filenames"
)
def test_non_wsl_skips_powershell_fallback(self, monkeypatch, sample_wav):
"""On non-WSL Linux, the PowerShell player must not be inserted."""
from unittest.mock import patch, MagicMock
from tools import voice_mode as vm
captured_players = []
def _capture_popen(cmd, **kw):
captured_players.append(cmd)
m = MagicMock()
m.returncode = 0
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), \
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), \
patch("tools.voice_mode._playback_lock"), \
patch("tools.voice_mode._active_playback", None):
vm.play_audio_file(str(sample_wav))
assert captured_players, "No players were tried"
for cmd in captured_players:
assert not (cmd[0] == "sh" and "powershell" in " ".join(str(c) for c in cmd)), (
"PowerShell player must not appear on non-WSL Linux"
)
class TestWSLAudioEnvironmentGate:
"""Regression tests (review of #63768) for detect_audio_environment()'s
WSL gate: when the PowerShell TTS fallback is viable, voice mode must
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
monkeypatch.delenv("PULSE_SERVER", raising=False)
monkeypatch.delenv("PIPEWIRE_REMOTE", raising=False)
monkeypatch.setattr("tools.voice_mode._import_audio",
lambda: (MagicMock(), MagicMock()))
with patch("builtins.open", side_effect=self._fake_open_wsl), \
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):
result = vm.detect_audio_environment()
assert result["available"] is True, (
"PowerShell TTS fallback must keep voice mode enabled even "
"without a PulseAudio bridge: " + str(result["warnings"])
)
assert any("PowerShell" in n or "Media.SoundPlayer" in n for n in result["notices"]), (
"The PowerShell fallback path must be mentioned in notices"
)
assert any("recording" in n.lower() or "PulseAudio" in n for n in result["notices"]), (
"The recording/STT PulseAudio caveat must still be surfaced"
)
def test_wsl_no_pulse_no_powershell_still_blocked(self, monkeypatch):
from unittest.mock import patch
from tools import voice_mode as vm
monkeypatch.delenv("PULSE_SERVER", raising=False)
monkeypatch.delenv("PIPEWIRE_REMOTE", raising=False)
monkeypatch.setattr("tools.voice_mode._import_audio",
lambda: (MagicMock(), MagicMock()))
with patch("builtins.open", side_effect=self._fake_open_wsl), \
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):
result = vm.detect_audio_environment()
assert result["available"] is False, (
"Without PulseAudio AND without the PowerShell fallback, WSL "
"must still be hard-blocked as before"
)
def test_wsl_with_pulse_server_unaffected(self, monkeypatch):
"""PULSE_SERVER already configured: existing behavior unchanged."""
from unittest.mock import patch
from tools import voice_mode as vm
monkeypatch.setenv("PULSE_SERVER", "unix:/mnt/wslg/PulseServer")
monkeypatch.setattr("tools.voice_mode._import_audio",
lambda: (MagicMock(), MagicMock()))
with patch("builtins.open", side_effect=self._fake_open_wsl), \
patch("hermes_constants.is_container", return_value=False):
result = vm.detect_audio_environment()
assert result["available"] is True
assert any("PulseAudio bridge" in n for n in result["notices"])

View file

@ -14,6 +14,7 @@ import math
import os
import platform
import re
import shlex
import shutil
import subprocess
import sys
@ -209,13 +210,28 @@ def detect_audio_environment() -> dict:
# WSL detection — a reachable sound server makes audio work in WSL.
# Honor any forwarding (PulseAudio bridge OR a forwarded PipeWire/Pulse
# socket), mirroring the SSH and container blocks above. Only block when
# no forwarding is configured.
# socket), mirroring the SSH and container blocks above. When no
# forwarding is configured, only hard-block if the WSL2 PowerShell TTS
# fallback (Media.SoundPlayer via powershell.exe, see play_audio_file)
# isn't available either. The PowerShell path only covers OUTPUT (TTS
# playback) -- microphone recording genuinely still needs the
# PulseAudio bridge -- so when it's the only thing available we
# downgrade to a notice (keeps the same recording guidance visible,
# but doesn't block /voice on for TTS-only usage).
try:
with open('/proc/version', 'r', encoding="utf-8") as f:
if 'microsoft' in f.read().lower():
if has_forwarded_audio:
notices.append("Running in WSL with a reachable PulseAudio/PipeWire sound server")
elif _wsl_powershell_tts_available():
notices.append(
"Running in WSL without a PulseAudio bridge -- TTS playback "
"will use the PowerShell/Media.SoundPlayer fallback. "
"Voice INPUT (recording) still requires a PulseAudio bridge:\n"
" 1. Set PULSE_SERVER=unix:/mnt/wslg/PulseServer\n"
" 2. Create ~/.asoundrc pointing ALSA at PulseAudio\n"
" 3. Verify with: arecord -d 3 /tmp/test.wav && aplay /tmp/test.wav"
)
else:
warnings.append(
"Running in WSL -- audio requires a forwarded sound server.\n"
@ -1188,6 +1204,36 @@ def _is_wsl() -> bool:
return False
def _is_wsl2_env() -> bool:
"""Return True when running inside WSL2 (Windows Subsystem for Linux 2).
Reads /proc/version and checks for the Microsoft kernel signature.
Returns False on any error (non-WSL Linux, Docker, SSH, etc.).
Extracted as a module-level function so tests can patch it directly
without fighting builtins.open patching complexity.
"""
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:
"""Return True when the WSL2 PowerShell TTS playback fallback can be used.
This only covers OUTPUT (TTS playback via Media.SoundPlayer on the
Windows host) -- it does NOT make microphone recording work. A caller
using this to relax the audio-environment gate must still surface the
existing PulseAudio-bridge guidance for recording/STT.
"""
return bool(
_is_wsl2_env()
and shutil.which("powershell.exe")
and shutil.which("ffmpeg")
)
def play_audio_file(file_path: str) -> bool:
"""Play an audio file through the default output device.
@ -1256,6 +1302,60 @@ def play_audio_file(file_path: str) -> bool:
if system == "Darwin":
players.append(["afplay", file_path])
# WSL2 PowerShell fallback: when running in WSL without a PulseAudio
# bridge, ffplay and aplay have no audio device. If powershell.exe and
# ffmpeg are available, convert the audio to a uniquely-named WAV in the
# Windows %TEMP% directory and play it via Media.SoundPlayer -- which
# always has a working audio device on the Windows host (#17608).
# A unique suffix prevents concurrent Hermes TTS calls from colliding on
# the same filename. The WAV is deleted in the shell pipeline
# unconditionally (success or failure), and the ORIGINAL ffmpeg/
# powershell exit status is preserved past that cleanup so the player
# loop below can correctly fall through to ffplay/aplay on failure.
if system == "Linux" and shutil.which("powershell.exe") and shutil.which("ffmpeg"):
if _is_wsl2_env():
try:
import uuid
_win_tmp_raw = subprocess.check_output(
["cmd.exe", "/c", "echo %TEMP%"],
stderr=subprocess.DEVNULL, timeout=3,
).decode(errors="replace").strip()
_win_tmp_wsl = subprocess.check_output(
["wslpath", "-u", _win_tmp_raw],
stderr=subprocess.DEVNULL, timeout=3,
).decode(errors="replace").strip()
if _win_tmp_wsl:
# Unique suffix prevents concurrent TTS playback collision.
_unique = uuid.uuid4().hex[:8]
_wsl_wav = os.path.join(_win_tmp_wsl, f"hermes-tts-{_unique}.wav")
_win_wav = subprocess.check_output(
["wslpath", "-w", _wsl_wav],
stderr=subprocess.DEVNULL, timeout=3,
).decode(errors="replace").strip()
if _win_wav:
_win_wav_safe = _win_wav.replace("'", "''")
_ps_script = (
f"(New-Object Media.SoundPlayer '{_win_wav_safe}').PlaySync()"
)
_ps_cmd = " && ".join([
shlex.join(["ffmpeg", "-i", file_path, "-f", "wav",
_wsl_wav, "-loglevel", "quiet", "-y"]),
shlex.join(["powershell.exe", "-NoProfile", "-Command",
_ps_script]),
])
_cleanup = shlex.join(["rm", "-f", _wsl_wav])
# Capture the (ffmpeg && powershell) exit status into
# $rc BEFORE cleanup runs, then exit with that status
# instead of rm -f's (rm -f always exits 0, which
# would otherwise mask a conversion/playback failure
# and prevent falling through to the next player).
_full_cmd = f"( {_ps_cmd} ); rc=$?; {_cleanup}; exit $rc"
# Use full path so the which(cmd[0]) check in the player loop passes.
players.insert(0, ["/bin/sh", "-c", _full_cmd])
except Exception:
pass # WSL path resolution failed; fall through to ffplay/aplay
players.append(["ffplay", "-nodisp", "-autoexit", "-loglevel", "quiet", file_path])
if system == "Linux":
players.append(["aplay", "-q", file_path])
@ -1268,9 +1368,15 @@ def play_audio_file(file_path: str) -> bool:
with _playback_lock:
_active_playback = proc
proc.wait(timeout=300)
rc = proc.returncode
with _playback_lock:
_active_playback = None
return True
if rc == 0:
return True
# Non-zero exit: player failed (e.g. WSL ffplay/aplay with no
# audio device, or the PowerShell fallback's ffmpeg/playback
# step failing). Fall through to the next player in the list.
logger.debug("System player %s exited with code %d, trying next", cmd[0], rc)
except subprocess.TimeoutExpired:
logger.warning("System player %s timed out, killing process", cmd[0])
proc.kill()