Voice API

Join voice channels, receive and send audio, record and play

from cordslite.core import *
from cordslite.gateway import *
from contextlib import suppress
from fastcore.utils import *
from pathlib import Path
from nacl.bindings import crypto_aead_xchacha20poly1305_ietf_decrypt as xchacha_decrypt
from nacl.bindings import crypto_aead_xchacha20poly1305_ietf_encrypt as xchacha_encrypt
from nacl.exceptions import CryptoError

import asyncio,davey,ffmpeg,json,os,random,struct,time
import websockets
try: import opuslib_next
except: print("Failed to import opuslib-next")

dc = DiscordClient()
gid = '1493461895615873044'
gld = await dc.guild(gid)
chs = await gld.channels()
intents = (1 << 0) | (1 << 7) | (1 << 9) | (1 << 15) # GUILDS | VOICE_EVENTS | GUILD_MESSAGES | MESSAGE_CONTENT
gc = GatewayClient(intents, dc)

await gc.start(debug=True)
vch = next(c for c in chs if c.id == '1493461896139903029'); vch

Channel(id=1327046393453613080, name='General', type=2)

DEBUG: Received Opcode: 10
DEBUG: Received Opcode: 0
DEBUG: Received Opcode: 0
DEBUG: Received Opcode: 0
DEBUG: Received Opcode: 0
DEBUG: Received Opcode: 0
DEBUG: Received Opcode: 11
DEBUG: Received Opcode: 0
DEBUG: Received Opcode: 11
DEBUG: Received Opcode: 0
DEBUG: Received Opcode: 0
DEBUG: Received Opcode: 0
DEBUG: Received Opcode: 0
DEBUG: Received Opcode: 11
DEBUG: Received Opcode: 0
DEBUG: Received Opcode: 11
DEBUG: Received Opcode: 11
DEBUG: Received Opcode: 0
DEBUG: Received Opcode: 11
DEBUG: Received Opcode: 0
DEBUG: Received Opcode: 0
DEBUG: Received Opcode: 0
DEBUG: Received Opcode: 0
DEBUG: Received Opcode: 11
DEBUG: Received Opcode: 0

Voice requires three simultaneous connections:

1. Main Gateway WebSocket (gc) — to request joining a voice channel and receive voice server info
2. Voice Gateway WebSocket — a separate WebSocket to a dedicated voice server for session coordination
3. Voice UDP — a UDP (datagram) connection for actual audio data. UDP is used over TCP because real-time audio needs low latency more than guaranteed delivery

VoiceClient manages the voice gateway WebSocket and UDP connections for a single voice channel, coordinated through the main GatewayClient.

class VoiceClient:
    def __init__(self, gc, gid, ch):
        store_attr()
        self.state_ready = asyncio.Event()
        self.server_ready = asyncio.Event()
        self.rdy = asyncio.Event()
        self.sess = asyncio.Event()

        self.session_id = self.token = self.endpoint = None
        self.ws = self.udp = None
        self._listen_task = self._hb_task = self._ka_task = None
        self.v_seq = -1
        self._tries = 0
        self.resuming = self._rejoining = False
        self.resumed = asyncio.Event()

        self.ssrc_to_user = {}
        self.decoders = {}
        self.dave_pending_transitions = {}
        self.dave_session = None
        self.dave_protocol_version = 0

        self.gc.on("VOICE_STATE_UPDATE", self.on_state_update)
        self.gc.on("VOICE_SERVER_UPDATE", self.on_server_update)
    
    async def on_state_update(self, data):
        if data["user_id"] != self.gc.user_id: return
        self.session_id = data["session_id"]
        ch = data.get("channel_id")
        if ch is None:
            if getattr(self, 'running', False) and not self._rejoining:
                print('removed from voice channel; disconnecting')
                await self.disconnect()
            return
        if ch != self.ch.id: self.ch['id'] = ch # moved to another channel
        self.state_ready.set()

    async def on_server_update(self, data):
        if data["guild_id"] != self.gid: return
        tok,ep = data["token"],data["endpoint"]
        if ep and ep.startswith("wss://"): ep = ep[6:]
        moved = self.server_ready.is_set() and (ep,tok) != (self.endpoint,self.token)
        self.token,self.endpoint = tok,ep
        if not ep: return self.server_ready.clear() # server went away; a follow-up update names the new one
        self.server_ready.set()
        if moved and getattr(self, 'running', False):
            asyncio.create_task(self._reconnect(resume=False, rejoin=False))

    def __repr__(self): return f'VoiceClient({self.ch=})'

vc = VoiceClient(gc, gid, vch); vc

VoiceClient(self.ch=Channel(id=1327046393453613080, name='General', type=2))

Voice requires additional Op opcodes beyond the main gateway’s identify/heartbeat. These handle the voice-specific protocol: requesting to join a channel (voice_state), authenticating with the voice server (voice_identify), negotiating encryption (select_protocol), signaling audio intent (speaking), and keeping the voice connection alive (voice_heartbeat).

