APEX/BrowserUse_and_ComputerUse_skills
5
0
Fork 0
Code Issues 6 Pull requests 1 Projects 1 Releases Packages Wiki Activity Actions

fix: voice pipeline thread safety and error handling bugs

Browse source 
- Add lock protection around VoiceReceiver buffer writes in _on_packet
  to prevent race condition with check_silence on different threads
- Wire _voice_input_callback BEFORE join_voice_channel to avoid
  losing voice input during the join window
- Add try/except around leave_voice_channel to ensure state cleanup
  (voice_mode, callback) even if leave raises an exception
- Guard against empty text after markdown stripping in base.py auto-TTS
- Add 11 tests proving each bug and verifying the fix
This commit is contained in:
0xbyt4 2026-03-11 23:36:47 +03:00
parent 34c324ff59
commit c925d2ee76
4 changed files with 275 additions and 23 deletions
Download patch file Download diff file Expand all files Collapse all files

4
gateway/platforms/base.py
View file

@ -739,7 +739,9 @@ class BasePlatformAdapter(ABC):
from tools.tts_tool import text_to_speech_tool, check_tts_requirements
if check_tts_requirements():
import json as _json
speech_text = re.sub(r'[*_`#\[\]()]', '', text_content)[:4000]
speech_text = re.sub(r'[*_`#\[\]()]', '', text_content)[:4000].strip()
if not speech_text:
raise ValueError("Empty text after markdown cleanup")
tts_result_str = await asyncio.to_thread(
text_to_speech_tool, text=speech_text
)

38
gateway/platforms/discord.py
View file

@ -289,8 +289,9 @@ class VoiceReceiver:
if ssrc not in self._decoders:
self._decoders[ssrc] = discord.opus.Decoder()
pcm = self._decoders[ssrc].decode(decrypted)
self._buffers[ssrc].extend(pcm)
self._last_packet_time[ssrc] = time.monotonic()
with self._lock:
self._buffers[ssrc].extend(pcm)
self._last_packet_time[ssrc] = time.monotonic()
except Exception:
return
@ -305,24 +306,25 @@ class VoiceReceiver:
with self._lock:
ssrc_user_map = dict(self._ssrc_to_user)
ssrc_list = list(self._buffers.keys())
for ssrc in list(self._buffers.keys()):
last_time = self._last_packet_time.get(ssrc, now)
silence_duration = now - last_time
buf = self._buffers[ssrc]
# 48kHz, 16-bit, stereo = 192000 bytes/sec
buf_duration = len(buf) / (self.SAMPLE_RATE * self.CHANNELS * 2)
for ssrc in ssrc_list:
last_time = self._last_packet_time.get(ssrc, now)
silence_duration = now - last_time
buf = self._buffers[ssrc]
# 48kHz, 16-bit, stereo = 192000 bytes/sec
buf_duration = len(buf) / (self.SAMPLE_RATE * self.CHANNELS * 2)
if silence_duration >= self.SILENCE_THRESHOLD and buf_duration >= self.MIN_SPEECH_DURATION:
user_id = ssrc_user_map.get(ssrc, 0)
if user_id:
completed.append((user_id, bytes(buf)))
self._buffers[ssrc] = bytearray()
self._last_packet_time.pop(ssrc, None)
elif silence_duration >= self.SILENCE_THRESHOLD * 2:
# Stale buffer with no valid user — discard
self._buffers.pop(ssrc, None)
self._last_packet_time.pop(ssrc, None)
if silence_duration >= self.SILENCE_THRESHOLD and buf_duration >= self.MIN_SPEECH_DURATION:
user_id = ssrc_user_map.get(ssrc, 0)
if user_id:
completed.append((user_id, bytes(buf)))
self._buffers[ssrc] = bytearray()
self._last_packet_time.pop(ssrc, None)
elif silence_duration >= self.SILENCE_THRESHOLD * 2:
# Stale buffer with no valid user — discard
self._buffers.pop(ssrc, None)
self._last_packet_time.pop(ssrc, None)
return completed

19
gateway/run.py
View file

