# Voice API


<!-- WARNING: THIS FILE WAS AUTOGENERATED! DO NOT EDIT! -->

``` python
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")
```

``` python
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)
```

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

<div class="prose" markdown="1">

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

</div>

    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`](https://AnswerDotAI.github.io/cordslite/voice.html#voiceclient)
manages the voice gateway WebSocket and UDP connections for a single
voice channel, coordinated through the main
[`GatewayClient`](https://AnswerDotAI.github.io/cordslite/gateway.html#gatewayclient).

``` python
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=})'
```

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

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

Voice requires additional
[`Op`](https://AnswerDotAI.github.io/cordslite/gateway.html#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).

``` python
@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)
```

``` python
# 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:

<table>
<thead>
<tr>
<th>Field</th>
<th>Size</th>
<th>Description</th>
</tr>
</thead>
<tbody>
<tr>
<td>Type</td>
<td>2 bytes</td>
<td>1=request, 2=response</td>
</tr>
<tr>
<td>Length</td>
<td>2 bytes</td>
<td>70 (size of remaining fields)</td>
</tr>
<tr>
<td>SSRC</td>
<td>4 bytes</td>
<td>Our audio stream identifier</td>
</tr>
<tr>
<td>Address</td>
<td>64 bytes</td>
<td>Blank in request; our external IP in response</td>
</tr>
<tr>
<td>Port</td>
<td>2 bytes</td>
<td>Blank in request; our external port in response</td>
</tr>
</tbody>
</table>

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

``` python
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

``` python
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()
```

``` python
@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).

``` python
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.

``` python
@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](https://docs.discord.com/developers/topics/opcodes-and-status-codes#voice-voice-close-event-codes):
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.

``` python
@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()
```

``` python
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.

``` python
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.

``` python
@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).

``` python
_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.

``` python
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.

``` python
@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.

``` python
@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}
```

``` python
vc.start_recording()
```

    '/tmp/recording.mp3'

``` python
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.

``` python
@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.

``` python
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](https://docs.discord.com/developers/topics/voice-connections#voice-data-interpolation)
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.

``` python
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.

``` python
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)
```

``` python
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

``` python
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.

``` python
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)
```

``` python
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](https://docs.discord.com/developers/topics/opcodes-and-status-codes#voice-voice-close-event-codes)
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.

``` python
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.

``` python
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`:

``` python
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))
```

``` python
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.

``` python
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.

``` python
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()
```

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

``` python
await vc.leave()
```