discord_radio: desired-state streams, logout on release, no mention controls (node-26#25)

- Rewritten as RadioManager reconciling C2's retained
  nodes/{id}/discord/desired (token ids) and publishing
  nodes/{id}/discord/actual {applied_seq, sessions, streams}.
- One gateway session per leased token, one voice connection per
  (token, guild); a token no stream needs is logged OUT (client.close()),
  which is what lets C2 free it safely (item 1).
- Watchdog is a separate reconcile loop; recovering a dead session can no
  longer cancel it (item 2). Presence kept across rejoins.
- No message-content intent and no @bot leave/join mention controls (item 3).
- Local dashboard: /api/discord/join returns 409 with an explanation instead
  of a TypeError 500; /api/discord/leave pauses locally; GET /api/discord
  shows state (item 4). Join/leave results are reported to C2 (item 5).
- Token strings fetched over HTTPS from C2 (lease holder only), memory only.
- MQTT-loss fail-safe: logs every bot out after
  DISCORD_MQTT_LOSS_STOP_SECONDS (300, below C2's 600s lease grace).
- Legacy discord_join/discord_leave MQTT commands are ignored.
- Audio path unchanged: FFmpeg -f pulse on settings.pulse_source per
  connection, never Icecast.

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

drb-edge-node/app/config.py

@@ -54,6 +54,13 @@ class Settings(BaseSettings):
     # Bounded wait for the shared PulseAudio socket before launching FFmpeg.
     pulse_wait_timeout: float = 30.0
 
+    # Discord radio (node-26#25): with the broker unreachable this long, log
+    # every radio bot out. MUST stay below C2's
+    # DISCORD_LEASE_OFFLINE_RELEASE_SECONDS (default 600), after which C2 hands
+    # this node's bot tokens to other nodes — so a token is never logged in on
+    # two nodes at once.
+    discord_mqtt_loss_stop_seconds: int = 300
+
     # ------------------------------------------------------------------
     # Call segmentation
     #

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

@@ -1,299 +1,484 @@
+"""
+Discord radio streams on the edge, driven by desired state (node-26#25,
+server-26#203/#204).
+
+C2 retains ``nodes/{id}/discord/desired`` = the FULL list of streams this node
+should carry: ``{seq, streams: [{stream_id, guild_id, channel_id, token_id,
+system_name, ...}]}``. This module reconciles toward that list and publishes
+``nodes/{id}/discord/actual`` = ``{applied_seq, sessions: [token_id],
+streams: {stream_id: {state, error, code}}}`` after every change.
+
+Rules this enforces by construction:
+
+* One gateway session per leased token; a token no stream needs any more is
+  logged OUT (``client.close()``), not just disconnected from voice. C2 frees
+  the token only after an ``actual`` with ``applied_seq`` >= the drop and the
+  token absent from ``sessions`` — so one token is never logged in on two
+  nodes (node-26#25 item 1).
+* One voice connection per (token, guild); one token can serve many guilds.
+* The watchdog is a separate reconcile loop that never closes itself: closing
+  a session cannot cancel it (node-26#25 item 2).
+* No message-content intent, no ``@bot leave/join`` mention controls — voice
+  is controlled only through C2 (node-26#25 item 3).
+* MQTT-loss fail-safe: if the broker has been unreachable for
+  ``discord_mqtt_loss_stop_seconds`` every radio bot is logged out, because C2
+  returns this node's tokens to the pool after its own (longer) grace.
+
+Bot token strings never travel over MQTT: they are fetched over HTTPS from C2
+(``GET /discord/node/{node}/tokens/{id}``, node API key), which hands a token
+only to the node holding its lease, and are kept in memory only.
+
+Audio: each voice connection plays the PulseAudio monitor (``-f pulse`` on
+``settings.pulse_source``) — the accuracy path, never Icecast.
+"""
+from __future__ import annotations
+
 import asyncio
-from typing import Optional
+import time
+from dataclasses import dataclass
+from typing import Awaitable, Callable, Optional
+
 import discord
-from discord.ext import commands
+
 from app.config import settings
 from app.internal import pulse
 from app.internal.logger import logger
 
-BOT_READY_TIMEOUT = 15   # seconds to wait for Discord bot to become ready
-WATCHDOG_INTERVAL = 30   # seconds between voice-connection health checks
-REJOIN_DELAY = 5  # seconds to wait before attempting a rejoin
-STREAM_RETRY_DELAY = 5   # seconds to back off before re-arming the audio source
+BOT_READY_TIMEOUT = 15     # seconds to wait for a Discord login to become ready
+WATCHDOG_INTERVAL = 15     # seconds between reconcile passes
+STREAM_RETRY_DELAY = 5     # seconds to back off before re-arming an audio source
+REPORT_EVERY_TICKS = 4     # unconditional actual-state report every ~60s
+_VOICE_TYPES = (discord.ChannelType.voice, discord.ChannelType.stage_voice)
 
 
-class RadioBot:
-    def __init__(self):
-        self._bot: Optional[commands.Bot] = None
-        self._voice_client: Optional[discord.VoiceClient] = None
-        self._task: Optional[asyncio.Task] = None
-        self._watchdog_task: Optional[asyncio.Task] = None
-        self._ready_event: Optional[asyncio.Event] = None
-        self._current_token: Optional[str] = None
+@dataclass
+class _Session:
+    token_id: str
+    client: object
+    task: Optional[asyncio.Task] = None
+    presence: Optional[str] = None
+
+    def alive(self) -> bool:
+        return self.task is not None and not self.task.done() and not self.client.is_closed()
+
+
+@dataclass
+class _Voice:
+    stream_id: str
+    token_id: str
+    guild_id: int
+    channel_id: int
+    vc: Optional[object] = None
+    state: str = "connecting"
+    error: Optional[str] = None
+    code: Optional[str] = None
+
+
+def _default_client_factory():
+    intents = discord.Intents.none()
+    intents.guilds = True
+    intents.voice_states = True
+    return discord.Client(intents=intents)
+
+
+def _default_audio_factory():
+    # before_options land ahead of -i:  ffmpeg -f pulse -i drb_sink.monitor …
+    source = discord.FFmpegPCMAudio(settings.pulse_source, before_options="-f pulse")
+    return discord.PCMVolumeTransformer(source, volume=1.0)
+
+
+async def _default_token_fetcher(token_id: str) -> Optional[str]:
+    import httpx
+    from app.internal import credentials
+    api_key = credentials.get_api_key()
+    if not settings.c2_url or not api_key:
+        logger.error("Cannot fetch a radio-bot token: C2_URL or the node API key is missing.")
+        return None
+    url = f"{settings.c2_url.rstrip('/')}/discord/node/{settings.node_id}/tokens/{token_id}"
+    try:
+        async with httpx.AsyncClient(timeout=10) as client:
+            r = await client.get(url, headers={"Authorization": f"Bearer {api_key}"})
+            if r.status_code != 200:
+                logger.error(f"C2 refused radio-bot token {token_id}: HTTP {r.status_code}")
+                return None
+            return r.json().get("token")
+    except Exception as e:
+        logger.error(f"Fetching radio-bot token {token_id} failed: {e}")
+        return None
+
+
+async def _default_publisher(payload: dict) -> None:
+    from app.internal.mqtt_manager import mqtt_manager
+    mqtt_manager.publish_discord_actual(payload)
+
+
+def _default_mqtt_up() -> bool:
+    from app.internal.mqtt_manager import mqtt_manager
+    return mqtt_manager.is_connected
+
+
+class RadioManager:
+    def __init__(
+        self,
+        client_factory: Callable[[], object] = _default_client_factory,
+        token_fetcher: Callable[[str], Awaitable[Optional[str]]] = _default_token_fetcher,
+        publisher: Callable[[dict], Awaitable[None]] = _default_publisher,
+        audio_factory: Callable[[], object] = _default_audio_factory,
+        mqtt_up: Callable[[], bool] = _default_mqtt_up,
+        ready_timeout: float = BOT_READY_TIMEOUT,
+    ):
+        self._client_factory = client_factory
+        self._fetch_token = token_fetcher
+        self._publish = publisher
+        self._audio_factory = audio_factory
+        self._mqtt_up = mqtt_up
+        self._ready_timeout = ready_timeout
+
+        self._desired: dict[str, dict] = {}
+        self._applied_seq: Optional[int] = None
+        self._sessions: dict[str, _Session] = {}
+        self._voices: dict[str, _Voice] = {}
+        self._session_errors: dict[str, str] = {}
+        self._tokens: dict[str, str] = {}          # token_id -> string, memory only
+        self._lock = asyncio.Lock()
         self._loop: Optional[asyncio.AbstractEventLoop] = None
+        self._watchdog: Optional[asyncio.Task] = None
+        self._mqtt_down_since: Optional[float] = None
+        self._failsafe = False
+        self._paused = False
+        self._ticks = 0
 
-        # Remembered so we can rejoin after an unexpected disconnect
-        self._guild_id: Optional[int] = None
-        self._channel_id: Optional[int] = None
-        self._was_streaming: bool = False
+    # ------------------------------------------------------------------
+    # Inputs
+    # ------------------------------------------------------------------
 
-    async def join(self, guild_id: int, channel_id: int, token: str,
-                   call_active: bool = False, system_name: str = None) -> bool:
-        # (Re)start the bot if the token changed or the bot isn't running
-        if self._current_token != token or not self._is_bot_running():
-            if not await self._start_bot(token):
-                return False
-
-        guild = self._bot.get_guild(guild_id)
-        if not guild:
-            logger.error(f"Guild {guild_id} not found — bot may not be a member.")
-            return False
-
-        channel = guild.get_channel(channel_id)
-        if not isinstance(channel, discord.VoiceChannel):
-            logger.error(f"Channel {channel_id} is not a voice channel.")
-            return False
-
-        try:
-            if self._voice_client and self._voice_client.is_connected():
-                await self._voice_client.disconnect(force=True)
-            self._voice_client = await channel.connect()
-            # Remember where we are so the watchdog can rejoin if we drop
-            self._guild_id = guild_id
-            self._channel_id = channel_id
-            # Bounded wait for the shared PulseAudio socket. Historically FFmpeg was
-            # launched before the op25 container had created it, failed instantly,
-            # and the bot sat silently connected forever.
-            await pulse.wait_until_ready()
-            self._play_stream()
-            if system_name:
-                await self._bot.change_presence(
-                    activity=discord.Activity(
-                        type=discord.ActivityType.listening,
-                        name=system_name,
-                    )
-                )
-            logger.info(f"Joined #{channel.name} in {guild.name}")
-            return True
-        except Exception as e:
-            logger.error(f"Failed to join voice channel: {e}")
-            return False
-
-    async def leave(self) -> bool:
-        # Clear remembered channel so the watchdog doesn't rejoin
-        self._guild_id = None
-        self._channel_id = None
-        self._was_streaming = False
-
-        if self._voice_client and self._voice_client.is_connected():
+    async def apply_desired(self, payload: dict) -> None:
+        """Handle a (retained) desired-state message from C2."""
+        if not isinstance(payload, dict):
+            return
+        seq = payload.get("seq")
+        seq = seq if isinstance(seq, int) and not isinstance(seq, bool) else None
+        if seq is not None and self._applied_seq is not None and seq < self._applied_seq:
+            logger.warning(f"Ignoring stale Discord desired state seq {seq} (applied {self._applied_seq})")
+            return
+        wanted = {}
+        for s in payload.get("streams") or []:
             try:
-                self._stop_stream()
-                await self._voice_client.disconnect(force=True)
-                self._voice_client = None
-                await self._bot.change_presence(activity=None)
-                logger.info("Disconnected from voice channel.")
-                return True
-            except Exception as e:
-                logger.error(f"Failed to disconnect: {e}")
-        return False
+                wanted[str(s["stream_id"])] = {
+                    "stream_id": str(s["stream_id"]), "guild_id": int(s["guild_id"]),
+                    "channel_id": int(s["channel_id"]), "token_id": str(s["token_id"]),
+                    "system_name": s.get("system_name"),
+                }
+            except (KeyError, TypeError, ValueError):
+                logger.error(f"Malformed Discord desired stream entry ignored: {s!r}")
+        self._ensure_watchdog()
+        self._desired = wanted
+        self._paused = False
+        await self.reconcile()
+        if seq is not None:
+            self._applied_seq = seq
+        await self.publish_actual()
 
+    async def pause_locally(self) -> None:
+        """Local-dashboard stop: log every bot out until C2 sends a new list."""
+        self._paused = True
+        await self.reconcile()
+        await self.publish_actual()
+
+    # ------------------------------------------------------------------
+    # Reconcile
+    # ------------------------------------------------------------------
+
+    def _target(self) -> dict[str, dict]:
+        return {} if (self._paused or self._failsafe) else self._desired
+
+    async def reconcile(self) -> None:
+        async with self._lock:
+            target = self._target()
+            # 1) voice connections no longer wanted (or re-pointed)
+            for sid, v in list(self._voices.items()):
+                d = target.get(sid)
+                if not d or (d["token_id"], d["guild_id"], d["channel_id"]) != (v.token_id, v.guild_id, v.channel_id):
+                    await self._disconnect(v)
+                    del self._voices[sid]
+            # 2) gateway sessions no stream needs: LOG OUT, not just leave voice
+            needed = {d["token_id"] for d in target.values()}
+            for tid in list(self._sessions):
+                if tid not in needed:
+                    await self._close_session(tid)
+            for tid in list(self._tokens):
+                if tid not in needed:
+                    self._tokens.pop(tid, None)
+            # 3) sessions + voice for everything wanted
+            for sid, d in sorted(target.items()):
+                v = self._voices.get(sid)
+                if v is None:
+                    v = self._voices[sid] = _Voice(sid, d["token_id"], d["guild_id"], d["channel_id"])
+                sess = await self._ensure_session(d["token_id"])
+                if sess is None:
+                    v.state, v.code = "error", "login_failed"
+                    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)
+
+    async def _ensure_session(self, token_id: str) -> Optional[_Session]:
+        sess = self._sessions.get(token_id)
+        if sess and sess.alive():
+            return sess
+        if sess:
+            await self._close_session(token_id)
+        token = self._tokens.get(token_id) or await self._fetch_token(token_id)
+        if not token:
+            self._session_errors[token_id] = "DRB would not hand this node the radio bot's token."
+            return None
+        client = self._client_factory()
+        sess = _Session(token_id, client)
+        ready = asyncio.Event()
+
+        async def on_ready():
+            ready.set()
+        client.event(on_ready)
+        sess.task = asyncio.create_task(client.start(token))
+        waiter = asyncio.create_task(ready.wait())
+        done, _ = await asyncio.wait({waiter, sess.task}, timeout=self._ready_timeout,
+                                     return_when=asyncio.FIRST_COMPLETED)
+        if waiter not in done:
+            waiter.cancel()
+            err = sess.task.exception() if sess.task.done() and not sess.task.cancelled() else None
+            msg = f"{type(err).__name__}: {err}" if err else f"not ready after {self._ready_timeout:.0f}s"
+            logger.error(f"Radio bot {token_id} failed to log in: {msg}")
+            self._session_errors[token_id] = f"The radio bot could not log in to Discord ({msg[:120]})."
+            await self._shutdown_client(sess)
+            if isinstance(err, discord.LoginFailure):
+                self._tokens.pop(token_id, None)
+            return None
+        self._tokens[token_id] = token
+        self._sessions[token_id] = sess
+        self._session_errors.pop(token_id, None)
+        logger.info(f"Radio bot {token_id} logged in as {getattr(client, 'user', None)}")
+        return sess
+
+    async def _ensure_voice(self, sess: _Session, v: _Voice, d: dict) -> None:
+        client = sess.client
+        guild = client.get_guild(v.guild_id)
+        if guild is None:
+            v.state, v.code = "error", "not_in_guild"
+            v.error = "This radio bot is not a member of that Discord server — invite it first."
+            return
+        channel = guild.get_channel(v.channel_id)
+        if channel is None or getattr(channel, "type", None) not in _VOICE_TYPES:
+            v.state, v.code = "error", "bad_channel"
+            v.error = "That channel is not a voice channel the radio bot can see."
+            return
+        vc = v.vc
+        if vc is not None and vc.is_connected() and getattr(vc.channel, "id", None) == v.channel_id:
+            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
+        v.state = "connecting"
+        try:
+            existing = guild.voice_client
+            if existing is not None and existing.is_connected():
+                await existing.move_to(channel)
+                v.vc = existing
+            else:
+                if existing is not None:
+                    await existing.disconnect(force=True)
+                v.vc = await channel.connect()
+        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}")
+            return
+        # Bounded wait for the shared PulseAudio socket (op25 container creates it).
+        if not await pulse.wait_until_ready():
+            v.state, v.code = "error", "no_audio"
+            v.error = "The node's audio path (PulseAudio) is not up."
+            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:
+            return
+        try:
+            await sess.client.change_presence(
+                activity=discord.Activity(type=discord.ActivityType.listening, name=name))
+            sess.presence = name
+        except Exception as e:
+            logger.warning(f"Presence update failed: {e}")
+
+    # ------------------------------------------------------------------
+    # Audio
+    # ------------------------------------------------------------------
+
+    def _play(self, v: _Voice, pulse_checked: bool = False) -> None:
+        vc = v.vc
+        if vc is None or not vc.is_connected():
+            return
+        if not pulse_checked and not pulse.is_ready():
+            logger.error(f"PulseAudio socket {pulse.socket_path()} missing — retrying audio in {STREAM_RETRY_DELAY}s.")
+            self._schedule_replay(v.stream_id)
+            return
+        try:
+            vc.play(self._audio_factory(), after=lambda err, sid=v.stream_id: self._on_audio_end(sid, err))
+        except Exception as e:
+            logger.error(f"Stream {v.stream_id}: could not start audio: {e}")
+            self._schedule_replay(v.stream_id)
+
+    def _on_audio_end(self, stream_id: str, error) -> None:
+        """discord.py calls this from its player thread."""
+        if error:
+            logger.error(f"Stream {stream_id} audio ended with error: {error}")
+        self._schedule_replay(stream_id, STREAM_RETRY_DELAY if error else 0.5)
+
+    def _schedule_replay(self, stream_id: str, delay: float = STREAM_RETRY_DELAY) -> None:
+        loop = self._loop
+        if loop is None or loop.is_closed():
+            return
+
+        async def _later():
+            await asyncio.sleep(delay)
+            v = self._voices.get(stream_id)
+            if v and v.vc and v.vc.is_connected() and not v.vc.is_playing():
+                self._play(v)
+
+        loop.call_soon_threadsafe(lambda: asyncio.ensure_future(_later()))
+
+    # ------------------------------------------------------------------
+    # Teardown
+    # ------------------------------------------------------------------
+
+    async def _disconnect(self, v: _Voice) -> None:
+        vc, v.vc = v.vc, None
+        if vc is None:
+            return
+        try:
+            if vc.is_playing():
+                vc.stop()
+            await vc.disconnect(force=True)
+        except Exception as e:
+            logger.warning(f"Stream {v.stream_id}: voice disconnect error: {e}")
+
+    async def _close_session(self, token_id: str) -> None:
+        sess = self._sessions.pop(token_id, None)
+        if sess is None:
+            return
+        for v in self._voices.values():
+            if v.token_id == token_id:
+                await self._disconnect(v)
+        await self._shutdown_client(sess)
+        logger.info(f"Radio bot {token_id} logged out")
+
+    async def _shutdown_client(self, sess: _Session) -> None:
+        try:
+            await sess.client.close()
+        except Exception as e:
+            logger.warning(f"Radio bot {sess.token_id} close error: {e}")
+        if sess.task and not sess.task.done():
+            sess.task.cancel()
+            try:
+                await sess.task
+            except BaseException:
+                pass
+
+    async def stop(self) -> None:
+        """Node shutdown: log every radio bot out."""
+        if self._watchdog:
+            self._watchdog.cancel()
+            self._watchdog = None
+        async with self._lock:
+            for v in list(self._voices.values()):
+                await self._disconnect(v)
+            self._voices.clear()
+            for tid in list(self._sessions):
+                await self._close_session(tid)
+            self._tokens.clear()
+
+    # ------------------------------------------------------------------
+    # Watchdog + MQTT-loss fail-safe
+    # ------------------------------------------------------------------
+
+    def _ensure_watchdog(self) -> None:
+        self._loop = asyncio.get_running_loop()
+        if self._watchdog is None or self._watchdog.done():
+            self._watchdog = asyncio.create_task(self._watchdog_loop())
+
+    async def _watchdog_loop(self) -> None:
+        while True:
+            await asyncio.sleep(WATCHDOG_INTERVAL)
+            try:
+                await self.tick()
+            except asyncio.CancelledError:
+                raise
+            except Exception as e:
+                logger.error(f"Discord watchdog error: {e}")
+
+    async def tick(self, now: Optional[float] = None) -> None:
+        """One watchdog pass: fail-safe check, then reconcile + report if needed."""
+        now = time.monotonic() if now is None else now
+        before = self.snapshot()
+        if self._mqtt_up():
+            self._mqtt_down_since = None
+            if self._failsafe:
+                logger.info("MQTT back — resuming Discord radio streams.")
+                self._failsafe = False
+        else:
+            if self._mqtt_down_since is None:
+                self._mqtt_down_since = now
+            elif (not self._failsafe and self._sessions
+                  and now - self._mqtt_down_since >= settings.discord_mqtt_loss_stop_seconds):
+                logger.warning("MQTT unreachable past the fail-safe — logging every radio bot out "
+                               "(C2 will hand these tokens to other nodes).")
+                self._failsafe = True
+        await self.reconcile()
+        self._ticks += 1
+        # On change, and every few passes regardless: a periodic report is what
+        # lets C2 settle a release whose first ack was lost.
+        if self.snapshot() != before or self._ticks % REPORT_EVERY_TICKS == 0:
+            await self.publish_actual()
+
+    # ------------------------------------------------------------------
+    # Reporting
+    # ------------------------------------------------------------------
+
+    def snapshot(self) -> dict:
+        streams = {}
+        for sid in self._desired:
+            v = self._voices.get(sid)
+            if self._paused:
+                streams[sid] = {"state": "paused_locally", "error": None, "code": None}
+            elif v is None:
+                streams[sid] = {"state": "connecting", "error": None, "code": None}
+            else:
+                streams[sid] = {"state": v.state, "error": v.error, "code": v.code}
+        return {
+            "applied_seq": self._applied_seq,
+            "sessions": sorted(tid for tid, s in self._sessions.items() if s.alive()),
+            "streams": streams,
+        }
+
+    async def publish_actual(self) -> None:
+        try:
+            await self._publish(self.snapshot())
+        except Exception as e:
+            logger.warning(f"Publishing Discord actual state failed: {e}")
+
+    # metadata_watcher hooks — audio plays continuously, nothing to do per call.
     def start_stream(self):