The voice heartbeat uses a different format from the main gateway — v8 requires seq_ack tracking the last received sequence number. It must start immediately after receiving Hello, before UDP setup, or the voice WebSocket will time out (close code 4006).

@patch(cls_method=True)
def voice_state(cls:Op, guild_id, channel_id):
    return cls(op=4, d=AttrDict(guild_id=guild_id, channel_id=channel_id, self_mute=False, self_deaf=False))
@patch(cls_method=True)
def voice_identify(cls:Op, server_id, user_id, session_id, token):
    return cls(op=0, d=AttrDict(server_id=server_id, user_id=user_id,
                                session_id=session_id, token=token,
                                max_dave_protocol_version=davey.DAVE_PROTOCOL_VERSION))
@patch(cls_method=True)
def select_protocol(cls:Op, ip, port, mode='aead_xchacha20_poly1305_rtpsize'):
    return cls(op=1, d=AttrDict(protocol='udp', data=dict(address=ip, port=port, mode=mode)))
@patch(cls_method=True)
def speaking(cls:Op, ssrc, speaking=0):
    return cls(op=5, d=AttrDict(speaking=speaking, delay=0, ssrc=ssrc))
@patch(cls_method=True)
def voice_heartbeat(cls:Op, seq_ack=-1):
    return cls(op=3, d=AttrDict(t=int(time.time() * 1000), seq_ack=seq_ack))
@patch(cls_method=True)
def voice_resume(cls:Op, server_id, session_id, token, seq_ack=-1):
    return cls(op=7, d=AttrDict(server_id=server_id, session_id=session_id, token=token, seq_ack=seq_ack))

@patch
async def _join(self:VoiceClient, timeout=10):
    self.state_ready.clear()
    self.server_ready.clear()
    await self.gc.ws.send(Op.voice_state(self.gid, self.ch.id))
    await asyncio.wait_for( asyncio.gather(self.state_ready.wait(), self.server_ready.wait()),
                            timeout)

# await vc._join()

Op 2 - Ready Event

IP Discovery: Discord needs your external IP/port (what the internet sees after NAT), not your local one. We send a 74-byte UDP request containing our SSRC, and Discord responds with our external address filled in:

Field	Size	Description
Type	2 bytes	1=request, 2=response
Length	2 bytes	70 (size of remaining fields)
SSRC	4 bytes	Our audio stream identifier
Address	64 bytes	Blank in request; our external IP in response
Port	2 bytes	Blank in request; our external port in response

Note: Discord requires a Speaking event (even with speaking=0) before it will send audio packets to you.

async def _ip(trans, proto, ssrc):
    pkt = bytearray(74)
    pkt[0:2],pkt[2:4],pkt[4:8] = (1).to_bytes(2,'big'),(70).to_bytes(2,'big'),ssrc.to_bytes(4,'big')
    trans.sendto(pkt)
    r = await proto.packets.get()
    return r[8:72].decode('utf-8').rstrip('\x00'), int.from_bytes(r[72:74], 'big')

class VoiceUDP(asyncio.DatagramProtocol):
    def __init__(self): self.packets = asyncio.Queue()
    def datagram_received(self, data, addr): self.packets.put_nowait(data)

@patch
async def _keepalive(self:VoiceClient, interval=5):
    "Ping the voice UDP socket so NAT mappings survive listen-only sessions"
    n = 0
    while self.running:
        try: self.trans.sendto(n.to_bytes(8, 'big'))
        except Exception: pass
        n = (n + 1) & 0xffffffffffffffff
        await asyncio.sleep(interval)

@patch
async def _udp(self:VoiceClient):
    self.loop = asyncio.get_running_loop()
    self.trans, self.proto = await self.loop.create_datagram_endpoint(VoiceUDP, remote_addr=(self.vip, self.vport))
    self.ip, self.port = await _ip(self.trans, self.proto, self.ssrc)
    if self._ka_task: self._ka_task.cancel()
    self._ka_task = asyncio.create_task(self._keepalive())
    await self.ws.send(Op.select_protocol(self.ip, self.port))

@patch
async def _handle_rdy(self:VoiceClient, d):
    self.ssrc,self.vip,self.vport,self.modes = d["ssrc"],d["ip"],d["port"],d["modes"]
    self.rdy.set()
    await self._udp()

One more thing the UDP path needs: while a bot only listens (like a recorder), it never sends a packet after IP discovery, so NAT/conntrack mappings expire after a few minutes and inbound audio silently stops while the websocket stays healthy. A tiny 8-byte counter ping every 5s (the same trick discord.js uses) keeps the mapping alive; the voice server ignores unknown packets.

Op 4 - Session Description

spf = 960
sr = 48_000
n_chs = 2

