drb-c2-core/app/config.py

@@ -258,8 +258,10 @@ class Settings(BaseSettings):
 
     # --- Discord permission checks, REST client, OAuth linking (server-26#225)
     # Every setting the #225 slices use lives here, added once (S2).
-    # Kill switch (S3): False restores today's token picking exactly — no
-    # REST permission check before a bot is handed to a node.
+    # Kill switch (S3): False restores token picking without the REST
+    # permission check — no Discord call before a bot is handed to a node.
+    # (The system's preferred bot is still tried first either way: that is
+    # a separate fix, plan finding 1, and makes no Discord call.)
     discord_perm_check_enabled: bool = True
     # Cache lifetimes for per-(bot, server) results (app/internal/discord_rest.py):
     # a result that lets a bot in; denied / not-a-member / invalid token; and

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

@@ -49,6 +49,7 @@ from dataclasses import dataclass, field
 from datetime import datetime, timezone
 from typing import Optional
 
+from app.config import settings
 from app.internal import firestore as fstore
 
 TOKENS = "bot_tokens"
@@ -92,6 +93,25 @@ def guild_of(stream_id: str) -> str:
     return stream_id.split(":", 1)[0]
 
 
+def invitable(tok: dict) -> bool:
+    """May DRB suggest inviting this bot to one more server? A pool bot
+    (``org_id`` None) at ``discord_pool_bot_guild_cap`` servers is full: an
+    unverified Discord bot can't join more than 100, and the cap leaves room
+    (owner decision Q3, 2026-10-05: cap 90, add bots when full). An org's own
+    BYO bot is the org's business, never capped here."""
+    if tok.get("org_id") is not None:
+        return True
+    members = tok.get("guild_ids")
+    return members is None or len(members) < settings.discord_pool_bot_guild_cap
+
+
+# Permission-check outcomes the lease code acts on (server-26#225 S3). The
+# full per-bot result is built by discord_streams.check_bots; pick_token reads
+# only ``status`` (plus name/missing/reason/fix for the refusal).
+PERM_OK = "ok"
+PERM_DENIED = "denied"
+
+
 def _now() -> datetime:
     return datetime.now(timezone.utc)
 
@@ -105,7 +125,11 @@ class LeaseResult:
     # error codes: pool_empty | no_bot_in_guild | pool_exhausted |
     #   preferred_busy (the chosen bot already has a stream in this guild) |
     #   preferred_other_node (the chosen bot is logged in on another node) |
-    #   preferred_unavailable (unknown, invalid, or another org's bot)
+    #   preferred_unavailable (unknown, invalid, or another org's bot) |
+    #   missing_permissions (every bot that could take it lacks a Discord
+    #     permission in the channel — ``detail["denied"]`` names each) |
+    #   pool_full (no bot is in this server and every pool bot that could be
+    #     invited is at discord_pool_bot_guild_cap)
     error: Optional[str] = None
     # bots that would work if invited to this guild: [{name, invite_url}]
     invite: list = field(default_factory=list)
@@ -128,12 +152,23 @@ def _legacy_mirror(lease: Optional[dict]) -> dict:
     }
 
 