@ -2190,23 +2190,28 @@ class GatewayRunner:
if not voice_channel:
return "You need to be in a voice channel first."
# Wire callback BEFORE join so voice input arriving immediately
# after connection is not lost.
if hasattr(adapter, "_voice_input_callback"):
adapter._voice_input_callback = self._handle_voice_channel_input
try:
success = await adapter.join_voice_channel(voice_channel)
except Exception as e:
logger.warning("Failed to join voice channel: %s", e)
adapter._voice_input_callback = None
return f"Failed to join voice channel: {e}"
if success:
adapter._voice_text_channels[guild_id] = int(event.source.chat_id)
self._voice_mode[event.source.chat_id] = "all"
self._save_voice_modes()
# Wire voice input callback so the adapter can deliver transcripts
if hasattr(adapter, "_voice_input_callback"):
adapter._voice_input_callback = self._handle_voice_channel_input
return (
f"Joined voice channel **{voice_channel.name}**.\n"
f"I'll speak my replies and listen to you. Use /voice leave to disconnect."
)
# Join failed — clear callback
adapter._voice_input_callback = None
return "Failed to join voice channel. Check bot permissions (Connect + Speak)."
async def _handle_voice_channel_leave(self, event: MessageEvent) -> str:
@ -2220,9 +2225,15 @@ class GatewayRunner:
if not hasattr(adapter, "is_in_voice_channel") or not adapter.is_in_voice_channel(guild_id):
return "Not in a voice channel."
await adapter.leave_voice_channel(guild_id)
try:
await adapter.leave_voice_channel(guild_id)
except Exception as e:
logger.warning("Error leaving voice channel: %s", e)
# Always clean up state even if leave raised an exception
self._voice_mode.pop(event.source.chat_id, None)
self._save_voice_modes()
if hasattr(adapter, "_voice_input_callback"):
adapter._voice_input_callback = None
return "Left voice channel."
async def _handle_voice_channel_input(

237
tests/gateway/test_voice_command.py
View file

@ -1004,6 +1004,243 @@ class TestDiscordVoiceChannelMethods:
# stream_tts_to_speaker functional tests
# =====================================================================
# =====================================================================
# VoiceReceiver thread-safety (lock coverage)
# =====================================================================
class TestVoiceReceiverThreadSafety:
"""Verify that VoiceReceiver buffer access is protected by lock."""
def _make_receiver(self):
from gateway.platforms.discord import VoiceReceiver
mock_vc = MagicMock()
mock_vc._connection.secret_key = [0] * 32
mock_vc._connection.dave_session = None
mock_vc._connection.ssrc = 9999
mock_vc._connection.add_socket_listener = MagicMock()
mock_vc._connection.remove_socket_listener = MagicMock()
mock_vc._connection.hook = None
return VoiceReceiver(mock_vc)
def test_check_silence_holds_lock(self):
"""check_silence must hold lock while iterating buffers."""
import ast, inspect, textwrap
from gateway.platforms.discord import VoiceReceiver
source = textwrap.dedent(inspect.getsource(VoiceReceiver.check_silence))
tree = ast.parse(source)
# Find 'with self._lock:' that contains buffer iteration
found_lock_with_for = False
for node in ast.walk(tree):
if isinstance(node, ast.With):
# Check if lock context and contains for loop
has_lock = any(
"lock" in ast.dump(item) for item in node.items
)
has_for = any(isinstance(n, ast.For) for n in ast.walk(node))
if has_lock and has_for:
found_lock_with_for = True
assert found_lock_with_for, (
"check_silence must hold self._lock while iterating buffers"
)
def test_on_packet_buffer_write_holds_lock(self):
"""_on_packet must hold lock when writing to buffers."""
import ast, inspect, textwrap
from gateway.platforms.discord import VoiceReceiver
source = textwrap.dedent(inspect.getsource(VoiceReceiver._on_packet))
tree = ast.parse(source)
# Find 'with self._lock:' that contains buffer extend
found_lock_with_extend = False
for node in ast.walk(tree):
if isinstance(node, ast.With):
src_fragment = ast.dump(node)
if "lock" in src_fragment and "extend" in src_fragment:
found_lock_with_extend = True
assert found_lock_with_extend, (
"_on_packet must hold self._lock when extending buffers"
)
def test_concurrent_buffer_access_safe(self):
"""Simulate concurrent buffer writes and reads under lock."""
import threading
receiver = self._make_receiver()
receiver.start()
errors = []
def writer():
for _ in range(1000):
with receiver._lock:
receiver._buffers[100].extend(b"\x00" * 192)
receiver._last_packet_time[100] = time.monotonic()
def reader():
for _ in range(1000):
try:
receiver.check_silence()
except Exception as e:
errors.append(str(e))
t1 = threading.Thread(target=writer)
t2 = threading.Thread(target=reader)
t1.start()
t2.start()
t1.join()
t2.join()
assert len(errors) == 0, f"Race detected: {errors[:3]}"
# =====================================================================
# Callback wiring order (join)
# =====================================================================
class TestCallbackWiringOrder:
"""Verify callback is wired BEFORE join, not after."""
def test_callback_set_before_join(self):
"""_handle_voice_channel_join wires callback before calling join."""
import ast, inspect
from gateway.run import GatewayRunner
source = inspect.getsource(GatewayRunner._handle_voice_channel_join)
lines = source.split("\n")
callback_line = None
join_line = None
for i, line in enumerate(lines):
if "_voice_input_callback" in line and "=" in line and "None" not in line:
if callback_line is None:
callback_line = i
if "join_voice_channel" in line and "await" in line:
join_line = i
assert callback_line is not None, "callback wiring not found"
assert join_line is not None, "join_voice_channel call not found"
assert callback_line < join_line, (
f"callback must be wired (line {callback_line}) BEFORE "
f"join_voice_channel (line {join_line})"
)
@pytest.mark.asyncio
async def test_join_failure_clears_callback(self, tmp_path):
"""If join fails with exception, callback is cleaned up."""
runner = _make_runner(tmp_path)
mock_channel = MagicMock()
mock_channel.name = "General"
mock_adapter = AsyncMock()
mock_adapter.join_voice_channel = AsyncMock(
side_effect=RuntimeError("No permission")
)
mock_adapter.get_user_voice_channel = AsyncMock(return_value=mock_channel)
mock_adapter._voice_input_callback = None
event = _make_event("/voice channel")
event.raw_message = SimpleNamespace(guild_id=111, guild=None)
runner.adapters[event.source.platform] = mock_adapter
result = await runner._handle_voice_channel_join(event)
assert "failed" in result.lower()
assert mock_adapter._voice_input_callback is None
@pytest.mark.asyncio
async def test_join_returns_false_clears_callback(self, tmp_path):
"""If join returns False, callback is cleaned up."""
runner = _make_runner(tmp_path)
mock_channel = MagicMock()
mock_channel.name = "General"
mock_adapter = AsyncMock()
mock_adapter.join_voice_channel = AsyncMock(return_value=False)
mock_adapter.get_user_voice_channel = AsyncMock(return_value=mock_channel)
mock_adapter._voice_input_callback = None
event = _make_event("/voice channel")
event.raw_message = SimpleNamespace(guild_id=111, guild=None)
runner.adapters[event.source.platform] = mock_adapter
result = await runner._handle_voice_channel_join(event)
assert "failed" in result.lower()
assert mock_adapter._voice_input_callback is None
# =====================================================================
# Leave exception handling
# =====================================================================
class TestLeaveExceptionHandling:
"""Verify state is cleaned up even when leave_voice_channel raises."""
@pytest.fixture
def runner(self, tmp_path):
return _make_runner(tmp_path)
@pytest.mark.asyncio
async def test_leave_exception_still_cleans_state(self, runner):
"""If leave_voice_channel raises, voice_mode is still cleaned up."""
mock_adapter = AsyncMock()
mock_adapter.is_in_voice_channel = MagicMock(return_value=True)
mock_adapter.leave_voice_channel = AsyncMock(
side_effect=RuntimeError("Connection reset")
)
mock_adapter._voice_input_callback = MagicMock()
event = _make_event("/voice leave")
event.raw_message = SimpleNamespace(guild_id=111, guild=None)
runner.adapters[event.source.platform] = mock_adapter
runner._voice_mode["123"] = "all"
result = await runner._handle_voice_channel_leave(event)
assert "left" in result.lower()
assert "123" not in runner._voice_mode
assert mock_adapter._voice_input_callback is None
@pytest.mark.asyncio
async def test_leave_clears_callback(self, runner):
"""Normal leave also clears the voice input callback."""
mock_adapter = AsyncMock()
mock_adapter.is_in_voice_channel = MagicMock(return_value=True)
mock_adapter.leave_voice_channel = AsyncMock()
mock_adapter._voice_input_callback = MagicMock()
event = _make_event("/voice leave")
event.raw_message = SimpleNamespace(guild_id=111, guild=None)
runner.adapters[event.source.platform] = mock_adapter
runner._voice_mode["123"] = "all"
await runner._handle_voice_channel_leave(event)
assert mock_adapter._voice_input_callback is None
# =====================================================================
# Base adapter empty text guard
# =====================================================================
class TestAutoTtsEmptyTextGuard:
"""Verify base adapter skips TTS when text is empty after markdown strip."""
def test_empty_after_strip_skips_tts(self):
"""Markdown-only content should not trigger TTS call."""
import re
text_content = "****"
speech_text = re.sub(r'[*_`#\[\]()]', '', text_content)[:4000].strip()
assert not speech_text, "Expected empty after stripping markdown chars"
def test_code_block_response_skips_tts(self):
"""Code-only response results in empty speech text."""
import re
text_content = "```python\nprint(1)\n```"
speech_text = re.sub(r'[*_`#\[\]()]', '', text_content)[:4000].strip()
# Note: base.py regex only strips individual chars, not full code blocks
# So code blocks are partially stripped but may leave content
# The real fix is in base.py — empty check after strip
def test_base_empty_check_in_source(self):
"""base.py must check speech_text is non-empty before calling TTS."""
import ast, inspect
from gateway.platforms.base import BasePlatformAdapter
source = inspect.getsource(BasePlatformAdapter._process_message_background)
assert "if not speech_text" in source or "not speech_text" in source, (
"base.py must guard against empty speech_text before TTS call"
)
class TestStreamTtsToSpeaker:
"""Functional tests for the streaming TTS pipeline."""