@patch
async def _dave(self:VoiceClient):
    if self.dave_version > 0:
        if hasattr(self, 'dave'): self.dave.reinit(self.dave_version, int(self.gc.user_id), int(self.ch.id))
        else: self.dave = davey.DaveSession(self.dave_version, int(self.gc.user_id), int(self.ch.id))
        await self.ws.send(bytes([26]) + self.dave.get_serialized_key_package())
    elif getattr(self, 'dave', None):
        self.dave.reset()
        self.dave.set_passthrough_mode(True, 10)

@patch
async def _handle_sess_desc(self:VoiceClient, d):
    self.mode,self.secret_key = d["mode"],bytes(d["secret_key"])
    self.dave_version = d.get("dave_protocol_version", 0)
    await self._dave()
    await self.ws.send(Op.speaking(self.ssrc, 0))
    self.send_seq = random.randrange(0, 65536)
    self.send_ts = random.randrange(0, 2**32)
    self.send_nonce = 0
    self.encoder = opuslib_next.Encoder(sr, n_chs, opuslib_next.APPLICATION_AUDIO)
    self._tries = 0
    self.sess.set()

@patch
def execute_transition(self:VoiceClient, tid):
    if tid not in self.dave_pending_transitions: return
    self.dave_version = self.dave_pending_transitions.pop(tid)

@patch
async def _handle_trans(self:VoiceClient, op, d):
    if op == 21:  # DAVE_PREPARE_TRANSITION
        tid = d["transition_id"]
        self.dave_pending_transitions[tid] = d["protocol_version"]
        if tid == 0: self.execute_transition(tid)
        else:
            if d["protocol_version"] == 0 and self.dave:
                self.dave.set_passthrough_mode(True, 120)
            await self.ws.send({"op": 23, "d": {"transition_id": tid}})
    elif op == 22:  # DAVE_EXECUTE_TRANSITION
        self.execute_transition(d["transition_id"])
    elif op == 13:  # CLIENT_DISCONNECT: a user left, drop their decode state
        uid = int(d["user_id"])
        self.decoders.pop(uid, None)
        self.last_ts = {k:v for k,v in getattr(self, 'last_ts', {}).items() if self.ssrc_to_user.get(k) != uid}
        self.last_seq = {k:v for k,v in getattr(self, 'last_seq', {}).items() if self.ssrc_to_user.get(k) != uid}
        self.ssrc_to_user = {k:v for k,v in self.ssrc_to_user.items() if v != uid}
    elif op == 24:  # DAVE_PREPARE_EPOCH
        if d["epoch"] == 1:
            self.dave_version = d["protocol_version"]
            await self._dave()
    else: print("voice json", op, d)

@patch
async def _handle_mls(self:VoiceClient, msg):
    self.v_seq = int.from_bytes(msg[:2], "big")
    op,data = msg[2],msg[3:]

    if op == 25: self.dave.set_external_sender(data)
    elif op == 27:
        typ = data[0]
        res = self.dave.process_proposals(davey.ProposalsOperationType.append if typ == 0
                                                  else davey.ProposalsOperationType.revoke, data[1:])
        if isinstance(res, davey.CommitWelcome):
            await self.ws.send(bytes([28]) + res.commit + (res.welcome or b""))
    elif op == 29:
        tid = int.from_bytes(data[:2], "big")
        self.dave.process_commit(data[2:])
        if tid:
            self.dave_pending_transitions[tid] = self.dave_version
            await self.ws.send({"op": 23, "d": {"transition_id": tid}})
    elif op == 30:
        tid = int.from_bytes(data[:2], "big")
        self.dave.process_welcome(data[2:])
        if tid:
            self.dave_pending_transitions[tid] = self.dave_version
            await self.ws.send({"op": 23, "d": {"transition_id": tid}})
    else: print("voice bin", self.v_seq, op, len(msg))

Op 13 (CLIENT_DISCONNECT) tells us a user left the channel — without dropping their decoder and SSRC mapping, a rejoining user could inherit stale jitter state (and the maps grow forever in long sessions).

fvc = object.__new__(VoiceClient)
fvc.decoders,fvc.ssrc_to_user,fvc.last_ts = {5:'dec'},{111:5, 222:9},{111:1, 222:2}
await fvc._handle_trans(13, dict(user_id='5'))
assert fvc.decoders == {} and fvc.ssrc_to_user == {222:9} and fvc.last_ts == {222:2}

The voice gateway runs on two cooperating loops. _hb sends heartbeats at the interval Discord specifies and watches for ACK responses; a missed ACK closes with code 4000, which preserves the session for a resume attempt. _reconnect uses one path for both flavors: a resume keeps session_id, token, endpoint, UDP, and DAVE state intact, while a fresh reconnect resets the audio session and re-runs _join to get new server info from the main gateway. Because _listen schedules _reconnect as its own task, _reconnect must never cancel the task it is running in (that was a bug that killed reconnects mid-flight), so it only cancels _listen_task when called from elsewhere. Failed connection attempts retry with exponential backoff (capped at 30s), and the counter resets once a session is established or resumed. Recording state is preserved in either path, so per-user files keep accumulating across reconnects with only a silence gap during the swap.

