Discord: check bot permissions before handing a token to a node (server-26#225 S3)

start_stream now asks Discord (cached REST snapshots, S2) whether each bot
it could pick may View, Connect and Speak in the voice channel, before the
lease. Covers all four entry points: bot POST /discord/streams, web picker,
nodes discord_join, and the reconciler's re-place.

- pick_token: denied bots skipped; confirmed-ok sorts before unknown;
  unknown is allowed (fail open); all denied -> missing_permissions naming
  each bot, missing permission, trace reason and fix. The reuse branch
  falls through when its token is denied.
- Bounded: discord_perm_check_concurrency at once, within
  discord_perm_check_budget_seconds; unanswered = unknown.
- Non-voice channel -> 400 bad_channel (not missing_permissions).
- systems/{id}.preferred_token_id is now read (plan finding 1), as a hint.
- Pool bots at discord_pool_bot_guild_cap are never offered as invites;
  pool_full when nothing else is left (owner Q3).
- handle_actual drops cached access on missing_permissions, channel_full,
  not_in_guild, bad_channel.
- New GET /discord/nodes/{node_id}/guilds/{guild_id}/voice-check (org
  member, names only) and GET /discord/guilds/{guild_id}/voice-check
  (service key or admin).
- PUT /tokens/{id}/prefer/{system_id}: org owners for their own systems
  and own BYO bots; shared systems and pool bots stay admin-only (owner Q2).
  Keeps both pointers single-valued; audited.
- Kill switch: DISCORD_PERM_CHECK_ENABLED=false.
- tests/conftest.py: autouse fixture so no test ever reaches discord.com.

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

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
+    # Kill switch (S3): False restores token picking without the REST
-    # REST permission check before a bot is handed to a node.
+    # 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:
+        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 existing.get("system_id") == system_id else None
-        cur = existing.get("node_id")
+        held_on = None
-        if cur and existing.get("system_id") == system_id:
+        if preferred_token_id or soft_pref:
-            candidates.sort(key=lambda n: 0 if n["node_id"] == cur else 1)
+            # A bot can only run on the node it is logged in on.
-        if preferred_token_id:
+            tok = await fstore.doc_get(leases.TOKENS, preferred_token_id or soft_pref) or {}
-            # 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 {}
             held_on = (tok.get("lease") or {}).get("node_id")
-            if held_on:
+        # Keep the current node when it is still a valid choice (reuse the
-                candidates.sort(key=lambda n: 0 if n["node_id"] == held_on else 1)
+        # 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,12 +610,22 @@ 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)
+                                            preferred_token_id=preferred_token_id, stream_fields=fields,
+                                            soft_preferred_token_id=soft_pref, perms=perms)
         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.
@@ -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"):
+        chan = _channel_ref(perms, channel_name or existing.get("channel_name"))
-            raise await _lease_error(res)                 # true on every node — no point trying the rest
+        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 member 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.
+    Mark this token as the preferred bot for a system: ``start_stream`` tries
-    When a discord_join is issued for any node in that system, this token
+    it first for that system's radio channels when nobody named a bot
-    is tried first before falling back to the general pool.
+    (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:
+    uid, email = describe_actor(decoded)
-        raise HTTPException(404, "Token not found.")
     if system_id == "_none":
-        # Clear any existing preference on the system that pointed to this token.
+        _check_prefer_access(decoded, existing, None, None)
-        system_doc = await fstore.doc_get("systems", existing.get("preferred_for_system_id", ""))
+        old_id = existing.get("preferred_for_system_id") or ""
-        if system_doc:
+        system_doc = await fstore.doc_get("systems", old_id) if old_id else None
-            await fstore.doc_set("systems", existing["preferred_for_system_id"], {"preferred_token_id": 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:
+    _check_prefer_access(decoded, existing, system_doc, system_id)
-        raise HTTPException(404, "System not found.")
+
-    # Set preference on both sides for easy lookup in either direction.
+    # 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,557 @@
+"""
+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_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