Merge pull request 'checkin: report host disk free/total (#41)' (#43) from feat/41-disk-reporting into main

/recordings is a bind mount of the host checkout, so shutil.disk_usage on it
is the disk the recorder writes to and Docker fills. Explicit nulls on error
so C2 never keeps a stale reading. Server half: server-26 feat/41-node-disk.

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:
-        if not name or sess.presence == name:
+    async def _sync_presence(self, sess: _Session, target: dict[str, dict]) -> None:
+        """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))
+            activity = discord.Activity(type=discord.ActivityType.listening, name=name) if name else None
+            await sess.client.change_presence(activity=activity)
             sess.presence = name
         except Exception as e:
             logger.warning(f"Presence update failed: {e}")

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

@@ -1,5 +1,6 @@
 import asyncio
 import json
+import shutil
 from datetime import datetime, timezone
 from collections import deque
 from typing import Optional, Callable, Awaitable, Dict, Any
@@ -8,6 +9,22 @@ from app.config import settings
 from app.internal.logger import logger
 from app.internal import credentials, key_pickup
 
+
+def disk_report(path: str) -> Dict[str, Optional[int]]:
+    """Free/total bytes of the filesystem holding `path` (node-26#41).
+
+    /recordings is a bind mount of the host checkout, so this is the host disk
+    the recorder writes to and Docker's images/build cache live on. A full disk
+    stopped node-002 recording for 17 days while it still showed Online. Nulls
+    on error, so C2 overwrites a stale reading instead of keeping it.
+    """
+    try:
+        usage = shutil.disk_usage(path)
+    except OSError:
+        return {"disk_free_bytes": None, "disk_total_bytes": None}
+    return {"disk_free_bytes": usage.free, "disk_total_bytes": usage.total}
+
+
 CommandCallback = Callable[[Dict[str, Any]], Awaitable[None]]
 ConfigCallback = Callable[[Dict[str, Any]], Awaitable[None]]
 ApiKeyCallback = Callable[[Dict[str, Any]], Awaitable[None]]
@@ -372,6 +389,7 @@ class MQTTManager:
             count = devices.get("count") or None
         if count is not None:
             payload["sdr_count"] = count
+        payload.update(disk_report(settings.recordings_path))
         self._publish(self._t_checkin, payload, qos=1)
 
     def _publish(self, topic: str, payload: dict, qos: int = 0, retain: bool = False):

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"

drb-edge-node/tests/test_host_updater.py

@@ -151,6 +151,17 @@ def test_mqtt_sentinel_triggers_update_and_is_consumed(env):
     assert "evil" not in calls and "rm -rf" not in calls
 
 
+def test_update_prunes_build_cache_and_stale_images_after_up(env):
+    # node-26#41: the build cache was never pruned and filled node-002's disk.
+    (env["stub"] / "response").write_text(json.dumps({"update_requested_at": "2026-10-04T12:00:00+00:00"}))
+    assert _run(env).returncode == 0
+    lines = _calls(env).splitlines()
+    up = next(i for i, ln in enumerate(lines) if ln.startswith("docker compose up -d"))
+    builder = next(i for i, ln in enumerate(lines) if ln.startswith("docker builder prune -af --filter until="))
+    images = next(i for i, ln in enumerate(lines) if ln.startswith("docker image prune -af --filter until="))
+    assert up < builder and up < images  # only after the new containers are running
+
+
 def test_c2_response_is_never_used_as_a_ref_or_command(env):
     (env["stub"] / "response").write_text(json.dumps({"update_requested_at": "$(touch /tmp/pwned); main"}))
     r = _run(env)

drb-edge-node/tests/test_mqtt_manager.py

@@ -140,3 +140,16 @@ def test_build_client_does_not_touch_offline_buffer(monkeypatch):
 
     assert list(mqtt_manager._offline_buffer) == [("nodes/test/metadata", {"call_id": "sentinel"})]
     mqtt_manager._offline_buffer.clear()
+
+
+def test_disk_report_reads_free_and_total(tmp_path):
+    from app.internal.mqtt_manager import disk_report
+    rep = disk_report(str(tmp_path))
+    assert rep["disk_total_bytes"] > 0
+    assert 0 <= rep["disk_free_bytes"] <= rep["disk_total_bytes"]
+
+
+def test_disk_report_nulls_when_path_unreadable(tmp_path):
+    # node-26#41: nulls (not omitted keys) so C2 never keeps a stale reading.
+    from app.internal.mqtt_manager import disk_report
+    assert disk_report(str(tmp_path / "missing")) == {"disk_free_bytes": None, "disk_total_bytes": None}

scripts/drb-update.sh

@@ -225,6 +225,14 @@ do_update() {
         done || true
   fi
   docker image prune -f >/dev/null 2>&1 || true
+
+  # node-26#41: neither of the above ever freed BuildKit cache or untagged-
+  # but-unused images, so a BUILD=1 node grew ~40 GB of build cache plus
+  # 16 GB of stale images until the root disk hit 100% and recording silently
+  # stopped (node-002, 17 days). Both prunes skip anything a container uses;
+  # the age filters keep the cache warm for the next build.
+  docker builder prune -af --filter until=72h >/dev/null 2>&1 || true
+  docker image prune -af --filter until=168h >/dev/null 2>&1 || true
 }
 
 main "$@"