@patch
async def _hb(self:VoiceClient):
    await asyncio.sleep(self.hb_int * random.random()) # jitter
    while self.running:
        self._got_ack = False
        try: await self.ws.send(Op.voice_heartbeat(self.v_seq))
        except Exception: return
        await asyncio.sleep(self.hb_int)
        if not self._got_ack: return await self.ws.close(code=4000)

@patch
async def _open_ws(self:VoiceClient):
    self.ws = await websockets.connect(f"wss://{self.endpoint}/?v=8", max_size=None)
    self._listen_task = asyncio.create_task(self._listen())

@patch
def reset_audio(self:VoiceClient):
    self.decoders,self.last_seq,self.last_ts,self.ssrc_to_user,self.dave_pending_transitions = {},{},{},{},{}

@patch
async def _reconnect(self:VoiceClient, resume=True, rejoin=None):
    if rejoin is None: rejoin = not resume
    self.resuming = resume
    if resume: self.resumed.clear()
    t = self._listen_task
    if t and t is not asyncio.current_task(): t.cancel()
    if self._hb_task: self._hb_task.cancel()
    if self.ws:
        try: await self.ws.close(code=4000 if resume else 1000)
        except Exception: pass
    if not resume:
        if getattr(self, 'trans', None): self.trans.close()
        self.reset_audio()
        self.v_seq = -1
        self.sess.clear(); self.rdy.clear()
    if rejoin:
        self._rejoining = True
        while self.running: # the main gateway may itself be mid-reconnect; keep trying until it can carry us
            try:
                self.state_ready.clear(); self.server_ready.clear()
                await self.gc.ws.send(Op.voice_state(self.gid, None))
                await asyncio.sleep(0.5)
                await self._join()
                break
            except Exception as e:
                print('voice rejoin failed, retrying:', repr(e))
                await asyncio.sleep(min(2 ** self._tries, 30))
                self._tries += 1
        self._rejoining = False
    while self.running:
        await asyncio.sleep(min(2 ** self._tries, 30))
        self._tries += 1
        try: return await self._open_ws()
        except OSError as e: print('voice reconnect failed:', repr(e))

@patch
async def _wait_dave_ready(self:VoiceClient, timeout=5):
    start = time.time()
    while getattr(self, 'dave', None) and self.dave.epoch and not self.dave.ready: # no epoch = no E2EE to wait for
        if time.time() - start > timeout: raise TimeoutError("DAVE not ready after resume")
        await asyncio.sleep(0.05)

The listener is where the voice gateway’s state machine actually runs. HELLO (op 8) branches on self.resuming to choose between voice_resume and voice_identify, which is how a single reconnect path supports both flavors. Opcode 9 confirms resume at the gateway level, but media encryption is a separate handshake; DAVE has to catch up before outgoing audio can be encrypted again, so we wait on _wait_dave_ready before setting resumed.

When the socket closes, the close code decides the recovery per the docs: the auth/session family (4003–4006, 4009, 4011) means this session’s credentials are dead, and since voice tokens are minted per-session by the gateway, the right move is a full rejoin with fresh ones — resuming just loops (live testing showed a bad resume earns 4003 Not authenticated, not the 4006 you might expect). 4014 means we were kicked/moved (or the voice server is being swapped) so we must not touch this endpoint again and instead wait for gateway events. Anything else (like 4015, voice server crashed) is transient so we resume. Recovery runs as a separate task so the dying listener never cancels its own reconnect. Handler errors are logged and skipped rather than silently killing the listener.

@patch
async def _listen(self:VoiceClient):
    while self.running:
        try:
            msg = await self.ws.recv()
            if isinstance(msg, bytes):
                await self._handle_mls(msg)
                continue

            msg = dict2obj(json.loads(msg))
            if (seq := msg.get("seq")) is not None: self.v_seq = seq
            op, d = msg.op, msg.d
            if op == 2: await self._handle_rdy(d)
            elif op == 4: await self._handle_sess_desc(d)
            elif op == 5: # SPEAKING
                if "ssrc" in d and "user_id" in d: self.ssrc_to_user[d["ssrc"]] = int(d["user_id"])
            elif op == 6: self._got_ack = True
            elif op == 8: # HELLO
                self.hb_int = d.heartbeat_interval / 1_000
                if self._hb_task: self._hb_task.cancel()
                self._hb_task = asyncio.create_task(self._hb())
                if self.resuming: await self.ws.send(Op.voice_resume(self.gid, self.session_id, self.token, self.v_seq))
                else: await self.ws.send(Op.voice_identify(self.gid, self.gc.user_id, self.session_id, self.token))
            elif op == 9:
                self.resuming = False
                self._tries = 0
                await self._wait_dave_ready()
                self.resumed.set()
            elif op >= 13: await self._handle_trans(op, d)
            else: print("voice json", op, d)
        except websockets.exceptions.ConnectionClosed as e:
            if not self.running: break
            print(f'voice gateway closed: {e.code} {e.reason}')
            if e.code == 4014: break # kicked/moved or server swap: only gateway events can revive us
            asyncio.create_task(self._reconnect(resume=e.code not in (4003, 4004, 4005, 4006, 4009, 4011)))
            break
        except Exception as e:
            print('voice listen error:', repr(e))