-        """Called when an OP25 call starts — stream plays continuously, nothing to do."""
         pass
 
     def stop_stream(self):
-        """Called when an OP25 call ends — stream plays continuously, nothing to do."""
         pass
 
-    async def stop(self):
-        self._guild_id = None
-        self._channel_id = None
-        self._was_streaming = False
-
-        if self._watchdog_task:
-            self._watchdog_task.cancel()
-            self._watchdog_task = None
-
-        await self.leave()
-
-        if self._task:
-            self._task.cancel()
-        if self._bot:
-            await self._bot.close()
-        self._bot = None
-        self._task = None
-        self._current_token = None
-        self._ready_event = None
-
-    def _play_stream(self):
-        """
-        Feed Discord voice straight from the PulseAudio monitor.
-
-        Icecast is NOT used here: it lags ~1 s at connect and drifts to 100 s+,
-        which is unusable for live listening. Icecast remains the frontend/mobile
-        listening path only.
-        """
-        if not self._voice_client:
-            return
-
-        if not pulse.is_ready():
-            logger.error(
-                f"PulseAudio socket {pulse.socket_path()} missing — "
-                f"cannot start Discord audio; retrying in {STREAM_RETRY_DELAY}s."
-            )
-            self._schedule_restart()
-            return
-
-        # before_options land ahead of -i, so this becomes:
-        #   ffmpeg -f pulse -i drb_sink.monitor …
-        source = discord.FFmpegPCMAudio(
-            settings.pulse_source,
-            before_options="-f pulse",
-        )
-        self._voice_client.play(
-            discord.PCMVolumeTransformer(source, volume=1.0),
-            after=self._on_stream_end,
-        )
-
-    def _schedule_restart(self, delay: float = STREAM_RETRY_DELAY):
-        """Re-arm the audio source after a delay — safe to call from any thread."""
-        if not self._loop:
-            return
-
-        async def _delayed_restart():
-            await asyncio.sleep(delay)
-            vc = self._voice_client
-            if vc and vc.is_connected() and not vc.is_playing():
-                self._play_stream()
-
-        self._loop.call_soon_threadsafe(lambda: asyncio.ensure_future(_delayed_restart()))
-
-    def _on_stream_end(self, error):
-        if error:
-            logger.error(f"Stream ended with error: {error}")
-        vc = self._voice_client
-        if not (self._loop and vc and vc.is_connected() and not vc.is_playing()):
-            return
-        if error:
-            # Back off before retrying — prevents a tight loop when PulseAudio is
-            # unavailable (FFmpeg exits immediately in that case).
-            self._schedule_restart()
-        else:
-            self._loop.call_soon_threadsafe(self._play_stream)
-
-    def _stop_stream(self):
-        if self._voice_client and self._voice_client.is_playing():
-            self._voice_client.stop()
-
-    async def _watchdog_loop(self):
-        """Periodically verify the voice connection is alive; rejoin silently if not."""
-        await asyncio.sleep(WATCHDOG_INTERVAL)  # give initial join time to settle
-        while True:
-            try:
-                await asyncio.sleep(WATCHDOG_INTERVAL)
-
-                # Only act if we're supposed to be in a channel
-                if not self._guild_id or not self._channel_id:
-                    continue
-
-                connected = self._voice_client is not None and self._voice_client.is_connected()
-                if not connected:
-                    logger.warning("Watchdog: voice connection lost — attempting rejoin.")
-                    await asyncio.sleep(REJOIN_DELAY)
-                    rejoined = await self.join(
-                        self._guild_id,
-                        self._channel_id,
-                        self._current_token,
-                    )
-                    if rejoined:
-                        logger.info("Watchdog: successfully rejoined voice channel.")
-                    else:
-                        logger.error("Watchdog: rejoin failed — will retry next cycle.")
-
-            except asyncio.CancelledError:
-                return
-            except Exception as e:
-                logger.error(f"Watchdog error: {e}")
-
-    async def _start_bot(self, token: str) -> bool:
-        await self.stop()  # clean up any previous instance
-
-        intents = discord.Intents.default()
-        intents.voice_states = True
-        intents.message_content = True
-        intents.messages = True
-        self._bot = commands.Bot(command_prefix="!", intents=intents)
-        self._ready_event = asyncio.Event()
-        self._current_token = token
-        self._loop = asyncio.get_event_loop()
-
-        @self._bot.event
-        async def on_ready():
-            logger.info(f"Discord bot ready: {self._bot.user} ({self._bot.user.id})")
-            self._ready_event.set()
-
-        @self._bot.event
-        async def on_voice_state_update(
-            member: discord.Member,
-            before: discord.VoiceState,
-            after: discord.VoiceState,
-        ):
-            """Detect when our own bot gets disconnected from a voice channel."""
-            if self._bot.user and member.id != self._bot.user.id:
-                return
-            if before.channel is not None and after.channel is None:
-                # Bot was disconnected (kicked or server drop).
-                # Don't null _voice_client here — Discord.py reconnects voice
-                # transparently and would leave _voice_client stale as None.
-                # The watchdog detects real disconnects via is_connected().
-                logger.warning("Bot was disconnected from voice channel — watchdog will rejoin if needed.")
-
-        @self._bot.event
-        async def on_message(message: discord.Message):
-            if message.author.bot:
-                return
-            if self._bot.user not in message.mentions:
-                return
-            content = message.content.lower()
-            if "leave" in content:
-                await self.leave()
-                try:
-                    await message.reply("Disconnected.")
-                except Exception:
-                    pass
-            elif "joinme" in content or "join" in content:
-                member = message.guild.get_member(message.author.id) if message.guild else None
-                vc = member.voice.channel if member and member.voice else None
-                if not vc:
-                    try:
-                        await message.reply("You're not in a voice channel.")
-                    except Exception:
-                        pass
-                    return
-                try:
-                    if self._voice_client and self._voice_client.is_connected():
-                        await self._voice_client.move_to(vc)
-                    else:
-                        self._voice_client = await vc.connect()
-                    self._channel_id = vc.id
-                    # A fresh connect() has no audio source attached yet.
-                    if not self._voice_client.is_playing():
-                        self._play_stream()
-                    await message.reply(f"Joined {vc.name}.")
-                except Exception as e:
-                    logger.error(f"joinme failed: {e}")
-
-        self._task = asyncio.create_task(self._bot.start(token))
-
-        try:
-            await asyncio.wait_for(self._ready_event.wait(), timeout=BOT_READY_TIMEOUT)
-            self._watchdog_task = asyncio.create_task(self._watchdog_loop())
-            return True
-        except asyncio.TimeoutError:
-            logger.error("Timed out waiting for Discord bot to become ready.")
-            await self.stop()
-            return False
-
-    def _is_bot_running(self) -> bool:
-        return (
-            self._bot is not None
-            and self._task is not None
-            and not self._task.done()
-        )
-
     @property
     def is_connected(self) -> bool:
-        return self._voice_client is not None and self._voice_client.is_connected()
+        return any(v.vc is not None and v.vc.is_connected() for v in self._voices.values())
 
 
-radio_bot = RadioBot()
+radio_bot = RadioManager()

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

@@ -11,6 +11,7 @@ from app.internal import credentials, key_pickup
 CommandCallback = Callable[[Dict[str, Any]], Awaitable[None]]
 ConfigCallback = Callable[[Dict[str, Any]], Awaitable[None]]
 ApiKeyCallback = Callable[[Dict[str, Any]], Awaitable[None]]
+DesiredCallback = Callable[[Dict[str, Any]], Awaitable[None]]
 
 # CONNACK codes meaning "your credentials are wrong", as paho reports them:
 # MQTT 3.1.1 return codes 4/5, and the MQTT 5 reason codes paho maps them to
@@ -46,6 +47,7 @@ class MQTTManager:
         self.on_command: Optional[CommandCallback] = None
         self.on_config_push: Optional[ConfigCallback] = None
         self.on_api_key: Optional[ApiKeyCallback] = None
+        self.on_discord_desired: Optional[DesiredCallback] = None
 
         self._offline_buffer = deque(maxlen=settings.offline_call_buffer_size)
 
@@ -55,6 +57,10 @@ class MQTTManager:
         self._t_metadata = f"nodes/{nid}/metadata"
         self._t_commands = f"nodes/{nid}/commands"
         self._t_config = f"nodes/{nid}/config"