+def _perm_status(perms: Optional[dict], tok_id: str) -> Optional[str]:
+    return ((perms or {}).get(tok_id) or {}).get("status")
+
+
+def _denied_row(tok_id: str, tok: dict, perms: dict) -> dict:
+    p = perms.get(tok_id) or {}
+    return {"name": tok.get("name") or tok_id, "missing": list(p.get("missing") or []),
+            "reason": p.get("reason"), "fix": p.get("fix")}
+
+
 def _candidate_order(tok_id: str, tok: dict, node_id: str, org_id: Optional[str],
-                     preferred: Optional[str]) -> tuple:
+                     preferred: Optional[str], perms: Optional[dict] = None) -> tuple:
     lease = tok.get("lease") or {}
     return (
-        0 if tok_id == preferred else 1,
+        0 if tok_id == preferred else 1,                        # the system's preferred bot
         0 if org_id and tok.get("org_id") == org_id else 1,     # BYO before pool
+        0 if _perm_status(perms, tok_id) == PERM_OK else 1,     # confirmed-ok before unknown
         0 if lease.get("node_id") == node_id else 1,            # reuse a session
         0 if tok.get("guild_ids") is not None else 1,           # known member first
         str(tok.get("name") or ""),
@@ -149,18 +184,32 @@ def pick_token(
     org_id: Optional[str],
     preferred_token_id: Optional[str] = None,
     exclude_stream_id: Optional[str] = None,
+    soft_preferred_token_id: Optional[str] = None,
+    perms: Optional[dict] = None,
 ) -> LeaseResult:
     """Pure choice over a snapshot of bot_tokens. No I/O — unit tested directly.
 
     A ``preferred_token_id`` is a demand, not a hint: if that bot cannot take
-    the stream the result says why instead of quietly using another bot."""
+    the stream the result says why instead of quietly using another bot.
+    ``soft_preferred_token_id`` (the system's preferred bot,
+    ``systems/{id}.preferred_token_id``) is a hint: tried first, but any other
+    bot that fits is used when it can't.
+
+    ``perms`` ({token_id: {status, missing, reason, fix}}, server-26#225) is
+    the Discord permission check for this voice channel. A ``denied`` bot is
+    skipped; a confirmed ``ok`` bot sorts before one whose check was
+    ``unknown`` (or absent), and unknown is allowed (fail open: a Discord
+    outage must never block a start). None = no check (kill switch).
+    """
     if not tokens:
         return LeaseResult(ok=False, error="pool_empty")
     if preferred_token_id:
         return _pick_preferred(tokens, preferred_token_id, node_id=node_id, guild_id=guild_id,
-                               org_id=org_id, exclude_stream_id=exclude_stream_id)
+                               org_id=org_id, exclude_stream_id=exclude_stream_id, perms=perms)
     usable: list[tuple] = []
     not_member: list[dict] = []
+    denied: list[dict] = []
+    pool_full = False
     for tok_id, tok in tokens.items():
         if tok.get("valid") is False:
             continue
@@ -174,24 +223,41 @@ def pick_token(
         if guild_id in busy_guilds:
             continue  # this bot already has a voice connection in this guild
         members = tok.get("guild_ids")
-        if members is not None and guild_id not in members:
-            not_member.append({"name": tok.get("name") or tok_id,
-                               "invite_url": invite_url(tok.get("application_id"), guild_id)})
+        if (members is not None and guild_id not in members) or _perm_status(perms, tok_id) == "not_member":
+            if invitable(tok):
+                not_member.append({"name": tok.get("name") or tok_id,
+                                   "invite_url": invite_url(tok.get("application_id"), guild_id)})
+            else:
+                pool_full = True
             continue
-        usable.append((_candidate_order(tok_id, tok, node_id, org_id, preferred_token_id), tok_id))
+        if _perm_status(perms, tok_id) == "invalid":
+            continue  # Discord answered 401 during the check
+        if _perm_status(perms, tok_id) == PERM_DENIED:
+            denied.append(_denied_row(tok_id, tok, perms or {}))
+            continue
+        usable.append((_candidate_order(tok_id, tok, node_id, org_id, soft_preferred_token_id, perms), tok_id))
     if usable:
         usable.sort()
         return LeaseResult(ok=True, token_id=usable[0][1])
+    if denied:
+        # Fixing a permission is the shortest way to radio; inviting another
+        # bot is the other, so the invite links ride along.
+        return LeaseResult(ok=False, error="missing_permissions", invite=not_member,
+                           detail={"denied": denied})
     if not_member:
         return LeaseResult(ok=False, error="no_bot_in_guild", invite=not_member)
+    if pool_full:
+        return LeaseResult(ok=False, error="pool_full")
     return LeaseResult(ok=False, error="pool_exhausted")
 
 
 def _pick_preferred(tokens: dict[str, dict], tok_id: str, *, node_id: str, guild_id: str,
-                    org_id: Optional[str], exclude_stream_id: Optional[str]) -> LeaseResult:
+                    org_id: Optional[str], exclude_stream_id: Optional[str],
+                    perms: Optional[dict] = None) -> LeaseResult:
     tok = tokens.get(tok_id)
     owner = (tok or {}).get("org_id")
-    if not tok or tok.get("valid") is False or (owner is not None and owner != org_id):
+    if (not tok or tok.get("valid") is False or (owner is not None and owner != org_id)
+            or _perm_status(perms, tok_id) == "invalid"):
         return LeaseResult(ok=False, error="preferred_unavailable",
                            detail={"token_name": (tok or {}).get("name") or tok_id})
     name = tok.get("name") or tok_id
@@ -206,9 +272,14 @@ def _pick_preferred(tokens: dict[str, dict], tok_id: str, *, node_id: str, guild
         return LeaseResult(ok=False, error="preferred_other_node",
                            detail={"token_name": name, "node_id": lease["node_id"]})
     members = tok.get("guild_ids")
-    if members is not None and guild_id not in members:
+    if (members is not None and guild_id not in members) or _perm_status(perms, tok_id) == "not_member":
+        if not invitable(tok):
+            return LeaseResult(ok=False, error="pool_full", detail={"token_name": name})
         return LeaseResult(ok=False, error="no_bot_in_guild",
                            invite=[{"name": name, "invite_url": invite_url(tok.get("application_id"), guild_id)}])
+    if _perm_status(perms, tok_id) == PERM_DENIED:
+        return LeaseResult(ok=False, error="missing_permissions",
+                           detail={"token_name": name, "denied": [_denied_row(tok_id, tok, perms or {})]})
     return LeaseResult(ok=True, token_id=tok_id)
 
 
@@ -219,12 +290,18 @@ async def lease_for_stream(
     org_id: Optional[str],
     preferred_token_id: Optional[str] = None,
     stream_fields: Optional[dict] = None,
+    soft_preferred_token_id: Optional[str] = None,
+    perms: Optional[dict] = None,
 ) -> LeaseResult:
     """Bind ``stream_id`` to a token leased to ``node_id``, atomically.
 
     Also writes ``stream_fields`` (desired=on, node_id, token_id, lease_id...)
     onto the stream doc in the same transaction, so a stream doc can never name
     a token whose lease does not list it.
+
+    ``perms`` / ``soft_preferred_token_id``: see ``pick_token``. A stream's
+    current token is reused only while its permission check is not
+    ``denied`` — otherwise the stream moves to a bot that can speak there.
     """
     guild_id = guild_of(stream_id)
     db = fstore.db
@@ -243,6 +320,7 @@ async def lease_for_stream(
             and (prev_tok.get("lease") or {}).get("node_id") == node_id
             and stream_id in _lease_streams(prev_tok)
             and prev_tok.get("valid") is not False
+            and _perm_status(perms, prev_tok_id) not in (PERM_DENIED, "invalid", "not_member")
             # An explicit edit to a different bot moves the stream to it.
             and (not preferred_token_id or preferred_token_id == prev_tok_id)
         ):
@@ -255,7 +333,8 @@ async def lease_for_stream(
             return LeaseResult(ok=True, token_id=prev_tok_id, lease_id=lease["lease_id"], reused=True)
 
         result = pick_token(tokens, node_id=node_id, guild_id=guild_id, org_id=org_id,
-                            preferred_token_id=preferred_token_id, exclude_stream_id=stream_id)
+                            preferred_token_id=preferred_token_id, exclude_stream_id=stream_id,
+                            soft_preferred_token_id=soft_preferred_token_id, perms=perms)
         if not result.ok:
             return result
 

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

@@ -23,10 +23,20 @@ the same system in the same channel (idempotent re-add / restart). Fields:
       state: pending | connecting | connected | error | stopping |
              node_offline | paused_locally
       code:  the node's machine-readable reason on error (login_failed,
-             connect_failed, not_in_guild, bad_channel, no_audio) or None
+             connect_failed, not_in_guild, bad_channel, no_audio,
+             missing_permissions, channel_full) or None
   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
 
+Before a token is leased, ``start_stream`` asks Discord (REST, cached —
+discord_rest.guild_snapshot) whether each bot it could pick may View, Connect
+and Speak in the voice channel (server-26#225 S3). A bot that is denied is
+never handed to a node; when every bot is denied the start is refused with
+the bot, the channel, the missing permission and what removed it. A check
+that can't be completed is "unknown" and allowed (fail open). Kill switch:
+``DISCORD_PERM_CHECK_ENABLED=false``. The system's preferred bot
+(``systems/{id}.preferred_token_id``) is tried first when nobody named one.
+
 Flow: ``start_stream`` picks a node + leases a token -> ``publish_desired``
 retains ``nodes/{node}/discord/desired`` (full list for that node, with a
 monotonically increasing ``seq``) -> the edge reconciles and publishes
@@ -49,6 +59,7 @@ from typing import Optional
 from app.config import settings
 from app.internal import firestore as fstore
 from app.internal import discord_leases as leases
+from app.internal import discord_perms
 from app.internal import discord_rest
 from app.internal.logger import logger
 
@@ -203,6 +214,178 @@ async def publish_desired(node_id: str) -> bool:
     return bool(ok)
 
 
+# ---------------------------------------------------------------------------
+# Discord permission check (server-26#225 S3)
+# ---------------------------------------------------------------------------
+
+# Trace sentences (discord_perms) that only say a permission went with
+# another one. The fix is for the other one, so they are not what we report.
+_IMPLIED = ("without View Channel", "without Connect")
+
+
+def _join_labels(labels: list) -> str:
+    if len(labels) <= 1:
+        return "".join(labels)
+    return ", ".join(labels[:-1]) + " or " + labels[-1]
+
+
+def _fix_text(bot: str, missing: list, reason: str) -> str:
+    if "timed out" in reason:
+        return (f"In Discord: remove {bot}'s timeout (right-click it in the member list → Remove Timeout), "
+                "then try again.")
+    allow = " and ".join(missing) if len(missing) <= 2 else ", ".join(missing[:-1]) + " and " + missing[-1]
+    return f"In Discord: Edit Channel → Permissions → add {bot} → allow {allow}, then try again."
+
+
+def evaluate_access(snap: "discord_rest.GuildSnapshot", channel_id: str, *, bot: str) -> dict:
+    """One bot's verdict for one voice channel, from its cached snapshot.
+
+    ``status``: ok | denied | unknown | not_member | invalid | bad_channel
+    (the channel exists but is not a voice channel). ``missing`` lists the
+    edge's labels ("View Channel", "Connect", "Speak"); ``reason`` is the
+    trace sentence for the first one that isn't just implied by another;
+    ``fix`` is what to change in Discord. Never raises."""
+    out: dict = {"status": discord_rest.UNKNOWN, "missing": [], "reason": None, "fix": None,
+                 "user_limit": 0, "limit_bypass": False, "channel_name": None}
+    if snap.status != discord_rest.OK:
+        out["status"] = snap.status
+        out["error"] = snap.error
+        return out
+    channel = next((c for c in snap.channels or []
+                    if isinstance(c, dict) and str(c.get("id")) == str(channel_id)), None)
+    if channel is None:
+        # Deleted, or hidden from the bot's list: the edge's join is the judge.
+        out["error"] = "channel not in the server's channel list"
+        return out
+    out["channel_name"] = channel.get("name")
+    try:
+        bits, trace = discord_perms.effective_permissions(snap.guild or {}, snap.member or {}, channel)
+    except (ValueError, TypeError, AttributeError) as e:
+        out["error"] = f"could not read permissions ({type(e).__name__})"
+        return out
+    verdict = discord_perms.check_voice(bits, channel, trace)
+    out["user_limit"] = verdict["user_limit"]
+    out["limit_bypass"] = verdict["limit_bypass"]
+    if not verdict["is_voice"]:
+        out["status"] = "bad_channel"
+        return out
+    if verdict["ok"]:
+        out["status"] = leases.PERM_OK
+        return out
+    missing = verdict["missing"]
+    reasons = verdict["reasons"]
+    primary = [m for m in missing if not any(x in (reasons.get(m) or "") for x in _IMPLIED)] or missing
+    out.update(status=leases.PERM_DENIED, missing=missing, reasons=reasons,
+               reason=reasons.get(primary[0]) or "not granted",
+               fix=_fix_text(bot, primary, reasons.get(primary[0]) or ""))
+    return out
+
+
+def tokens_to_check(tokens: dict[str, dict], *, guild_id: str, org_id: Optional[str],
+                    node_ids: Optional[set] = None, exclude_stream_id: Optional[str] = None,
+                    preferred_token_id: Optional[str] = None) -> list[str]:
+    """The bots ``pick_token`` could hand this stream to, best first — the
+    only ones worth asking Discord about. ``node_ids`` None = any node."""
+    out = []
+    for tok_id, tok in tokens.items():
+        if tok.get("valid") is False or not tok.get("token"):
+            continue
+        if tok.get("org_id") is not None and tok.get("org_id") != org_id:
+            continue  # another org's bot is never asked about this org's server
+        members = tok.get("guild_ids")
+        if members is not None and guild_id not in members:
+            continue
+        if preferred_token_id:
+            if tok_id == preferred_token_id:
+                out.append(tok_id)
+            continue
+        lease_node = (tok.get("lease") or {}).get("node_id")
+        if lease_node and node_ids is not None and lease_node not in node_ids:
+            continue
+        if any(leases.guild_of(s) == guild_id
+               for s in leases._lease_streams(tok) if s != exclude_stream_id):
+            continue
+        out.append(tok_id)
+    out.sort(key=lambda t: leases._candidate_order(t, tokens[t], "", org_id, None))
+    return out
+
+
+def _unanswered() -> dict:
+    return {"status": discord_rest.UNKNOWN, "missing": [], "reason": None, "fix": None,
+            "error": "Discord did not answer in time"}
+
+
+async def gather_snapshots(tokens: dict[str, dict], token_ids: list[str], guild_id: str, *,
+                           refresh: bool = False, budget: Optional[float] = None) -> dict:
+    """``discord_rest.guild_snapshot`` for each bot, at most
+    ``discord_perm_check_concurrency`` at a time, all within
+    ``discord_perm_check_budget_seconds``. A bot not answered in time maps
+    to None — unknown (its load keeps running and fills the cache for next
+    time: the cache shields it from this cancellation)."""
+    out: dict = {}
+    if not token_ids:
+        return out
+    sem = asyncio.Semaphore(max(1, settings.discord_perm_check_concurrency))
+
+    async def _one(tok_id: str) -> None:
+        async with sem:
+            out[tok_id] = await discord_rest.guild_snapshot({**tokens[tok_id], "token_id": tok_id}, guild_id,
+                                                            refresh=refresh)
+
+    tasks = [asyncio.create_task(_one(t)) for t in token_ids]
+    _, pending = await asyncio.wait(tasks, timeout=budget if budget is not None
+                                    else settings.discord_perm_check_budget_seconds)
+    for t in pending:
+        t.cancel()
+    for t in tasks:
+        if t.done() and not t.cancelled() and t.exception() is not None:
+            logger.warning(f"Discord permission check failed: {type(t.exception()).__name__}")
+    return {t: out.get(t) for t in token_ids}
+
+
+async def check_bots(tokens: dict[str, dict], token_ids: list[str], guild_id: str, channel_id: str,
+                     *, refresh: bool = False, budget: Optional[float] = None) -> dict[str, dict]:
+    """``evaluate_access`` for each bot in one voice channel. Never raises."""
+    snaps = await gather_snapshots(tokens, token_ids, guild_id, refresh=refresh, budget=budget)
+    return {t: (evaluate_access(snaps[t], channel_id, bot=tokens[t].get("name") or t) if snaps.get(t)
+                else _unanswered())
+            for t in token_ids}
+
+
+async def _perm_check(*, org_id: str, guild_id: str, channel_id: str, stream_id: str,
+                      preferred_token_id: Optional[str], node_ids: set) -> Optional[dict]:
+    if not settings.discord_perm_check_enabled:
+        return None
+    tokens = {t["token_id"]: t for t in await fstore.collection_list(leases.TOKENS) if t.get("token_id")}
+    ids = tokens_to_check(tokens, guild_id=guild_id, org_id=org_id, node_ids=node_ids,
+                          exclude_stream_id=stream_id, preferred_token_id=preferred_token_id)
+    return await check_bots(tokens, ids, guild_id, channel_id)
+
+
+def _channel_ref(perms: Optional[dict], channel_name: Optional[str]) -> str:
+    name = next((p.get("channel_name") for p in (perms or {}).values() if p.get("channel_name")), None)
+    name = name or channel_name
+    return f"#{name}" if name else "that voice channel"
+
+
+def _missing_permissions_error(denied: list, invite: list, chan: str) -> "StreamError":
+    def line(d: dict) -> str:
+        return f"{d['name']} can't {_join_labels(d.get('missing') or ['Speak'])} in {chan}: {d.get('reason')}."
+    if len(denied) == 1:
+        d = denied[0]
+        text = f"{line(d)} {d.get('fix') or ''}".strip()
+    else:
+        shown = denied[:3]
+        text = f"No radio bot can play in {chan}:\n" + "\n".join(f"- {line(d)}" for d in shown)
+        if len(denied) > len(shown):
+            text += f"\n- …and {len(denied) - len(shown)} more."
+        text += "\nIn Discord: Edit Channel → Permissions → add one of these bots → allow the missing " \
+                "permission, then try again."
+    if invite:
+        text += " Or invite another radio bot with a link below."
+    return StreamError("missing_permissions", text, invite=invite, extra={"denied": denied})
+
+
 # ---------------------------------------------------------------------------
 # Start / stop
 # ---------------------------------------------------------------------------
@@ -229,6 +412,10 @@ ERROR_TEXT = {
     "limit": "This server already has its maximum number of radio channels. Stop one with /leave first.",
     "preferred_unavailable": "That radio bot can't be used here (it was removed, failed validation, or belongs to "
                              "another organization). Pick another bot, or let DRB choose any free one.",
+    "pool_full": ("No radio bot is in this server, and every DRB radio bot that could be invited is already in its "
+                  "maximum number of servers. A platform admin needs to add another radio bot."),
+    "missing_permissions": "No radio bot has the Discord permissions it needs in that voice channel.",
+    "bad_channel": "That is not a voice channel. Radio can only play in a voice channel.",
     "not_found": "That radio channel no longer exists.",
 }
 
@@ -264,10 +451,12 @@ async def find_channel_stream(guild_id: str, channel_id: str, system_id: str) ->
     return same[0] if same else None
 
 
-async def _lease_error(res: leases.LeaseResult) -> "StreamError":
+async def _lease_error(res: leases.LeaseResult, chan: str = "that voice channel") -> "StreamError":
     code = res.error or "no_node"
     d = res.detail or {}
     bot = d.get("token_name") or "That radio bot"
+    if code == "missing_permissions" and d.get("denied"):
+        return _missing_permissions_error(d["denied"], res.invite, chan)
     if code == "preferred_busy":
         busy = (await fstore.doc_get(STREAMS, d["busy_stream_id"]) if d.get("busy_stream_id") else None) or {}
         what = _stream_label(busy) if busy else "another radio channel"
@@ -307,7 +496,14 @@ async def start_stream(
       the same channel is never touched.
 
     ``preferred_token_id`` is strict: a bot already in this server, on another
-    node, or unusable is refused with a reason, never silently swapped.
+    node, or unusable is refused with a reason, never silently swapped. With
+    none given, the system's ``preferred_token_id`` is tried first (a hint:
+    another bot is used when it can't take the stream).
+
+    Every bot that could be picked is permission-checked against the voice
+    channel first (``_perm_check``; server-26#225). Every entry point — the
+    bot's /join, the web picker, the node command and the reconciler —
+    comes through here, so none can hand a node a bot Discord will refuse.
     """
     if mode not in ("persistent", "adhoc"):
         raise StreamError("bad_mode", "mode must be persistent or adhoc", status=400)
@@ -362,6 +558,14 @@ async def _start_locked(*, org_id: str, guild_id: str, channel_id: str, system_i
             raise StreamError("limit", ERROR_TEXT["limit"])
 
     system = await fstore.doc_get_cached("systems", system_id) or {}
+    # Finding 1 (#225 plan): "preferred for this system" was written by
+    # PUT /tokens/{id}/prefer/{system} and never read. It is a hint, not a
+    # demand — an explicit preferred_token_id still wins and stays strict.
+    # Read uncached: the systems cache lives 5 min, and a preference the
+    # owner just set should apply to the next start.
+    soft_pref = None
+    if not preferred_token_id:
+        soft_pref = ((await fstore.doc_get("systems", system_id)) or {}).get("preferred_token_id") or None
 
     if node_id:
         node = await fstore.doc_get("nodes", node_id)
@@ -372,16 +576,18 @@ async def _start_locked(*, org_id: str, guild_id: str, channel_id: str, system_i
         candidates = [node]
     else:
         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:
-            candidates.sort(key=lambda n: 0 if n["node_id"] == cur else 1)
-        if preferred_token_id:
-            # A chosen bot can only run on the node it is logged in on.
-            tok = await fstore.doc_get(leases.TOKENS, preferred_token_id) or {}
+        cur = existing.get("node_id") if existing.get("system_id") == system_id else None
+        held_on = None
+        if preferred_token_id or soft_pref:
+            # A bot can only run on the node it is logged in on.
+            tok = await fstore.doc_get(leases.TOKENS, preferred_token_id or soft_pref) or {}
             held_on = (tok.get("lease") or {}).get("node_id")
-            if held_on:
-                candidates.sort(key=lambda n: 0 if n["node_id"] == held_on else 1)
+        # Keep the current node when it is still a valid choice (reuse the
+        # lease) — unless a bot was demanded, whose node then comes first.
+        order = [held_on, cur] if preferred_token_id else [cur, held_on]
+        for nid in reversed(order):
+            if nid:
+                candidates.sort(key=lambda n, nid=nid: 0 if n["node_id"] == nid else 1)
     if not candidates:
         raise StreamError("no_node", ERROR_TEXT["no_node"])
 
@@ -404,13 +610,23 @@ async def _start_locked(*, org_id: str, guild_id: str, channel_id: str, system_i
         fields["created_by_discord_id"] = created_by
         fields["now_playing_message_id"] = None
 
+    node_ids = {n["node_id"] for n in candidates}
+    perms = await _perm_check(org_id=org_id, guild_id=guild_id, channel_id=channel_id, stream_id=stream_id,
+                              preferred_token_id=preferred_token_id, node_ids=node_ids)
+    if perms and any(p.get("status") == "bad_channel" for p in perms.values()):
+        raise StreamError("bad_channel", f"{_channel_ref(perms, channel_name)} is not a voice channel. "
+                                         "Radio can only play in a voice channel — pick one from the list.",
+                          status=400)
+
     last: Optional[leases.LeaseResult] = None
+    perm_refusal: 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:
+                                            preferred_token_id=preferred_token_id, stream_fields=fields,
+                                            soft_preferred_token_id=soft_pref, perms=perms)
+        if not res.ok and res.error in ("no_bot_in_guild", "pool_full") 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
@@ -418,8 +634,14 @@ async def _start_locked(*, org_id: str, guild_id: str, channel_id: str, system_i
             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 discord_rest.refresh_membership(stale):
+                if perms is not None:
+                    # Bots that just turned out to be members get checked too.
+                    perms = await _perm_check(org_id=org_id, guild_id=guild_id, channel_id=channel_id,
+                                              stream_id=stream_id, preferred_token_id=preferred_token_id,
+                                              node_ids=node_ids)
                 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)
+                                                    preferred_token_id=preferred_token_id, stream_fields=fields,
+                                                    soft_preferred_token_id=soft_pref, perms=perms)
         if res.ok:
             delivered = await publish_desired(node["node_id"])
             if prev_node and prev_node != node["node_id"]:
@@ -429,12 +651,17 @@ async def _start_locked(*, org_id: str, guild_id: str, channel_id: str, system_i
             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,
                     "delivered": delivered, "already_on": already_on}
-        if res.error in ("preferred_busy", "preferred_unavailable"):
-            raise await _lease_error(res)                 # true on every node — no point trying the rest
+        chan = _channel_ref(perms, channel_name or existing.get("channel_name"))
+        if res.error in ("preferred_busy", "preferred_unavailable") or (
+                preferred_token_id and res.error == "missing_permissions"):
+            raise await _lease_error(res, chan)           # true on every node — no point trying the rest
+        if res.error == "missing_permissions":
+            perm_refusal = res                            # more useful than another node's "all busy"
         last = res
     if last is None:
         raise StreamError("no_node", ERROR_TEXT["no_node"])
-    raise await _lease_error(last)
+    raise await _lease_error(perm_refusal or last,
+                             _channel_ref(perms, channel_name or existing.get("channel_name")))
 
 
 async def stop_stream(stream_id: str, reason: str = "stopped") -> bool:
@@ -550,6 +777,8 @@ async def stop_node(node_id: str, reason: str) -> list[str]:
 # ---------------------------------------------------------------------------
 
 _VALID_STATES = {"connecting", "connected", "error", "stopped", "paused_locally"}
+# Edge error codes that mean the cached permission/membership view is stale.
+_INVALIDATE_ON = {"missing_permissions", "channel_full", "not_in_guild", "bad_channel"}
 
 
 def _log_transition(node_id: str, doc: dict, prev_state: Optional[str], state: str,
@@ -595,6 +824,10 @@ async def handle_actual(node_id: str, payload: dict) -> None:
             code = str(code)[:40] if code else None
             if code == "not_in_guild" and doc.get("token_id"):
                 recheck.add(doc["token_id"])
+            if code in _INVALIDATE_ON and doc.get("token_id"):
+                # The edge saw Discord refuse; a cached "ok" must not keep
+                # handing this bot the same channel (server-26#225).
+                discord_rest.invalidate(token_id=doc["token_id"], guild_id=str(doc.get("guild_id")))
             prev = doc.get("actual") or {}
             if prev.get("state") == state and prev.get("error") == error and prev.get("code") == code:
                 since = prev.get("since") or now

drb-c2-core/app/main.py

@@ -24,6 +24,7 @@ from app.routers import discord_link as discord_link_router
 from app.internal import dynsec
 from app.internal import firestore as fstore
 from app.routers import discord_web
+from app.routers import discord_checks
 
 
 async def _clear_legacy_token_flags():
@@ -164,6 +165,7 @@ app.include_router(discord_web.router)
 # the token-fetch route takes the calling node's own API key.
 app.include_router(discord_router.router)
 app.include_router(discord_link_router.router)  # auth per-endpoint (org owner / bot service key) — server-26#213
+app.include_router(discord_checks.router)  # auth per-endpoint (org member / service key or admin) — server-26#225
 # 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/routers/discord.py

@@ -93,7 +93,9 @@ async def guild_overview(guild_id: str, _: dict = Depends(require_service_key_or
             "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),
+            # A pool bot at discord_pool_bot_guild_cap is never offered (#225, owner Q3).
+            "invite_url": (leases.invite_url(t.get("application_id"), guild_id)
+                           if guild_id in (member or []) or leases.invitable(t) else None),
         })
     return {
         "guild_id": guild_id,

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

@@ -0,0 +1,215 @@
+"""
+Discord voice-channel permission status for the picker (server-26#225 S3).
+
+Read-only. For one Discord server and its voice channels, says for each radio
+bot that could serve it whether Discord lets that bot View, Connect and Speak
+there, and when not, which permission is missing, what removed it and what to
+change. Built on the same cached REST snapshots ``start_stream`` checks before
+it hands a bot to a node (discord_streams.gather_snapshots / evaluate_access),
+so the badge the UI shows and the refusal a start gets agree.
+
+Auth:
+  * GET /discord/nodes/{node_id}/guilds/{guild_id}/voice-check
+      any signed-in member of the node's org, or a platform admin
+      (``discord_web._node_for_caller``); the server must be one of the org's
+      (``discord_web._org_guilds``). Anything else is a 404. Bot NAMES only —
+      no token ids for non-admins, never a token string.
+  * GET /discord/guilds/{guild_id}/voice-check
+      the Discord bot's service key or a platform admin; rows carry token ids.
+
+``refresh=1`` skips the cache, at most once per 10 s per caller and server;
+past that the cached answer is served with ``refreshed: false``.
+
+User limits are advisory: REST can't see who is connected, so the edge's
+``channel_full`` stays the enforcement.
+"""
+from __future__ import annotations
+
+import time
+from datetime import datetime, timezone
+from typing import Optional
+
+from fastapi import APIRouter, Depends, HTTPException, Query
+
+from app.config import settings
+from app.internal import discord_leases as leases
+from app.internal import discord_rest
+from app.internal import discord_streams as streams
+from app.internal import firestore as fstore
+from app.internal.auth import get_role, require_firebase_token, require_service_key_or_admin
+from app.routers.discord_web import _guild_name, _node_for_caller, _org_guilds
+
+router = APIRouter(prefix="/discord", tags=["discord"])
+
+REFRESH_EVERY_SECONDS = 10.0
+_SNOWFLAKE = r"^[0-9]{1,25}$"
+_last_refresh: dict[tuple[str, str], float] = {}
+
+
+def _refresh_allowed(caller: str, guild_id: str) -> bool:
+    now = time.monotonic()
+    key = (caller, guild_id)
+    if now - _last_refresh.get(key, -1e9) < REFRESH_EVERY_SECONDS:
+        return False
+    _last_refresh[key] = now
+    if len(_last_refresh) > 5000:  # bounded: drop entries past the window
+        for k in [k for k, t in _last_refresh.items() if now - t >= REFRESH_EVERY_SECONDS]:
+            _last_refresh.pop(k, None)
+    return True
+
+
+def _availability(tok: dict, *, guild_id: str, node_id: Optional[str]) -> Optional[str]:
+    """None when the bot could take a new stream here, else why not."""
+    if any(leases.guild_of(s) == guild_id for s in leases._lease_streams(tok)):
+        return "busy_here"        # one voice connection per bot per server
+    lease_node = (tok.get("lease") or {}).get("node_id")
+    if node_id and lease_node and lease_node != node_id:
+        return "other_node"       # logged in on another node
+    return None
+
+
+def _channel_state(rows: list[dict], enabled: bool) -> str:
+    if not enabled:
+        return "unchecked"
+    if any(r["status"] == "bad_channel" for r in rows):
+        return "bad_channel"
+    if not rows:
+        return "no_bot"
+    free = [r for r in rows if r["available"]]
+    if any(r["ok"] for r in free):
+        return "ready"
+    if any(r["status"] == discord_rest.UNKNOWN for r in free):
+        return "unknown"          # fail open: a start is still tried
+    if any(r["status"] == leases.PERM_DENIED for r in free):
+        return "needs_fix"        # a start is refused with missing_permissions
+    if any(r["ok"] for r in rows):
+        return "busy"
+    return "no_bot"
+
+
+async def voice_check(*, guild_id: str, org_id: Optional[str], node_id: Optional[str],
+                      channels: list[dict], channel_id: Optional[str], refresh: bool,
+                      with_token_ids: bool) -> dict:
+    """The JSON contract (see the PR for server-26#225 S3). ``channels`` is
+    the server's directory (``[{channel_id, name}]``); when it is empty the
+    voice channels a bot's snapshot lists are used instead."""
+    enabled = settings.discord_perm_check_enabled
+    tokens = {t["token_id"]: t for t in await fstore.collection_list(leases.TOKENS) if t.get("token_id")}
+    eligible = {tid: t for tid, t in tokens.items()
+                if t.get("valid") is not False and t.get("org_id") in (None, org_id)}
+    members = [tid for tid, t in eligible.items()
+               if t.get("guild_ids") is None or guild_id in t["guild_ids"]]
+    members.sort(key=lambda t: leases._candidate_order(t, eligible[t], node_id or "", org_id, None))
+    invite = [{"name": t.get("name") or tid, "invite_url": leases.invite_url(t.get("application_id"), guild_id)}
+              for tid, t in sorted(eligible.items(), key=lambda kv: str(kv[1].get("name") or kv[0]))
+              if tid not in members and leases.invitable(t) and t.get("application_id")][:5]
+
+    snaps = (await streams.gather_snapshots(tokens, [m for m in members if tokens[m].get("token")], guild_id,
+                                            refresh=refresh)
+             if enabled else {})
+
+    picked = [{"channel_id": str(c.get("channel_id")), "name": c.get("name")} for c in channels]
+    if not picked:
+        seen = next((s for s in snaps.values() if s is not None and s.status == discord_rest.OK), None)
+        picked = [{"channel_id": str(c.get("id")), "name": c.get("name")}
+                  for c in sorted((seen.channels if seen else []) or [],
+                                  key=lambda c: (c.get("position") or 0, str(c.get("name") or "")))
+                  if isinstance(c, dict) and c.get("type") == 2]
+    if channel_id:
+        picked = [c for c in picked if c["channel_id"] == channel_id] or [{"channel_id": channel_id, "name": None}]
+
+    # Bots whose rows say "not a member" after the check move to the invite list.
+    newly_out = {t for t, s in snaps.items() if s is not None and s.status == discord_rest.NOT_MEMBER}
+    out_channels = []
+    for c in picked:
+        rows = []
+        for tid in members:
+            if tid in newly_out:
+                continue
+            tok = tokens[tid]
+            name = tok.get("name") or tid
+            if not enabled:
+                res = {"status": "unchecked", "missing": [], "reason": None, "fix": None}
+            elif snaps.get(tid) is None:
+                res = streams._unanswered()
+            else:
+                res = streams.evaluate_access(snaps[tid], c["channel_id"], bot=name)
+            why = _availability(tok, guild_id=guild_id, node_id=node_id)
+            row = {
+                "name": name,
+                "ok": res["status"] == leases.PERM_OK,
+                "status": res["status"],
+                "missing": res.get("missing") or [],
+                "reason": res.get("reason") or (res.get("error") if res["status"] == discord_rest.UNKNOWN else None),
+                "fix": res.get("fix"),
+                "available": why is None,
+                "unavailable_reason": why,
+                "user_limit": res.get("user_limit") or 0,
+                "limit_bypass": bool(res.get("limit_bypass")),
+            }
+            if with_token_ids:
+                row["token_id"] = tid
+            rows.append(row)
+            if not c["name"] and res.get("channel_name"):
+                c["name"] = res["channel_name"]
+        limit = max((r["user_limit"] for r in rows), default=0)
+        out_channels.append({
+            "channel_id": c["channel_id"],
+            "name": c["name"],
+            "state": _channel_state(rows, enabled),
+            "any_ok": any(r["ok"] for r in rows),
+            "user_limit": limit,
+            "limit_bypass": any(r["ok"] and r["limit_bypass"] for r in rows),
+            "bots": [{k: v for k, v in r.items() if k not in ("user_limit", "limit_bypass")} for r in rows],
+        })
+    for tid in sorted(newly_out):
+        t = tokens[tid]
+        if leases.invitable(t) and t.get("application_id") and len(invite) < 5:
+            invite.append({"name": t.get("name") or tid,
+                           "invite_url": leases.invite_url(t.get("application_id"), guild_id)})
+    return {
+        "guild_id": guild_id,
+        "node_id": node_id,
+        "enabled": enabled,
+        "checked_at": datetime.now(timezone.utc).isoformat(),
+        "refreshed": refresh,
+        "channels": out_channels,
+        "invite": invite,
+    }
+
+
+@router.get("/nodes/{node_id}/guilds/{guild_id}/voice-check")
+async def node_voice_check(node_id: str, guild_id: str,
+                           refresh: int = Query(0, ge=0, le=1),
+                           channel_id: Optional[str] = Query(None, pattern=_SNOWFLAKE),
+                           decoded: dict = Depends(require_firebase_token)):
+    node, _ = await _node_for_caller(node_id, decoded)
+    org_id = node.get("org_id")
+    guild = next((g for g in (await _org_guilds(org_id) if org_id else []) if g["guild_id"] == str(guild_id)), None)
+    if not guild:
+        raise HTTPException(404, "That Discord server is not linked to this organization.")
+    do_refresh = bool(refresh) and _refresh_allowed(f"uid:{decoded.get('uid') or '?'}", guild["guild_id"])
+    out = await voice_check(guild_id=guild["guild_id"], org_id=org_id, node_id=node_id,
+                            channels=guild.get("voice_channels") or [], channel_id=channel_id,
+                            refresh=do_refresh, with_token_ids=get_role(decoded) == "admin")
+    out["guild_name"] = _guild_name(guild)
+    return out
+
+
+@router.get("/guilds/{guild_id}/voice-check")
+async def guild_voice_check(guild_id: str,
+                            channel_id: Optional[str] = Query(None, pattern=_SNOWFLAKE),
+                            refresh: int = Query(0, ge=0, le=1),
+                            decoded: dict = Depends(require_service_key_or_admin)):
+    if not guild_id.isdigit():
+        raise HTTPException(400, "guild_id must be a Discord server id (digits only).")
+    org_id = await streams.resolve_guild_org(guild_id)
+    link = await fstore.doc_get(streams.GUILDS, guild_id) or {}
+    caller = "service" if decoded.get("service") else f"uid:{decoded.get('uid') or '?'}"
+    do_refresh = bool(refresh) and _refresh_allowed(caller, guild_id)
+    out = await voice_check(guild_id=guild_id, org_id=org_id, node_id=None,
+                            channels=link.get("voice_channels") or [], channel_id=channel_id,
+                            refresh=do_refresh, with_token_ids=True)
+    out["guild_name"] = _guild_name(link)
+    out["org_id"] = org_id
+    return out

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

@@ -5,7 +5,14 @@ from typing import Optional
 from app.internal import firestore as fstore
 from app.internal import discord_leases as leases
 from app.internal import discord_rest
-from app.internal.auth import require_admin_token
+from app.internal.audit import write_audit
+from app.internal.auth import (
+    describe_actor,
+    get_org_role,
+    get_role,
+    require_admin_token,
+    require_firebase_token,
+)
 from app.internal.logger import logger
 
 router = APIRouter(prefix="/tokens", tags=["tokens"])
@@ -94,35 +101,99 @@ async def flush_tokens(_: dict = Depends(require_admin_token)):
     return {"released": released}
 
 
+def _may_edit_system(decoded: dict, system: Optional[dict]) -> bool:
+    """Platform admin: any system. Org owner: only their own org's systems —
+    a shared system (no org_id) serves every org, so a preference there is
+    platform-admin only (owner decision Q2, 2026-10-05)."""
+    if get_role(decoded) == "admin":
+        return True
+    return bool(system and system.get("org_id") and system.get("org_id") == decoded.get("org_id"))
+
+
+def _check_prefer_access(decoded: dict, token: Optional[dict], system: Optional[dict],
+                         system_id: Optional[str]) -> None:
+    """Raise unless the caller may point ``system`` at ``token``. Another
+    org's bot or system is a 404, never a 403 (no existence leak)."""
+    if get_role(decoded) == "admin":
+        if not token:
+            raise HTTPException(404, "Token not found.")
+        if system_id is not None and not system:
+            raise HTTPException(404, "System not found.")
+        return
+    org_id = decoded.get("org_id")
+    if not org_id or get_org_role(decoded) != "owner":
+        raise HTTPException(403, "Only the organization owner or a platform admin can choose a system's radio bot.")
+    if not token or (token.get("org_id") is not None and token.get("org_id") != org_id):
+        raise HTTPException(404, "Token not found.")
+    if token.get("org_id") is None:
+        raise HTTPException(403, "Only a platform admin can make a shared DRB radio bot preferred. "
+                                 "Organization owners can choose their own organization's radio bots.")
+    if system_id is None:
+        return
+    if not system or (system.get("org_id") and system.get("org_id") != org_id):
+        raise HTTPException(404, "System not found.")
+    if not system.get("org_id"):
+        raise HTTPException(403, "That system is shared by every organization; only a platform admin can "
+                                 "choose its radio bot.")
+
+
 @router.put("/{token_id}/prefer/{system_id}", status_code=200)
 async def set_preferred_system(
     token_id: str,
     system_id: str,
-    _: dict = Depends(require_admin_token),
+    decoded: dict = Depends(require_firebase_token),
 ):
     """
-    Mark this token as the preferred bot for a system.
-    When a discord_join is issued for any node in that system, this token
-    is tried first before falling back to the general pool.
+    Mark this token as the preferred bot for a system: ``start_stream`` tries
+    it first for that system's radio channels when nobody named a bot
+    (server-26#225 S3), and uses another bot when it can't take the stream.
     Pass system_id="_none" to clear the preference.
+
+    Who (owner decision Q2, 2026-10-05): a platform admin for any bot and any
+    system; an org owner for their OWN org's systems and their OWN
+    bring-your-own bots. Shared systems (no org_id) and pool bots stay
+    platform-admin only.
     """
+    if "/" in token_id or "/" in system_id:
+        raise HTTPException(400, "Bad id.")
     existing = await fstore.doc_get("bot_tokens", token_id)
-    if not existing:
-        raise HTTPException(404, "Token not found.")
+    uid, email = describe_actor(decoded)
     if system_id == "_none":
-        # Clear any existing preference on the system that pointed to this token.
-        system_doc = await fstore.doc_get("systems", existing.get("preferred_for_system_id", ""))
-        if system_doc:
-            await fstore.doc_set("systems", existing["preferred_for_system_id"], {"preferred_token_id": None})
+        _check_prefer_access(decoded, existing, None, None)
+        old_id = existing.get("preferred_for_system_id") or ""
+        system_doc = await fstore.doc_get("systems", old_id) if old_id else None
+        if system_doc and system_doc.get("preferred_token_id") == token_id:
+            if not _may_edit_system(decoded, system_doc):
+                raise HTTPException(403, "That system is shared by every organization; only a platform admin "
+                                         "can change its radio bot.")
+            await fstore.doc_set("systems", old_id, {"preferred_token_id": None})
         await fstore.doc_set("bot_tokens", token_id, {"preferred_for_system_id": None})
+        await write_audit(uid, email, "discord.token.prefer", details={
+            "token_id": token_id, "system_id": None, "previous_system_id": old_id or None})
         return {"ok": True, "preferred_for_system_id": None}
 
     system_doc = await fstore.doc_get("systems", system_id)
-    if not system_doc:
-        raise HTTPException(404, "System not found.")
-    # Set preference on both sides for easy lookup in either direction.
+    _check_prefer_access(decoded, existing, system_doc, system_id)
+
+    # Keep both pointers single-valued: this bot stops being preferred for the
+    # system it was preferred for (when the caller may edit that system), and
+    # the bot this system preferred before loses its back-pointer.
+    old_sys_id = existing.get("preferred_for_system_id")
+    if old_sys_id and old_sys_id != system_id:
+        old_sys = await fstore.doc_get("systems", old_sys_id)
+        if old_sys and old_sys.get("preferred_token_id") == token_id and _may_edit_system(decoded, old_sys):
+            await fstore.doc_set("systems", old_sys_id, {"preferred_token_id": None})
+    old_tok_id = system_doc.get("preferred_token_id")
+    if old_tok_id and old_tok_id != token_id:
+        old_tok = await fstore.doc_get("bot_tokens", old_tok_id)
+        if old_tok and old_tok.get("preferred_for_system_id") == system_id:
+            await fstore.doc_set("bot_tokens", old_tok_id, {"preferred_for_system_id": None})
+
     await fstore.doc_set("systems", system_id, {"preferred_token_id": token_id})
     await fstore.doc_set("bot_tokens", token_id, {"preferred_for_system_id": system_id})
+    await write_audit(uid, email, "discord.token.prefer", details={
+        "token_id": token_id, "system_id": system_id, "previous_token_id": old_tok_id,
+        "org_id": system_doc.get("org_id")})
     return {"ok": True, "preferred_for_system_id": system_id}
 
 

drb-c2-core/tests/conftest.py

@@ -72,3 +72,27 @@ except ModuleNotFoundError:
     ):
         sys.modules.setdefault(_name, ModuleType(_name))
     sys.modules["google.cloud.firestore_v1.base_query"].FieldFilter = MagicMock()
+
+
+import pytest  # noqa: E402
+
+
+@pytest.fixture(autouse=True)
+def _no_real_discord(monkeypatch):
+    """No test reaches discord.com (server-26#225 S3).
+
+    ``start_stream`` now asks Discord, over REST, whether a bot may speak in
+    the channel before it hands the bot to a node. Every Discord call in a
+    test is answered by a transport that fails to connect, so the check is
+    "unknown" and fails open — today's token picking — unless the test
+    installs its own transport or stubs ``discord_rest.guild_snapshot``."""
+    import httpx
+    from app.internal import discord_rest
+
+    def _offline(request):
+        raise httpx.ConnectError("tests never reach Discord", request=request)
+
+    discord_rest.reset()
+    monkeypatch.setattr(discord_rest, "_transport", httpx.MockTransport(_offline))
+    yield
+    discord_rest.reset()

drb-c2-core/tests/test_discord_perm_check.py

@@ -0,0 +1,583 @@
+"""
+Discord permission check wired into token picking (server-26#225 S3).
+
+Before a radio bot is handed to a node, C2 asks Discord (cached REST
+snapshots, discord_rest.guild_snapshot — stubbed here) whether the bot may
+View, Connect and Speak in the voice channel. These tests pin:
+
+* pick_token: a denied bot is skipped, confirmed-ok sorts before unknown,
+  unknown is allowed (fail open), all-denied is ``missing_permissions``
+  naming each bot; a full pool bot (discord_pool_bot_guild_cap) is never
+  offered as an invite.
+* start_stream: the check runs for every entry point (bot POST
+  /discord/streams, web picker, node discord_join, reconciler), inside a time
+  budget, never when the kill switch is off; a non-voice channel is
+  ``bad_channel``; the system's preferred bot is tried first.
+* GET voice-check: org-scoped, bot names only for non-admins.
+* PUT /tokens/{id}/prefer/{system}: owner decision Q2.
+"""
+import asyncio
+
+import pytest
+from fastapi import HTTPException
+
+from app.config import settings
+from app.internal import discord_leases as leases
+from app.internal import discord_perms as perms_mod
+from app.internal import discord_rest
+from app.internal import discord_streams as streams
+from app.models import CommandPayload
+from app.routers import discord as discord_router
+from app.routers import discord_checks as checks
+from app.routers import discord_web as web
+from app.routers import nodes as nodes_router
+from app.routers import tokens as tokens_router
+from tests.fake_firestore import FakeFirestore
+
+ORG, OTHER, SYS = "org-a", "org-b", "sys-1"
+GUILD = "100"
+ADMIN = {"uid": "u-admin", "role": "admin"}
+OWNER = {"uid": "u-own", "org_id": ORG, "org_role": "owner"}
+MEMBER = {"uid": "u-mem", "org_id": ORG, "org_role": "member"}
+STRANGER = {"uid": "u-x", "org_id": OTHER, "org_role": "owner"}
+
+VOICE = perms_mod.VIEW_CHANNEL | perms_mod.CONNECT | perms_mod.SPEAK
+
+
+def _guild():
+    return {"id": GUILD, "owner_id": "owner-user",
+            "roles": [{"id": GUILD, "name": "@everyone", "permissions": str(VOICE)},
+                      {"id": "r-muted", "name": "Muted", "permissions": "0"}]}
+
+
+def _channels():
+    return [
+        {"id": "1001", "name": "dispatch", "type": 2, "user_limit": 0,
+         "permission_overwrites": [{"id": "r-muted", "type": 0, "allow": "0", "deny": str(perms_mod.SPEAK)}]},
+        {"id": "1002", "name": "fireground", "type": 2, "user_limit": 5, "permission_overwrites": []},
+        {"id": "1003", "name": "chat", "type": 0, "permission_overwrites": []},
+    ]
+
+
+def _snap(tok_id, *, muted=False, status=discord_rest.OK):
+    if status != discord_rest.OK:
+        return discord_rest.GuildSnapshot(token_id=tok_id, guild_id=GUILD, status=status, error="x")
+    return discord_rest.GuildSnapshot(
+        token_id=tok_id, guild_id=GUILD, status=discord_rest.OK, bot_user_id=f"u-{tok_id}",
+        guild=_guild(), member={"user": {"id": f"u-{tok_id}"}, "roles": ["r-muted"] if muted else []},
+        channels=_channels())
+
+
+@pytest.fixture
+def fs(monkeypatch):
+    fake = FakeFirestore.install(monkeypatch)
+    published: list = []
+    from app.internal.mqtt_handler import mqtt_handler
+    monkeypatch.setattr(mqtt_handler, "publish_discord_desired", lambda n, p: published.append((n, p)) or True)
+    streams._publish_locks.clear()
+    streams._guild_locks.clear()
+    checks._last_refresh.clear()
+
+    async def _no_discord(token_ids):
+        return 0
+    monkeypatch.setattr(discord_rest, "refresh_membership", _no_discord)
+    monkeypatch.setattr(settings, "discord_unlinked_guild_org_id", "")
+    monkeypatch.setattr(settings, "discord_perm_check_enabled", True)
+
+    # Per-token snapshot behaviour: "ok" | "muted" | "unknown" | "slow" | "not_member"
+    fake.snap_mode = {}
+    fake.snap_calls = []
+
+    async def _snapshot(token_doc, guild_id, *, refresh=False):
+        tid = token_doc["token_id"]
+        fake.snap_calls.append((tid, guild_id, refresh))
+        assert "token" in token_doc            # the real call needs the string; it never leaves C2
+        mode = fake.snap_mode.get(tid, "ok")
+        if mode == "slow":
+            await asyncio.sleep(5)
+        if mode == "unknown":
+            return _snap(tid, status=discord_rest.UNKNOWN)
+        if mode == "not_member":
+            return _snap(tid, status=discord_rest.NOT_MEMBER)
+        return _snap(tid, muted=(mode == "muted"))
+    monkeypatch.setattr(discord_rest, "guild_snapshot", _snapshot)
+
+    for nid, org in (("n1", ORG), ("n2", ORG), ("nx", OTHER)):
+        fake.put("nodes", nid, {"node_id": nid, "name": f"Node {nid}", "org_id": org, "status": "online",
+                                "approval_status": "approved", "assigned_system_id": SYS})
+    fake.put("systems", SYS, {"system_id": SYS, "name": "County P25", "org_id": ORG})
+    fake.put("systems", "shared", {"system_id": "shared", "name": "State Net", "org_id": None})
+    fake.put("systems", "sys-b", {"system_id": "sys-b", "name": "Theirs", "org_id": OTHER})
+    fake.put("discord_guilds", GUILD, {"guild_id": GUILD, "org_id": ORG, "name": "Fire Co. 7", "bot_present": True,
+                                       "voice_channels": [{"channel_id": "1001", "name": "dispatch"},
+                                                          {"channel_id": "1002", "name": "fireground"}]})
+    fake.put("discord_guilds", "200", {"guild_id": "200", "org_id": OTHER, "name": "Other", "bot_present": True,
+                                       "voice_channels": [{"channel_id": "2001", "name": "theirs"}]})
+    fake.published = published
+    return fake
+
+
+def _token(fake, tid, *, name=None, org=None, guilds=(GUILD,), **extra):
+    fake.put("bot_tokens", tid, {"token_id": tid, "name": name or f"DRB Radio {tid}", "token": f"secret-{tid}",
+                                 "org_id": org, "guild_ids": list(guilds) if guilds is not None else None,
+                                 "valid": True, "application_id": f"app-{tid}", "lease": None, **extra})
+
+
+def run(coro):
+    return asyncio.run(coro)
+
+
+async def _start(channel="1001", **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)
+
+
+# ---------------------------------------------------------------------------
+# pick_token — pure
+# ---------------------------------------------------------------------------
+
+def _toks(*ids, **kw):
+    return {t: {"name": f"Radio {t}", "guild_ids": [GUILD], "application_id": f"a{t}", **kw} for t in ids}
+
+
+def test_pick_skips_denied_and_prefers_confirmed_ok_over_unknown():
+    toks = _toks("a", "b", "c")
+    perms = {"a": {"status": "denied", "missing": ["Speak"], "reason": "r", "fix": "f"},
+             "b": {"status": "unknown"}, "c": {"status": "ok"}}
+    assert leases.pick_token(toks, node_id="n1", guild_id=GUILD, org_id=ORG, perms=perms).token_id == "c"
+    # Unknown is allowed: fail open.
+    perms["c"] = {"status": "denied", "missing": ["Speak"]}
+    assert leases.pick_token(toks, node_id="n1", guild_id=GUILD, org_id=ORG, perms=perms).token_id == "b"
+
+
+def test_pick_all_denied_is_missing_permissions_naming_each_bot():
+    toks = {**_toks("a", "b"), "out": {"name": "Radio out", "guild_ids": [], "application_id": "9"}}
+    perms = {t: {"status": "denied", "missing": ["Speak"], "reason": f"why {t}", "fix": "fix"} for t in ("a", "b")}
+    res = leases.pick_token(toks, node_id="n1", guild_id=GUILD, org_id=ORG, perms=perms)
+    assert res.error == "missing_permissions"
+    assert [d["name"] for d in res.detail["denied"]] == ["Radio a", "Radio b"]
+    assert res.detail["denied"][0] == {"name": "Radio a", "missing": ["Speak"], "reason": "why a", "fix": "fix"}
+    assert res.invite and res.invite[0]["name"] == "Radio out"      # the other way out rides along
+
+
+def test_pick_without_perms_is_unchanged():
+    toks = _toks("a", "b")
+    assert leases.pick_token(toks, node_id="n1", guild_id=GUILD, org_id=ORG).token_id == "a"
+
+
+def test_preferred_bot_denied_is_refused_by_name_not_swapped():
+    toks = _toks("a", "b")
+    perms = {"a": {"status": "denied", "missing": ["Connect", "Speak"], "reason": "r", "fix": "f"}}
+    res = leases.pick_token(toks, node_id="n1", guild_id=GUILD, org_id=ORG, preferred_token_id="a", perms=perms)
+    assert res.error == "missing_permissions" and res.detail["token_name"] == "Radio a"
+
+
+def test_soft_preferred_goes_first_but_falls_back():
+    toks = _toks("a", "z")
+    assert leases.pick_token(toks, node_id="n1", guild_id=GUILD, org_id=ORG,
+                             soft_preferred_token_id="z").token_id == "z"
+    toks["z"]["lease"] = {"node_id": "n1", "streams": [f"{GUILD}:9:abcdef012345"]}   # busy in this server
+    assert leases.pick_token(toks, node_id="n1", guild_id=GUILD, org_id=ORG,
+                             soft_preferred_token_id="z").token_id == "a"
+
+
+def test_full_pool_bot_is_never_offered_as_an_invite(monkeypatch):
+    monkeypatch.setattr(settings, "discord_pool_bot_guild_cap", 3)
+    toks = {"full": {"name": "Full", "guild_ids": ["1", "2", "3"], "application_id": "1"},
+            "room": {"name": "Room", "guild_ids": ["1"], "application_id": "2"},
+            "byo": {"name": "Ours", "org_id": ORG, "guild_ids": ["1", "2", "3", "4"], "application_id": "3"}}
+    res = leases.pick_token(toks, node_id="n1", guild_id=GUILD, org_id=ORG)
+    assert res.error == "no_bot_in_guild" and {i["name"] for i in res.invite} == {"Room", "Ours"}
+    res = leases.pick_token({"full": toks["full"]}, node_id="n1", guild_id=GUILD, org_id=ORG)
+    assert res.error == "pool_full" and not res.invite
+    # A full pool bot already IN the server is still used: the cap is about joining more.
+    toks["full"]["guild_ids"].append(GUILD)
+    assert leases.pick_token({"full": toks["full"]}, node_id="n1", guild_id=GUILD, org_id=ORG).ok
+
+
+# ---------------------------------------------------------------------------
+# evaluate_access — trace -> reason/fix text
+# ---------------------------------------------------------------------------
+
+def test_evaluate_names_missing_permission_reason_and_fix():
+    r = streams.evaluate_access(_snap("t", muted=True), "1001", bot="DRB Radio 2")
+    assert r["status"] == "denied" and r["missing"] == ["Speak"]
+    assert r["reason"] == "denied by #dispatch's override for @Muted"
+    assert "DRB Radio 2" in r["fix"] and "allow Speak" in r["fix"]
+    assert streams.evaluate_access(_snap("t"), "1001", bot="x")["status"] == "ok"
+    assert streams.evaluate_access(_snap("t"), "1003", bot="x")["status"] == "bad_channel"
+    assert streams.evaluate_access(_snap("t"), "9999", bot="x")["status"] == "unknown"     # not in list: fail open
+    bad = _snap("t")
+    bad.guild = {"id": GUILD, "roles": []}                                                   # partial payload
+    assert streams.evaluate_access(bad, "1001", bot="x")["status"] == "unknown"
+
+
+# ---------------------------------------------------------------------------
+# start_stream
+# ---------------------------------------------------------------------------
+
+async def test_denied_bot_is_skipped_for_one_that_can_speak(fs):
+    _token(fs, "t1"); _token(fs, "t2")
+    fs.snap_mode["t1"] = "muted"
+    doc = await _start(node_id="n1")
+    assert doc["token_id"] == "t2"
+
+
+async def test_every_bot_denied_is_a_409_naming_bot_channel_permission_and_reason(fs):
+    _token(fs, "t1", name="DRB Radio 2")
+    fs.snap_mode["t1"] = "muted"
+    with pytest.raises(streams.StreamError) as e:
+        await _start(node_id="n1")
+    err = e.value
+    assert err.code == "missing_permissions" and err.status == 409
+    assert err.message.startswith("DRB Radio 2 can't Speak in #dispatch: denied by #dispatch's override for @Muted.")
+    assert "Edit Channel → Permissions" in err.message
+    assert err.extra["denied"][0]["name"] == "DRB Radio 2" and "token" not in str(err.extra)
+    assert not fs.get("bot_tokens", "t1")["lease"]                      # nothing handed to a node
+    assert not fs.data.get("discord_streams")
+
+
+async def test_unknown_fails_open(fs):
+    _token(fs, "t1")
+    fs.snap_mode["t1"] = "unknown"
+    assert (await _start(node_id="n1"))["token_id"] == "t1"
+
+
+async def test_check_is_bounded_by_the_time_budget(fs, monkeypatch):
+    monkeypatch.setattr(settings, "discord_perm_check_budget_seconds", 0.05)
+    _token(fs, "t1")
+    fs.snap_mode["t1"] = "slow"
+    loop = asyncio.get_running_loop()
+    t0 = loop.time()
+    doc = await _start(node_id="n1")
+    assert doc["token_id"] == "t1" and loop.time() - t0 < 2
+
+
+async def test_concurrency_is_bounded(fs, monkeypatch):
+    monkeypatch.setattr(settings, "discord_perm_check_concurrency", 2)
+    for i in range(6):
+        _token(fs, f"t{i}")
+    live = {"now": 0, "max": 0}
+
+    async def _snapshot(token_doc, guild_id, *, refresh=False):
+        live["now"] += 1
+        live["max"] = max(live["max"], live["now"])
+        await asyncio.sleep(0.01)
+        live["now"] -= 1
+        return _snap(token_doc["token_id"])
+    monkeypatch.setattr(discord_rest, "guild_snapshot", _snapshot)
+    await _start(node_id="n1")
+    assert live["max"] == 2
+
+
+async def test_kill_switch_restores_unchecked_picking(fs, monkeypatch):
+    monkeypatch.setattr(settings, "discord_perm_check_enabled", False)
+    _token(fs, "t1")
+    fs.snap_mode["t1"] = "muted"
+    assert (await _start(node_id="n1"))["token_id"] == "t1"
+    assert fs.snap_calls == []
+
+
+async def test_text_channel_is_bad_channel_not_missing_permissions(fs):
+    _token(fs, "t1")
+    with pytest.raises(streams.StreamError) as e:
+        await _start(channel="1003", node_id="n1")
+    assert e.value.code == "bad_channel" and e.value.status == 400 and "#chat" in e.value.message
+
+
+async def test_only_this_orgs_and_member_bots_are_asked(fs):
+    _token(fs, "pool"); _token(fs, "theirs", org=OTHER); _token(fs, "elsewhere", guilds=("999",))
+    _token(fs, "dead", valid=False)
+    await _start(node_id="n1")
+    assert {c[0] for c in fs.snap_calls} == {"pool"}
+
+
+async def test_system_preferred_bot_is_used_first(fs):
+    _token(fs, "a"); _token(fs, "z")
+    fs.put("systems", SYS, {**fs.get("systems", SYS), "preferred_token_id": "z"})
+    assert (await _start(node_id="n1"))["token_id"] == "z"
+
+
+async def test_system_preferred_bot_is_a_hint_not_a_demand(fs):
+    _token(fs, "a"); _token(fs, "z")
+    fs.put("systems", SYS, {**fs.get("systems", SYS), "preferred_token_id": "z"})
+    fs.snap_mode["z"] = "muted"
+    assert (await _start(node_id="n1"))["token_id"] == "a"
+
+
+async def test_one_bot_setup_invite_mark_preferred_stream(fs):
+    """Owner requirement 1: only the system's own BYO radio bot is in the server."""
+    _token(fs, "byo", org=ORG)
+    _token(fs, "pool", guilds=("555",))
+    fs.put("systems", SYS, {**fs.get("systems", SYS), "preferred_token_id": "byo"})
+    assert (await _start())["token_id"] == "byo"
+
+
+async def test_running_stream_moves_off_a_bot_that_lost_its_permission(fs):
+    _token(fs, "t1"); _token(fs, "t2")
+    first = await _start(node_id="n1")
+    assert first["token_id"] == "t1"
+    discord_rest.invalidate()
+    fs.snap_mode["t1"] = "muted"
+    again = await streams.start_stream(org_id=ORG, stream_id=first["stream_id"], system_id=SYS, node_id="n1")
+    assert again["token_id"] == "t2" and again["reused"] is False
+    assert first["stream_id"] not in (fs.get("bot_tokens", "t1")["lease"] or {}).get("streams", [])
+
+
+async def test_handle_actual_drops_cached_access_on_edge_refusals(fs, monkeypatch):
+    _token(fs, "t1")
+    doc = await _start(node_id="n1")
+    dropped = []
+    monkeypatch.setattr(discord_rest, "invalidate", lambda token_id=None, guild_id=None: dropped.append(
+        (token_id, guild_id)) or 1)
+    for code in ("missing_permissions", "channel_full", "bad_channel", "no_audio"):
+        await streams.handle_actual("n1", {"streams": {doc["stream_id"]: {"state": "error", "code": code}}})
+    assert dropped == [("t1", GUILD)] * 3
+
+
+# ---------------------------------------------------------------------------
+# Every entry point gets the same refusal
+# ---------------------------------------------------------------------------
+
+def _all_denied(fs):
+    _token(fs, "t1", name="DRB Radio 2")
+    fs.snap_mode["t1"] = "muted"
+
+
+def test_bot_join_route_returns_409_with_text(fs):
+    _all_denied(fs)
+    r = run(discord_router.start_stream(discord_router.StreamStart(guild_id=GUILD, channel_id="1001",
+                                                                   system_id=SYS), {"service": True}))
+    assert r.status_code == 409
+    import json
+    body = json.loads(r.body)
+    assert body["code"] == "missing_permissions" and "DRB Radio 2 can't Speak in #dispatch" in body["detail"]
+    assert body["denied"][0]["missing"] == ["Speak"] and "secret" not in r.body.decode()
+
+
+def test_web_picker_returns_409_with_text(fs):
+    _all_denied(fs)
+    r = run(web.start_node_stream("n1", web.NodeStreamStart(guild_id=GUILD, channel_id="1001"), OWNER))
+    assert r.status_code == 409 and b"missing_permissions" in r.body and b"#dispatch" in r.body
+
+
+def test_node_discord_join_command_is_refused(fs):
+    _all_denied(fs)
+    with pytest.raises(HTTPException) as e:
+        run(nodes_router.send_command("n1", CommandPayload(action="discord_join", guild_id=GUILD,
+                                                           channel_id="1001"), ADMIN))
+    assert e.value.status_code == 409 and "can't Speak in #dispatch" in e.value.detail
+
+
+async def test_reconciler_marks_the_stream_with_the_refusal(fs):
+    _token(fs, "t1", name="DRB Radio 2")
+    doc = await _start(node_id="n1", mode="persistent")
+    # Node goes away; the bot meanwhile lost Speak.
+    await streams.release_offline_node("n1")
+    fs.put("nodes", "n1", {**fs.get("nodes", "n1"), "status": "offline", "last_seen": "2000-01-01T00:00:00+00:00"})
+    discord_rest.invalidate()
+    fs.snap_mode["t1"] = "muted"
+    await streams.reconcile_once()
+    s = fs.get("discord_streams", doc["stream_id"])
+    assert s["actual"]["state"] == "error" and "DRB Radio 2 can't Speak in #dispatch" in s["actual"]["error"]
+    assert not s.get("token_id")
+
+
+# ---------------------------------------------------------------------------
+# GET voice-check
+# ---------------------------------------------------------------------------
+
+def test_voice_check_reports_each_channel_for_an_org_member(fs):
+    _token(fs, "t1", name="DRB Radio 1"); _token(fs, "t2", name="DRB Radio 2")
+    fs.snap_mode["t2"] = "muted"
+    out = run(checks.node_voice_check("n1", GUILD, refresh=0, channel_id=None, decoded=MEMBER))
+    assert out["enabled"] is True and out["guild_name"] == "Fire Co. 7"
+    by = {c["channel_id"]: c for c in out["channels"]}
+    assert by["1001"]["state"] == "ready" and by["1001"]["any_ok"] is True
+    bots = {b["name"]: b for b in by["1001"]["bots"]}
+    assert bots["DRB Radio 2"]["ok"] is False and bots["DRB Radio 2"]["missing"] == ["Speak"]
+    assert bots["DRB Radio 2"]["reason"] == "denied by #dispatch's override for @Muted"
+    assert all("token_id" not in b for c in out["channels"] for b in c["bots"])       # names only
+    assert "secret" not in str(out)
+    assert by["1002"]["user_limit"] == 5
+
+
+def test_voice_check_needs_fix_when_every_free_bot_is_denied(fs):
+    _token(fs, "t1")
+    fs.snap_mode["t1"] = "muted"
+    out = run(checks.node_voice_check("n1", GUILD, refresh=0, channel_id="1001", decoded=OWNER))
+    assert [c["state"] for c in out["channels"]] == ["needs_fix"]
+    fs.snap_mode["t1"] = "unknown"
+    discord_rest.invalidate()
+    out = run(checks.node_voice_check("n1", GUILD, refresh=0, channel_id="1001", decoded=OWNER))
+    assert out["channels"][0]["state"] == "unknown"        # can't tell -> start still allowed
+
+
+def test_voice_check_is_org_scoped(fs):
+    _token(fs, "t1")
+    for node, guild, who in (("n1", "200", OWNER), ("nx", GUILD, OWNER), ("n1", GUILD, STRANGER)):
+        with pytest.raises(HTTPException) as e:
+            run(checks.node_voice_check(node, guild, refresh=0, channel_id=None, decoded=who))
+        assert e.value.status_code == 404
+
+
+def test_voice_check_admin_sees_token_ids_and_service_route(fs):
+    _token(fs, "t1")
+    out = run(checks.node_voice_check("n1", GUILD, refresh=0, channel_id=None, decoded=ADMIN))
+    assert out["channels"][0]["bots"][0]["token_id"] == "t1"
+    out = run(checks.guild_voice_check(GUILD, channel_id="1001", refresh=0, decoded={"service": True}))
+    assert out["org_id"] == ORG and out["channels"][0]["bots"][0]["token_id"] == "t1"
+
+
+def test_voice_check_refresh_is_rate_limited_per_caller_and_server(fs):
+    _token(fs, "t1")
+    first = run(checks.node_voice_check("n1", GUILD, refresh=1, channel_id=None, decoded=OWNER))
+    second = run(checks.node_voice_check("n1", GUILD, refresh=1, channel_id=None, decoded=OWNER))
+    assert first["refreshed"] is True and second["refreshed"] is False
+    assert [c[2] for c in fs.snap_calls] == [True, False]
+
+
+def test_voice_check_lists_invites_within_the_pool_cap(fs, monkeypatch):
+    monkeypatch.setattr(settings, "discord_pool_bot_guild_cap", 2)
+    _token(fs, "in")
+    _token(fs, "room", guilds=("1",))
+    _token(fs, "full", guilds=("1", "2"))
+    out = run(checks.node_voice_check("n1", GUILD, refresh=0, channel_id=None, decoded=OWNER))
+    assert [i["name"] for i in out["invite"]] == ["DRB Radio room"]
+
+
+def test_pool_full_rereads_membership_once(fs, monkeypatch):
+    # A full pool bot an admin just invited by hand: guild_ids is stale, so
+    # the first pick says pool_full; one membership re-read must still find it.
+    monkeypatch.setattr(settings, "discord_pool_bot_guild_cap", 2)
+    _token(fs, "full", guilds=("1", "2"))
+
+    async def _refresh(token_ids):
+        assert token_ids == ["full"]
+        fs.put("bot_tokens", "full", {**fs.get("bot_tokens", "full"), "guild_ids": ["1", "2", GUILD]})
+        return 1
+    monkeypatch.setattr(discord_rest, "refresh_membership", _refresh)
+    assert run(_start())["token_id"] == "full"
+
+
+def test_bot_overview_never_offers_a_full_pool_bot(fs, monkeypatch):
+    # A pool bot in other servers (non-empty guild_ids) at the cap must not get
+    # an invite link; one already in this server keeps its link.
+    monkeypatch.setattr(settings, "discord_pool_bot_guild_cap", 2)
+    _token(fs, "full", guilds=("1", "2"))
+    _token(fs, "room", guilds=("1",))
+    _token(fs, "here", guilds=("1", GUILD))
+    out = run(discord_router.guild_overview(GUILD, {"service": True}))
+    links = {b["token_id"]: b["invite_url"] for b in out["bots"]}
+    assert links["full"] is None and links["room"] and links["here"]
+
+
+def test_voice_check_busy_and_other_node(fs):
+    _token(fs, "t1", lease={"node_id": "n2", "streams": [], "lease_id": "L"})
+    out = run(checks.node_voice_check("n1", GUILD, refresh=0, channel_id="1001", decoded=OWNER))
+    b = out["channels"][0]["bots"][0]
+    assert b["ok"] is True and b["available"] is False and b["unavailable_reason"] == "other_node"
+    assert out["channels"][0]["state"] == "busy"
+
+
+def test_voice_check_kill_switch(fs, monkeypatch):
+    monkeypatch.setattr(settings, "discord_perm_check_enabled", False)
+    _token(fs, "t1")
+    out = run(checks.node_voice_check("n1", GUILD, refresh=0, channel_id=None, decoded=OWNER))
+    assert out["enabled"] is False and {c["state"] for c in out["channels"]} == {"unchecked"}
+    assert fs.snap_calls == []
+
+
+# ---------------------------------------------------------------------------
+# PUT /tokens/{id}/prefer/{system_id} — owner decision Q2
+# ---------------------------------------------------------------------------
+
+def _prefer(tid, sid, who):
+    return run(tokens_router.set_preferred_system(tid, sid, who))
+
+
+def test_owner_may_prefer_own_byo_bot_for_own_system(fs):
+    _token(fs, "byo", org=ORG)
+    assert _prefer("byo", SYS, OWNER)["preferred_for_system_id"] == SYS
+    assert fs.get("systems", SYS)["preferred_token_id"] == "byo"
+    assert fs.get("bot_tokens", "byo")["preferred_for_system_id"] == SYS
+    assert any(a.get("action") == "discord.token.prefer" for a in (fs.data.get("audit_log") or {}).values())
+    assert _prefer("byo", "_none", OWNER)["preferred_for_system_id"] is None
+    assert fs.get("systems", SYS)["preferred_token_id"] is None
+
+
+@pytest.mark.parametrize("tid,sid,who,status", [
+    ("pool", SYS, OWNER, 403),          # pool bots: platform admin only
+    ("byo", "shared", OWNER, 403),      # shared system: platform admin only
+    ("theirs", SYS, OWNER, 404),        # another org's bot: not found
+    ("byo", "sys-b", OWNER, 404),       # another org's system: not found
+    ("byo", SYS, MEMBER, 403),          # members can't
+    ("byo", SYS, STRANGER, 404),
+    ("nope", SYS, OWNER, 404),
+])
+def test_prefer_refusals(fs, tid, sid, who, status):
+    _token(fs, "pool"); _token(fs, "byo", org=ORG); _token(fs, "theirs", org=OTHER)
+    with pytest.raises(HTTPException) as e:
+        _prefer(tid, sid, who)
+    assert e.value.status_code == status
+    assert not (fs.get("systems", sid) or {}).get("preferred_token_id")
+
+
+def test_admin_may_prefer_any_bot_for_any_system(fs):
+    _token(fs, "pool")
+    assert _prefer("pool", "shared", ADMIN)["preferred_for_system_id"] == "shared"
+
+
+def test_prefer_keeps_both_pointers_single_valued(fs):
+    _token(fs, "byo", org=ORG); _token(fs, "byo2", org=ORG)
+    fs.put("systems", "sys-2", {"system_id": "sys-2", "name": "City", "org_id": ORG})
+    _prefer("byo", SYS, OWNER)
+    _prefer("byo", "sys-2", OWNER)               # moved: the old system no longer points at it
+    assert fs.get("systems", SYS)["preferred_token_id"] is None
+    _prefer("byo2", "sys-2", OWNER)              # replaced: the old bot loses its back-pointer
+    assert fs.get("bot_tokens", "byo")["preferred_for_system_id"] is None
+    assert fs.get("systems", "sys-2")["preferred_token_id"] == "byo2"
+
+
+# ---------------------------------------------------------------------------
+# End to end through the real REST client (Discord answered by MockTransport)
+# ---------------------------------------------------------------------------
+
+async def test_real_snapshot_path_denies_and_never_logs_the_token(fs, monkeypatch, caplog):
+    import httpx
+    from app.internal import discord_rest as rest
+    monkeypatch.undo()                     # drop the fixture's guild_snapshot stub ...
+    fs = FakeFirestore.install(monkeypatch)   # ... and re-install only what this test needs
+    from app.internal.mqtt_handler import mqtt_handler
+    monkeypatch.setattr(mqtt_handler, "publish_discord_desired", lambda n, p: True)
+    monkeypatch.setattr(settings, "discord_perm_check_enabled", True)
+    streams._guild_locks.clear()
+    fs.put("nodes", "n1", {"node_id": "n1", "name": "Node n1", "org_id": ORG, "status": "online",
+                           "approval_status": "approved", "assigned_system_id": SYS})
+    fs.put("systems", SYS, {"system_id": SYS, "name": "County P25", "org_id": ORG})
+    _token(fs, "t1", name="DRB Radio 2", bot_user_id="u-t1")
+    seen = []
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        seen.append((request.url.path, request.headers.get("authorization")))
+        path = request.url.path.removeprefix("/api/v10")
+        if path == f"/guilds/{GUILD}":
+            return httpx.Response(200, json=_guild())
+        if path == f"/guilds/{GUILD}/members/u-t1":
+            return httpx.Response(200, json={"user": {"id": "u-t1"}, "roles": ["r-muted"]})
+        if path == f"/guilds/{GUILD}/channels":
+            return httpx.Response(200, json=_channels())
+        return httpx.Response(404, json={"code": 0})
+    rest.reset()
+    monkeypatch.setattr(rest, "_transport", httpx.MockTransport(handler))
+    with pytest.raises(streams.StreamError) as e:
+        await _start(node_id="n1")
+    assert e.value.code == "missing_permissions" and "can't Speak in #dispatch" in e.value.message
+    assert {p for p, _ in seen} >= {f"/api/v10/guilds/{GUILD}", f"/api/v10/guilds/{GUILD}/channels"}
+    assert all(a == "Bot secret-t1" for _, a in seen)
+    assert "secret-t1" not in caplog.text and "secret-t1" not in e.value.message
+    # Cached: a second start asks Discord nothing new.
+    n = len(seen)
+    with pytest.raises(streams.StreamError):
+        await _start(node_id="n1")
+    assert len(seen) == n