@patch
async def join(self:VoiceClient, debug=False):
    self.debug = debug
    self.running = True
    try: await self._join()
    except TimeoutError: # Discord may still hold a stale voice state (unclean shutdown): detach, then join again
        self._rejoining = True # so our own detach isn't mistaken for a kick
        await self.gc.ws.send(Op.voice_state(self.gid, None))
        await asyncio.sleep(1)
        await self._join()
        self._rejoining = False
    await self._open_ws()

await vc.join(debug=True)

voice json 15 {'any': 100}
voice json 11 {'user_ids': ['346450717025894400']}
voice json 18 {'user_id': '346450717025894400', 'flags': 2}
voice json 20 {'user_id': '346450717025894400', 'platform': 0}

With the session up we can watch the UDP keepalive doing its job on the real socket — wrapping (not replacing) sendto to observe the 8-byte pings that keep NAT mappings alive during listen-only sessions.

await asyncio.wait_for(vc.sess.wait(), 15)
assert vc._ka_task and not vc._ka_task.done()
pings = []
_orig_sendto = vc.trans.sendto
vc.trans.sendto = lambda d, addr=None: (pings.append(d), _orig_sendto(d))[1]
await asyncio.sleep(5.5)
del vc.trans.sendto
assert pings and all(len(p) == 8 for p in pings)

Audio arrives as UDP packets using the RTP (Real-time Transport Protocol) format. Each packet contains: - RTP header (12+ bytes) — version, sequence number, timestamp, SSRC (identifies which user is speaking) - Encrypted payload — Opus audio wrapped first by DAVE, then by Discord’s voice transport encryption using the secret_key - Nonce (4 bytes at end) — used for the transport decrypt

Decryption is two-stage: - First, decrypt the Discord transport layer with aead_xchacha20_poly1305_rtpsize - The cipher expects a 24-byte nonce, but only 4 bytes are transmitted — we pad with 20 zero bytes - The unencrypted RTP header is used as AAD (Additional Authenticated Data) — it’s verified but not encrypted - For rtpsize mode, the AAD includes the 12-byte base header, any CSRC entries (4 bytes each, usually 0), and only the 4-byte extension preamble — the extension data is part of the encrypted payload - After transport decryption, skip past the RTP extension data to get the DAVE-protected audio frame — its length comes from the extension preamble (a count of 32-bit words), not a fixed size, and packets without the extension bit have nothing to skip - Then decrypt that frame with DAVE: self.dave.decrypt(uid, davey.MediaType.audio, data), where uid comes from the SSRC → user mapping learned from voice SPEAKING events - The result is the Opus payload, ready for Opus decoding

Packets with byte[1] == 0x78 are RTP voice data. Other packets (like 0xC9) are RTCP control packets used for connection quality reporting — we skip those.

@patch
def decrypt(self:VoiceClient, pkt, uid):
    csrc_cnt = pkt[0] & 0x0F
    has_ext = bool(pkt[0] & 0x10)
    hdr_sz = 12 + csrc_cnt * 4 + (4 if has_ext else 0)
    nonce = bytearray(24) # only use 4, but pad to 24 bytes as expected by the cipher
    nonce[:4] = pkt[-4:]
    d, hdr, nonce = bytes(pkt[hdr_sz:-4]), bytes(pkt[:hdr_sz]), bytes(nonce)
    d = xchacha_decrypt(d, hdr, nonce, bytes(self.secret_key))
    if has_ext: d = d[4 * int.from_bytes(pkt[hdr_sz-2:hdr_sz], 'big'):] # ext payload length is in 32-bit words
    return self.dave.decrypt(uid, davey.MediaType.audio, d) if getattr(self, 'dave', None) else d

A hardcoded 8-byte skip here used to corrupt audio whenever the extension wasn’t exactly two words (and ate real Opus data on extension-less packets). Discord won’t produce a specific extension size on demand, so we build real RTP packets ourselves and round-trip them through the real cipher — varying extension sizes, no extension, and no DAVE session (where the transport plaintext is the Opus frame).