+        # Discord radio desired state (retained by C2) and our actual state —
+        # node-26#25. Token IDs only; the strings come over HTTPS.
+        self._t_discord_desired = f"nodes/{nid}/discord/desired"
+        self._t_discord_actual = f"nodes/{nid}/discord/actual"
         # TODO(mqtt-cutover): dead once enrollment lands client-side. This
         # was the pre-dynsec key-delivery path (server retain-publishes the
         # api_key here after admin approval; node asks for redelivery via
@@ -131,6 +137,7 @@ class MQTTManager:
             self._connected = True
             client.subscribe(self._t_commands, qos=1)
             client.subscribe(self._t_config, qos=1)
+            client.subscribe(self._t_discord_desired, qos=1)
             client.subscribe(self._t_api_key, qos=2)  # TODO(mqtt-cutover): see _t_api_key comment above
             client.subscribe(self._t_discovery, qos=0)
             logger.info("MQTT connected.")
@@ -159,6 +166,8 @@ class MQTTManager:
             asyncio.run_coroutine_threadsafe(self.on_command(payload), self._loop)
         elif msg.topic == self._t_config and self.on_config_push:
             asyncio.run_coroutine_threadsafe(self.on_config_push(payload), self._loop)
+        elif msg.topic == self._t_discord_desired and self.on_discord_desired:
+            asyncio.run_coroutine_threadsafe(self.on_discord_desired(payload), self._loop)
         elif msg.topic == self._t_api_key and self.on_api_key:
             asyncio.run_coroutine_threadsafe(self.on_api_key(payload), self._loop)
         elif msg.topic == self._t_discovery:
@@ -285,6 +294,12 @@ class MQTTManager:
         else:
             self._publish(self._t_metadata, payload, qos=1)
 
+    def publish_discord_actual(self, payload: dict) -> None:
+        """What the Discord radio manager is actually doing. Not buffered while
+        offline: a fresh report is sent after every change and periodically,
+        and only the newest one means anything."""
+        self._publish(self._t_discord_actual, payload, qos=1)
+
     async def _flush_offline_buffer(self):
         if not self._offline_buffer:
             return

drb-edge-node/app/main.py

@@ -144,20 +144,11 @@ async def on_command(payload: dict):
     action = payload.get("action")
     logger.info(f"Command received: {action}")
 
-    if action == "discord_join":
-        token = payload.get("token")
-        if not token:
-            logger.error("discord_join command missing token — ignoring.")
-            return
-        await radio_bot.join(
-            guild_id=int(payload["guild_id"]),
-            channel_id=int(payload["channel_id"]),
-            token=token,
-            call_active=metadata_watcher.is_active,
-            system_name=payload.get("system_name"),
-        )
-    elif action == "discord_leave":
-        await radio_bot.leave()
+    if action in ("discord_join", "discord_leave"):
+        # Pre-desired-state C2. Discord streams now arrive on
+        # nodes/{id}/discord/desired (node-26#25); a raw-token join is refused
+        # so a bot token is never taken from an MQTT command again.
+        logger.warning(f"Ignoring legacy {action} command — Discord radio is driven by desired state now.")
     elif action in ("set_sdr_config", "set_secondary_priority"):
         from app.internal.sdr_settings import set_sdr_settings
         try:
