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
-    # 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,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"):
-            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 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.
-    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}