_key = bytes(range(32))
def _mk_pkt(payload, ext_words=2, has_ext=True):
    hdr = struct.pack('>BBHII', 0x80 | (0x10 if has_ext else 0), 0x78, 1, 960, 1234)
    if has_ext: hdr += struct.pack('>HH', 0xBEDE, ext_words)
    n4 = (7).to_bytes(4, 'big')
    enc = xchacha_encrypt((bytes(4 * ext_words) if has_ext else b'') + payload, hdr, n4 + b'\0'*20, _key)
    return hdr + enc + n4

fvc = object.__new__(VoiceClient)
fvc.secret_key = _key
for kw in [dict(ext_words=0), dict(ext_words=2), dict(ext_words=3), dict(has_ext=False)]:
    assert fvc.decrypt(_mk_pkt(b'opus!', **kw), 0) == b'opus!', kw

decode turns incoming RTP voice packets into PCM while keeping each speaker’s stream ordered without adding a jitter buffer. RTP sequence numbers are compared modulo 16 bits, so normal wraparound from 65535 to 0 is accepted; duplicates and packets that arrive behind the last accepted sequence are dropped. A forward sequence gap represents genuine packet loss: small gaps use Opus packet-loss concealment and larger gaps use raw silence.

Packets are decrypted and decoded before their sequence and timestamp become accepted state. A corrupt packet therefore cannot move the stream backwards or poison the next gap calculation. The tradeoff is intentional: a reordered packet is discarded instead of briefly buffering for it, which keeps recording simple at the cost of an occasional 20 ms loss.

def silence(n_smpls:int): return b'\x00' * (n_smpls * n_chs * 2) # 2 bytes per sample (s16le)

@patch
def decode(self:VoiceClient, pkt):
    if (pkt[1] & 0x7F) != 0x78: return # skip RTCP/control packets
    ssrc = int.from_bytes(pkt[8:12], 'big')
    uid = self.ssrc_to_user.get(ssrc)
    if uid is None or uid == int(self.gc.user_id): return

    seq = int.from_bytes(pkt[2:4], 'big')
    ts = int.from_bytes(pkt[4:8], 'big')
    self.last_seq = getattr(self, 'last_seq', {})
    self.last_ts = getattr(self, 'last_ts', {})
    prev_seq = self.last_seq.get(ssrc)
    delta = (seq - prev_seq) & 0xffff if prev_seq is not None else 1
    if delta == 0 or delta >= 0x8000: return # duplicate or late

    dec = self.decoders.setdefault(uid, opuslib_next.Decoder(48000, 2))
    try: opus = self.decrypt(pkt, uid)
    except (ValueError, CryptoError): return

    missing = delta - 1
    fill = [dec.decode(b'', spf) for _ in range(missing)] if missing <= 3 else [silence(missing * spf)]
    pcm = b''.join(fill) + dec.decode(opus, 5760)
    self.last_seq[ssrc],self.last_ts[ssrc] = seq,ts
    return pcm

Recording is split per speaker by opening one ffmpeg process per user and writing that user’s decoded PCM to its stdin. If someone starts speaking late, we remember the offset from _rec_start and apply it when mixing, instead of pushing a long silence prefix through ffmpeg.

@patch
async def _get_proc(self:VoiceClient, uid):
    if uid not in self._rec_procs:
        path = str(self._rec_path.with_stem(f'{self._rec_path.stem}_{uid}'))
        p = ( ffmpeg.input('pipe:', f='s16le', ar=sr, ac=n_chs)
                    .output(path)
                    .overwrite_output()
                    .run_async(pipe_stdin=True, quiet=True))
        self._rec_procs[uid] = (p, path)
        self._rec_offsets[uid] = max(time.time() - self._rec_start, 0)
    return self._rec_procs[uid]

@patch
async def _recv_audio(self:VoiceClient):
    while self._recording:
        try: pkt = await asyncio.wait_for(self.proto.packets.get(), timeout=0.1)
        except asyncio.TimeoutError: continue
        if not (pcm := self.decode(pkt)): continue

        ssrc = int.from_bytes(pkt[8:12], 'big')
        uid = self.ssrc_to_user.get(ssrc, ssrc)
        p, path = await self._get_proc(uid)
        await asyncio.to_thread(p.stdin.write, pcm)

start_recording drains the UDP queue first so stale packets from before the recording do not get written into the new files.

stop_recording has to close each ffmpeg stdin and wait for the process. That finalizes the containers; otherwise the files can look valid but be missing buffered audio at the end.

@patch
def start_recording(self:VoiceClient, path='/tmp/recording.mp3'):
    while not self.proto.packets.empty(): self.proto.packets.get_nowait()

    self.decoders = {}
    self.last_seq,self.last_ts = {},{}
    self._rec_path = Path(path)
    self._rec_procs = {}
    self._rec_offsets = {}
    self._rec_start = time.time()
    self._recording = True
    self._rec_task = asyncio.create_task(self._recv_audio())
    return path

def _delayed_input(path, delay):
    inp = ffmpeg.input(path)
    if delay <= 0: return inp
    delay = max(round(delay * 1_000), 0)
    return inp.filter('adelay', '|'.join([str(delay)] * n_chs))

