voice: one macOS output policy across WAV, beeps, streaming TTS + tests

Addresses review on #62601. Applies a single rule — no sounddevice for
audio OUTPUT on macOS (PortAudio/CoreAudio init triggers a
kTCCServiceMediaLibrary prompt) — consistently at all three output sites:

- play_audio_file: WAV playback (already routed to afplay) now uses the
  shared _sounddevice_output_allowed() helper.
- play_beep: synthesize the tone with numpy only, then on macOS play it
  via a temp WAV through afplay instead of sounddevice.
- stream_tts_to_speaker (tts_tool): on macOS, skip the sounddevice
  OutputStream so playback falls through to the existing tempfile/afplay path.

Audio INPUT (recording) is untouched — it legitimately needs mic permission.

Tests: TestMacOSAudioOutputPolicy (voice_mode) proves WAV + beep routing
does not import sounddevice on Darwin and still uses it off Darwin;
test_tts_macos_output proves streaming TTS skips the OutputStream on Darwin.
Existing test_play_wav_via_sounddevice pinned to non-Darwin for determinism.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
simonmmafs 2026-07-11 22:06:15 +02:00 committed by Teknium
parent 0179ff9738
commit a1bc12f191
4 changed files with 294 additions and 19 deletions

View file

@ -0,0 +1,100 @@
"""macOS output policy for streaming TTS.
On macOS, stream_tts_to_speaker must NOT open a sounddevice OutputStream
(PortAudio/CoreAudio init triggers a kTCCServiceMediaLibrary prompt). It
should route audio through the tempfile/afplay fallback instead.
See PR #62601 / #13291.
"""
import queue
import threading
import pytest
def _run_stream(monkeypatch, system_name):
"""Drive stream_tts_to_speaker once with a mock client on *system_name*.
Returns True if _import_sounddevice was called during the run.
"""
import tools.tts_tool as tts
monkeypatch.setattr("tools.tts_tool.platform.system", lambda: system_name)
monkeypatch.setattr("tools.tts_tool.get_env_value",
lambda name, default=None: "fake-key"
if name == "ELEVENLABS_API_KEY" else default)
monkeypatch.setattr("tools.tts_tool._load_tts_config", lambda: {})
class _FakeTTS:
def __init__(self, *a, **k):
self.text_to_speech = self
def convert(self, *a, **k):
return iter([]) # no audio chunks needed for setup assertion
monkeypatch.setattr("tools.tts_tool._import_elevenlabs", lambda: _FakeTTS)
sd_called = {"hit": False}
def _spy_import_sd():
sd_called["hit"] = True
raise AssertionError("sounddevice must not be imported for output on macOS")
monkeypatch.setattr("tools.tts_tool._import_sounddevice", _spy_import_sd)
text_queue: queue.Queue = queue.Queue()
text_queue.put(None) # end-of-text sentinel: no sentence spoken
stop_event = threading.Event()
done_event = threading.Event()
tts.stream_tts_to_speaker(text_queue, stop_event, done_event)
assert done_event.is_set()
return sd_called["hit"]
def test_streaming_tts_skips_sounddevice_on_macos(monkeypatch):
assert _run_stream(monkeypatch, "Darwin") is False
def test_streaming_tts_uses_sounddevice_off_macos(monkeypatch):
# Off macOS the OutputStream setup runs; _import_sounddevice raising here
# is caught by the function's own guard, so the call itself is what we assert.
called = _run_stream_offmac(monkeypatch)
assert called is True
def _run_stream_offmac(monkeypatch):
"""Like _run_stream but tolerant of the sounddevice import being attempted."""
import tools.tts_tool as tts
monkeypatch.setattr("tools.tts_tool.platform.system", lambda: "Linux")
monkeypatch.setattr("tools.tts_tool.get_env_value",
lambda name, default=None: "fake-key"
if name == "ELEVENLABS_API_KEY" else default)
monkeypatch.setattr("tools.tts_tool._load_tts_config", lambda: {})
class _FakeTTS:
def __init__(self, *a, **k):
self.text_to_speech = self
def convert(self, *a, **k):
return iter([])
monkeypatch.setattr("tools.tts_tool._import_elevenlabs", lambda: _FakeTTS)
sd_called = {"hit": False}
def _spy_import_sd():
sd_called["hit"] = True
raise OSError("no audio device in test") # handled by the function's guard
monkeypatch.setattr("tools.tts_tool._import_sounddevice", _spy_import_sd)
text_queue: queue.Queue = queue.Queue()
text_queue.put(None)
stop_event = threading.Event()
done_event = threading.Event()
tts.stream_tts_to_speaker(text_queue, stop_event, done_event)
assert done_event.is_set()
return sd_called["hit"]

View file

