Discord radio channels: wire leases + desired-state streams (server-26#203, #204)

c2-core
- discord_streams/discord_leases wired in: POST /discord/streams,
  POST /discord/guilds/{id}/stop, GET /discord/guilds/{id}/overview,
  GET /discord/streams/{id}, PUT /discord/guilds/{id}/link (admin),
  GET /discord/node/{node}/tokens/{id} (node key, lease holder only).
- Retained nodes/{id}/discord/desired (token ids, never token strings),
  nodes/+/discord/actual subscribed and handled; per-node publish lock keeps
  the retained list the newest; failed publishes retried by the reconciler.
- Tokens freed only after the node acks logout (or offline past
  discord_lease_offline_release_seconds) - sweeper and C2 startup no longer
  release leases.
- /nodes/{id}/command discord_join/leave map onto streams; CommandPayload
  keeps preferred_token_id. /tokens: admin-only list, validate on add against
  Discord, refresh, flush via transactional force release.
- Guild -> org via discord_guilds, unlinked guilds fall back to
  DISCORD_UNLINKED_GUILD_ORG_ID (default "founding").
- tests/fake_firestore.py: optimistic-transaction fake; 27 lease/race tests.

discord bot
- /join, /leave, /status, autocomplete scoped to the calling guild; /leave
  never touches another server; Manage Channels default permission;
  /join edits its reply with the node's real result; /help copy fixed.

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

drb-c2-core/app/config.py

@@ -210,6 +210,24 @@ class Settings(BaseSettings):
     # deployment will set this up, and skipping it must be silent.
     ai_alert_webhook_url: str = ""
 
+    # --- Discord radio channels (server-26#203/#204, node-26#25) -----------
+    # Voice connections one node may carry at once (≈64 kbps upstream each —
+    # this is what protects an LTE node).
+    discord_node_max_streams: int = 4
+    # Radio channels one Discord server may have on at once.
+    discord_guild_max_streams: int = 4
+    # How often the reconciler re-places persistent streams / retries publishes.
+    discord_reconcile_interval_seconds: int = 30
+    # A node offline this long has its radio-bot tokens returned to the pool.
+    # MUST stay above the edge's DISCORD_MQTT_LOSS_STOP_SECONDS (default 300):
+    # by then the edge has logged every radio bot out by itself, so the same
+    # token can never end up logged in on two nodes.
+    discord_lease_offline_release_seconds: int = 600
+    # Org a Discord server that was never linked to an org (discord_guilds/{id})
+    # acts for. Default keeps today's single-org deployment working unchanged;
+    # set it to "" to require every server to be linked first.
+    discord_unlinked_guild_org_id: str = "founding"
+
     class Config:
         env_file = ".env"
 

drb-c2-core/app/internal/discord_leases.py

@@ -318,7 +318,11 @@ async def release_node_leases(node_id: str) -> list[str]:
 
 
 async def force_release_token(token_id: str) -> Optional[dict]:
-    """Admin override (/tokens/flush, Force release). Returns the old lease."""
+    """Admin override (/tokens/flush, delete). Frees the token and detaches
+    every stream that named it, in one transaction, so no stream doc is left
+    pointing at a token whose lease no longer lists it. Returns the old lease
+    (``None`` if the token was not leased or does not exist); the caller
+    republishes desired state for ``lease["node_id"]`` so the edge logs out."""
     db = fstore.db
 
     def _txn(txn):
@@ -327,7 +331,83 @@ async def force_release_token(token_id: str) -> Optional[dict]:
         if not snap.exists:
             return None
         lease = snap.to_dict().get("lease")
+        stream_refs = []
+        for sid in (lease or {}).get("streams") or []:
+            sref = db.collection(STREAMS).document(sid)
+            ssnap = sref.get(transaction=txn)
+            if ssnap.exists and (ssnap.to_dict() or {}).get("token_id") == token_id:
+                stream_refs.append(sref)
         txn.update(ref, _legacy_mirror(None))
+        for sref in stream_refs:
+            txn.set(sref, {"token_id": None, "lease_id": None}, merge=True)
         return lease
 
     return await fstore.run_transaction(_txn)
+
+
+# ---------------------------------------------------------------------------
+# Token validation / guild membership (server-26#204 item 5)
+# ---------------------------------------------------------------------------
+
+DISCORD_API = "https://discord.com/api/v10"
+
+
+class TokenRejected(Exception):
+    """Discord answered 401: the string is not a working bot token."""
+
+
+async def inspect_token(token: str) -> dict:
+    """Ask Discord who this token is and which servers it is in.
+
+    Returns ``{bot_user_id, bot_name, application_id, guild_ids, valid,
+    validated_at, check_error}``. Raises TokenRejected on a 401. Any other
+    failure (Discord down, rate limit) is reported in ``check_error`` with
+    ``valid``/``guild_ids`` = None — unknown, not bad: a token is never
+    rejected because Discord had a bad minute.
+    """
+    import httpx
+    headers = {"Authorization": f"Bot {token}"}
+    out: dict = {"valid": None, "guild_ids": None, "check_error": None,
+                 "validated_at": _now().isoformat()}
+    try:
+        async with httpx.AsyncClient(timeout=10, base_url=DISCORD_API, headers=headers) as client:
+            me = await client.get("/users/@me")
+            if me.status_code == 401:
+                raise TokenRejected("Discord rejected this bot token (401).")
+            me.raise_for_status()
+            user = me.json()
+            out["bot_user_id"] = str(user.get("id"))
+            out["bot_name"] = user.get("username")
+            app = await client.get("/oauth2/applications/@me")
+            if app.status_code == 200:
+                out["application_id"] = str(app.json().get("id"))
+            guilds = await client.get("/users/@me/guilds", params={"limit": 200})
+            guilds.raise_for_status()
+            out["guild_ids"] = [str(g["id"]) for g in guilds.json()]
+            out["valid"] = True
+    except TokenRejected:
+        raise
+    except Exception as e:  # network / 5xx / 429 — unknown, not invalid
+        out["check_error"] = str(e)[:200]
+    return out
+
+
+async def refresh_membership(token_ids: list[str]) -> int:
+    """Re-read guild membership for these tokens. Returns how many changed."""
+    changed = 0
+    for tok_id in token_ids:
+        tok = await fstore.doc_get(TOKENS, tok_id)
+        if not tok or not tok.get("token"):
+            continue
+        try:
+            info = await inspect_token(tok["token"])
+        except TokenRejected as e:
+            await fstore.doc_update(TOKENS, tok_id, {"valid": False, "check_error": str(e),
+                                                     "validated_at": _now().isoformat()})
+            changed += 1
+            continue
+        updates = {k: v for k, v in info.items() if v is not None or k == "check_error"}
+        if updates.get("guild_ids") != tok.get("guild_ids") or updates.get("valid") != tok.get("valid"):
+            changed += 1
+        await fstore.doc_update(TOKENS, tok_id, updates)
+    return changed

drb-c2-core/app/internal/discord_streams.py

@@ -1,5 +1,5 @@
 """
-Discord radio channels as desired state (server-26#203 MVP).
+Discord radio channels as desired state (server-26#203, #204; node-26#25).
 
 One ``discord_streams/{guild_id}:{channel_id}`` doc per Discord voice channel
 that should carry radio. Fields:
@@ -7,14 +7,15 @@ that should carry radio. Fields:
   org_id, guild_id, channel_id, channel_name, guild_name,
   system_id, system_name,
   mode: "persistent" | "adhoc"
-      persistent — stays connected and reconnects by itself (node reboot,
-                   Discord voice drop, C2 restart) until someone stops it.
-      adhoc      — the Discord bot sets desired=off after ``idle_minutes``
-                   with no humans in the voice channel.
+      persistent — stays connected and is re-placed by the reconciler (node
+                   reboot, node lost, C2 restart) until someone stops it.
+      adhoc      — stopped instead of re-placed when its node goes away.
+                   (Idle auto-dismiss is not built yet — see the PR.)
   desired: "on" | "off"
   node_id, token_id, lease_id — placement (written with the lease, atomically)
   actual: {state, error, since, at} — reported by the node, never guessed here
-      state: pending | connecting | connected | error | stopped | node_offline
+      state: pending | connecting | connected | error | stopping |
+             node_offline | paused_locally
   tg_filter: None — phase-2 seam (per-channel talkgroup gate); always None now
   now_playing_message_id, created_by_discord_id, created_at, updated_at
 
@@ -23,12 +24,13 @@ retains ``nodes/{node}/discord/desired`` (full list for that node, with a
 monotonically increasing ``seq``) -> the edge reconciles and publishes
 ``nodes/{node}/discord/actual`` -> ``handle_actual`` updates ``actual`` and
 settles token releases. ``reconcile_loop`` re-places persistent streams whose
-node went away.
+node went away and retries desired-state publishes that never reached the
+broker.
 
 The desired payload carries token **ids**, never token strings: the edge
-fetches the string over HTTPS (GET /discord/node/tokens/{id}) with its own
-node API key, and C2 hands it out only to the node holding the lease. A
-retained MQTT message therefore never stores a bot credential on the broker.
+fetches the string over HTTPS (GET /discord/node/{node_id}/tokens/{id}) with
+its own node API key, and C2 hands it out only to the node holding the lease.
+A retained MQTT message therefore never stores a bot credential on the broker.
 """
 from __future__ import annotations
 
@@ -42,7 +44,8 @@ from app.internal import discord_leases as leases
 from app.internal.logger import logger
 
 STREAMS = leases.STREAMS
-NODE_STATE = "discord_nodes"   # discord_nodes/{node_id}: {seq, applied_seq, actual_at}
+GUILDS = "discord_guilds"      # discord_guilds/{guild_id}: {org_id, linked_by, linked_at}
+NODE_STATE = "discord_nodes"   # discord_nodes/{node_id}: {seq, published, applied_seq, actual_at}
 ONLINE = ("online", "recording")
 
 
@@ -56,33 +59,56 @@ def _effective_system(node: dict) -> Optional[str]:
     return node.get("assigned_system_id")
 
 
-async def _node_stream_counts(org_id: Optional[str]) -> dict[str, int]:
+def _parse_ts(value) -> Optional[datetime]:
+    try:
+        ts = datetime.fromisoformat(value) if isinstance(value, str) else value
+    except (TypeError, ValueError):
+        return None
+    if isinstance(ts, datetime) and ts.tzinfo is None:
+        ts = ts.replace(tzinfo=timezone.utc)
+    return ts if isinstance(ts, datetime) else None
+
+
+# ---------------------------------------------------------------------------
+# Guild -> org
+# ---------------------------------------------------------------------------
+
+async def resolve_guild_org(guild_id: str) -> Optional[str]:
+    """The org a Discord server acts for: its ``discord_guilds`` link, else
+    ``settings.discord_unlinked_guild_org_id`` (None when that is empty)."""
+    link = await fstore.doc_get(GUILDS, str(guild_id))
+    if link and link.get("org_id"):
+        return link["org_id"]
+    return settings.discord_unlinked_guild_org_id or None
+
+
+async def _node_stream_counts(exclude_stream_id: Optional[str] = None) -> dict[str, int]:
     streams = await fstore.collection_list(STREAMS, desired="on")
     counts: dict[str, int] = {}
     for s in streams:
-        if s.get("node_id"):
+        if s.get("node_id") and s.get("stream_id") != exclude_stream_id:
             counts[s["node_id"]] = counts.get(s["node_id"], 0) + 1
     return counts
 
 
-async def candidate_nodes(org_id: str, system_id: str, exclude: Optional[str] = None) -> list[dict]:
+def _node_usable(n: dict, org_id: str, system_id: str) -> bool:
+    return (
+        n.get("org_id") == org_id
+        and n.get("status") in ONLINE
+        and n.get("approval_status") in (None, "approved")
+        and _effective_system(n) == system_id
+    )
+
+
+async def candidate_nodes(org_id: str, system_id: str, exclude_stream_id: Optional[str] = None) -> list[dict]:
     """Online, approved nodes of this org currently listening to ``system_id``,
     best first: nodes already carrying streams (one gateway session can serve
     several guilds) but under the per-node cap, then the rest."""
     nodes = await fstore.collection_list("nodes", org_id=org_id)
-    counts = await _node_stream_counts(org_id)
+    counts = await _node_stream_counts(exclude_stream_id)
     cap = settings.discord_node_max_streams
-    out = []
-    for n in nodes:
-        if n.get("node_id") == exclude:
-            continue
-        if n.get("status") not in ONLINE or n.get("approval_status") not in (None, "approved"):
-            continue
-        if _effective_system(n) != system_id:
-            continue
-        if counts.get(n["node_id"], 0) >= cap:
-            continue
-        out.append(n)
+    out = [n for n in nodes
+           if n.get("node_id") and _node_usable(n, org_id, system_id) and counts.get(n["node_id"], 0) < cap]
     out.sort(key=lambda n: (-counts.get(n["node_id"], 0), n.get("name") or n["node_id"]))
     return out
 
@@ -91,13 +117,27 @@ async def candidate_nodes(org_id: str, system_id: str, exclude: Optional[str] =
 # Desired state -> node
 # ---------------------------------------------------------------------------
 
+# One c2-core process publishes desired state. Serialising bump->build->publish
+# per node is what keeps the RETAINED message the newest one: two unserialised
+# publishes could land seq 7 then seq 6, and a reconnecting edge would then be
+# handed the stale list.
+_publish_locks: dict[str, asyncio.Lock] = {}
+
+
+def _lock_for(node_id: str) -> asyncio.Lock:
+    lock = _publish_locks.get(node_id)
+    if lock is None:
+        lock = _publish_locks[node_id] = asyncio.Lock()
+    return lock
+
+
 async def _bump_seq(node_id: str) -> int:
     db = fstore.db
 
     def _txn(txn):
         ref = db.collection(NODE_STATE).document(node_id)
         snap = ref.get(transaction=txn)
-        seq = int((snap.to_dict() or {}).get("seq") or 0) + 1 if snap.exists else 1
+        seq = int(((snap.to_dict() or {}) if snap.exists else {}).get("seq") or 0) + 1
         txn.set(ref, {"node_id": node_id, "seq": seq}, merge=True)
         return seq
 
@@ -125,14 +165,23 @@ async def desired_for_node(node_id: str) -> list[dict]:
     return out
 
 
-async def publish_desired(node_id: str) -> int:
-    """Retain the node's full stream list; stamp releasing tokens with the seq."""
+async def publish_desired(node_id: str) -> bool:
+    """Retain the node's full stream list under a fresh ``seq``.
+
+    Releasing tokens are stamped with that seq BEFORE the publish, so the
+    earliest ack that can free them is one for a list that already lacks them.
+    Returns whether the broker took the message; a miss is recorded and the
+    reconciler retries it (nothing is lost — the streams stay desired)."""
     from app.internal.mqtt_handler import mqtt_handler
-    seq = await _bump_seq(node_id)
-    payload = {"seq": seq, "streams": await desired_for_node(node_id), "issued_at": _now_iso()}
-    mqtt_handler.publish_discord_desired(node_id, payload)
-    await leases.stamp_release_seq(node_id, seq)
-    return seq
+    async with _lock_for(node_id):
+        seq = await _bump_seq(node_id)
+        payload = {"seq": seq, "streams": await desired_for_node(node_id), "issued_at": _now_iso()}
+        await leases.stamp_release_seq(node_id, seq)
+        ok = mqtt_handler.publish_discord_desired(node_id, payload)
+        await fstore.doc_set(NODE_STATE, node_id, {"published": bool(ok), "published_at": _now_iso()})
+    if not ok:
+        logger.warning(f"Discord desired state for {node_id} (seq {seq}) not delivered — reconciler will retry")
+    return bool(ok)
 
 
 # ---------------------------------------------------------------------------
@@ -149,11 +198,12 @@ class StreamError(Exception):
 
 
 ERROR_TEXT = {
+    "no_org": "This Discord server is not linked to a DRB organization yet.",
     "no_node": "No online node in your organization is listening to that system right now.",
     "pool_empty": "No radio bots have been added to DRB yet. A platform admin adds them under Radio bots.",
     "no_bot_in_guild": "No free radio bot is a member of this server. Invite one with the link below, then try again.",
     "pool_exhausted": "Every radio bot that can join this server is already busy. Stop another radio channel or add a bot.",
-    "limit": "This server already has its maximum number of radio channels. Stop one with /radio stop first.",
+    "limit": "This server already has its maximum number of radio channels. Stop one with /leave first.",
 }
 
 
@@ -175,6 +225,9 @@ async def start_stream(
     it for a channel that is already on reuses the node and the lease."""
     if mode not in ("persistent", "adhoc"):
         raise StreamError("bad_mode", "mode must be persistent or adhoc", status=400)
+    if not org_id:
+        raise StreamError("no_org", ERROR_TEXT["no_org"], status=403)
+    guild_id, channel_id = str(guild_id), str(channel_id)
     stream_id = leases.stream_id_for(guild_id, channel_id)
     existing = await fstore.doc_get(STREAMS, stream_id) or {}
 
@@ -186,9 +239,14 @@ async def start_stream(
     system = await fstore.doc_get_cached("systems", system_id) or {}
 
     if node_id:
-        candidates = [await fstore.doc_get("nodes", node_id) or {"node_id": node_id}]
+        node = await fstore.doc_get("nodes", node_id)
+        # A pinned node still has to be this org's and approved; it may be
+        # listening to anything (the web "Join Discord" button pins a node).
+        if not node or node.get("org_id") != org_id or node.get("approval_status") not in (None, "approved"):
+            raise StreamError("no_node", ERROR_TEXT["no_node"])
+        candidates = [node]
     else:
-        candidates = await candidate_nodes(org_id, system_id)
+        candidates = await candidate_nodes(org_id, system_id, exclude_stream_id=stream_id)
         # Keep the current node when it is still a valid choice (reuse the lease).
         cur = existing.get("node_id")
         if cur and existing.get("system_id") == system_id:
@@ -200,10 +258,12 @@ async def start_stream(
     fields = {
         "stream_id": stream_id, "org_id": org_id, "guild_id": guild_id, "channel_id": channel_id,
         "system_id": system_id, "system_name": system.get("name") or system_id,
-        "mode": mode, "desired": "on", "tg_filter": None,
-        "actual": {"state": "pending", "error": None, "since": now, "at": now},
+        "mode": mode, "desired": "on", "tg_filter": None, "stop_reason": None,
         "updated_at": now,
     }
+    prev_actual = existing.get("actual") or {}
+    if existing.get("desired") != "on" or prev_actual.get("state") not in ("connected", "connecting"):
+        fields["actual"] = {"state": "pending", "error": None, "since": now, "at": now}
     if channel_name:
         fields["channel_name"] = channel_name
     if guild_name:
@@ -215,19 +275,31 @@ async def start_stream(
 
     last: Optional[leases.LeaseResult] = None
     prev_node = existing.get("node_id")
+    refreshed = False
     for node in candidates:
         res = await leases.lease_for_stream(stream_id, node_id=node["node_id"], org_id=org_id,
                                             preferred_token_id=preferred_token_id, stream_fields=fields)
+        if not res.ok and res.error == "no_bot_in_guild" and not refreshed:
+            # Membership is cached on the token doc; someone may have just
+            # invited a bot. Re-read it from Discord once, then retry.
+            refreshed = True
+            tokens = await fstore.collection_list(leases.TOKENS)
+            stale = [t["token_id"] for t in tokens
+                     if t.get("token_id") and t.get("guild_ids") is not None and guild_id not in t["guild_ids"]]
+            if await leases.refresh_membership(stale):
+                res = await leases.lease_for_stream(stream_id, node_id=node["node_id"], org_id=org_id,
+                                                    preferred_token_id=preferred_token_id, stream_fields=fields)
         if res.ok:
-            await publish_desired(node["node_id"])
+            delivered = await publish_desired(node["node_id"])
             if prev_node and prev_node != node["node_id"]:
                 await publish_desired(prev_node)
             logger.info(f"Discord stream {stream_id} -> node {node['node_id']} token {res.token_id}"
                         f" ({'reused' if res.reused else 'new'} lease)")
             return {**existing, **fields, "node_id": node["node_id"], "node_name": node.get("name"),
-                    "token_id": res.token_id, "lease_id": res.lease_id, "reused": res.reused}
+                    "token_id": res.token_id, "lease_id": res.lease_id, "reused": res.reused,
+                    "delivered": delivered}
         last = res
-    code = last.error if last else "no_node"
+    code = last.error if last and last.error else "no_node"
     raise StreamError(code, ERROR_TEXT.get(code, code), invite=last.invite if last else [])
 
 
@@ -246,22 +318,27 @@ async def stop_stream(stream_id: str, reason: str = "stopped") -> bool:
     return True
 
 
-async def stop_guild(guild_id: str, reason: str) -> int:
-    n = 0
-    for s in await fstore.collection_list(STREAMS, guild_id=guild_id):
-        if s.get("desired") == "on":
-            await stop_stream(s["stream_id"], reason)
-            n += 1
-    return n
+async def stop_guild(guild_id: str, reason: str, channel_id: Optional[str] = None) -> list[str]:
+    """Stop every live stream in ONE Discord server (optionally one channel).
+    The /leave fix for server-26#203: never touches another server's streams."""
+    stopped = []
+    for s in await fstore.collection_list(STREAMS, guild_id=str(guild_id)):
+        if s.get("desired") != "on":
+            continue
+        if channel_id and s.get("channel_id") != str(channel_id):
+            continue
+        await stop_stream(s["stream_id"], reason)
+        stopped.append(s["stream_id"])
+    return stopped
 
 
-async def stop_node(node_id: str, reason: str) -> int:
-    n = 0
+async def stop_node(node_id: str, reason: str) -> list[str]:
+    stopped = []
     for s in await fstore.collection_list(STREAMS, node_id=node_id):
         if s.get("desired") == "on":
             await stop_stream(s["stream_id"], reason)
-            n += 1
-    return n
+            stopped.append(s["stream_id"])
+    return stopped
 
 
 # ---------------------------------------------------------------------------
@@ -275,8 +352,11 @@ async def handle_actual(node_id: str, payload: dict) -> None:
     """``nodes/{node_id}/discord/actual``. A node may only report on streams
     currently placed on it — anything else is ignored, so a compromised node
     cannot rewrite another node's (or org's) stream state."""
+    if not isinstance(payload, dict):
+        return
     reported = payload.get("streams") or {}
     now = _now_iso()
+    recheck: set = set()
     if isinstance(reported, dict):
         for stream_id, st in list(reported.items())[:50]:
             if not isinstance(st, dict):
@@ -287,16 +367,23 @@ async def handle_actual(node_id: str, payload: dict) -> None:
             state = st.get("state") if st.get("state") in _VALID_STATES else "error"
             error = st.get("error")
             error = str(error)[:300] if error else None
+            if st.get("code") == "not_in_guild" and doc.get("token_id"):
+                recheck.add(doc["token_id"])
             prev = doc.get("actual") or {}
-            since = prev.get("since") if prev.get("state") == state else now
-            await fstore.doc_update(STREAMS, stream_id, {
+            if prev.get("state") == state and prev.get("error") == error:
+                since = prev.get("since") or now
+            else:
+                since = now
+            await fstore.doc_update(STREAMS, str(stream_id), {
                 "actual": {"state": state, "error": error, "since": since, "at": now},
             })
+    if recheck:
+        await leases.refresh_membership(sorted(recheck))
 
     if "sessions" in payload:
         sessions = {str(t) for t in (payload.get("sessions") or [])}
         applied = payload.get("applied_seq")
-        applied = int(applied) if isinstance(applied, (int, float)) else None
+        applied = int(applied) if isinstance(applied, (int, float)) and not isinstance(applied, bool) else None
         freed = await leases.settle_releases(node_id, sessions, applied)
         for tok in freed:
             logger.info(f"Radio bot token {tok} released — node {node_id} confirmed it logged out")
@@ -309,7 +396,7 @@ async def handle_actual(node_id: str, payload: dict) -> None:
 
 async def release_offline_node(node_id: str) -> None:
     """Node offline past the lease grace: free its tokens (the edge already
-    logged them out — see release_node_leases) and re-place its streams."""
+    logged them out — see release_node_leases) and unplace its streams."""
     orphaned = await leases.release_node_leases(node_id)
     if not orphaned:
         return
@@ -324,38 +411,57 @@ async def release_offline_node(node_id: str) -> None:
     logger.warning(f"Node {node_id} offline past lease grace — released {len(orphaned)} radio stream(s)")
 
 
+async def _unplace(s: dict, why: str) -> None:
+    """Take a stream off its node (token detached, node republished) but leave
+    it desired, so the reconciler can place it somewhere valid."""
+    now = _now_iso()
+    old = await leases.detach_stream(s["stream_id"], {
+        "node_id": None,
+        "actual": {"state": "error", "error": why, "since": now, "at": now},
+    })
+    if old:
+        await publish_desired(old)
+
+
 async def reconcile_once() -> None:
     nodes = {n["node_id"]: n for n in await fstore.collection_list("nodes") if n.get("node_id")}
     grace = timedelta(seconds=settings.discord_lease_offline_release_seconds)
     now = datetime.now(timezone.utc)
+    released: set = set()
     for s in await fstore.collection_list(STREAMS, desired="on"):
         node = nodes.get(s.get("node_id") or "")
-        online = bool(node) and node.get("status") in ONLINE and _effective_system(node) == s.get("system_id")
-        if online and s.get("token_id"):
-            continue
         if node and s.get("token_id"):
-            # Node offline (or retuned): hold the lease until the grace passes.
-            last_seen = node.get("last_seen")
-            try:
-                ls = datetime.fromisoformat(last_seen) if isinstance(last_seen, str) else last_seen
-                if ls and ls.tzinfo is None:
-                    ls = ls.replace(tzinfo=timezone.utc)
-            except Exception:
-                ls = None
-            if node.get("status") in ONLINE or (ls and now - ls < grace):
-                if node.get("status") not in ONLINE:
+            if node.get("status") in ONLINE:
+                if _effective_system(node) == s.get("system_id"):
+                    continue  # healthy placement
+                # Node retuned to another system under a live stream.
+                if s.get("mode") != "persistent":
+                    await stop_stream(s["stream_id"], "node_retuned")
+                    continue
+                await _unplace(s, "The node carrying this channel switched systems; finding another.")
+            else:
+                ls = _parse_ts(node.get("last_seen"))
+                if ls and now - ls < grace:
                     await _mark(s, "node_offline", "The node carrying this channel is offline; waiting for it.")
-                continue
-            await release_offline_node(node["node_id"])
+                    continue  # hold the lease through the grace window
+                if node["node_id"] not in released:
+                    await release_offline_node(node["node_id"])
+                    released.add(node["node_id"])
         if s.get("mode") != "persistent":
             await stop_stream(s["stream_id"], "node_lost")
             continue
+        org_id = s.get("org_id")
         try:
-            await start_stream(org_id=s["org_id"], guild_id=s["guild_id"], channel_id=s["channel_id"],
+            await start_stream(org_id=org_id, guild_id=s["guild_id"], channel_id=s["channel_id"],
                                system_id=s["system_id"], mode="persistent")
         except StreamError as e:
             await _mark(s, "error", e.message)
 
+    # Desired-state publishes the broker never took (MQTT was down).
+    for st in await fstore.collection_list(NODE_STATE, published=False):
+        if st.get("node_id"):
+            await publish_desired(st["node_id"])
+
 
 async def _mark(s: dict, state: str, error: str) -> None:
     prev = s.get("actual") or {}

drb-c2-core/app/internal/mqtt_handler.py

@@ -35,6 +35,8 @@ class MQTTHandler:
             client.subscribe("nodes/+/checkin",     qos=1)
             client.subscribe("nodes/+/status",      qos=1)
             client.subscribe("nodes/+/metadata",    qos=1)
+            # Edge -> C2 actual state of its Discord radio streams (node-26#25).
+            client.subscribe("nodes/+/discord/actual", qos=1)
             # TODO(mqtt-cutover): drop this subscribe once the enrollment/HTTP
             # credentials flow (routers/enrollment.py) is stable in prod and
             # node-26 (the one live node) has been migrated. See
@@ -61,6 +63,14 @@ class MQTTHandler:
 
     async def _dispatch(self, topic: str, payload: dict):
         parts = topic.split("/")
+        # nodes/{node_id}/discord/actual — the one four-level node topic.
+        if len(parts) == 4 and parts[0] == "nodes" and parts[2:] == ["discord", "actual"]:
+            from app.internal import discord_streams
+            try:
+                await discord_streams.handle_actual(parts[1], payload)
+            except Exception as e:
+                logger.error(f"MQTT dispatch error [discord/actual] from {parts[1]}: {e}")
+            return
         # Expected: nodes/{node_id}/{type}
         if len(parts) != 3 or parts[0] != "nodes":
             return
@@ -179,11 +189,11 @@ class MQTTHandler:
 
             await fstore.doc_update("nodes", node_id, updates)
 
-        # NOTE: discord_connected in checkins is informational only — do NOT release the
-        # token here.  The bot watchdog reconnects on transient Discord drops, so a single
-        # checkin with discord_connected=False during a brief reconnect window would
-        # incorrectly free the token while the bot is still active.  Token release is
-        # handled exclusively by the discord_leave command and the node offline sweeper.
+        # NOTE: discord_connected in checkins is informational only — never release a
+        # token from it. Radio-bot tokens are freed only when the node acks (over
+        # nodes/{id}/discord/actual) that it logged the bot out, or by the
+        # reconciler once the node has been offline past the lease grace — see
+        # app/internal/discord_leases.py.
 
     # ------------------------------------------------------------------
     # Status update
@@ -335,6 +345,19 @@ class MQTTHandler:
         else:
             logger.warning(f"MQTT not connected — could not push config to {node_id}")
 
+    def publish_discord_desired(self, node_id: str, payload: dict) -> bool:
+        """Retained full list of the Discord streams this node should carry
+        (app/internal/discord_streams.py). Carries token IDs, never token
+        strings — the broker never stores a bot credential."""
+        topic = f"nodes/{node_id}/discord/desired"
+        if self._client and self._connected:
+            info = self._client.publish(topic, json.dumps(payload), qos=1, retain=True)
+            if getattr(info, "rc", 0) == mqtt.MQTT_ERR_SUCCESS:
+                logger.info(f"Discord desired state seq {payload.get('seq')} -> {node_id}")
+                return True
+        logger.warning(f"MQTT not connected — Discord desired state for {node_id} not published")
+        return False
+
     def publish_node_key(self, node_id: str, api_key: str):
         """Publish the provisioned API key to the node (retained so it survives reconnects).
         TODO(mqtt-cutover): dead once nodes.py's callers switch to the HTTP

drb-c2-core/app/internal/node_sweeper.py

@@ -53,8 +53,11 @@ async def _sweep():
             node_id = node.get("node_id")
             await fstore.doc_update("nodes", node_id, {"status": "offline"})
             logger.info(f"Node {node_id} marked offline (last seen: {last_seen.isoformat()})")
-            from app.routers.tokens import release_token
-            await release_token(node_id)
+            # Radio-bot tokens are NOT freed here any more (node-26#25): the
+            # node may still be logged in to Discord on them. The Discord
+            # reconciler (internal/discord_streams.py) frees them once the node
+            # has been offline past discord_lease_offline_release_seconds — by
+            # which point the edge has logged its bots out by itself.
             continue
 
         # Check for expired system overrides (only for fixed nodes with timeout enforced)

drb-c2-core/app/main.py

@@ -10,6 +10,7 @@ from app.internal.summarizer import summarizer_loop
 from app.internal.vocabulary_learner import vocabulary_induction_loop
 from app.internal.recorrelation_sweep import recorrelation_loop
 from app.internal import ai_health
+from app.internal import discord_streams
 from app.config import settings
 from app.internal.auth import (
     require_firebase_token,
@@ -18,31 +19,42 @@ from app.internal.auth import (
 )
 from app.routers import nodes, systems, calls, upload, tokens, incidents, alerts, admin, trips, places, links, users
 from app.routers import enrollment, media, org, waitlist, telemetry, replay, traffic
+from app.routers import discord as discord_router
 from app.internal import dynsec
 from app.internal import firestore as fstore
 
 
-async def _release_orphaned_tokens():
-    """Release all in-use tokens on startup — voice connections don't survive server restarts."""
+async def _clear_legacy_token_flags():
+    """Tokens written by the pre-lease code carry ``in_use`` with no ``lease``.
+
+    Nothing reads ``in_use`` for decisions any more (discord_leases.py decides
+    on ``lease``), but the /tokens page shows it, so a legacy "in use" with no
+    lease behind it would be a permanent false label. A token that HAS a lease
+    is left alone: voice connections now belong to the edge and survive a C2
+    restart, so releasing leases at startup (what this used to do) would hand
+    a still-logged-in token to a second node.
+    """
     def _find():
         from app.internal.firestore import db
         return [d for d in db.collection("bot_tokens").where("in_use", "==", True).stream()]
 
     results = await asyncio.to_thread(_find)
+    cleared = 0
     for doc in results:
+        if (doc.to_dict() or {}).get("lease"):
+            continue
         await fstore.doc_update("bot_tokens", doc.id, {
-            "in_use": False,
-            "assigned_node_id": None,
-            "assigned_at": None,
+            "lease": None, "in_use": False, "assigned_node_id": None, "assigned_at": None,
         })
-    if results:
-        logger.info(f"Released {len(results)} orphaned token(s) on startup.")
+        cleared += 1
+    if cleared:
+        logger.info(f"Cleared legacy in_use flag on {cleared} token(s) with no lease.")
 
 
 @asynccontextmanager
 async def lifespan(app: FastAPI):
     logger.info("DRB C2 Core starting.")
-    await _release_orphaned_tokens()
+    await _clear_legacy_token_flags()
 
     # dynsec bootstrap + reconcile — must happen before mqtt_handler.connect()
     # so that by the time the app is serving requests, c2-core's own dynsec
@@ -65,6 +77,7 @@ async def lifespan(app: FastAPI):
     summarizer_task   = asyncio.create_task(summarizer_loop())
     induction_task    = asyncio.create_task(vocabulary_induction_loop())
     recorrelation_task = asyncio.create_task(recorrelation_loop())
+    discord_task      = asyncio.create_task(discord_streams.reconcile_loop())
 
     yield  # --- app running ---
 
@@ -73,6 +86,7 @@ async def lifespan(app: FastAPI):
     summarizer_task.cancel()
     induction_task.cancel()
     recorrelation_task.cancel()
+    discord_task.cancel()
     await mqtt_handler.disconnect()
 
 
@@ -139,6 +153,9 @@ app.include_router(waitlist.router)    # public — no auth, source-IP rate limi
 # public by necessity — an <audio src> can't send a bearer token, so the
 # short-lived HMAC in the URL is the credential. Checked inline in media.py.
 app.include_router(media.router)
+# auth is per-endpoint: bot routes take the service key (or a platform admin);
+# the token-fetch route takes the calling node's own API key.
+app.include_router(discord_router.router)
 # NOTE: there used to be an app.routers.mqtt_auth router here (an HTTP
 # backend for the mosquitto-go-auth plugin). That plugin's upstream project
 # is archived (no CVE patches) and was rejected for an internet-facing

drb-c2-core/app/models.py

@@ -112,6 +112,9 @@ class CommandPayload(BaseModel):
     action: str                      # discord_join / discord_leave / op25_restart
     guild_id: Optional[str] = None
     channel_id: Optional[str] = None
+    # discord_join only: try this pool bot first (server-26#204 item 1 — it
+    # used to be dropped by pydantic before the router ever read it).
+    preferred_token_id: Optional[str] = None
 
 
 # ---------------------------------------------------------------------------

drb-c2-core/app/routers/discord.py

@@ -0,0 +1,184 @@
+"""
+Discord radio channels — the HTTP face of internal/discord_streams.py.
+
+Every bot-facing route is scoped to ONE Discord server (``guild_id`` in the
+path or body) and acts for the org that server is linked to
+(server-26#203): /leave in server A can no longer touch server B, and the
+system/node/bot lists a server sees are its own org's.
+
+Auth:
+  * bot routes — the Discord bot's service key, or a platform admin.
+  * PUT /discord/guilds/{id}/link — platform admin only.
+  * GET /discord/node/{node_id}/tokens/{token_id} — that node's own API key;
+    the token string is handed only to the node currently holding its lease.
+"""
+from __future__ import annotations
+
+import secrets
+from datetime import datetime, timezone
+from typing import Optional
+
+from fastapi import APIRouter, Depends, HTTPException, Security
+from fastapi.responses import JSONResponse
+from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
+from pydantic import BaseModel
+
+from app.config import settings
+from app.internal import discord_leases as leases
+from app.internal import discord_streams as streams
+from app.internal import firestore as fstore
+from app.internal.auth import require_admin_token, require_service_key_or_admin
+from app.internal.logger import logger
+
+router = APIRouter(prefix="/discord", tags=["discord"])
+_bearer = HTTPBearer(auto_error=False)
+
+
+def _error(e: streams.StreamError) -> JSONResponse:
+    return JSONResponse(status_code=e.status,
+                        content={"detail": e.message, "code": e.code, "invite": e.invite})
+
+
+async def _org_or_403(guild_id: str) -> str:
+    org_id = await streams.resolve_guild_org(guild_id)
+    if not org_id:
+        raise HTTPException(403, streams.ERROR_TEXT["no_org"])
+    return org_id
+
+
+def _public_stream(s: dict) -> dict:
+    keep = ("stream_id", "guild_id", "channel_id", "channel_name", "system_id", "system_name", "mode",
+            "desired", "node_id", "token_id", "actual", "stop_reason", "updated_at", "delivered", "reused")
+    return {k: s.get(k) for k in keep if k in s}
+
+
+# ---------------------------------------------------------------------------
+# Bot routes
+# ---------------------------------------------------------------------------
+
+@router.get("/guilds/{guild_id}/overview")
+async def guild_overview(guild_id: str, _: dict = Depends(require_service_key_or_admin)):
+    """Everything one Discord server may see: its org's systems and nodes, its
+    own streams, and which radio bots could serve it. No token strings."""
+    org_id = await streams.resolve_guild_org(guild_id)
+    if not org_id:
+        return {"guild_id": guild_id, "org_id": None, "linked": False,
+                "systems": [], "nodes": [], "streams": [], "bots": []}
+    link = await fstore.doc_get(streams.GUILDS, guild_id)
+    systems = await fstore.collection_list("systems", org_id=org_id)
+    nodes = await fstore.collection_list("nodes", org_id=org_id)
+    gstreams = [s for s in await fstore.collection_list(streams.STREAMS, guild_id=guild_id)
+                if s.get("desired") == "on"]
+    bots = []
+    for t in await fstore.collection_list(leases.TOKENS):
+        if t.get("org_id") not in (None, org_id) or t.get("valid") is False:
+            continue
+        lease = t.get("lease") or {}
+        member = t.get("guild_ids")
+        bots.append({
+            "token_id": t.get("token_id"),
+            "name": t.get("name"),
+            "busy_here": any(leases.guild_of(s) == guild_id for s in lease.get("streams") or []),
+            "member": None if member is None else guild_id in member,
+            "invite_url": leases.invite_url(t.get("application_id"), guild_id),
+        })
+    return {
+        "guild_id": guild_id,
+        "org_id": org_id,
+        "linked": bool(link and link.get("org_id")),
+        "systems": [{"system_id": s.get("system_id"), "name": s.get("name")} for s in systems],
+        "nodes": [{"node_id": n.get("node_id"), "name": n.get("name"), "status": n.get("status"),
+                   "system_id": streams._effective_system(n)} for n in nodes],
+        "streams": [_public_stream(s) for s in gstreams],
+        "bots": bots,
+    }
+
+
+class StreamStart(BaseModel):
+    guild_id: str
+    channel_id: str
+    system_id: str
+    mode: str = "adhoc"
+    preferred_token_id: Optional[str] = None
+    channel_name: Optional[str] = None
+    guild_name: Optional[str] = None
+    created_by: Optional[str] = None
+
+
+@router.post("/streams")
+async def start_stream(body: StreamStart, _: dict = Depends(require_service_key_or_admin)):
+    org_id = await _org_or_403(body.guild_id)
+    system = await fstore.doc_get_cached("systems", body.system_id)
+    if not system or system.get("org_id") not in (None, org_id):
+        raise HTTPException(404, "System not found.")
+    try:
+        doc = await streams.start_stream(
+            org_id=org_id, guild_id=body.guild_id, channel_id=body.channel_id, system_id=body.system_id,
+            mode=body.mode, channel_name=body.channel_name, guild_name=body.guild_name,
+            created_by=body.created_by, preferred_token_id=body.preferred_token_id,
+            max_streams=settings.discord_guild_max_streams,
+        )
+    except streams.StreamError as e:
+        return _error(e)
+    return _public_stream(doc)
+
+
+@router.get("/streams/{stream_id}")
+async def get_stream(stream_id: str, _: dict = Depends(require_service_key_or_admin)):
+    doc = await fstore.doc_get(streams.STREAMS, stream_id)
+    if not doc:
+        raise HTTPException(404, "Stream not found.")
+    return _public_stream(doc)
+
+
+class GuildStop(BaseModel):
+    channel_id: Optional[str] = None
+    reason: str = "stopped"
+
+
+@router.post("/guilds/{guild_id}/stop")
+async def stop_guild(guild_id: str, body: GuildStop, _: dict = Depends(require_service_key_or_admin)):
+    stopped = await streams.stop_guild(guild_id, body.reason[:60], channel_id=body.channel_id)
+    return {"stopped": stopped}
+
+
+class GuildLink(BaseModel):
+    org_id: str
+
+
+@router.put("/guilds/{guild_id}/link")
+async def link_guild(guild_id: str, body: GuildLink, decoded: dict = Depends(require_admin_token)):
+    org = await fstore.doc_get("organizations", body.org_id)
+    if not org:
+        raise HTTPException(404, "Organization not found.")
+    await fstore.doc_set(streams.GUILDS, guild_id, {
+        "guild_id": guild_id, "org_id": body.org_id,
+        "linked_by": decoded.get("uid"), "linked_at": datetime.now(timezone.utc).isoformat(),
+    })
+    return {"ok": True, "guild_id": guild_id, "org_id": body.org_id}
+
+
+# ---------------------------------------------------------------------------
+# Node route — token pickup
+# ---------------------------------------------------------------------------
+
+@router.get("/node/{node_id}/tokens/{token_id}")
+async def node_token(
+    node_id: str,
+    token_id: str,
+    credentials: Optional[HTTPAuthorizationCredentials] = Security(_bearer),
+):
+    """The bot token string, for the node that holds its lease and nobody else."""
+    if not credentials:
+        raise HTTPException(401, "Missing authorization")
+    key_doc = await fstore.doc_get("node_keys", node_id)
+    stored = (key_doc or {}).get("api_key") or ""
+    if not stored or not secrets.compare_digest(stored, credentials.credentials):
+        raise HTTPException(401, "Invalid node API key")
+    tok = await fstore.doc_get(leases.TOKENS, token_id)
+    lease = (tok or {}).get("lease") or {}
+    if not tok or lease.get("node_id") != node_id or lease.get("releasing"):
+        logger.warning(f"Node {node_id} asked for token {token_id} it does not hold a lease on")
+        raise HTTPException(403, "This node does not hold a lease on that token.")
+    return {"token_id": token_id, "token": tok["token"], "lease_id": lease.get("lease_id"),
+            "name": tok.get("name")}

drb-c2-core/app/routers/nodes.py

@@ -14,7 +14,7 @@ from app.internal.auth import (
     require_service_or_firebase_token,
     resolve_caller_org_id,
 )
-from app.routers.tokens import assign_token, release_token
+from app.internal import discord_streams
 
 router = APIRouter(prefix="/nodes", tags=["nodes"])
 
@@ -138,27 +138,28 @@ async def send_command(
 
     payload = cmd.model_dump(exclude_none=True)
 
+    # Discord join/leave are desired state now (server-26#203/#204, node-26#25):
+    # they never go to the node as commands, and never carry a token string.
     if cmd.action == "discord_join":
-        # Resolve system doc once — used for preferred token and presence name.
-        system_doc = None
-        system_id = node.get("assigned_system_id")
-        if system_id:
-            system_doc = await fstore.doc_get_cached("systems", system_id)
+        if not cmd.guild_id or not cmd.channel_id:
+            raise HTTPException(400, "guild_id and channel_id are required.")
+        system_id = discord_streams._effective_system(node)
+        if not system_id:
+            raise HTTPException(409, "This node is not listening to a system yet.")
+        try:
+            doc = await discord_streams.start_stream(
+                org_id=node.get("org_id"), guild_id=cmd.guild_id, channel_id=cmd.channel_id,
+                system_id=system_id, node_id=node_id, preferred_token_id=cmd.preferred_token_id,
+                mode="adhoc",
+            )
+        except discord_streams.StreamError as e:
+            raise HTTPException(e.status, e.message)
+        return {"ok": True, "stream_id": doc["stream_id"], "token_id": doc.get("token_id"),
+                "delivered": doc.get("delivered")}
 
-        # Explicit preferred_token_id in the request beats the system-level preference.
-        preferred = payload.pop("preferred_token_id", None) or (system_doc or {}).get("preferred_token_id")
-        token = await assign_token(node_id, preferred_token_id=preferred)
-        if not token:
-            raise HTTPException(503, "No Discord bot tokens available in the pool.")
-        payload["token"] = token
-
-        # Pass system name so the bot can set its Discord presence on join.
-        system_name = (system_doc or {}).get("name")
-        if system_name:
-            payload["system_name"] = system_name
-
-    elif cmd.action == "discord_leave":
-        await release_token(node_id)
+    if cmd.action == "discord_leave":
+        stopped = await discord_streams.stop_node(node_id, "stopped_from_web")
+        return {"ok": True, "stopped": stopped}
 
     if not mqtt_handler.send_command(node_id, payload):
         raise HTTPException(503, "MQTT broker unavailable — command not delivered.")

drb-c2-core/app/routers/tokens.py

@@ -2,9 +2,10 @@ import uuid
 from fastapi import APIRouter, HTTPException, Depends
 from pydantic import BaseModel
 from typing import Optional
-from datetime import datetime, timezone
 from app.internal import firestore as fstore
+from app.internal import discord_leases as leases
 from app.internal.auth import require_admin_token
+from app.internal.logger import logger
 
 router = APIRouter(prefix="/tokens", tags=["tokens"])
 
@@ -12,53 +13,84 @@ router = APIRouter(prefix="/tokens", tags=["tokens"])
 class TokenCreate(BaseModel):
     name: str        # friendly label e.g. "DRB Bot 1"
     token: str       # the actual Discord bot token
+    # None = platform pool. An org id = that org's bring-your-own bot, never
+    # leased to anyone else (discord_leases.pick_token).
+    org_id: Optional[str] = None
+
+
+def _masked(t: dict) -> dict:
+    out = {**t, "token": "•••" + (t.get("token") or "")[-4:]}
+    out["invite_url"] = leases.invite_url(t.get("application_id"))
+    return out
+
+
+async def _republish_after_release(lease: Optional[dict]) -> None:
+    if lease and lease.get("node_id"):
+        from app.internal.discord_streams import publish_desired
+        await publish_desired(lease["node_id"])
 
 
 # ---------------------------------------------------------------------------
-# CRUD
+# CRUD — platform admin only (server-26#204 item 6: the list used to be open
+# to any signed-in user of any org)
 # ---------------------------------------------------------------------------
 
 @router.get("")
-async def list_tokens():
+async def list_tokens(_: dict = Depends(require_admin_token)):
     """List all tokens. The actual token string is masked for safety."""
-    tokens = await fstore.collection_list("bot_tokens")
-    return [
-        {**t, "token": "•••" + t["token"][-4:]}
-        for t in tokens
-    ]
+    return [_masked(t) for t in await fstore.collection_list("bot_tokens")]
 
 
 @router.post("", status_code=201)
 async def add_token(body: TokenCreate, _: dict = Depends(require_admin_token)):
+    """Add a radio bot. The token is checked against Discord first
+    (server-26#204 item 5): a string Discord answers 401 for is refused; if
+    Discord is merely unreachable the token is stored with membership unknown."""
+    try:
+        info = await leases.inspect_token(body.token.strip())
+    except leases.TokenRejected as e:
+        raise HTTPException(400, str(e))
     token_id = str(uuid.uuid4())
     doc = {
         "token_id": token_id,
         "name": body.name,
-        "token": body.token,
+        "token": body.token.strip(),
+        "org_id": body.org_id,
+        "lease": None,
         "in_use": False,
         "assigned_node_id": None,
         "assigned_at": None,
+        **info,
     }
     await fstore.doc_set("bot_tokens", token_id, doc, merge=False)
-    return {"token_id": token_id, "name": body.name}
+    return {"token_id": token_id, "name": body.name, "valid": info.get("valid"),
+            "guild_ids": info.get("guild_ids"), "check_error": info.get("check_error"),
+            "invite_url": leases.invite_url(info.get("application_id"))}
+
+
+@router.post("/{token_id}/refresh", status_code=200)
+async def refresh_token(token_id: str, _: dict = Depends(require_admin_token)):
+    """Re-check a bot with Discord (valid? which servers is it in?)."""
+    if not await fstore.doc_get("bot_tokens", token_id):
+        raise HTTPException(404, "Token not found.")
+    await leases.refresh_membership([token_id])
+    return _masked(await fstore.doc_get("bot_tokens", token_id) or {})
 
 
 @router.post("/flush", status_code=200)
 async def flush_tokens(_: dict = Depends(require_admin_token)):
-    """Force-release all in-use tokens (admin utility — use when tokens get orphaned)."""
-    def _find():
-        from app.internal.firestore import db
-        return [d for d in db.collection("bot_tokens").where("in_use", "==", True).stream()]
-
-    import asyncio
-    results = await asyncio.to_thread(_find)
-    for doc in results:
-        await fstore.doc_update("bot_tokens", doc.id, {
-            "in_use": False,
-            "assigned_node_id": None,
-            "assigned_at": None,
-        })
-    return {"released": len(results)}
+    """Force-release every leased token (admin last resort). Each holding node
+    is sent a desired-state list without the token, so it logs the bot out;
+    persistent streams that lost their token are re-placed by the reconciler."""
+    released = 0
+    for t in await fstore.collection_list("bot_tokens"):
+        if not (t.get("lease") or t.get("in_use")):
+            continue
+        lease = await leases.force_release_token(t["token_id"])
+        released += 1
+        await _republish_after_release(lease)
+    logger.warning(f"Admin flushed {released} radio-bot token lease(s)")
+    return {"released": released}
 
 
 @router.put("/{token_id}/prefer/{system_id}", status_code=200)
@@ -98,62 +130,6 @@ async def delete_token(token_id: str, _: dict = Depends(require_admin_token)):
     existing = await fstore.doc_get("bot_tokens", token_id)
     if not existing:
         raise HTTPException(404, "Token not found.")
-    if existing.get("in_use"):
-        raise HTTPException(409, "Token is currently in use by a node.")
+    if existing.get("lease"):
+        raise HTTPException(409, "Token is currently leased to a node. Stop its radio channels (or flush) first.")
     await fstore.doc_delete("bot_tokens", token_id)
-
-
-# ---------------------------------------------------------------------------
-# Internal helpers — used by the nodes router, not exposed via HTTP
-# ---------------------------------------------------------------------------
-
-async def assign_token(node_id: str, preferred_token_id: Optional[str] = None) -> Optional[str]:
-    """
-    Find a free token, mark it as in-use, return the token string.
-    If preferred_token_id is given, try that token first (only if it's free).
-    Returns None if no tokens are available.
-    """
-    def _find_free(preferred: Optional[str]):
-        from app.internal.firestore import db
-        if preferred:
-            snap = db.collection("bot_tokens").document(preferred).get()
-            if snap.exists and not snap.to_dict().get("in_use"):
-                return [snap]
-        docs = db.collection("bot_tokens").where("in_use", "==", False).limit(1).stream()
-        return [d for d in docs]
-
-    import asyncio
-    results = await asyncio.to_thread(_find_free, preferred_token_id)
-    if not results:
-        return None
-
-    doc = results[0]
-    token_id = doc.id
-    token_value = doc.to_dict()["token"]
-
-    await fstore.doc_update("bot_tokens", token_id, {
-        "in_use": True,
-        "assigned_node_id": node_id,
-        "assigned_at": datetime.now(timezone.utc),
-    })
-    return token_value
-
-
-async def release_token(node_id: str) -> None:
-    """Free whichever token is currently assigned to this node."""
-    def _find_assigned():
-        from app.internal.firestore import db
-        return [
-            d for d in db.collection("bot_tokens")
-            .where("assigned_node_id", "==", node_id)
-            .stream()
-        ]
-
-    import asyncio
-    results = await asyncio.to_thread(_find_assigned)
-    for doc in results:
-        await fstore.doc_update("bot_tokens", doc.id, {
-            "in_use": False,
-            "assigned_node_id": None,
-            "assigned_at": None,
-        })

drb-c2-core/tests/fake_firestore.py

@@ -0,0 +1,172 @@
+"""
+In-memory Firestore stand-in for the Discord lease/stream tests.
+
+Models the one property those modules depend on: a transaction is
+optimistic — every document (and every collection listing) it read is
+re-checked at commit, and if anything changed underneath it the whole
+function is re-run, exactly like the real SDK's ``transactional`` retry. A
+``before_commit`` hook lets a test hold two transactions at the commit point
+together, so a race the code must survive actually happens in the test.
+
+Install with ``FakeFirestore.install(monkeypatch)``; it patches
+``app.internal.firestore``'s db + async helpers in place.
+"""
+from __future__ import annotations
+
+import asyncio
+import copy
+import threading
+from typing import Callable, Optional
+
+
+class Conflict(Exception):
+    pass
+
+
+class _Snap:
+    def __init__(self, doc_id: str, data: Optional[dict]):
+        self.id = doc_id
+        self._data = data
+        self.exists = data is not None
+
+    def to_dict(self):
+        return copy.deepcopy(self._data) if self._data is not None else None
+
+
+class _DocRef:
+    def __init__(self, db: "FakeFirestore", coll: str, doc_id: str):
+        self._db, self.coll, self.id = db, coll, doc_id
+
+    def get(self, transaction=None):
+        with self._db.lock:
+            data = self._db.data.get(self.coll, {}).get(self.id)
+            ver = self._db.versions.get((self.coll, self.id), 0)
+        if transaction is not None:
+            transaction.reads[("doc", self.coll, self.id)] = ver
+        return _Snap(self.id, copy.deepcopy(data))
+
+
+class _Collection:
+    def __init__(self, db: "FakeFirestore", name: str):
+        self._db, self.name = db, name
+
+    def document(self, doc_id: str) -> _DocRef:
+        return _DocRef(self._db, self.name, doc_id)
+
+    def stream(self, transaction=None):
+        with self._db.lock:
+            docs = {k: copy.deepcopy(v) for k, v in self._db.data.get(self.name, {}).items()}
+            vers = {k: self._db.versions.get((self.name, k), 0) for k in docs}
+        if transaction is not None:
+            transaction.reads[("coll", self.name)] = frozenset(docs)
+            for k, v in vers.items():
+                transaction.reads[("doc", self.name, k)] = v
+        return [_Snap(k, v) for k, v in docs.items()]
+
+
+class _Txn:
+    def __init__(self):
+        self.reads: dict = {}
+        self.writes: list = []
+
+    def set(self, ref: _DocRef, data: dict, merge: bool = False):
+        self.writes.append(("set", ref.coll, ref.id, copy.deepcopy(data), merge))
+
+    def update(self, ref: _DocRef, data: dict):
+        self.writes.append(("update", ref.coll, ref.id, copy.deepcopy(data), True))
+
+
+class FakeFirestore:
+    def __init__(self):
+        self.data: dict[str, dict[str, dict]] = {}
+        self.versions: dict[tuple, int] = {}
+        self.lock = threading.RLock()
+        self.before_commit: Optional[Callable[[int], None]] = None
+        self.commits = 0
+        self.retries = 0
+
+    # --- sync surface used inside transactions ---------------------------
+    def collection(self, name: str) -> _Collection:
+        return _Collection(self, name)
+
+    def _write(self, kind: str, coll: str, doc_id: str, data: dict, merge: bool):
+        docs = self.data.setdefault(coll, {})
+        if kind == "update" and doc_id not in docs:
+            raise KeyError(f"No document to update: {coll}/{doc_id}")
+        if merge and doc_id in docs:
+            docs[doc_id] = {**docs[doc_id], **data}
+        else:
+            docs[doc_id] = dict(data)
+        self.versions[(coll, doc_id)] = self.versions.get((coll, doc_id), 0) + 1
+
+    def run_transaction_sync(self, fn, max_attempts: int = 20):
+        for attempt in range(max_attempts):
+            txn = _Txn()
+            result = fn(txn)
+            if self.before_commit is not None:
+                self.before_commit(attempt)
+            with self.lock:
+                stale = False
+                for key, seen in txn.reads.items():
+                    if key[0] == "doc":
+                        if self.versions.get((key[1], key[2]), 0) != seen:
+                            stale = True
+                            break
+                    elif frozenset(self.data.get(key[1], {})) != seen:
+                        stale = True
+                        break
+                if not stale:
+                    for w in txn.writes:
+                        self._write(*w)
+                    self.commits += 1
+                    return result
+            self.retries += 1
+        raise Conflict("transaction kept conflicting")
+
+    # --- async helpers mirroring app.internal.firestore ------------------
+    async def run_transaction(self, fn):
+        return await asyncio.to_thread(self.run_transaction_sync, fn)
+
+    async def doc_get(self, coll: str, doc_id: str):
+        with self.lock:
+            d = self.data.get(coll, {}).get(doc_id)
+            return copy.deepcopy(d) if d is not None else None
+
+    async def doc_get_cached(self, coll: str, doc_id: str, ttl: float = 300.0):
+        return await self.doc_get(coll, doc_id)
+
+    async def doc_set(self, coll: str, doc_id: str, data: dict, merge: bool = True):
+        with self.lock:
+            self._write("set", coll, doc_id, copy.deepcopy(data), merge)
+
+    async def doc_update(self, coll: str, doc_id: str, data: dict):
+        with self.lock:
+            self._write("update", coll, doc_id, copy.deepcopy(data), True)
+
+    async def doc_delete(self, coll: str, doc_id: str):
+        with self.lock:
+            self.data.get(coll, {}).pop(doc_id, None)
+            self.versions[(coll, doc_id)] = self.versions.get((coll, doc_id), 0) + 1
+
+    async def collection_list(self, coll: str, **filters):
+        with self.lock:
+            docs = [copy.deepcopy(d) for d in self.data.get(coll, {}).values()]
+        return [d for d in docs if all(d.get(k) == v for k, v in filters.items())]
+
+    # --- test conveniences ------------------------------------------------
+    def put(self, coll: str, doc_id: str, data: dict):
+        with self.lock:
+            self._write("set", coll, doc_id, copy.deepcopy(data), False)
+
+    def get(self, coll: str, doc_id: str) -> Optional[dict]:
+        return copy.deepcopy(self.data.get(coll, {}).get(doc_id))
+
+    @classmethod
+    def install(cls, monkeypatch) -> "FakeFirestore":
+        from app.internal import firestore as fstore
+        fake = cls()
+        monkeypatch.setattr(fstore, "db", fake)
+        for name in ("run_transaction", "doc_get", "doc_get_cached", "doc_set", "doc_update",
+                     "doc_delete", "collection_list"):
+            monkeypatch.setattr(fstore, name, getattr(fake, name))
+        return fake

drb-c2-core/tests/test_discord_leases.py

@@ -0,0 +1,423 @@
+"""
+Radio-bot token leases + Discord desired-state streams
+(server-26#203, server-26#204, node-26#25).
+
+The race tests run real concurrent transactions against tests/fake_firestore.py,
+whose commit re-checks every read like Firestore does; ``before_commit`` holds
+both transactions at the commit point so the interleaving actually happens.
+"""
+import asyncio
+import threading
+from datetime import datetime, timezone, timedelta
+
+import pytest
+
+from app.internal import discord_leases as leases
+from app.internal import discord_streams as streams
+from tests.fake_firestore import FakeFirestore
+
+ORG = "org-a"
+SYS = "sys-1"
+
+
+@pytest.fixture
+def fs(monkeypatch):
+    fake = FakeFirestore.install(monkeypatch)
+    published: list = []
+
+    def _publish(node_id, payload):
+        published.append((node_id, payload))
+        return fake.mqtt_up
+
+    fake.mqtt_up = True
+    fake.published = published
+    from app.internal.mqtt_handler import mqtt_handler
+    monkeypatch.setattr(mqtt_handler, "publish_discord_desired", _publish)
+    streams._publish_locks.clear()
+
+    async def _no_discord(token_ids):
+        return 0
+    monkeypatch.setattr(leases, "refresh_membership", _no_discord)
+    return fake
+
+
+def _node(fake, node_id, *, org=ORG, system=SYS, status="online", age=0):
+    fake.put("nodes", node_id, {
+        "node_id": node_id, "name": node_id, "org_id": org, "status": status,
+        "approval_status": "approved", "assigned_system_id": system,
+        "last_seen": (datetime.now(timezone.utc) - timedelta(seconds=age)).isoformat(),
+    })
+
+
+def _token(fake, token_id, *, org=None, guilds=None, valid=True, name=None):
+    fake.put("bot_tokens", token_id, {
+        "token_id": token_id, "name": name or token_id, "token": f"secret-{token_id}",
+        "org_id": org, "guild_ids": guilds, "valid": valid, "application_id": f"app-{token_id}",
+        "lease": None, "in_use": False, "assigned_node_id": None,
+    })
+
+
+def _system(fake, system_id=SYS, org=ORG):
+    fake.put("systems", system_id, {"system_id": system_id, "name": "County P25", "org_id": org})
+
+
+async def _start(guild="g1", channel="c1", **kw):
+    return await streams.start_stream(org_id=kw.pop("org_id", ORG), guild_id=guild, channel_id=channel,
+                                      system_id=kw.pop("system_id", SYS), **kw)
+
+
+def _lease(fake, token_id):
+    return (fake.get("bot_tokens", token_id) or {}).get("lease")
+
+
+# ---------------------------------------------------------------------------
+# pick_token — pure choice
+# ---------------------------------------------------------------------------
+
+def test_pick_prefers_byo_then_reuse_and_never_crosses_orgs():
+    toks = {
+        "pool": {"name": "a", "guild_ids": ["g1"]},
+        "mine": {"name": "z", "org_id": ORG, "guild_ids": ["g1"]},
+        "theirs": {"name": "b", "org_id": "org-b", "guild_ids": ["g1"]},
+    }
+    assert leases.pick_token(toks, node_id="n1", guild_id="g1", org_id=ORG).token_id == "mine"
+    # Another org's branded bot is never offered, even when it is the only one.
+    res = leases.pick_token({"theirs": toks["theirs"]}, node_id="n1", guild_id="g1", org_id=ORG)
+    assert not res.ok and res.error == "pool_exhausted"
+
+
+def test_pick_skips_other_nodes_busy_guilds_and_invalid_tokens():
+    toks = {
+        "other-node": {"lease": {"node_id": "n2", "streams": []}},
+        "busy-here": {"lease": {"node_id": "n1", "streams": ["g1:c9"]}},
+        "dead": {"valid": False},
+    }
+    res = leases.pick_token(toks, node_id="n1", guild_id="g1", org_id=ORG)
+    assert not res.ok and res.error == "pool_exhausted"
+    # The same session can serve a different guild (fan-out).
+    assert leases.pick_token(toks, node_id="n1", guild_id="g2", org_id=ORG).token_id == "busy-here"
+
+
+def test_pick_reports_invite_links_when_no_bot_is_in_the_guild():
+    toks = {"t1": {"name": "Radio 1", "guild_ids": ["other"], "application_id": "123"}}
+    res = leases.pick_token(toks, node_id="n1", guild_id="g1", org_id=ORG)
+    assert res.error == "no_bot_in_guild"
+    assert res.invite[0]["name"] == "Radio 1"
+    assert "client_id=123" in res.invite[0]["invite_url"] and "guild_id=g1" in res.invite[0]["invite_url"]
+    assert leases.pick_token({}, node_id="n1", guild_id="g1", org_id=ORG).error == "pool_empty"
+
+
+# ---------------------------------------------------------------------------
+# Leasing through start_stream
+# ---------------------------------------------------------------------------
+
+async def test_rejoin_reuses_the_lease_instead_of_taking_a_second_token(fs):
+    _system(fs); _node(fs, "n1"); _token(fs, "t1"); _token(fs, "t2")
+    first = await _start()
+    again = await _start()
+    assert again["token_id"] == first["token_id"] and again["reused"] is True
+    leased = [t for t in ("t1", "t2") if _lease(fs, t)]
+    assert leased == [first["token_id"]]                     # server-26#204 item 2
+    assert _lease(fs, first["token_id"])["streams"] == ["g1:c1"]
+
+
+async def test_preferred_token_is_honoured(fs):
+    _system(fs); _node(fs, "n1"); _token(fs, "t1"); _token(fs, "t2")
+    doc = await _start(preferred_token_id="t2")
+    assert doc["token_id"] == "t2"                          # server-26#204 item 1
+
+
+async def test_one_session_fans_out_across_guilds_but_not_within_one(fs):
+    _system(fs); _node(fs, "n1"); _token(fs, "t1"); _token(fs, "t2")
+    a = await _start("g1", "c1")
+    b = await _start("g2", "c1")
+    c = await _start("g1", "c2")
+    assert a["token_id"] == b["token_id"]                  # one gateway session, two guilds
+    assert c["token_id"] != a["token_id"]                  # one voice connection per bot per guild
+    with pytest.raises(streams.StreamError) as e:
+        await _start("g1", "c3")
+    assert e.value.code == "pool_exhausted"
+
+
+async def test_join_never_moves_another_guilds_stream(fs):
+    """server-26#203 item 2: a /join from guild B used to steal guild A's node."""
+    _system(fs); _node(fs, "n1"); _token(fs, "t1")
+    await _start("gA", "c1")
+    await _start("gB", "c1")
+    a = fs.get("discord_streams", "gA:c1")
+    assert a["desired"] == "on" and a["token_id"] == "t1"
+    assert sorted(_lease(fs, "t1")["streams"]) == ["gA:c1", "gB:c1"]
+
+
+async def test_desired_state_never_carries_the_token_string(fs):
+    _system(fs); _node(fs, "n1"); _token(fs, "t1")
+    await _start()
+    node_id, payload = fs.published[-1]
+    assert node_id == "n1" and payload["streams"][0]["token_id"] == "t1"
+    assert "secret-t1" not in repr(payload)
+
+
+async def test_no_node_in_org_is_reported(fs):
+    _system(fs); _node(fs, "n1", org="org-b"); _token(fs, "t1")
+    with pytest.raises(streams.StreamError) as e:
+        await _start()
+    assert e.value.code == "no_node"
+    assert _lease(fs, "t1") is None
+
+
+async def test_guild_stream_limit(fs):
+    _system(fs); _node(fs, "n1"); _token(fs, "t1"); _token(fs, "t2")
+    await _start("g1", "c1", max_streams=1)
+    with pytest.raises(streams.StreamError) as e:
+        await _start("g1", "c2", max_streams=1)
+    assert e.value.code == "limit"
+    # Re-requesting the channel that is already on is not a new stream.
+    await _start("g1", "c1", max_streams=1)
+
+
+# ---------------------------------------------------------------------------
+# Races — two concurrent requests can never both win (server-26#204 item 4)
+# ---------------------------------------------------------------------------
+
+def _hold_both_at_commit(fake):
+    barrier = threading.Barrier(2, timeout=5)
+
+    def _hook(attempt):
+        if attempt == 0:
+            barrier.wait()
+    fake.before_commit = _hook
+
+
+async def test_two_nodes_racing_for_the_last_token(fs):
+    _node(fs, "n1"); _node(fs, "n2"); _token(fs, "t1")
+    _hold_both_at_commit(fs)
+    r1, r2 = await asyncio.gather(
+        leases.lease_for_stream("g1:c1", node_id="n1", org_id=ORG),
+        leases.lease_for_stream("g2:c1", node_id="n2", org_id=ORG),
+    )
+    assert fs.retries >= 1                                   # the race really happened
+    assert sorted([r1.ok, r2.ok]) == [False, True]
+    winner = "n1" if r1.ok else "n2"
+    loser_stream = "g2:c1" if r1.ok else "g1:c1"
+    lease = _lease(fs, "t1")
+    assert lease["node_id"] == winner and len(lease["streams"]) == 1
+    assert (fs.get("discord_streams", loser_stream) or {}).get("token_id") is None
+
+
+async def test_same_channel_requested_twice_concurrently_takes_one_token(fs):
+    _node(fs, "n1"); _token(fs, "t1"); _token(fs, "t2")
+    _hold_both_at_commit(fs)
+    r1, r2 = await asyncio.gather(
+        leases.lease_for_stream("g1:c1", node_id="n1", org_id=ORG),
+        leases.lease_for_stream("g1:c1", node_id="n1", org_id=ORG),
+    )
+    assert r1.ok and r2.ok and r1.token_id == r2.token_id
+    held = [t for t in ("t1", "t2") if _lease(fs, t)]
+    assert held == [r1.token_id]
+    assert _lease(fs, r1.token_id)["streams"] == ["g1:c1"]
+
+
+async def test_two_guilds_same_node_racing_share_one_session(fs):
+    _node(fs, "n1"); _token(fs, "t1")
+    _hold_both_at_commit(fs)
+    r1, r2 = await asyncio.gather(
+        leases.lease_for_stream("g1:c1", node_id="n1", org_id=ORG),
+        leases.lease_for_stream("g2:c1", node_id="n1", org_id=ORG),
+    )
+    assert r1.ok and r2.ok and r1.token_id == r2.token_id == "t1"
+    assert r1.lease_id == r2.lease_id
+    assert sorted(_lease(fs, "t1")["streams"]) == ["g1:c1", "g2:c1"]
+
+
+# ---------------------------------------------------------------------------
+# Ordered release (node-26#25): free only after the node confirms logout
+# ---------------------------------------------------------------------------
+
+async def test_stop_marks_releasing_and_holds_until_the_node_acks(fs):
+    _system(fs); _node(fs, "n1"); _node(fs, "n2"); _token(fs, "t1")
+    await _start()
+    await streams.stop_stream("g1:c1")
+    lease = _lease(fs, "t1")
+    assert lease["releasing"] is True and lease["streams"] == []
+    drop_seq = lease["release_seq"]
+    assert drop_seq == fs.published[-1][1]["seq"]          # stamped with the list that dropped it
+    assert fs.published[-1][1]["streams"] == []
+
+    # Another node cannot take a token that may still be logged in on n1.
+    res = await leases.lease_for_stream("g9:c1", node_id="n2", org_id=ORG)
+    assert not res.ok
+
+    # A stale report (from before the node saw the drop) frees nothing.
+    assert await leases.settle_releases("n1", set(), drop_seq - 1) == []
+    # Up to date, but the bot is still logged in: nothing.
+    assert await leases.settle_releases("n1", {"t1"}, drop_seq) == []
+    # Up to date and logged out: freed.
+    assert await leases.settle_releases("n1", set(), drop_seq) == ["t1"]
+    tok = fs.get("bot_tokens", "t1")
+    assert tok["lease"] is None and tok["in_use"] is False and tok["assigned_node_id"] is None
+
+
+async def test_restart_before_release_reuses_the_logged_in_session(fs):
+    _system(fs); _node(fs, "n1"); _token(fs, "t1"); _token(fs, "t2")
+    first = await _start()
+    await streams.stop_stream("g1:c1")
+    again = await _start()
+    assert again["token_id"] == first["token_id"]
+    lease = _lease(fs, first["token_id"])
+    assert lease["releasing"] is False and lease["streams"] == ["g1:c1"]
+
+
+async def test_handle_actual_settles_and_ignores_other_nodes_streams(fs):
+    _system(fs); _node(fs, "n1"); _node(fs, "n2"); _token(fs, "t1"); _token(fs, "t2")
+    await _start("g1", "c1", node_id="n1")
+    await _start("g2", "c1", node_id="n2")
+    await streams.handle_actual("n1", {
+        "applied_seq": 1, "sessions": ["t1"],
+        "streams": {"g1:c1": {"state": "connected"}, "g2:c1": {"state": "error", "error": "spoof"}},
+    })
+    assert fs.get("discord_streams", "g1:c1")["actual"]["state"] == "connected"
+    assert fs.get("discord_streams", "g2:c1")["actual"]["state"] == "pending"   # not n1's to report
+
+    await streams.stop_stream("g1:c1")
+    seq = _lease(fs, "t1")["release_seq"]
+    await streams.handle_actual("n1", {"applied_seq": seq, "sessions": [], "streams": {}})
+    assert _lease(fs, "t1") is None
+
+
+async def test_failed_publish_is_retried_by_the_reconciler(fs):
+    """server-26#204 item 3: a send failure used to leak the token forever."""
+    _system(fs); _node(fs, "n1"); _token(fs, "t1")
+    fs.mqtt_up = False
+    doc = await _start()
+    assert doc["delivered"] is False
+    assert fs.get("discord_nodes", "n1")["published"] is False
+    fs.mqtt_up = True
+    n = len(fs.published)
+    await streams.reconcile_once()
+    assert len(fs.published) > n and fs.get("discord_nodes", "n1")["published"] is True
+    assert fs.published[-1][1]["streams"][0]["stream_id"] == "g1:c1"
+
+
+async def test_concurrent_publishes_keep_the_retained_list_newest(fs):
+    _system(fs); _node(fs, "n1"); _token(fs, "t1")
+    await _start()
+    await asyncio.gather(*(streams.publish_desired("n1") for _ in range(5)))
+    seqs = [p["seq"] for nid, p in fs.published if nid == "n1"]
+    assert seqs == sorted(seqs) and len(set(seqs)) == len(seqs)
+
+
+# ---------------------------------------------------------------------------
+# Guild scoping (server-26#203 item 1)
+# ---------------------------------------------------------------------------
+
+async def test_leave_in_one_guild_never_touches_another(fs):
+    _system(fs); _node(fs, "n1"); _token(fs, "t1"); _token(fs, "t2")
+    await _start("gA", "c1")
+    await _start("gB", "c1")
+    stopped = await streams.stop_guild("gA", "leave")
+    assert stopped == ["gA:c1"]
+    assert fs.get("discord_streams", "gB:c1")["desired"] == "on"
+    assert fs.get("discord_streams", "gA:c1")["desired"] == "off"
+
+
+async def test_guild_org_resolution(fs, monkeypatch):
+    monkeypatch.setattr(streams.settings, "discord_unlinked_guild_org_id", "founding")
+    fs.put("discord_guilds", "g-linked", {"org_id": "org-b"})
+    assert await streams.resolve_guild_org("g-linked") == "org-b"
+    assert await streams.resolve_guild_org("g-other") == "founding"
+    monkeypatch.setattr(streams.settings, "discord_unlinked_guild_org_id", "")
+    assert await streams.resolve_guild_org("g-other") is None
+
+
+# ---------------------------------------------------------------------------
+# Reconciler — offline grace, re-placement
+# ---------------------------------------------------------------------------
+
+async def test_offline_node_keeps_its_lease_inside_the_grace(fs, monkeypatch):
+    monkeypatch.setattr(streams.settings, "discord_lease_offline_release_seconds", 600)
+    _system(fs); _node(fs, "n1"); _token(fs, "t1")
+    await _start(mode="persistent")
+    _node(fs, "n1", status="offline", age=120)
+    await streams.reconcile_once()
+    assert _lease(fs, "t1")["node_id"] == "n1"
+    assert fs.get("discord_streams", "g1:c1")["actual"]["state"] == "node_offline"
+
+
+async def test_persistent_stream_moves_after_grace_adhoc_stops(fs, monkeypatch):
+    monkeypatch.setattr(streams.settings, "discord_lease_offline_release_seconds", 600)
+    _system(fs); _node(fs, "n1"); _token(fs, "t1"); _token(fs, "t2")
+    await _start("g1", "c1", mode="persistent")
+    await _start("g2", "c1", mode="adhoc")
+    _node(fs, "n1", status="offline", age=900)
+    _node(fs, "n2")
+    await streams.reconcile_once()
+    moved = fs.get("discord_streams", "g1:c1")
+    assert moved["desired"] == "on" and moved["node_id"] == "n2" and moved["token_id"]
+    assert _lease(fs, moved["token_id"])["node_id"] == "n2"
+    assert fs.get("discord_streams", "g2:c1")["desired"] == "off"
+    assert all((_lease(fs, t) or {}).get("node_id") != "n1" for t in ("t1", "t2"))
+
+
+async def test_retuned_node_gives_up_a_persistent_stream(fs):
+    _system(fs); _system(fs, "sys-2"); _node(fs, "n1"); _token(fs, "t1")
+    await _start(mode="persistent")
+    _node(fs, "n1", system="sys-2")
+    await streams.reconcile_once()
+    s = fs.get("discord_streams", "g1:c1")
+    assert s["desired"] == "on" and s["token_id"] is None and s["actual"]["state"] == "error"
+    assert _lease(fs, "t1")["releasing"] is True
+
+
+# ---------------------------------------------------------------------------
+# Admin force release + token pickup endpoint
+# ---------------------------------------------------------------------------
+
+async def test_force_release_detaches_streams(fs):
+    _system(fs); _node(fs, "n1"); _token(fs, "t1")
+    await _start()
+    old = await leases.force_release_token("t1")
+    assert old["node_id"] == "n1"
+    assert _lease(fs, "t1") is None
+    assert fs.get("discord_streams", "g1:c1")["token_id"] is None
+
+
+def test_node_token_pickup_only_for_the_lease_holder(fs):
+    from fastapi import FastAPI
+    from fastapi.testclient import TestClient
+    from app.routers import discord as discord_router
+
+    fs.put("node_keys", "n1", {"node_id": "n1", "api_key": "key-1"})
+    fs.put("node_keys", "n2", {"node_id": "n2", "api_key": "key-2"})
+    _token(fs, "t1")
+    fs.put("bot_tokens", "t1", {**fs.get("bot_tokens", "t1"),
+                                "lease": {"node_id": "n1", "lease_id": "L", "streams": ["g:c"], "releasing": False}})
+    app = FastAPI()
+    app.include_router(discord_router.router)
+    client = TestClient(app)
+
+    ok = client.get("/discord/node/n1/tokens/t1", headers={"Authorization": "Bearer key-1"})
+    assert ok.status_code == 200 and ok.json()["token"] == "secret-t1"
+    assert client.get("/discord/node/n2/tokens/t1", headers={"Authorization": "Bearer key-2"}).status_code == 403
+    assert client.get("/discord/node/n1/tokens/t1", headers={"Authorization": "Bearer key-2"}).status_code == 401
+    assert client.get("/discord/node/n1/tokens/t1").status_code == 401
+
+
+# ---------------------------------------------------------------------------
+# Wiring
+# ---------------------------------------------------------------------------
+
+async def test_mqtt_dispatch_routes_discord_actual(monkeypatch):
+    from app.internal.mqtt_handler import MQTTHandler
+    seen = []
+
+    async def _fake(node_id, payload):
+        seen.append((node_id, payload))
+    monkeypatch.setattr(streams, "handle_actual", _fake)
+    await MQTTHandler()._dispatch("nodes/n1/discord/actual", {"applied_seq": 3})
+    assert seen == [("n1", {"applied_seq": 3})]
+
+
+def test_command_payload_keeps_preferred_token_id():
+    from app.models import CommandPayload
+    cmd = CommandPayload(action="discord_join", guild_id="g", channel_id="c", preferred_token_id="t9")
+    assert cmd.preferred_token_id == "t9"

drb-c2-core/tests/test_node_sweeper.py

@@ -35,16 +35,10 @@ def _node_naive(node_id, status, age_seconds):
 async def test_stale_online_node_marked_offline():
     nodes = [_node("node-01", "online", age_seconds=120)]
 
-    # A stale node also triggers app.routers.tokens.release_token(node_id) —
-    # added by the PulseAudio/Discord-token work (commit 2a690ec). It's
-    # imported inline inside _sweep, so it must be patched at its source
-    # module rather than relying on the global asyncio.to_thread patch above,
-    # which is scoped to the node-query call and would otherwise feed
-    # release_token's own internal to_thread call the wrong shape of data
-    # (raw node dicts instead of Firestore doc snapshots with .id).
+    # The sweeper no longer frees Discord tokens (node-26#25) — the Discord
+    # reconciler does, after the lease grace. See test_discord_leases.py.
     with patch("asyncio.to_thread", new=AsyncMock(return_value=nodes)), \
-         patch("app.internal.node_sweeper.fstore") as mock_fstore, \
-         patch("app.routers.tokens.release_token", new=AsyncMock()):
+         patch("app.internal.node_sweeper.fstore") as mock_fstore:
         mock_fstore.doc_update = AsyncMock()
         await _sweep()
 
@@ -58,8 +52,7 @@ async def test_stale_recording_node_marked_offline():
     nodes = [_node("node-02", "recording", age_seconds=200)]
 
     with patch("asyncio.to_thread", new=AsyncMock(return_value=nodes)), \
-         patch("app.internal.node_sweeper.fstore") as mock_fstore, \
-         patch("app.routers.tokens.release_token", new=AsyncMock()):
+         patch("app.internal.node_sweeper.fstore") as mock_fstore:
         mock_fstore.doc_update = AsyncMock()
         await _sweep()
 
@@ -115,8 +108,7 @@ async def test_tz_naive_last_seen_is_handled():
     nodes = [_node_naive("node-06", "online", age_seconds=120)]
 
     with patch("asyncio.to_thread", new=AsyncMock(return_value=nodes)), \
-         patch("app.internal.node_sweeper.fstore") as mock_fstore, \
-         patch("app.routers.tokens.release_token", new=AsyncMock()):
+         patch("app.internal.node_sweeper.fstore") as mock_fstore:
         mock_fstore.doc_update = AsyncMock()
         await _sweep()
 
@@ -151,8 +143,7 @@ async def test_only_stale_nodes_updated_in_batch():
     ]
 
     with patch("asyncio.to_thread", new=AsyncMock(return_value=nodes)), \
-         patch("app.internal.node_sweeper.fstore") as mock_fstore, \
-         patch("app.routers.tokens.release_token", new=AsyncMock()) as mock_release:
+         patch("app.internal.node_sweeper.fstore") as mock_fstore:
         mock_fstore.doc_update = AsyncMock()
         await _sweep()
 
@@ -160,7 +151,6 @@ async def test_only_stale_nodes_updated_in_batch():
     updated_ids = {call.args[1] for call in mock_fstore.doc_update.call_args_list}
     assert updated_ids == {"node-08", "node-11"}
 
-    # Both newly-offline nodes should have their Discord token freed.
-    assert mock_release.call_count == 2
-    released_ids = {call.args[0] for call in mock_release.call_args_list}
-    assert released_ids == {"node-08", "node-11"}
+    # Marking a node offline must NOT free its radio-bot tokens: the node may
+    # still be logged in on them (node-26#25). Only status was written.
+    assert all(call.args[2] == {"status": "offline"} for call in mock_fstore.doc_update.call_args_list)

drb-server-discord-bot/app/commands/radio.py

@@ -1,144 +1,223 @@
+import asyncio
+import time
+from typing import Optional
+
 import discord
 from discord import app_commands
 from discord.ext import commands
-from typing import Optional
+
 from app.internal.c2_client import c2
 from app.internal.logger import logger
 
+# How long /join keeps editing its reply while the node reports back.
+JOIN_WATCH_SECONDS = 30
+JOIN_POLL_SECONDS = 2.5
+OVERVIEW_TTL = 30.0
+
+STATE_TEXT = {
+    "pending": "Waiting for the node to pick it up…",
+    "connecting": "Connecting…",
+    "connected": "Connected.",
+    "node_offline": "The node carrying this channel is offline.",
+    "paused_locally": "Paused from the node's local dashboard.",
+}
+
+
+def _refusal_text(body: dict) -> str:
+    text = body.get("detail") or "DRB could not start radio in that channel."
+    if isinstance(text, dict):  # FastAPI HTTPException detail
+        text = text.get("detail") or str(text)
+    invites = [i for i in (body.get("invite") or []) if i.get("invite_url")]
+    if invites:
+        text += "\n" + "\n".join(f"- [Invite {i['name']}]({i['invite_url']})" for i in invites[:5])
+    return text
+
 
 class RadioCommands(commands.Cog):
+    """Radio channels for ONE Discord server at a time (server-26#203).
+
+    Every C2 call carries this interaction's guild_id; C2 resolves the org that
+    server is linked to and never returns or touches anything outside it.
+    /join and /leave default to members with Manage Channels — a server admin
+    can hand them to a DJ role under Server Settings > Integrations.
+    """
+
     def __init__(self, bot: commands.Bot):
         self.bot = bot
+        self._overview: dict[int, tuple[float, dict]] = {}
+
+    async def _guild(self, guild_id: int, fresh: bool = False) -> dict:
+        hit = self._overview.get(guild_id)
+        if hit and not fresh and time.monotonic() - hit[0] < OVERVIEW_TTL:
+            return hit[1]
+        data = await c2.guild_overview(guild_id)
+        if not data.get("error"):
+            self._overview[guild_id] = (time.monotonic(), data)
+        return data
 
     # ------------------------------------------------------------------
-    # Autocomplete helpers
+    # Autocomplete — this server's org only
     # ------------------------------------------------------------------
 
     async def system_autocomplete(
         self, interaction: discord.Interaction, current: str
     ) -> list[app_commands.Choice[str]]:
-        systems = await c2.get_systems()
+        if not interaction.guild_id:
+            return []
+        data = await self._guild(interaction.guild_id)
         return [
             app_commands.Choice(name=s["name"], value=s["system_id"])
-            for s in systems
-            if current.lower() in s["name"].lower()
+            for s in data.get("systems", [])
+            if s.get("name") and current.lower() in s["name"].lower()
         ][:25]
 
-    async def token_autocomplete(
+    async def bot_autocomplete(
         self, interaction: discord.Interaction, current: str
     ) -> list[app_commands.Choice[str]]:
-        tokens = await c2.get_tokens()
-        return [
-            app_commands.Choice(name=f"{t['name']} {'(in use)' if t.get('in_use') else '(free)'}", value=t["token_id"])
-            for t in tokens
-            if current.lower() in t["name"].lower()
-        ][:25]
+        if not interaction.guild_id:
+            return []
+        data = await self._guild(interaction.guild_id)
+        out = []
+        for b in data.get("bots", []):
+            name = b.get("name") or b.get("token_id")
+            if current.lower() not in name.lower():
+                continue
+            if b.get("busy_here"):
+                label = "busy here"
+            elif b.get("member") is False:
+                label = "not in this server"
+            else:
+                label = "free"
+            out.append(app_commands.Choice(name=f"{name} ({label})", value=b["token_id"]))
+        return out[:25]
 
     # ------------------------------------------------------------------
     # /join
     # ------------------------------------------------------------------
 
     @app_commands.command(name="join", description="Stream a radio system to your voice channel.")
+    @app_commands.guild_only()
+    @app_commands.default_permissions(manage_channels=True)
     @app_commands.describe(
         system="The radio system to listen to.",
-        token="Optionally pick a specific bot token from the pool.",
+        bot="Optionally pick which radio bot joins.",
+        persistent="Keep this channel on radio until someone runs /leave (reconnects by itself).",
     )
-    @app_commands.autocomplete(system=system_autocomplete, token=token_autocomplete)
-    async def join(self, interaction: discord.Interaction, system: str, token: Optional[str] = None):
+    @app_commands.autocomplete(system=system_autocomplete, bot=bot_autocomplete)
+    async def join(self, interaction: discord.Interaction, system: str,
+                   bot: Optional[str] = None, persistent: bool = False):
         await interaction.response.defer(ephemeral=True)
 
-        if not interaction.user.voice or not interaction.user.voice.channel:
-            await interaction.followup.send("You need to be in a voice channel first.")
+        voice = getattr(interaction.user, "voice", None)
+        if not voice or not voice.channel:
+            await interaction.followup.send("You need to be in a voice channel first.", ephemeral=True)
+            return
+        channel = voice.channel
+
+        ok, body = await c2.start_stream({
+            "guild_id": str(interaction.guild_id),
+            "channel_id": str(channel.id),
+            "system_id": system,
+            "mode": "persistent" if persistent else "adhoc",
+            "preferred_token_id": bot,
+            "channel_name": channel.name,
+            "guild_name": interaction.guild.name if interaction.guild else None,
+            "created_by": str(interaction.user.id),
+        })
+        self._overview.pop(interaction.guild_id, None)
+        if not ok:
+            await interaction.followup.send(_refusal_text(body), ephemeral=True)
             return
 
-        channel = interaction.user.voice.channel
-        guild_id = str(interaction.guild_id)
-        channel_id = str(channel.id)
+        system_name = body.get("system_name") or system
+        head = f"**{system_name}** → {channel.mention}"
+        if body.get("delivered") is False:
+            head += "\nDRB's message broker is down; the node will get this as soon as it is back."
+        msg = await interaction.followup.send(f"{head}\n{STATE_TEXT['pending']}", ephemeral=True, wait=True)
+        # The node reports what actually happened (node-26#25 item 5) — show
+        # that, not an optimistic "Streaming…" (server-26#203 item 4).
+        await self._watch(msg, body["stream_id"], head)
 
-        node = await c2.find_node_for_system(system)
-        if not node:
-            await interaction.followup.send(
-                "No online node is assigned to that system. Check `/status` for availability."
-            )
-            return
+    async def _watch(self, msg: discord.WebhookMessage, stream_id: str, head: str) -> None:
+        last = None
+        deadline = time.monotonic() + JOIN_WATCH_SECONDS
+        while time.monotonic() < deadline:
+            await asyncio.sleep(JOIN_POLL_SECONDS)
+            s = await c2.get_stream(stream_id)
+            if not s:
+                continue
+            actual = s.get("actual") or {}
+            state = actual.get("state") or "pending"
+            text = actual.get("error") if state == "error" else STATE_TEXT.get(state, state)
+            if text != last:
+                last = text
+                try:
+                    await msg.edit(content=f"{head}\n{text}")
+                except discord.HTTPException as e:
+                    logger.warning(f"/join status edit failed: {e}")
+                    return
+            if state in ("connected", "error"):
+                return
+        if last == STATE_TEXT["pending"]:
+            try:
+                await msg.edit(content=f"{head}\nThe node has not answered yet — it may be offline. "
+                                       "Check `/status`.")
+            except discord.HTTPException:
+                pass
 
-        cmd: dict = {
-            "action": "discord_join",
-            "guild_id": guild_id,
-            "channel_id": channel_id,
-        }
-        if token:
-            cmd["preferred_token_id"] = token
+    # ------------------------------------------------------------------
+    # /leave — this server only
+    # ------------------------------------------------------------------
 
-        ok = await c2.send_command(node["node_id"], cmd)
-
-        if ok:
-            systems = await c2.get_systems()
-            system_name = next((s["name"] for s in systems if s["system_id"] == system), system)
-            await interaction.followup.send(
-                f"Streaming **{system_name}** from node `{node['node_id']}` to {channel.mention}."
-            )
+    @app_commands.command(name="leave", description="Stop radio in this server (or in one channel).")
+    @app_commands.guild_only()
+    @app_commands.default_permissions(manage_channels=True)
+    @app_commands.describe(channel="Only stop radio in this voice channel.")
+    async def leave(self, interaction: discord.Interaction,
+                    channel: Optional[discord.VoiceChannel] = None):
+        await interaction.response.defer(ephemeral=True)
+        stopped = await c2.stop_guild(interaction.guild_id, str(channel.id) if channel else None)
+        self._overview.pop(interaction.guild_id, None)
+        if stopped is None:
+            await interaction.followup.send("Could not reach DRB. Try again in a minute.", ephemeral=True)
+        elif not stopped:
+            where = channel.mention if channel else "this server"
+            await interaction.followup.send(f"No radio is playing in {where}.", ephemeral=True)
         else:
-            await interaction.followup.send("Failed to contact the node. It may be offline.")
+            await interaction.followup.send(f"Stopped radio in {len(stopped)} channel(s).", ephemeral=True)
 
     # ------------------------------------------------------------------
-    # /leave
+    # /status — this server's org only
     # ------------------------------------------------------------------
 
-    @app_commands.command(name="leave", description="Stop streaming radio in this server.")
-    async def leave(self, interaction: discord.Interaction):
-        await interaction.response.defer(ephemeral=True)
-
-        nodes = await c2.get_nodes()
-        streaming_nodes = [
-            n for n in nodes if n.get("status") in ("online", "recording")
-        ]
-
-        if not streaming_nodes:
-            await interaction.followup.send("No nodes appear to be active right now.")
-            return
-
-        for node in streaming_nodes:
-            await c2.send_command(node["node_id"], {"action": "discord_leave"})
-
-        await interaction.followup.send("Disconnected.")
-
-    # ------------------------------------------------------------------
-    # /status
-    # ------------------------------------------------------------------
-
-    @app_commands.command(name="status", description="Show all node and system status.")
+    @app_commands.command(name="status", description="Show this server's radio channels and your nodes.")
+    @app_commands.guild_only()
     async def status(self, interaction: discord.Interaction):
-        await interaction.response.defer()
-
-        nodes = await c2.get_nodes()
-        systems = await c2.get_systems()
-
-        system_map = {s["system_id"]: s["name"] for s in systems}
-
-        status_emoji = {
-            "online": "🟢",
-            "recording": "🔴",
-            "offline": "⚫",
-            "unconfigured": "🟡",
-        }
-
-        embed = discord.Embed(title="DRB Node Status", color=0x2b2d31)
-
-        if not nodes:
-            embed.description = "No nodes registered."
+        await interaction.response.defer(ephemeral=True)
+        data = await self._guild(interaction.guild_id, fresh=True)
+        embed = discord.Embed(title="DRB status", color=0x2b2d31)
+        if data.get("error"):
+            embed.description = "Could not reach DRB."
+        elif not data.get("org_id"):
+            embed.description = "This server is not linked to a DRB organization yet."
         else:
-            for node in sorted(nodes, key=lambda n: n.get("name", "")):
-                s = node.get("status", "offline")
-                emoji = status_emoji.get(s, "⚪")
-                system_name = system_map.get(node.get("assigned_system_id", ""), "Unassigned")
+            systems = {s["system_id"]: s.get("name") for s in data.get("systems", [])}
+            lines = []
+            for s in data.get("streams", []):
+                actual = (s.get("actual") or {})
+                state = actual.get("state") or "pending"
+                lines.append(f"<#{s['channel_id']}> — {s.get('system_name') or s.get('system_id')} · `{state}`"
+                             + (" · persistent" if s.get("mode") == "persistent" else ""))
+            embed.add_field(name="Radio channels", value="\n".join(lines) or "None playing.", inline=False)
+            dot = {"online": "🟢", "recording": "🔴", "offline": "⚫", "unconfigured": "🟡"}
+            for n in sorted(data.get("nodes", []), key=lambda n: n.get("name") or n["node_id"])[:20]:
+                st = n.get("status") or "offline"
                 embed.add_field(
-                    name=f"{emoji} {node.get('name', node['node_id'])}",
-                    value=f"`{s}` — {system_name}",
+                    name=f"{dot.get(st, '⚪')} {n.get('name') or n['node_id']}",
+                    value=f"`{st}` — {systems.get(n.get('system_id'), 'Unassigned')}",
                     inline=True,
                 )
-
-        await interaction.followup.send(embed=embed)
+        await interaction.followup.send(embed=embed, ephemeral=True)
 
     # ------------------------------------------------------------------
     # /help
@@ -146,44 +225,38 @@ class RadioCommands(commands.Cog):
 
     @app_commands.command(name="help", description="How to use this bot.")
     async def help(self, interaction: discord.Interaction):
-        embed = discord.Embed(
-            title="DRB Radio Bot — Help",
-            color=0x5865f2,
-        )
-
+        embed = discord.Embed(title="DRB Radio Bot — Help", color=0x5865f2)
         embed.add_field(
-            name="/join `system`",
+            name="/join `system` [`bot`] [`persistent`]",
             value=(
-                "Join your current voice channel and start streaming a radio system.\n"
-                "Use the `token` option to pick a specific bot from the pool (optional).\n"
-                "You must be in a voice channel first."
+                "Put a radio system on your current voice channel. A radio bot joins and plays it.\n"
+                "`persistent` keeps the channel on radio until `/leave`, reconnecting by itself.\n"
+                "Needs Manage Channels unless a server admin grants it to another role."
             ),
             inline=False,
         )
         embed.add_field(
-            name="/leave",
-            value="Stop streaming and disconnect all active radio bots in this server.",
+            name="/leave [`channel`]",
+            value="Stop radio in this server, or only in one voice channel. Never affects other servers.",
             inline=False,
         )
         embed.add_field(
             name="/status",
-            value="Show all registered edge nodes and which systems they're monitoring.",
+            value="This server's radio channels and your organization's nodes.",
             inline=False,
         )
-
         embed.add_field(
-            name="Radio bot direct commands",
+            name="Radio bots",
             value=(
-                "Once a radio bot is in your voice channel, you can control it by **mentioning it** in any text channel:\n"
-                "- **@botname joinme** — Move the bot to your current voice channel\n"
-                "- **@botname leave** — Disconnect the bot from voice\n\n"
-                "The radio bot's green speaking ring lights up only when radio is actively transmitting."
+                "A radio bot has to be a member of this server before it can join a channel — "
+                "if none is, `/join` replies with an invite link.\n"
+                "The bot's green speaking ring lights up only while radio is transmitting."
             ),
             inline=False,
         )
-
-        embed.set_footer(text="Audio is streamed live from SDR edge nodes via Icecast.")
-        await interaction.response.send_message(embed=embed)
+        embed.set_footer(text="Audio comes live from DRB edge nodes' receivers, straight from their audio "
+                              "path — not a re-broadcast stream.")
+        await interaction.response.send_message(embed=embed, ephemeral=True)
 
 
 async def setup(bot: commands.Bot):

drb-server-discord-bot/app/internal/c2_client.py

@@ -13,60 +13,60 @@ class C2Client:
             return {"Authorization": f"Bearer {settings.c2_service_key}"}
         return {}
 
-    async def get_nodes(self) -> list:
+    # ------------------------------------------------------------------
+    # Discord radio channels — every call is scoped to one guild
+    # (server-26#203). C2 resolves the guild's org; nothing here sees another
+    # org's nodes, systems or bots.
+    # ------------------------------------------------------------------
+
+    async def guild_overview(self, guild_id: int | str) -> dict:
         try:
             async with httpx.AsyncClient(timeout=10) as client:
-                r = await client.get(f"{self.base}/nodes", headers=self._headers())
+                r = await client.get(f"{self.base}/discord/guilds/{guild_id}/overview", headers=self._headers())
                 r.raise_for_status()
                 return r.json()
         except Exception as e:
-            logger.error(f"C2 get_nodes failed: {e}")
-            return []
+            logger.error(f"C2 guild_overview failed: {e}")
+            return {"guild_id": str(guild_id), "org_id": None, "systems": [], "nodes": [],
+                    "streams": [], "bots": [], "error": True}
 
-    async def get_systems(self) -> list:
+    async def start_stream(self, payload: dict) -> tuple[bool, dict]:
+        """Returns (ok, body). On a refusal body carries detail/code/invite."""
+        try:
+            async with httpx.AsyncClient(timeout=30) as client:
+                r = await client.post(f"{self.base}/discord/streams", json=payload, headers=self._headers())
+                body = r.json() if r.headers.get("content-type", "").startswith("application/json") else {}
+                if r.status_code >= 400:
+                    return False, body if isinstance(body, dict) else {"detail": str(body)}
+                return True, body
+        except Exception as e:
+            logger.error(f"C2 start_stream failed: {e}")
+            return False, {"detail": "Could not reach DRB. Try again in a minute.", "code": "c2_unreachable"}
+
+    async def get_stream(self, stream_id: str) -> Optional[dict]:
         try:
             async with httpx.AsyncClient(timeout=10) as client:
-                r = await client.get(f"{self.base}/systems", headers=self._headers())
+                r = await client.get(f"{self.base}/discord/streams/{stream_id}", headers=self._headers())
                 r.raise_for_status()
                 return r.json()
         except Exception as e:
-            logger.error(f"C2 get_systems failed: {e}")
-            return []
+            logger.error(f"C2 get_stream failed: {e}")
+            return None
 
-    async def get_tokens(self) -> list:
+    async def stop_guild(self, guild_id: int | str, channel_id: Optional[str] = None,
+                         reason: str = "leave") -> Optional[list]:
         try:
-            async with httpx.AsyncClient(timeout=10) as client:
-                r = await client.get(f"{self.base}/tokens", headers=self._headers())
-                r.raise_for_status()
-                return r.json()
-        except Exception as e:
-            logger.error(f"C2 get_tokens failed: {e}")
-            return []
-
-    async def send_command(self, node_id: str, payload: dict) -> bool:
-        try:
-            async with httpx.AsyncClient(timeout=10) as client:
+            async with httpx.AsyncClient(timeout=30) as client:
                 r = await client.post(
-                    f"{self.base}/nodes/{node_id}/command",
-                    json=payload,
+                    f"{self.base}/discord/guilds/{guild_id}/stop",
+                    json={"channel_id": channel_id, "reason": reason},
                     headers=self._headers(),
                 )
                 r.raise_for_status()
-                return True
+                return r.json().get("stopped", [])
         except Exception as e:
-            logger.error(f"C2 send_command failed: {e}")
-            return False
-
-    async def find_node_for_system(self, system_id: str) -> Optional[dict]:
-        """Return the first online node assigned to the given system."""
-        nodes = await self.get_nodes()
-        for node in nodes:
-            if (
-                node.get("assigned_system_id") == system_id
-                and node.get("status") in ("online", "recording")
-            ):
-                return node
-        return None
+            logger.error(f"C2 stop_guild failed: {e}")
+            return None
 
     # ------------------------------------------------------------------
     # Trips