async def _wait_proc(p, timeout):
    try: return await asyncio.wait_for(asyncio.to_thread(p.wait), timeout)
    except asyncio.TimeoutError:
        with suppress(Exception): p.kill()
        return await asyncio.to_thread(p.wait)

@patch
async def stop_recording(self:VoiceClient, mix=True, mix_path=None, timeout=10):
    self._recording = False

    if getattr(self, '_rec_task', None):
        self._rec_task.cancel()
        try: await self._rec_task
        except asyncio.CancelledError: pass

    speaker_paths = {}
    for uid, (p, path) in self._rec_procs.items():
        with suppress(BrokenPipeError, OSError, ValueError): p.stdin.close()
        await _wait_proc(p, timeout)
        speaker_paths[uid] = path

    mixed_path = None
    if mix and speaker_paths:
        mixed_path = str(mix_path or self._rec_path.with_stem(f'{self._rec_path.stem}_mixed'))

        if len(speaker_paths) == 1:
            src = next(iter(speaker_paths.values()))
            out = ffmpeg.input(src).output(mixed_path).overwrite_output()
        else:
            inputs = [_delayed_input(path, self._rec_offsets.get(uid, 0)) for uid,path in speaker_paths.items()]
            mixed = ffmpeg.filter(inputs, 'amix', inputs=len(inputs), duration='longest')
            out = mixed.output(mixed_path).overwrite_output()
        await _wait_proc(out.run_async(quiet=True), timeout)

    return {'speakers': speaker_paths, 'mixed': mixed_path}

vc.start_recording()

'/tmp/recording.mp3'

out = await vc.stop_recording()
out

{'speakers': {}, 'mixed': None}

Outgoing audio is the receive path in reverse: encode PCM to Opus, wrap it with DAVE only when the E2EE session is actually ready, then encrypt it with Discord’s voice transport key and send it as an RTP packet. The readiness check matters: a bot alone in a channel negotiates a DAVE protocol version but no MLS group exists until the epoch starts, so clients send plain Opus until then (davey raises NotReady if asked to encrypt earlier) — the same applies briefly during epoch transitions.

The RTP sequence, timestamp, and nonce are advanced manually for each 20 ms frame. If those drift or repeat, Discord will hear either broken audio or nothing at all.

@patch
def _enc_opus(self:VoiceClient, opus):
    "DAVE-wrap an opus frame only when the E2EE session is active (epoch started and keys ready)"
    return self.dave.encrypt_opus(opus) if getattr(self, 'dave', None) and self.dave.ready else opus

@patch
def encode(self:VoiceClient, fr): return self._enc_opus(self.encoder.encode(fr, spf))

@patch
def send_pkt(self:VoiceClient, payload):
    hdr = struct.pack('>BBHII', 0x80, 0x78, self.send_seq, self.send_ts, self.ssrc)
    nonce4 = self.send_nonce.to_bytes(4, 'big')
    enc = xchacha_encrypt(payload, hdr, nonce4 + b'\0'*20, bytes(self.secret_key))
    self.trans.sendto(hdr + enc + nonce4)
    self.send_seq = (self.send_seq + 1) & 0xffff
    self.send_ts = (self.send_ts + spf) & 0xffffffff
    self.send_nonce = (self.send_nonce + 1) & 0xffffffff

We can check the readiness gating on the live session: DAVE-wrapped frames end with the 0xFAFA magic marker, so whether the encoder output carries it must always agree with dave.ready — a bot alone in the channel has a negotiated version but no epoch, so it sends plain Opus.

assert vc.encode(silence(spf)).endswith(b'\xfa\xfa') == vc.dave.ready

We must notify discord when our bot is speaking and stop speaking. When a transmission truly ends, the docs require five frames of Opus silence (0xF8 0xFF 0xFE) so receivers don’t interpolate the tail of our audio into the next utterance — skipping them is audible as a garbled blip at the end of playback. Streamed audio that arrives in chunks (e.g. a realtime TTS feed) should pass end=False for all but the last chunk so the transmission stays continuous. The finally keeps Discord from leaving the bot marked as speaking if playback errors halfway through.

opus_silence = b'\xf8\xff\xfe'

@patch
async def speaking(self:VoiceClient, on=True):
    await self.ws.send(Op.speaking(self.ssrc, int(on)))

@patch
def send_frame(self:VoiceClient, fr): self.send_pkt(self.encode(fr))

