discord_radio: don't count the bot's own voice state toward a full channel; timeout text points at the handshake

Permissions are pre-checked now, so a connect timeout is the voice handshake
(pre-2.7 discord.py: DAVE close 4017 retried to the 30s timeout) or UDP egress,
not permissions.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

drb-edge-node/app/internal/discord_radio.py

@@ -74,6 +74,43 @@ class _Voice:
     code: Optional[str] = None
 
 
+def _join_refusal(guild, channel) -> Optional[tuple[str, str]]:
+    """Why Discord would refuse this bot's voice join, or None.
+
+    Discord does not answer a join the bot lacks permission for — the gateway
+    just never sends the voice-server update, discord.py times out after 30s,
+    and the admin page used to read "(TimeoutError: )". Checking first turns
+    that into an instruction. Best effort: if the bot's own member or the
+    permission data is not cached, let the join go ahead.
+    """
+    me = getattr(guild, "me", None)
+    permissions_for = getattr(channel, "permissions_for", None)
+    if me is None or permissions_for is None:
+        return None
+    try:
+        perms = permissions_for(me)
+    except Exception:
+        return None
+    missing = [label for attr, label in (("view_channel", "View Channel"), ("connect", "Connect"))
+               if not getattr(perms, attr, True)]
+    if getattr(channel, "type", None) == discord.ChannelType.voice and not getattr(perms, "speak", True):
+        missing.append("Speak")
+    if missing:
+        return ("missing_permissions",
+                f"The radio bot lacks {', '.join(missing)} on that voice channel — "
+                "allow it in the channel's permission settings in Discord.")
+    limit = getattr(channel, "user_limit", 0) or 0
+    # The bot's own lingering voice state (a dropped connection) is not a
+    # slot it needs; counting it would block the rejoin of a full channel.
+    me_id = getattr(me, "id", None)
+    members = [m for m in (getattr(channel, "members", None) or [])
+               if me_id is None or getattr(m, "id", None) != me_id]
+    if limit and len(members) >= limit and not getattr(perms, "move_members", False):
+        return ("channel_full", f"That voice channel is full ({len(members)}/{limit}) — "
+                                "raise its user limit or free a slot.")
+    return None
+
+
 def _default_client_factory():
     intents = discord.Intents.none()
     intents.guilds = True
@@ -220,6 +257,10 @@ class RadioManager:
                     v.error = self._session_errors.get(d["token_id"], "The radio bot could not log in to Discord.")
                     continue
                 await self._ensure_voice(sess, v, d)
+            # 4) presence follows voice: "Listening to <system>" only while a
+            #    voice connection for that bot is actually up.
+            for sess in list(self._sessions.values()):
+                await self._sync_presence(sess, target)
 
     async def _ensure_session(self, token_id: str) -> Optional[_Session]:
         sess = self._sessions.get(token_id)
@@ -275,7 +316,12 @@ class RadioManager:
             if not vc.is_playing():
                 self._play(v)
             v.state, v.error, v.code = "connected", None, None
-            await self._set_presence(sess, d.get("system_name"))
+            return
+        refusal = _join_refusal(guild, channel)
+        if refusal:
+            if v.error != refusal[1]:   # once per change, not every watchdog pass
+                logger.error(f"Stream {v.stream_id}: {refusal[1]}")
+            v.state, v.code, v.error = "error", refusal[0], refusal[1]
             return
         v.state = "connecting"
         try:
@@ -287,11 +333,23 @@ class RadioManager:
                 if existing is not None:
                     await existing.disconnect(force=True)
                 v.vc = await channel.connect()
+        except asyncio.TimeoutError:
+            # discord.py's TimeoutError has no message, so the old text read
+            # "(TimeoutError: )" — useless on the admin page.
+            v.vc = None
+            v.state, v.code = "error", "connect_timeout"
+            # Permissions were checked above, so this is usually the voice
+            # handshake itself (e.g. a pre-2.7 discord.py refused with 4017 for
+            # lacking DAVE E2EE, then retried until the 30s timeout) or UDP egress.
+            v.error = ("Discord never completed the voice join (timed out). Permissions look fine, so "
+                       "check the node's discord.py voice log (DAVE close 4017, UDP blocked).")
+            logger.error(f"Stream {v.stream_id}: voice connect timed out")
+            return
         except Exception as e:
             v.vc = None
             v.state, v.code = "error", "connect_failed"
             v.error = f"Could not connect to the voice channel ({type(e).__name__}: {str(e)[:120]})."
-            logger.error(f"Stream {v.stream_id}: voice connect failed: {e}")
+            logger.error(f"Stream {v.stream_id}: voice connect failed: {type(e).__name__}: {e}")
             return
         # Bounded wait for the shared PulseAudio socket (op25 container creates it).
         if not await pulse.wait_until_ready():
@@ -300,15 +358,27 @@ class RadioManager:
             return
         self._play(v, pulse_checked=True)
         v.state, v.error, v.code = "connected", None, None
-        await self._set_presence(sess, d.get("system_name"))
         logger.info(f"Stream {v.stream_id} connected (bot {v.token_id})")
 
-    async def _set_presence(self, sess: _Session, name: Optional[str]) -> None:
+    async def _sync_presence(self, sess: _Session, target: dict[str, dict]) -> None:
-        if not name or sess.presence == name:
+        """Show "Listening to <system>" only while this bot is really in voice.
+
+        It used to be set once on the first successful join and never cleared,
+        so a bot whose voice later dropped (or kept failing to rejoin) still
+        looked healthy in Discord's member list.
+        """
+        name = None
+        for sid, v in sorted(self._voices.items()):
+            if (v.token_id == sess.token_id and v.state == "connected"
+                    and v.vc is not None and v.vc.is_connected()):
+                name = (target.get(sid) or {}).get("system_name") or name
+                if name:
+                    break
+        if sess.presence == name:
             return
         try:
