diff --git a/plugins/platforms/discord/adapter.py b/plugins/platforms/discord/adapter.py index ad1a6ab1191f..d93793156ac3 100644 --- a/plugins/platforms/discord/adapter.py +++ b/plugins/platforms/discord/adapter.py @@ -21,10 +21,8 @@ import struct import subprocess import tempfile -import threading import time import traceback -from collections import defaultdict from contextlib import suppress from typing import Callable, Dict, List, Optional, Any, Tuple from urllib.parse import quote, urljoin @@ -140,6 +138,8 @@ def __init__(self, id: int) -> None: # noqa: A002 - matches discord API except ImportError: from ffmpeg_utils import resolve_ffmpeg_executable +from .voice_receiver import VoiceReceiver # re-export (seam) — never del VoiceReceiver + from gateway.config import Platform, PlatformConfig from gateway.platforms.helpers import ( @@ -147,7 +147,7 @@ def __init__(self, id: int) -> None: # noqa: A002 - matches discord API ThreadParticipationTracker, convert_table_to_bullets, ) -from utils import atomic_json_write, env_float, env_int +from utils import atomic_json_write, env_float from gateway.platforms.base import ( BasePlatformAdapter, MessageEvent, @@ -562,391 +562,6 @@ def _discord_ready_timeout_seconds() -> float: return 30.0 -class VoiceReceiver: - """Captures and decodes voice audio from a Discord voice channel. - - Attaches to a VoiceClient's socket listener, decrypts RTP packets - (NaCl transport + DAVE E2EE), decodes Opus to PCM, and buffers - per-user audio. A polling loop detects silence and delivers - completed utterances via a callback. - """ - - SILENCE_THRESHOLD = 1.5 # seconds of silence → end of utterance - MIN_SPEECH_DURATION = 0.5 # minimum seconds to process (skip noise) - SAMPLE_RATE = 48000 # Discord native rate - CHANNELS = 2 # Discord sends stereo - - def __init__(self, voice_client, allowed_user_ids: set = None): - self._vc = voice_client - self._allowed_user_ids = allowed_user_ids or set() - self._running = False - - # Decryption - self._secret_key: Optional[bytes] = None - self._dave_session = None - self._bot_ssrc: int = 0 - - # SSRC -> user_id mapping (populated from SPEAKING events) - self._ssrc_to_user: Dict[int, int] = {} - self._lock = threading.Lock() - - # Per-user audio buffers - self._buffers: Dict[int, bytearray] = defaultdict(bytearray) - self._last_packet_time: Dict[int, float] = {} - - # Opus decoder per SSRC (each user needs own decoder state) - self._decoders: Dict[int, object] = {} - - # Pause flag: don't capture while bot is playing TTS - self._paused = False - - # Debug logging counter (instance-level to avoid cross-instance races) - self._packet_debug_count = 0 - - # ------------------------------------------------------------------ - # Lifecycle - # ------------------------------------------------------------------ - - def start(self): - """Start listening for voice packets.""" - conn = self._vc._connection - self._secret_key = bytes(conn.secret_key) - self._dave_session = conn.dave_session - self._bot_ssrc = conn.ssrc - - self._install_speaking_hook(conn) - conn.add_socket_listener(self._on_packet) - self._running = True - logger.info("VoiceReceiver started (bot_ssrc=%d)", self._bot_ssrc) - - def stop(self): - """Stop listening and clean up.""" - self._running = False - try: - self._vc._connection.remove_socket_listener(self._on_packet) - except Exception: - pass - with self._lock: - self._buffers.clear() - self._last_packet_time.clear() - self._decoders.clear() - self._ssrc_to_user.clear() - logger.info("VoiceReceiver stopped") - - def pause(self): - self._paused = True - - def resume(self): - self._paused = False - - # ------------------------------------------------------------------ - # SSRC -> user_id mapping via SPEAKING opcode hook - # ------------------------------------------------------------------ - - def map_ssrc(self, ssrc: int, user_id: int): - with self._lock: - self._ssrc_to_user[ssrc] = user_id - - def _install_speaking_hook(self, conn): - """Wrap the voice websocket hook to capture SPEAKING events (op 5). - - VoiceConnectionState stores the hook as ``conn.hook`` (public attr). - It is passed to DiscordVoiceWebSocket on each (re)connect, so we - must wrap it on the VoiceConnectionState level AND on the current - live websocket instance. - """ - original_hook = conn.hook - receiver_self = self - - async def wrapped_hook(ws, msg): - if isinstance(msg, dict) and msg.get("op") == 5: - data = msg.get("d", {}) - ssrc = data.get("ssrc") - user_id = data.get("user_id") - if ssrc and user_id: - logger.info("SPEAKING event: ssrc=%d -> user=%s", ssrc, user_id) - receiver_self.map_ssrc(int(ssrc), int(user_id)) - if original_hook: - await original_hook(ws, msg) - - # Set on connection state (for future reconnects) - conn.hook = wrapped_hook - # Set on the current live websocket (for immediate effect) - try: - from discord.utils import MISSING - if hasattr(conn, 'ws') and conn.ws is not MISSING: - conn.ws._hook = wrapped_hook - logger.info("Speaking hook installed on live websocket") - except Exception as e: - logger.warning("Could not install hook on live ws: %s", e) - - # ------------------------------------------------------------------ - # Packet handler (called from SocketReader thread) - # ------------------------------------------------------------------ - - def _on_packet(self, data: bytes): - if not self._running or self._paused: - return - - # Log first few raw packets for debugging - self._packet_debug_count += 1 - if self._packet_debug_count <= 5: - logger.debug( - "Raw UDP packet: len=%d, first_bytes=%s", - len(data), data[:4].hex() if len(data) >= 4 else "short", - ) - - if len(data) < 16: - return - - # RTP version check: top 2 bits must be 10 (version 2). - # Lower bits may vary (padding, extension, CSRC count). - # Payload type (byte 1 lower 7 bits) = 0x78 (120) for voice. - if (data[0] >> 6) != 2 or (data[1] & 0x7F) != 0x78: - if self._packet_debug_count <= 5: - logger.debug("Skipped non-RTP: byte0=0x%02x byte1=0x%02x", data[0], data[1]) - return - - first_byte = data[0] - _, _, seq, timestamp, ssrc = struct.unpack_from(">BBHII", data, 0) - - # Skip bot's own audio - if ssrc == self._bot_ssrc: - return - - # Calculate dynamic RTP header size (RFC 9335 / rtpsize mode) - cc = first_byte & 0x0F # CSRC count - has_extension = bool(first_byte & 0x10) # extension bit - has_padding = bool(first_byte & 0x20) # padding bit (RFC 3550 §5.1) - header_size = 12 + (4 * cc) + (4 if has_extension else 0) - - if len(data) < header_size + 4: # need at least header + nonce - return - - # Read extension length from preamble (for skipping after decrypt) - ext_data_len = 0 - if has_extension: - ext_preamble_offset = 12 + (4 * cc) - ext_words = struct.unpack_from(">H", data, ext_preamble_offset + 2)[0] - ext_data_len = ext_words * 4 - - if self._packet_debug_count <= 10: - with self._lock: - known_user = self._ssrc_to_user.get(ssrc, "unknown") - logger.debug( - "RTP packet: ssrc=%d, seq=%d, user=%s, hdr=%d, ext_data=%d", - ssrc, seq, known_user, header_size, ext_data_len, - ) - - header = bytes(data[:header_size]) - payload_with_nonce = data[header_size:] - - # --- NaCl transport decrypt (aead_xchacha20_poly1305_rtpsize) --- - if len(payload_with_nonce) < 4: - return - nonce = bytearray(24) - nonce[:4] = payload_with_nonce[-4:] - encrypted = bytes(payload_with_nonce[:-4]) - - try: - import nacl.secret # noqa: E402 — delayed import, only in voice path - box = nacl.secret.Aead(self._secret_key) - decrypted = box.decrypt(encrypted, header, bytes(nonce)) - except Exception as e: - if self._packet_debug_count <= 10: - logger.warning("NaCl decrypt failed: %s (hdr=%d, enc=%d)", e, header_size, len(encrypted)) - return - - # Skip encrypted extension data to get the actual opus payload - if ext_data_len and len(decrypted) > ext_data_len: - decrypted = decrypted[ext_data_len:] - - # --- Strip RTP padding (RFC 3550 §5.1) --- - # When the P bit is set, the last payload byte holds the count of - # trailing padding bytes (including itself) that must be removed - # before further processing. Skipping this passes padding-contaminated - # bytes into DAVE/Opus and corrupts inbound audio. - if has_padding: - if not decrypted: - if self._packet_debug_count <= 10: - logger.warning( - "RTP padding bit set but no payload (ssrc=%d)", ssrc, - ) - return - pad_len = decrypted[-1] - if pad_len == 0 or pad_len > len(decrypted): - if self._packet_debug_count <= 10: - logger.warning( - "Invalid RTP padding length %d for payload size %d (ssrc=%d)", - pad_len, len(decrypted), ssrc, - ) - return - decrypted = decrypted[:-pad_len] - if not decrypted: - # Padding consumed entire payload — nothing to decode - return - - # --- DAVE E2EE decrypt --- - if self._dave_session: - with self._lock: - user_id = self._ssrc_to_user.get(ssrc, 0) - if user_id: - try: - import davey - decrypted = self._dave_session.decrypt( - user_id, davey.MediaType.audio, decrypted - ) - except Exception as e: - # Unencrypted passthrough — use NaCl-decrypted data as-is - if "Unencrypted" not in str(e): - if self._packet_debug_count <= 10: - logger.warning("DAVE decrypt failed for ssrc=%d: %s", ssrc, e) - return - # If SSRC unknown (no SPEAKING event yet), skip DAVE and try - # Opus decode directly — audio may be in passthrough mode. - # Buffer will get a user_id when SPEAKING event arrives later. - - # --- Opus decode -> PCM --- - try: - if ssrc not in self._decoders: - self._decoders[ssrc] = discord.opus.Decoder() - pcm = self._decoders[ssrc].decode(decrypted) - with self._lock: - self._buffers[ssrc].extend(pcm) - self._last_packet_time[ssrc] = time.monotonic() - except Exception as e: - with self._lock: - self._decoders.pop(ssrc, None) - logger.debug( - "Opus decode error for SSRC %s; reset decoder: %s", - ssrc, - e, - ) - return - - # ------------------------------------------------------------------ - # Silence detection - # ------------------------------------------------------------------ - - def _infer_user_for_ssrc(self, ssrc: int) -> int: - """Try to infer user_id for an unmapped SSRC. - - When the bot rejoins a voice channel, Discord may not resend - SPEAKING events for users already speaking. If exactly one - allowed user is in the channel, map the SSRC to them. - """ - try: - channel = self._vc.channel - if not channel: - return 0 - bot_id = self._vc.user.id if self._vc.user else 0 - allowed = self._allowed_user_ids - candidates = [ - m.id for m in channel.members - if m.id != bot_id and (not allowed or str(m.id) in allowed) - ] - if len(candidates) == 1: - uid = candidates[0] - self._ssrc_to_user[ssrc] = uid - logger.info("Auto-mapped ssrc=%d -> user=%d (sole allowed member)", ssrc, uid) - return uid - except Exception: - pass - return 0 - - def check_silence(self) -> list: - """Return list of (user_id, pcm_bytes) for completed utterances.""" - now = time.monotonic() - completed = [] - - with self._lock: - ssrc_user_map = dict(self._ssrc_to_user) - ssrc_list = list(self._buffers.keys()) - - 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 not user_id: - # SSRC not mapped (SPEAKING event missing after bot rejoin). - # Infer from allowed users in the voice channel. - user_id = self._infer_user_for_ssrc(ssrc) - 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 - - def flush_pending(self) -> list: - """Return buffered utterances that have not yet reached silence.""" - completed = [] - - with self._lock: - ssrc_user_map = dict(self._ssrc_to_user) - for ssrc, buf in list(self._buffers.items()): - # 48kHz, 16-bit, stereo = 192000 bytes/sec - buf_duration = len(buf) / (self.SAMPLE_RATE * self.CHANNELS * 2) - if buf_duration >= self.MIN_SPEECH_DURATION: - user_id = ssrc_user_map.get(ssrc, 0) - if not user_id: - user_id = self._infer_user_for_ssrc(ssrc) - if user_id: - completed.append((user_id, bytes(buf))) - self._buffers.pop(ssrc, None) - self._last_packet_time.pop(ssrc, None) - - return completed - - # ------------------------------------------------------------------ - # PCM -> WAV conversion (for Whisper STT) - # ------------------------------------------------------------------ - - @staticmethod - def pcm_to_wav(pcm_data: bytes, output_path: str, - src_rate: int = 48000, src_channels: int = 2): - """Convert raw PCM to 16kHz mono WAV via ffmpeg. - - The PCM is fed straight to ffmpeg's stdin, which avoids staging it in a - temp file on every utterance. The WAV is still written to *output_path* - rather than captured from stdout: ffmpeg cannot seek on a pipe, so a - piped WAV carries placeholder (0xFFFFFFFF) RIFF/data sizes that make - strict readers misreport the length. - """ - from hermes_cli._subprocess_compat import windows_hide_flags - - subprocess.run( - [ - resolve_ffmpeg_executable(), "-y", "-loglevel", "error", - "-f", "s16le", - "-ar", str(src_rate), - "-ac", str(src_channels), - "-i", "pipe:0", - "-ar", "16000", - "-ac", "1", - output_path, - ], - input=pcm_data, - check=True, - timeout=10, - # Capture ffmpeg's -loglevel error output so a failure's - # CalledProcessError carries the actual message (parity with - # tools/transcription_tools' ffmpeg call sites) instead of - # "returned non-zero exit status N" with stderr detached. - stderr=subprocess.PIPE, - creationflags=windows_hide_flags(), - ) - - def _read_dm_role_auth_guild() -> Optional[int]: """Return the guild ID opted-in for DM role-based auth, or None. diff --git a/plugins/platforms/discord/voice_receiver.py b/plugins/platforms/discord/voice_receiver.py new file mode 100644 index 000000000000..b39bf9696242 --- /dev/null +++ b/plugins/platforms/discord/voice_receiver.py @@ -0,0 +1,424 @@ +from __future__ import annotations + +""" +Discord voice receiver. + +Captures and decodes voice audio from a Discord voice channel: RTP/NaCl +transport decryption, DAVE E2EE, Opus -> PCM decode, per-user silence +detection, and PCM -> WAV conversion for Whisper STT. + +Extracted verbatim from ``plugins/platforms/discord/adapter.py`` +(god-file slice R1-S1) and re-exported through ``adapter.VoiceReceiver`` +(identity-preserving seam — never delete the re-export). +""" + +import logging +import struct +import subprocess +import sys +import threading +import time +from collections import defaultdict +from pathlib import Path as _Path +from typing import Dict, Optional + +try: + import discord + DISCORD_AVAILABLE = True +except ImportError: + DISCORD_AVAILABLE = False + discord = None + +sys.path.insert(0, str(_Path(__file__).resolve().parents[3])) + +try: + from .ffmpeg_utils import resolve_ffmpeg_executable +except ImportError: + from ffmpeg_utils import resolve_ffmpeg_executable + +logger = logging.getLogger(__name__) + + +class VoiceReceiver: + """Captures and decodes voice audio from a Discord voice channel. + + Attaches to a VoiceClient's socket listener, decrypts RTP packets + (NaCl transport + DAVE E2EE), decodes Opus to PCM, and buffers + per-user audio. A polling loop detects silence and delivers + completed utterances via a callback. + """ + + SILENCE_THRESHOLD = 1.5 # seconds of silence → end of utterance + MIN_SPEECH_DURATION = 0.5 # minimum seconds to process (skip noise) + SAMPLE_RATE = 48000 # Discord native rate + CHANNELS = 2 # Discord sends stereo + + def __init__(self, voice_client, allowed_user_ids: set = None): + self._vc = voice_client + self._allowed_user_ids = allowed_user_ids or set() + self._running = False + + # Decryption + self._secret_key: Optional[bytes] = None + self._dave_session = None + self._bot_ssrc: int = 0 + + # SSRC -> user_id mapping (populated from SPEAKING events) + self._ssrc_to_user: Dict[int, int] = {} + self._lock = threading.Lock() + + # Per-user audio buffers + self._buffers: Dict[int, bytearray] = defaultdict(bytearray) + self._last_packet_time: Dict[int, float] = {} + + # Opus decoder per SSRC (each user needs own decoder state) + self._decoders: Dict[int, object] = {} + + # Pause flag: don't capture while bot is playing TTS + self._paused = False + + # Debug logging counter (instance-level to avoid cross-instance races) + self._packet_debug_count = 0 + + # ------------------------------------------------------------------ + # Lifecycle + # ------------------------------------------------------------------ + + def start(self): + """Start listening for voice packets.""" + conn = self._vc._connection + self._secret_key = bytes(conn.secret_key) + self._dave_session = conn.dave_session + self._bot_ssrc = conn.ssrc + + self._install_speaking_hook(conn) + conn.add_socket_listener(self._on_packet) + self._running = True + logger.info("VoiceReceiver started (bot_ssrc=%d)", self._bot_ssrc) + + def stop(self): + """Stop listening and clean up.""" + self._running = False + try: + self._vc._connection.remove_socket_listener(self._on_packet) + except Exception: + pass + with self._lock: + self._buffers.clear() + self._last_packet_time.clear() + self._decoders.clear() + self._ssrc_to_user.clear() + logger.info("VoiceReceiver stopped") + + def pause(self): + self._paused = True + + def resume(self): + self._paused = False + + # ------------------------------------------------------------------ + # SSRC -> user_id mapping via SPEAKING opcode hook + # ------------------------------------------------------------------ + + def map_ssrc(self, ssrc: int, user_id: int): + with self._lock: + self._ssrc_to_user[ssrc] = user_id + + def _install_speaking_hook(self, conn): + """Wrap the voice websocket hook to capture SPEAKING events (op 5). + + VoiceConnectionState stores the hook as ``conn.hook`` (public attr). + It is passed to DiscordVoiceWebSocket on each (re)connect, so we + must wrap it on the VoiceConnectionState level AND on the current + live websocket instance. + """ + original_hook = conn.hook + receiver_self = self + + async def wrapped_hook(ws, msg): + if isinstance(msg, dict) and msg.get("op") == 5: + data = msg.get("d", {}) + ssrc = data.get("ssrc") + user_id = data.get("user_id") + if ssrc and user_id: + logger.info("SPEAKING event: ssrc=%d -> user=%s", ssrc, user_id) + receiver_self.map_ssrc(int(ssrc), int(user_id)) + if original_hook: + await original_hook(ws, msg) + + # Set on connection state (for future reconnects) + conn.hook = wrapped_hook + # Set on the current live websocket (for immediate effect) + try: + from discord.utils import MISSING + if hasattr(conn, 'ws') and conn.ws is not MISSING: + conn.ws._hook = wrapped_hook + logger.info("Speaking hook installed on live websocket") + except Exception as e: + logger.warning("Could not install hook on live ws: %s", e) + + # ------------------------------------------------------------------ + # Packet handler (called from SocketReader thread) + # ------------------------------------------------------------------ + + def _on_packet(self, data: bytes): + if not self._running or self._paused: + return + + # Log first few raw packets for debugging + self._packet_debug_count += 1 + if self._packet_debug_count <= 5: + logger.debug( + "Raw UDP packet: len=%d, first_bytes=%s", + len(data), data[:4].hex() if len(data) >= 4 else "short", + ) + + if len(data) < 16: + return + + # RTP version check: top 2 bits must be 10 (version 2). + # Lower bits may vary (padding, extension, CSRC count). + # Payload type (byte 1 lower 7 bits) = 0x78 (120) for voice. + if (data[0] >> 6) != 2 or (data[1] & 0x7F) != 0x78: + if self._packet_debug_count <= 5: + logger.debug("Skipped non-RTP: byte0=0x%02x byte1=0x%02x", data[0], data[1]) + return + + first_byte = data[0] + _, _, seq, timestamp, ssrc = struct.unpack_from(">BBHII", data, 0) + + # Skip bot's own audio + if ssrc == self._bot_ssrc: + return + + # Calculate dynamic RTP header size (RFC 9335 / rtpsize mode) + cc = first_byte & 0x0F # CSRC count + has_extension = bool(first_byte & 0x10) # extension bit + has_padding = bool(first_byte & 0x20) # padding bit (RFC 3550 §5.1) + header_size = 12 + (4 * cc) + (4 if has_extension else 0) + + if len(data) < header_size + 4: # need at least header + nonce + return + + # Read extension length from preamble (for skipping after decrypt) + ext_data_len = 0 + if has_extension: + ext_preamble_offset = 12 + (4 * cc) + ext_words = struct.unpack_from(">H", data, ext_preamble_offset + 2)[0] + ext_data_len = ext_words * 4 + + if self._packet_debug_count <= 10: + with self._lock: + known_user = self._ssrc_to_user.get(ssrc, "unknown") + logger.debug( + "RTP packet: ssrc=%d, seq=%d, user=%s, hdr=%d, ext_data=%d", + ssrc, seq, known_user, header_size, ext_data_len, + ) + + header = bytes(data[:header_size]) + payload_with_nonce = data[header_size:] + + # --- NaCl transport decrypt (aead_xchacha20_poly1305_rtpsize) --- + if len(payload_with_nonce) < 4: + return + nonce = bytearray(24) + nonce[:4] = payload_with_nonce[-4:] + encrypted = bytes(payload_with_nonce[:-4]) + + try: + import nacl.secret # noqa: E402 — delayed import, only in voice path + box = nacl.secret.Aead(self._secret_key) + decrypted = box.decrypt(encrypted, header, bytes(nonce)) + except Exception as e: + if self._packet_debug_count <= 10: + logger.warning("NaCl decrypt failed: %s (hdr=%d, enc=%d)", e, header_size, len(encrypted)) + return + + # Skip encrypted extension data to get the actual opus payload + if ext_data_len and len(decrypted) > ext_data_len: + decrypted = decrypted[ext_data_len:] + + # --- Strip RTP padding (RFC 3550 §5.1) --- + # When the P bit is set, the last payload byte holds the count of + # trailing padding bytes (including itself) that must be removed + # before further processing. Skipping this passes padding-contaminated + # bytes into DAVE/Opus and corrupts inbound audio. + if has_padding: + if not decrypted: + if self._packet_debug_count <= 10: + logger.warning( + "RTP padding bit set but no payload (ssrc=%d)", ssrc, + ) + return + pad_len = decrypted[-1] + if pad_len == 0 or pad_len > len(decrypted): + if self._packet_debug_count <= 10: + logger.warning( + "Invalid RTP padding length %d for payload size %d (ssrc=%d)", + pad_len, len(decrypted), ssrc, + ) + return + decrypted = decrypted[:-pad_len] + if not decrypted: + # Padding consumed entire payload — nothing to decode + return + + # --- DAVE E2EE decrypt --- + if self._dave_session: + with self._lock: + user_id = self._ssrc_to_user.get(ssrc, 0) + if user_id: + try: + import davey + decrypted = self._dave_session.decrypt( + user_id, davey.MediaType.audio, decrypted + ) + except Exception as e: + # Unencrypted passthrough — use NaCl-decrypted data as-is + if "Unencrypted" not in str(e): + if self._packet_debug_count <= 10: + logger.warning("DAVE decrypt failed for ssrc=%d: %s", ssrc, e) + return + # If SSRC unknown (no SPEAKING event yet), skip DAVE and try + # Opus decode directly — audio may be in passthrough mode. + # Buffer will get a user_id when SPEAKING event arrives later. + + # --- Opus decode -> PCM --- + try: + if ssrc not in self._decoders: + self._decoders[ssrc] = discord.opus.Decoder() + pcm = self._decoders[ssrc].decode(decrypted) + with self._lock: + self._buffers[ssrc].extend(pcm) + self._last_packet_time[ssrc] = time.monotonic() + except Exception as e: + with self._lock: + self._decoders.pop(ssrc, None) + logger.debug( + "Opus decode error for SSRC %s; reset decoder: %s", + ssrc, + e, + ) + return + + # ------------------------------------------------------------------ + # Silence detection + # ------------------------------------------------------------------ + + def _infer_user_for_ssrc(self, ssrc: int) -> int: + """Try to infer user_id for an unmapped SSRC. + + When the bot rejoins a voice channel, Discord may not resend + SPEAKING events for users already speaking. If exactly one + allowed user is in the channel, map the SSRC to them. + """ + try: + channel = self._vc.channel + if not channel: + return 0 + bot_id = self._vc.user.id if self._vc.user else 0 + allowed = self._allowed_user_ids + candidates = [ + m.id for m in channel.members + if m.id != bot_id and (not allowed or str(m.id) in allowed) + ] + if len(candidates) == 1: + uid = candidates[0] + self._ssrc_to_user[ssrc] = uid + logger.info("Auto-mapped ssrc=%d -> user=%d (sole allowed member)", ssrc, uid) + return uid + except Exception: + pass + return 0 + + def check_silence(self) -> list: + """Return list of (user_id, pcm_bytes) for completed utterances.""" + now = time.monotonic() + completed = [] + + with self._lock: + ssrc_user_map = dict(self._ssrc_to_user) + ssrc_list = list(self._buffers.keys()) + + 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 not user_id: + # SSRC not mapped (SPEAKING event missing after bot rejoin). + # Infer from allowed users in the voice channel. + user_id = self._infer_user_for_ssrc(ssrc) + 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 + + def flush_pending(self) -> list: + """Return buffered utterances that have not yet reached silence.""" + completed = [] + + with self._lock: + ssrc_user_map = dict(self._ssrc_to_user) + for ssrc, buf in list(self._buffers.items()): + # 48kHz, 16-bit, stereo = 192000 bytes/sec + buf_duration = len(buf) / (self.SAMPLE_RATE * self.CHANNELS * 2) + if buf_duration >= self.MIN_SPEECH_DURATION: + user_id = ssrc_user_map.get(ssrc, 0) + if not user_id: + user_id = self._infer_user_for_ssrc(ssrc) + if user_id: + completed.append((user_id, bytes(buf))) + self._buffers.pop(ssrc, None) + self._last_packet_time.pop(ssrc, None) + + return completed + + # ------------------------------------------------------------------ + # PCM -> WAV conversion (for Whisper STT) + # ------------------------------------------------------------------ + + @staticmethod + def pcm_to_wav(pcm_data: bytes, output_path: str, + src_rate: int = 48000, src_channels: int = 2): + """Convert raw PCM to 16kHz mono WAV via ffmpeg. + + The PCM is fed straight to ffmpeg's stdin, which avoids staging it in a + temp file on every utterance. The WAV is still written to *output_path* + rather than captured from stdout: ffmpeg cannot seek on a pipe, so a + piped WAV carries placeholder (0xFFFFFFFF) RIFF/data sizes that make + strict readers misreport the length. + """ + from hermes_cli._subprocess_compat import windows_hide_flags + + subprocess.run( + [ + resolve_ffmpeg_executable(), "-y", "-loglevel", "error", + "-f", "s16le", + "-ar", str(src_rate), + "-ac", str(src_channels), + "-i", "pipe:0", + "-ar", "16000", + "-ac", "1", + output_path, + ], + input=pcm_data, + check=True, + timeout=10, + # Capture ffmpeg's -loglevel error output so a failure's + # CalledProcessError carries the actual message (parity with + # tools/transcription_tools' ffmpeg call sites) instead of + # "returned non-zero exit status N" with stderr detached. + stderr=subprocess.PIPE, + creationflags=windows_hide_flags(), + ) diff --git a/tests/gateway/test_voice_receiver_seam.py b/tests/gateway/test_voice_receiver_seam.py new file mode 100644 index 000000000000..47ce02d55971 --- /dev/null +++ b/tests/gateway/test_voice_receiver_seam.py @@ -0,0 +1,387 @@ +"""Seam-identity + aggressive tests for the VoiceReceiver extraction (R1-S1). + +``plugins/platforms/discord/voice_receiver.py`` holds the VoiceReceiver +class, moved verbatim out of ``plugins/platforms/discord/adapter.py`` +(god-file slice R1-S1, epic #78647) and re-exported through +``adapter.VoiceReceiver`` (identity-preserving seam — three existing test +files bind the adapter module global: test_voice_command.py, +test_voice_channel_flow.py, test_discord_race_polish.py). + +The seam-identity tests pin the regression this extraction is meant to +prevent: adapter must resolve every moved name to the *same object* the +voice_receiver module defines. The aggressive tests exercise the voice +path failure modes: RTP/NaCl decrypt, DAVE E2EE passthrough and failure, +RTP padding-strip edges, SPEAKING-hook chaining, sole-member SSRC +inference, and the pcm_to_wav ffmpeg argv contract. +""" + +import asyncio +import struct +import subprocess +import sys +from types import SimpleNamespace +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest + +from plugins.platforms.discord import adapter +from plugins.platforms.discord import voice_receiver + +# Every member of the moved cluster: class + 12 methods. All must resolve +# through the adapter re-export to the identical object in voice_receiver. +# (Class identity itself is pinned by test_class_is_seam_identical.) +MOVED_MEMBERS = ( + "__init__", + "start", + "stop", + "pause", + "resume", + "map_ssrc", + "_install_speaking_hook", + "_on_packet", + "_infer_user_for_ssrc", + "check_silence", + "flush_pending", + "pcm_to_wav", +) + + +# --------------------------------------------------------------------------- +# Seam identity +# --------------------------------------------------------------------------- + +def test_class_is_seam_identical(): + assert adapter.VoiceReceiver is voice_receiver.VoiceReceiver + assert voice_receiver.VoiceReceiver.__module__ == ( + "plugins.platforms.discord.voice_receiver" + ) + + +def test_all_moved_members_are_seam_identical(): + for name in MOVED_MEMBERS: + assert getattr(adapter.VoiceReceiver, name) is getattr( + voice_receiver.VoiceReceiver, name + ), name + + +def test_class_consts_survive_move(): + for const in ("SILENCE_THRESHOLD", "MIN_SPEECH_DURATION", "SAMPLE_RATE", "CHANNELS"): + assert getattr(adapter.VoiceReceiver, const) == getattr( + voice_receiver.VoiceReceiver, const + ) + assert getattr(adapter.VoiceReceiver, const) is getattr( + voice_receiver.VoiceReceiver, const + ) + + +def test_no_back_import_of_adapter(): + """voice_receiver must import standalone (module-level back-import of + adapter would be a cycle: adapter imports voice_receiver). + + The parent package __init__ imports adapter, so we first import the + package normally (caching adapter + voice_receiver), then None-out the + adapter module and re-execute voice_receiver from scratch: if it (or + anything in its import chain) back-imported adapter, the None entry in + sys.modules makes that import raise ImportError. + """ + code = ( + "import sys\n" + "import plugins.platforms.discord\n" + "import plugins.platforms.discord.voice_receiver\n" + "sys.modules['plugins.platforms.discord.adapter'] = None\n" + "del sys.modules['plugins.platforms.discord.voice_receiver']\n" + "import plugins.platforms.discord.voice_receiver as v\n" + "print(v.VoiceReceiver.__name__)\n" + ) + result = subprocess.run( + [sys.executable, "-c", code], + capture_output=True, + text=True, + timeout=60, + ) + assert result.returncode == 0, result.stderr + assert result.stdout.strip() == "VoiceReceiver" + + +# --------------------------------------------------------------------------- +# Receiver lifecycle +# --------------------------------------------------------------------------- + +def _make_receiver(): + mock_vc = MagicMock() + mock_vc._connection.secret_key = [0] * 32 + mock_vc._connection.dave_session = None + mock_vc._connection.ssrc = 9999 + mock_vc._connection.hook = None + mock_vc._connection.add_socket_listener = MagicMock() + mock_vc._connection.remove_socket_listener = MagicMock() + return voice_receiver.VoiceReceiver(mock_vc) + + +def test_start_installs_listener_and_hook(): + receiver = _make_receiver() + receiver.start() + conn = receiver._vc._connection + assert receiver._running is True + assert receiver._bot_ssrc == 9999 + conn.add_socket_listener.assert_called_once_with(receiver._on_packet) + # Speaking hook chained onto the connection state + assert conn.hook is not None + assert conn.hook.__name__ == "wrapped_hook" + + +def test_stop_cleans_up(): + receiver = _make_receiver() + receiver.start() + receiver._buffers[1111] = bytearray(b"\x01" * 64) + receiver._last_packet_time[1111] = 1.0 + receiver._decoders[1111] = object() + receiver._ssrc_to_user[1111] = 42 + receiver.stop() + assert receiver._running is False + receiver._vc._connection.remove_socket_listener.assert_called_once_with( + receiver._on_packet + ) + assert receiver._buffers == {} + assert receiver._last_packet_time == {} + assert receiver._decoders == {} + assert receiver._ssrc_to_user == {} + + +def test_pause_resume_gate_packet_processing(): + receiver = _make_receiver() + receiver.start() + receiver.pause() + assert receiver._paused is True + receiver._on_packet(b"\x80\x78" + b"\x00" * 20) + assert len(receiver._buffers) == 0 + receiver.resume() + assert receiver._paused is False + + +# --------------------------------------------------------------------------- +# Speaking hook (op-5 SPEAKING) chaining +# --------------------------------------------------------------------------- + +def test_speaking_hook_maps_ssrc_and_chains_original(): + receiver = _make_receiver() + original = AsyncMock() + receiver._vc._connection.hook = original + receiver.start() + hooked = receiver._vc._connection.hook + msg = {"op": 5, "d": {"ssrc": 1234, "user_id": 5678}} + asyncio.run(hooked(None, msg)) + assert receiver._ssrc_to_user[1234] == 5678 + original.assert_awaited_once_with(None, msg) + + +def test_speaking_hook_passthrough_non_speaking_op(): + receiver = _make_receiver() + original = AsyncMock() + receiver._vc._connection.hook = original + receiver.start() + hooked = receiver._vc._connection.hook + msg = {"op": 3, "d": {"ssrc": 999, "user_id": 1}} + asyncio.run(hooked(None, msg)) + assert 999 not in receiver._ssrc_to_user + original.assert_awaited_once_with(None, msg) + + +# --------------------------------------------------------------------------- +# Packet path: NaCl decrypt, DAVE E2EE, RTP padding edges +# --------------------------------------------------------------------------- + +class _FakeAead: + """nacl.secret.Aead stand-in. decrypt() returns ``plaintext`` verbatim.""" + + def __init__(self, key): + self.key = key + + def decrypt(self, encrypted, header, nonce): + return _FakeAead._plaintext + + +def _rtp_packet(ssrc, seq=42, payload=b"\xaa" * 40, flags=0x80): + header = struct.pack(">BBHII", flags, 0x78, seq, 1000, ssrc) + return header + payload + b"\xbb\xbb\xbb\xbb" + + +def _patched_receiver(decrypted_payload, **kwargs): + receiver = _make_receiver() + receiver.start() + _FakeAead._plaintext = decrypted_payload + fake_decoder = MagicMock() + fake_decoder.decode.return_value = b"\x00\x01" * 48 # 96 bytes PCM + fake_discord = MagicMock() + fake_discord.opus.Decoder.return_value = fake_decoder + fake_nacl_secret = SimpleNamespace(Aead=_FakeAead) + patches = [ + patch.dict( + sys.modules, + { + "nacl": SimpleNamespace(secret=fake_nacl_secret), + "nacl.secret": fake_nacl_secret, + }, + ), + patch.object(voice_receiver, "discord", fake_discord), + ] + if "dave_session" in kwargs: + patches.append( + patch.object(receiver, "_dave_session", kwargs["dave_session"]) + ) + for p in patches: + p.start() + receiver._patches = patches + receiver._fake_decoder = fake_decoder + return receiver + + +def test_on_packet_decrypts_buffers_pcm(): + ssrc = 1111 + receiver = _patched_receiver(b"\x01" * 40) + receiver.map_ssrc(ssrc, 42) + receiver._on_packet(_rtp_packet(ssrc)) + assert receiver._buffers[ssrc] == bytearray(b"\x00\x01" * 48) + assert ssrc in receiver._last_packet_time + for p in receiver._patches: + p.stop() + + +def test_on_packet_skips_bot_own_ssrc(): + receiver = _patched_receiver(b"\x01" * 40) + receiver._on_packet(_rtp_packet(9999)) # bot ssrc + assert receiver._buffers == {} + for p in receiver._patches: + p.stop() + + +def test_on_packet_skips_non_rtp_and_short_packets(): + receiver = _patched_receiver(b"\x01" * 40) + receiver._on_packet(b"\x00\x78" + b"\x00" * 20) # version != 2 + receiver._on_packet(b"\x80\x78" + b"\x00" * 10) # too short + assert receiver._buffers == {} + for p in receiver._patches: + p.stop() + + +def test_on_packet_drops_bad_rtp_padding(): + ssrc = 2222 + # Padding bit set; decrypted payload's last byte is the pad count. + # pad_len == 0 -> invalid -> drop. + receiver = _patched_receiver(b"\x01" * 40 + b"\x00") + receiver._on_packet(_rtp_packet(ssrc, flags=0xA0)) + assert receiver._buffers == {} + for p in receiver._patches: + p.stop() + + +def test_on_packet_strips_valid_rtp_padding(): + ssrc = 3333 + # pad_len == 4 (valid): 4 trailing bytes stripped before Opus decode. + receiver = _patched_receiver(b"\x01" * 40 + b"\x04") + receiver.map_ssrc(ssrc, 7) + receiver._on_packet(_rtp_packet(ssrc, flags=0xA0)) + assert receiver._buffers[ssrc] == bytearray(b"\x00\x01" * 48) + for p in receiver._patches: + p.stop() + + +def test_on_packet_dave_passthrough_on_unencrypted(): + ssrc = 4444 + dave_session = MagicMock() + dave_session.decrypt.side_effect = Exception("Unencrypted media") + receiver = _patched_receiver( + b"\x01" * 40, dave_session=dave_session + ) + receiver.map_ssrc(ssrc, 9) + with patch.dict(sys.modules, {"davey": SimpleNamespace( + MediaType=SimpleNamespace(audio="audio"))}): + receiver._on_packet(_rtp_packet(ssrc)) + assert receiver._buffers[ssrc] == bytearray(b"\x00\x01" * 48) + for p in receiver._patches: + p.stop() + + +def test_on_packet_drops_dave_hard_failure(): + ssrc = 5555 + dave_session = MagicMock() + dave_session.decrypt.side_effect = RuntimeError("E2EE exploded") + receiver = _patched_receiver( + b"\x01" * 40, dave_session=dave_session + ) + receiver.map_ssrc(ssrc, 9) + with patch.dict(sys.modules, {"davey": SimpleNamespace( + MediaType=SimpleNamespace(audio="audio"))}): + receiver._on_packet(_rtp_packet(ssrc)) + assert receiver._buffers == {} + for p in receiver._patches: + p.stop() + + +# --------------------------------------------------------------------------- +# Silence detection / SSRC inference +# --------------------------------------------------------------------------- + +def test_infer_user_for_ssrc_sole_allowed_member(): + receiver = _make_receiver() + receiver._allowed_user_ids = {"42"} + channel = MagicMock() + bot = SimpleNamespace(id=999) + member = SimpleNamespace(id=42) + receiver._vc.channel = channel + receiver._vc.user = bot + channel.members = [bot, member] + assert receiver._infer_user_for_ssrc(77) == 42 + assert receiver._ssrc_to_user[77] == 42 + + +def test_infer_user_for_ssrc_ambiguous_returns_zero(): + receiver = _make_receiver() + channel = MagicMock() + receiver._vc.channel = channel + receiver._vc.user = SimpleNamespace(id=999) + channel.members = [SimpleNamespace(id=1), SimpleNamespace(id=2)] + assert receiver._infer_user_for_ssrc(77) == 0 + assert 77 not in receiver._ssrc_to_user + + +# --------------------------------------------------------------------------- +# pcm_to_wav ffmpeg argv contract +# --------------------------------------------------------------------------- + +def test_pcm_to_wav_ffmpeg_argv_contract(): + run = MagicMock() + with patch.object(voice_receiver, "resolve_ffmpeg_executable", + return_value="/fake/ffmpeg"), \ + patch.object(voice_receiver.subprocess, "run", run), \ + patch("hermes_cli._subprocess_compat.windows_hide_flags", + return_value=0): + voice_receiver.VoiceReceiver.pcm_to_wav( + b"\x00\x01" * 100, "/tmp/out.wav" + ) + run.assert_called_once() + args, kwargs = run.call_args + assert args[0] == [ + "/fake/ffmpeg", "-y", "-loglevel", "error", + "-f", "s16le", "-ar", "48000", "-ac", "2", + "-i", "pipe:0", "-ar", "16000", "-ac", "1", + "/tmp/out.wav", + ] + assert kwargs["input"] == b"\x00\x01" * 100 + assert kwargs["check"] is True + assert kwargs["timeout"] == 10 + assert kwargs["stderr"] == subprocess.PIPE + assert kwargs["creationflags"] == 0 + + +def test_pcm_to_wav_propagates_ffmpeg_failure(): + run = MagicMock( + side_effect=subprocess.CalledProcessError(1, "ffmpeg", stderr=b"boom") + ) + with patch.object(voice_receiver, "resolve_ffmpeg_executable", + return_value="/fake/ffmpeg"), \ + patch.object(voice_receiver.subprocess, "run", run), \ + patch("hermes_cli._subprocess_compat.windows_hide_flags", + return_value=0): + with pytest.raises(subprocess.CalledProcessError): + voice_receiver.VoiceReceiver.pcm_to_wav(b"\x00", "/tmp/out.wav")