@@ -300,8 +291,9 @@ async def lifespan(app: FastAPI):
     mqtt_manager.on_command = on_command
     mqtt_manager.on_config_push = on_config_push
     mqtt_manager.on_api_key = on_api_key
+    mqtt_manager.on_discord_desired = radio_bot.apply_desired
 
-    # Start services (radio_bot starts on-demand when a discord_join command arrives)
+    # Start services (radio_bot starts when C2's retained discord/desired arrives)
     await mqtt_manager.connect()
     await metadata_watcher.start()
     await call_recorder.start()  # persistent PulseAudio ring buffer

drb-edge-node/app/routers/api.py

@@ -400,15 +400,25 @@ async def set_sdr(body: SdrSettingsBody):
     return {"ok": True, "running": running}
 
 
+@router.get("/discord")
+async def discord_status():
+    """What the node's radio bots are doing (desired state from C2, actual here)."""
+    return radio_bot.snapshot()
+
+
 @router.post("/discord/join")
-async def discord_join(guild_id: int, channel_id: int):
-    ok = await radio_bot.join(guild_id, channel_id)
-    if not ok:
-        raise HTTPException(500, "Failed to join voice channel")
-    return {"ok": True}
+async def discord_join():
+    # node-26#25 item 4: this used to call join() without a token and 500 every
+    # time. A node cannot lease a radio bot by itself — C2 owns the pool.
+    raise HTTPException(
+        409,
+        "Discord radio channels are started from Discord (/join) or the DRB web app; "
+        "the node cannot pick a radio bot on its own.",
+    )
 
 
 @router.post("/discord/leave")
 async def discord_leave():