-            await sess.client.change_presence(
+            activity = discord.Activity(type=discord.ActivityType.listening, name=name) if name else None
-                activity=discord.Activity(type=discord.ActivityType.listening, name=name))
+            await sess.client.change_presence(activity=activity)
             sess.presence = name
         except Exception as e:
             logger.warning(f"Presence update failed: {e}")

drb-edge-node/requirements.txt

@@ -3,7 +3,9 @@ uvicorn[standard]
 pydantic-settings
 paho-mqtt>=2.0.0
 httpx
-discord.py[voice]
+# >=2.7: Discord requires DAVE (E2EE) voice since 2026-03-01; 2.7 is the first
+# discord.py with it (the [voice] extra pulls `davey`). Older releases cannot join voice.
+discord.py[voice]>=2.7
 PyNaCl
 python-multipart
 pytest

drb-edge-node/tests/test_discord_radio.py

@@ -47,8 +47,18 @@ class FakeVC:
 class FakeChannel:
     def __init__(self, guild, cid, kind=discord.ChannelType.voice):
         self.guild, self.id, self.type = guild, cid, kind
+        self.perms = discord.Permissions(view_channel=True, connect=True, speak=True)
+        self.user_limit = 0
+        self.members = []
+        self.connect_error = None
+
+    def permissions_for(self, member):
+        assert member is self.guild.me
+        return self.perms
 
     async def connect(self):
+        if self.connect_error:
+            raise self.connect_error
         vc = FakeVC(self)
         self.guild.voice_client = vc
         return vc
@@ -57,6 +67,7 @@ class FakeChannel:
 class FakeGuild:
     def __init__(self, gid, channel_ids):
         self.id = gid
+        self.me = object()
         self.voice_client = None
         self.channels = {c: FakeChannel(self, c) for c in channel_ids}
 
@@ -149,6 +160,7 @@ async def test_apply_connects_and_reports(env):
     assert env.mgr.is_connected
     vc = env.guilds[1].voice_client
     assert vc.plays == 1 and vc.channel.id == 11
+    assert FakeClient.instances[0].presence.name == "County"
 
 
 async def test_one_session_serves_two_guilds(env):
@@ -258,3 +270,64 @@ async def test_no_message_content_intent_on_real_client():
         assert client.intents.voice_states and client.intents.guilds
     finally:
         await client.close()
+
+
+def _presence(client):
+    activity = getattr(client, "presence", None)
+    return activity.name if activity else None
+
+
+async def test_missing_connect_permission_is_named_not_a_timeout(env):
+    """Discord ignores a join the bot may not make; say which permission is missing."""
+    env.guilds[1].channels[11].perms = discord.Permissions(view_channel=True, speak=True)
+    await env.mgr.apply_desired(_desired(1, (1, 11, "t1")))
+    st = env.published[-1]["streams"]["1:11"]
+    assert st["state"] == "error" and st["code"] == "missing_permissions"
+    assert "Connect" in st["error"] and "View Channel" not in st["error"]
+    assert env.guilds[1].voice_client is None
+    assert _presence(FakeClient.instances[0]) is None   # not "Listening to …" while out of voice
+
+
+async def test_full_channel_is_reported(env):
+    ch = env.guilds[1].channels[11]
+    ch.user_limit, ch.members = 2, [object(), object()]
+    await env.mgr.apply_desired(_desired(1, (1, 11, "t1")))
+    st = env.published[-1]["streams"]["1:11"]
+    assert st["code"] == "channel_full" and "2/2" in st["error"]
+
+
+async def test_bots_own_lingering_voice_state_does_not_fill_the_channel(env):
+    """A dropped connection can leave the bot listed in the channel; that is
+    not a slot it needs to rejoin."""
+    guild = env.guilds[1]
+    guild.me = type("Me", (), {"id": 999})()
+    ch = guild.channels[11]
+    ch.user_limit, ch.members = 2, [object(), guild.me]
+    await env.mgr.apply_desired(_desired(1, (1, 11, "t1")))
+    assert env.published[-1]["streams"]["1:11"]["state"] == "connected"
+
+
+async def test_connect_timeout_has_a_readable_error(env):
+    env.guilds[1].channels[11].connect_error = asyncio.TimeoutError()
+    await env.mgr.apply_desired(_desired(1, (1, 11, "t1")))
+    st = env.published[-1]["streams"]["1:11"]
+    assert st["state"] == "error" and st["code"] == "connect_timeout"
+    assert "timed out" in st["error"] and "TimeoutError: )" not in st["error"]
+
+
+async def test_presence_is_cleared_when_voice_is_lost_and_rejoin_fails(env):
+    """The owner saw "Listening to <system>" on a bot that was not in voice:
+    presence was set once on join and never cleared."""
+    await env.mgr.apply_desired(_desired(1, (1, 11, "t1")))
+    client = FakeClient.instances[0]
+    assert _presence(client) == "County"
+    env.guilds[1].voice_client.connected = False
+    env.guilds[1].voice_client = None
+    env.guilds[1].channels[11].connect_error = RuntimeError("voice handshake failed")
+    await env.mgr.tick()
+    st = env.mgr.snapshot()["streams"]["1:11"]
+    assert st["state"] == "error" and st["code"] == "connect_failed"
+    assert _presence(client) is None
+    env.guilds[1].channels[11].connect_error = None
+    await env.mgr.tick()
+    assert _presence(client) == "County"