@ -1140,6 +1140,10 @@ class TestWhisperHallucinationFilter:
class TestPlayAudioFile:
def test_play_wav_via_sounddevice(self, monkeypatch, sample_wav):
np = pytest.importorskip("numpy")
# Pin to a non-macOS platform: on macOS WAV output deliberately skips
# sounddevice (see TestMacOSAudioOutputPolicy), so this path is only
# exercised off Darwin.
monkeypatch.setattr("tools.voice_mode.platform.system", lambda: "Linux")
mock_sd_obj = MagicMock()
# Simulate stream completing immediately (get_stream().active = False)
@ -1178,6 +1182,106 @@ class TestPlayAudioFile:
assert result is False
# ============================================================================
# macOS output policy (no sounddevice for OUTPUT -> avoids TCC prompt)
# ============================================================================
class TestMacOSAudioOutputPolicy:
def test_output_disallowed_on_macos(self, monkeypatch):
monkeypatch.setattr("tools.voice_mode.platform.system", lambda: "Darwin")
from tools.voice_mode import _sounddevice_output_allowed
assert _sounddevice_output_allowed() is False
def test_output_allowed_off_macos(self, monkeypatch):
monkeypatch.setattr("tools.voice_mode.platform.system", lambda: "Linux")
from tools.voice_mode import _sounddevice_output_allowed
assert _sounddevice_output_allowed() is True
def test_play_audio_file_skips_sounddevice_on_macos(self, monkeypatch, sample_wav):
"""On macOS, WAV playback must not import sounddevice; it routes to afplay."""
monkeypatch.setattr("tools.voice_mode.platform.system", lambda: "Darwin")
def _forbidden_import():
raise AssertionError("sounddevice must not be imported for output on macOS")
monkeypatch.setattr("tools.voice_mode._import_audio", _forbidden_import)
popen_cmds = []
class _FakeProc:
def wait(self, timeout=None):
return 0
def kill(self):
pass
def _fake_popen(cmd, **kwargs):
popen_cmds.append(cmd)
return _FakeProc()
monkeypatch.setattr("shutil.which", lambda exe: f"/usr/bin/{exe}")
monkeypatch.setattr("subprocess.Popen", _fake_popen)
from tools.voice_mode import play_audio_file
result = play_audio_file(sample_wav)
assert result is True
assert popen_cmds, "expected a system player to be invoked"
assert popen_cmds[0][0] == "afplay"
def test_play_beep_routes_through_afplay_on_macos(self, monkeypatch):
"""On macOS, beeps synthesize with numpy but play via the tempfile/afplay path."""
pytest.importorskip("numpy")
monkeypatch.setattr("tools.voice_mode.platform.system", lambda: "Darwin")
def _forbidden_import():
raise AssertionError("sounddevice must not be imported for beeps on macOS")
monkeypatch.setattr("tools.voice_mode._import_audio", _forbidden_import)
calls = []
monkeypatch.setattr(
"tools.voice_mode._play_int16_via_tempfile",
lambda audio, sample_rate: calls.append((len(audio), sample_rate)),
)
import tools.voice_mode as vm
vm.play_beep(frequency=880, count=1)
assert len(calls) == 1
n_samples, sample_rate = calls[0]
assert n_samples > 0
assert sample_rate == vm.SAMPLE_RATE
def test_play_beep_uses_sounddevice_off_macos(self, monkeypatch):
"""Off macOS, beeps go straight through sounddevice."""
np = pytest.importorskip("numpy")
monkeypatch.setattr("tools.voice_mode.platform.system", lambda: "Linux")
mock_sd = MagicMock()
mock_stream = MagicMock()
mock_stream.active = False
mock_sd.get_stream.return_value = mock_stream
monkeypatch.setattr("tools.voice_mode._import_audio", lambda: (mock_sd, np))
tempfile_calls = []
monkeypatch.setattr(
"tools.voice_mode._play_int16_via_tempfile",
lambda audio, sample_rate: tempfile_calls.append(True),
)
import tools.voice_mode as vm
vm.play_beep(frequency=880, count=1)
mock_sd.play.assert_called_once()
assert not tempfile_calls, "off macOS should not use the tempfile/afplay path"
# ============================================================================
# cleanup_temp_recordings
# ============================================================================

View file