-    await radio_bot.leave()
+    """Local stop: log every radio bot out until DRB sends a new channel list."""
+    await radio_bot.pause_locally()
     return {"ok": True}

drb-edge-node/tests/test_discord_radio.py

@@ -0,0 +1,260 @@
+"""
+Edge Discord radio manager (node-26#25): desired-state reconcile, logout on
+release, watchdog that survives a dead session, MQTT-loss fail-safe.
+
+Discord itself is faked: FakeClient logs "in" by firing on_ready, guilds and
+channels are plain objects, voice connections record play/disconnect.
+"""
+import asyncio
+
+import discord
+import pytest
+
+from app.internal import discord_radio
+from app.internal.discord_radio import RadioManager
+
+
+class FakeVC:
+    def __init__(self, channel):
+        self.channel = channel
+        self.connected = True
+        self.playing = False
+        self.plays = 0
+
+    def is_connected(self):
+        return self.connected
+
+    def is_playing(self):
+        return self.playing
+
+    def play(self, source, after=None):
+        self.playing = True
+        self.plays += 1
+
+    def stop(self):
+        self.playing = False
+
+    async def disconnect(self, force=False):
+        self.connected = False
+        self.playing = False
+        self.channel.guild.voice_client = None
+
+    async def move_to(self, channel):
+        self.channel = channel
+        channel.guild.voice_client = self
+
+
+class FakeChannel:
+    def __init__(self, guild, cid, kind=discord.ChannelType.voice):
+        self.guild, self.id, self.type = guild, cid, kind
+
+    async def connect(self):
+        vc = FakeVC(self)
+        self.guild.voice_client = vc
+        return vc
+
+
+class FakeGuild:
+    def __init__(self, gid, channel_ids):
+        self.id = gid
+        self.voice_client = None
+        self.channels = {c: FakeChannel(self, c) for c in channel_ids}
+
+    def get_channel(self, cid):
+        return self.channels.get(cid)
+
+
+class FakeClient:
+    instances: list = []
+
+    def __init__(self, guilds, fail_login=False):
+        self.guilds = guilds
+        self.fail_login = fail_login
+        self.closed = False
+        self.handlers = {}
+        self.token = None
+        self.user = "radio-bot"
+        FakeClient.instances.append(self)
+
+    def event(self, coro):
+        self.handlers[coro.__name__] = coro
+        return coro
+
+    async def start(self, token):
+        self.token = token
+        if self.fail_login:
+            raise discord.LoginFailure("Improper token has been passed.")
+        await self.handlers["on_ready"]()
+        while not self.closed:
+            await asyncio.sleep(0.01)
+
+    async def close(self):
+        self.closed = True
+
+    def is_closed(self):
+        return self.closed
+
+    def get_guild(self, gid):
+        return self.guilds.get(gid)
+
+    async def change_presence(self, activity=None):
+        self.presence = activity
+
+
+@pytest.fixture
+def env(monkeypatch):
+    FakeClient.instances = []
+    guilds = {1: FakeGuild(1, [11, 12]), 2: FakeGuild(2, [21])}
+    published = []
+    fetched = []
+    state = {"mqtt": True, "fail_login": False}
+
+    async def fetch(token_id):
+        fetched.append(token_id)
+        return f"secret-{token_id}"
+
+    async def publish(payload):
+        published.append(payload)
+
+    async def pulse_ready(timeout=None):
+        return True
+    monkeypatch.setattr(discord_radio.pulse, "wait_until_ready", pulse_ready)
+    monkeypatch.setattr(discord_radio.pulse, "is_ready", lambda: True)
+
+    mgr = RadioManager(
+        client_factory=lambda: FakeClient(guilds, fail_login=state["fail_login"]),
+        token_fetcher=fetch, publisher=publish, audio_factory=lambda: object(),
+        mqtt_up=lambda: state["mqtt"], ready_timeout=1,
+    )
+    yield type("Env", (), {"mgr": mgr, "guilds": guilds, "published": published,
+                           "fetched": fetched, "state": state})
+    # Leave no background tasks behind.
+    if mgr._watchdog:
+        mgr._watchdog.cancel()
+
+
+def _desired(seq, *streams):
+    return {"seq": seq, "streams": [
+        {"stream_id": f"{g}:{c}", "guild_id": str(g), "channel_id": str(c), "token_id": t, "system_name": "County"}
+        for g, c, t in streams
+    ]}
+
+
+async def test_apply_connects_and_reports(env):
+    await env.mgr.apply_desired(_desired(1, (1, 11, "t1")))
+    actual = env.published[-1]
+    assert actual["applied_seq"] == 1
+    assert actual["sessions"] == ["t1"]
+    assert actual["streams"] == {"1:11": {"state": "connected", "error": None, "code": None}}
+    assert env.mgr.is_connected
+    vc = env.guilds[1].voice_client
+    assert vc.plays == 1 and vc.channel.id == 11
+
+
+async def test_one_session_serves_two_guilds(env):
+    await env.mgr.apply_desired(_desired(1, (1, 11, "t1"), (2, 21, "t1")))
+    assert len(FakeClient.instances) == 1
+    assert env.published[-1]["sessions"] == ["t1"]
+    assert env.guilds[1].voice_client.is_connected() and env.guilds[2].voice_client.is_connected()
+
+
+async def test_dropping_the_last_stream_logs_the_bot_out(env):
+    """node-26#25 item 1: leave() used to keep the gateway session alive."""
+    await env.mgr.apply_desired(_desired(1, (1, 11, "t1"), (2, 21, "t1")))
+    client = FakeClient.instances[0]
+    await env.mgr.apply_desired(_desired(2, (2, 21, "t1")))
+    assert not client.closed                      # still needed for guild 2
+    assert env.guilds[1].voice_client is None
+    await env.mgr.apply_desired(_desired(3))
+    assert client.closed                          # logged OUT, not just out of voice
+    assert env.published[-1] == {"applied_seq": 3, "sessions": [], "streams": {}}
+
+
+async def test_stale_desired_is_ignored(env):
+    await env.mgr.apply_desired(_desired(5, (1, 11, "t1")))
+    await env.mgr.apply_desired(_desired(4))
+    assert env.mgr.snapshot()["applied_seq"] == 5 and env.mgr.is_connected
+
+
+async def test_dead_session_is_recovered_by_the_watchdog(env):
+    """node-26#25 item 2: the old watchdog cancelled itself on rejoin."""
+    await env.mgr.apply_desired(_desired(1, (1, 11, "t1")))
+    first = FakeClient.instances[0]
+    first.closed = True                           # bot task dies
+    await asyncio.sleep(0.05)
+    env.guilds[1].voice_client.connected = False
+    await env.mgr.tick()
+    assert len(FakeClient.instances) == 2 and not FakeClient.instances[1].closed
+    assert env.mgr.snapshot()["streams"]["1:11"]["state"] == "connected"
+    # The watchdog task itself is still alive after the recovery.
+    assert env.mgr._watchdog is not None and not env.mgr._watchdog.done()
+
+
+async def test_dropped_voice_is_reconnected(env):
+    await env.mgr.apply_desired(_desired(1, (1, 11, "t1")))
+    env.guilds[1].voice_client.connected = False
+    env.guilds[1].voice_client = None
+    await env.mgr.tick()
+    assert env.guilds[1].voice_client.is_connected()
+
+
+async def test_bot_not_in_guild_is_reported_not_swallowed(env):
+    await env.mgr.apply_desired(_desired(1, (9, 91, "t1")))
+    st = env.published[-1]["streams"]["9:91"]
+    assert st["state"] == "error" and st["code"] == "not_in_guild"
+
+
+async def test_login_failure_is_reported(env):
+    env.state["fail_login"] = 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"] == "login_failed"
+    assert env.published[-1]["sessions"] == []
+
+
+async def test_mqtt_loss_failsafe_logs_out_then_resumes(env, monkeypatch):
+    monkeypatch.setattr(discord_radio.settings, "discord_mqtt_loss_stop_seconds", 300)
+    await env.mgr.apply_desired(_desired(1, (1, 11, "t1")))
+    client = FakeClient.instances[0]
+    env.state["mqtt"] = False
+    await env.mgr.tick(now=1000.0)
+    assert not client.closed                       # inside the window
+    await env.mgr.tick(now=1299.0)
+    assert not client.closed
+    await env.mgr.tick(now=1300.0)
+    assert client.closed and env.mgr.snapshot()["sessions"] == []
+    # Token strings are not kept across a fail-safe logout: re-login refetches
+    # from C2, which refuses if the lease has moved on.
+    assert env.mgr._tokens == {}
+    env.state["mqtt"] = True
+    before = len(env.fetched)
+    await env.mgr.tick(now=1400.0)
+    assert len(env.fetched) == before + 1
+    assert env.mgr.snapshot()["streams"]["1:11"]["state"] == "connected"
+
+
+async def test_token_fetch_refused_reports_error(env):
+    async def refuse(token_id):
+        return None
+    env.mgr._fetch_token = refuse
+    await env.mgr.apply_desired(_desired(1, (1, 11, "t1")))
+    st = env.published[-1]["streams"]["1:11"]
+    assert st["state"] == "error" and FakeClient.instances == []
+
+
+async def test_local_pause_logs_out_until_next_desired(env):
+    await env.mgr.apply_desired(_desired(1, (1, 11, "t1")))
+    await env.mgr.pause_locally()
+    assert FakeClient.instances[0].closed
+    assert env.published[-1]["streams"]["1:11"]["state"] == "paused_locally"
+    await env.mgr.apply_desired(_desired(2, (1, 11, "t1")))
+    assert env.mgr.is_connected
+
+
+async def test_no_message_content_intent_on_real_client():
+    client = discord_radio._default_client_factory()
+    try:
+        assert client.intents.message_content is False
+        assert client.intents.voice_states and client.intents.guilds
+    finally:
+        await client.close()