@patch
async def send_frames(self:VoiceClient, frames, end=True):
    "Send 20ms Opus `frames`; unless `end=False`, finish with 5 silence frames and clear the speaking flag."
    loop = asyncio.get_running_loop()
    nxt = loop.time()
    await self.speaking(True)
    try:
        for fr in frames:
            self.send_frame(fr)
            nxt += spf / sr
            await asyncio.sleep(max(0, nxt - loop.time()))
    finally:
        if end:
            try:
                for _ in range(5): # per the docs, stops unintended Opus interpolation into the next transmission
                    self.send_pkt(self._enc_opus(opus_silence))
                    nxt += spf / sr
                    await asyncio.sleep(max(0, nxt - loop.time()))
                await self.speaking(False)
            except websockets.exceptions.ConnectionClosed: pass # ws died mid-send; nothing to clean up

Discord voice expects 20 ms Opus frames. Padding the final partial frame preserves the tail of short clips instead of silently dropping it.

frame_bytes = spf * n_chs * 2
def pcm2frames(pcm):
    for i in range(0, len(pcm), frame_bytes):
        fr = pcm[i:i+frame_bytes]
        yield fr if len(fr) == frame_bytes else fr + silence((frame_bytes - len(fr)) // (n_chs * 2))

@patch
async def send_pcm(self:VoiceClient, pcm, end=True): await self.send_frames(pcm2frames(pcm), end=end)

t = np.arange(sr) / sr
mono = (0.08 * np.sin(2*np.pi*440*t) * 32767).astype(np.int16)
pcm = np.column_stack([mono, mono]).ravel().tobytes()
len(pcm)

192000

await vc.send_pcm(pcm)

Spying on the live send path (wrapping send_pkt so packets still go out for real) shows the transmission shape on the wire: every send_pcm ends with exactly five extra packets — the silence frames — while end=False leaves the transmission open with no tail.

n_frames = len(list(pcm2frames(pcm)))
sent = []
_orig_send_pkt = vc.send_pkt
vc.send_pkt = lambda p: (sent.append(p), _orig_send_pkt(p))[1]
await vc.send_pcm(pcm)
assert len(sent) == n_frames + 5      # real transmission ends with 5 opus silence frames

sent.clear()
await vc.send_pcm(pcm, end=False)     # mid-stream chunk: no tail, transmission stays open
del vc.send_pkt
assert len(sent) == n_frames
await vc.speaking(False)

await vc._reconnect(resume=True)
await asyncio.wait_for(vc.resumed.wait(), 5)
await vc.send_pcm(pcm)

_reconnect was called directly above; real drops arrive as ConnectionClosed in the listener, which classifies the close code per the docs and schedules recovery as a separate task (a listener that ran the reconnect inline could cancel its own recovery — a bug this design retired). Killing the live socket with a transient code exercises the resume path end-to-end.

vc.resumed.clear()
await vc.ws.close(code=4000)                  # a real dropped connection
await asyncio.wait_for(vc.resumed.wait(), 20) # listener resumes unaided
await vc.send_pcm(pcm)

The auth/session close family (4003–4006, 4009, 4011) means this session’s credentials are dead and resuming would loop forever — the client must rejoin from scratch through the main gateway, which mints fresh ones. We can earn one for real by resuming with a corrupted token (Discord answers 4003 Not authenticated — a fake test here would have guessed 4006 and encoded the wrong policy): the client should come back with a fresh token and a working session, without us doing anything.

vc.token = 'not-a-real-token'
vc.resumed.clear()
await vc.ws.close(code=4000)   # resume attempt with the bad token -> 4003 -> automatic full rejoin
for _ in range(60):
    if vc.sess.is_set() and vc.token != 'not-a-real-token': break
    await asyncio.sleep(0.5)
assert vc.token != 'not-a-real-token' and vc.sess.is_set()
await vc.send_pcm(pcm)

To play audio back from a file, we’ve created a little helper using ffmpeg:

def file2pcm(path):
    return ffmpeg.input(path).output('pipe:', f='s16le', ar=sr, ac=n_chs).run(capture_stdout=True, quiet=True)[0]

@patch
async def play_file(self:VoiceClient, path): await self.send_pcm(file2pcm(path))

await vc.play_file("/tmp/recording_mixed.mp3")

Finally, the hostile ending: an admin disconnecting the bot. Discord sends our own VOICE_STATE_UPDATE with a null channel (and closes the voice socket with 4014) — the client must tear down cleanly, finalize any in-flight recording, and not fight its way back in. We can do this to ourselves over REST.

vc.start_recording('/tmp/kick_test.mp3')
await asyncio.sleep(1)
await vch('PATCH', f'/guilds/{gld.id}/members/{gc.user_id}', channel_id=None)  # kick ourselves
for _ in range(30):
    if not vc.running: break
    await asyncio.sleep(0.5)
assert not vc.running and not vc._recording   # clean teardown, recording finalized

Recording can continue across reconnect types.

vc.start_recording()
await asyncio.sleep(2)
procs_before, start_before = dict(vc._rec_procs), vc._rec_start
await vc._reconnect(resume=False)
await asyncio.sleep(3)
await vc.stop_recording()

await vc.play_file("/tmp/recording_mixed.mp3")

await vc.leave()