@ -41,6 +41,7 @@ import json
import logging
import os
import queue
import platform
import re
import shlex
import shutil
@ -3188,20 +3189,28 @@ def stream_tts_to_speaker(
)
except Exception:
stream_max_len = 0
try:
sd = _import_sounddevice()
output_stream = sd.OutputStream(
samplerate=streamer.sample_rate,
channels=streamer.channels,
dtype="int16",
)
output_stream.start()
except (ImportError, OSError) as exc:
logger.debug("sounddevice not available, streamer→tempfile: %s", exc)
output_stream = None
except Exception as exc:
logger.warning("sounddevice OutputStream failed: %s", exc)
# On macOS, skip the sounddevice OutputStream entirely: PortAudio/
# CoreAudio init triggers a kTCCServiceMediaLibrary permission
# prompt even though output needs no media-library access. Leaving
# output_stream=None routes each sentence through the tempfile
# -> play_audio_file -> afplay path. See PR #62601 / #13291.
if platform.system() == "Darwin":
output_stream = None
else:
try:
sd = _import_sounddevice()
output_stream = sd.OutputStream(
samplerate=streamer.sample_rate,
channels=streamer.channels,
dtype="int16",
)
output_stream.start()
except (ImportError, OSError) as exc:
logger.debug("sounddevice not available, streamer→tempfile: %s", exc)
output_stream = None
except Exception as exc:
logger.warning("sounddevice OutputStream failed: %s", exc)
output_stream = None
chunker = SentenceChunker()
long_flush_len = 100

View file

@ -42,6 +42,55 @@ def _import_audio():
return sd, np
def _import_numpy():
"""Lazy-import numpy only (no sounddevice). Returns the module.
Used where we need to synthesize/convert audio samples but must NOT
import sounddevice see _sounddevice_output_allowed.
"""
import numpy as np
return np
def _sounddevice_output_allowed() -> bool:
"""Whether sounddevice may be used for audio OUTPUT.
Returns False on macOS: importing/initializing sounddevice
(PortAudio/CoreAudio) for output triggers a kTCCServiceMediaLibrary
permission prompt, even though playback needs no media-library access.
On macOS all output is routed through ``afplay`` instead. This does NOT
affect audio *input* (recording), which legitimately needs microphone
permission. See PR #62601 / #13291.
"""
return platform.system() != "Darwin"
def _play_int16_via_tempfile(audio, sample_rate: int) -> None:
"""Write int16 mono PCM to a temp WAV and play it via play_audio_file.
Used on macOS so tone/beep output goes through ``afplay`` instead of
sounddevice (avoids the TCC media-library prompt).
"""
tmp_path = None
try:
tmp = tempfile.NamedTemporaryFile(suffix=".wav", delete=False)
tmp_path = tmp.name
with wave.open(tmp, "wb") as wf:
wf.setnchannels(1)
wf.setsampwidth(2) # 16-bit
wf.setframerate(sample_rate)
wf.writeframes(audio.tobytes())
play_audio_file(tmp_path)
except Exception as e:
logger.debug("Tone tempfile playback failed: %s", e)
finally:
if tmp_path:
try:
os.unlink(tmp_path)
except OSError:
pass
def _audio_available() -> bool:
"""Return True if audio libraries can be imported."""
try:
@ -419,9 +468,11 @@ def play_beep(frequency: int = 880, duration: float = 0.12, count: int = 1) -> N
duration: Duration of each beep in seconds.
count: Number of beeps to play (with short gap between).
"""
# Synthesize the tone with numpy only (no sounddevice import yet, so the
# macOS TCC prompt is not triggered on the synthesis step).
try:
sd, np = _import_audio()
except (ImportError, OSError):
np = _import_numpy()
except ImportError:
return
try:
gap = 0.06 # seconds between beeps
@ -442,6 +493,16 @@ def play_beep(frequency: int = 880, duration: float = 0.12, count: int = 1) -> N
parts.append(np.zeros(samples_per_gap, dtype=np.int16))
audio = np.concatenate(parts)
# On macOS, route the tone through afplay instead of sounddevice.
if not _sounddevice_output_allowed():
_play_int16_via_tempfile(audio, SAMPLE_RATE)
return
try:
sd, _ = _import_audio()
except (ImportError, OSError):
return
sd.play(audio, samplerate=SAMPLE_RATE)
# sd.wait() calls Event.wait() without timeout — hangs forever if the
# audio device stalls. Poll with a 2s ceiling and force-stop.
@ -1306,10 +1367,11 @@ def play_audio_file(file_path: str) -> bool:
logger.warning("Audio file not found: %s", file_path)
return False
# On macOS, skip sounddevice entirely — PortAudio/CoreAudio init triggers
# a kTCCServiceMediaLibrary permission prompt even though we don't need it.
# afplay handles all formats natively without touching the media stack.
if file_path.endswith(".wav") and platform.system() != "Darwin":
# Skip sounddevice for output where it is not allowed (macOS): PortAudio/
# CoreAudio init triggers a kTCCServiceMediaLibrary permission prompt even
# though playback needs no media-library access. afplay (added to the
# system-player list below) handles all formats natively instead.
if file_path.endswith(".wav") and _sounddevice_output_allowed():
try:
sd, np = _import_audio()
with wave.open(file_path, "rb") as wf: