Discord: channel list and presence without the main bot (server-26#225 S5)

A server linked to an org now shows in the node picker while a radio bot the
org may use is in it, even when the DRB server bot is absent or left. Its
voice-channel list is read from that radio bot's cached REST snapshot
(discord_rest.guild_directory, the same snapshot the permission check reads,
so no extra Discord request): inline when no list is stored, in the
background when the list is older than discord_directory_refresh_seconds.
The server bot's own fresh list still wins (back-compat) and its directory
push now drops the server's cached snapshots.

Guards: a server acting only through the unlinked-server fallback org still
needs the server bot; REST never creates a discord_guilds doc, never writes
org_id or bot_present, and never writes onto a server unlinked meanwhile.

Also fixes server-26#251: an interactive start (/join, admin page, web
picker, node discord_join) re-reads a cached "denied" snapshot older than
10 s once before refusing, so "fix it in Discord, then try again" works on
the next try. The reconciler never re-reads (no REST loop).

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

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

@@ -138,6 +138,7 @@ async def start_stream(body: StreamStart, _: dict = Depends(require_service_key_
             mode=body.mode, channel_name=body.channel_name, guild_name=body.guild_name,
             created_by=body.created_by, preferred_token_id=body.preferred_token_id,
             max_streams=settings.discord_guild_max_streams, stream_id=body.stream_id,
+            interactive=True,   # /join, the admin page: re-read a stale "denied" (server-26#251)
         )
     except streams.StreamError as e:
         return _error(e)

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

@@ -6,15 +6,38 @@ The node page used to take raw guild/channel IDs. It now picks from:
   * the Discord servers the caller's org owns — ``discord_guilds`` docs that
     resolve to the node's org (an explicit link, or the unlinked-server
     fallback org, same rule the bot uses: streams.resolve_guild_org), and
-  * their voice channels, from a directory the DRB server bot pushes
-    (``PUT /discord/guilds/{id}/directory``). The bot already holds its
-    gateway cache of every server it is in; C2 never calls Discord with a
-    token on the browser's behalf and no token string ever reaches the browser.
+  * their voice channels, from a directory kept on ``discord_guilds/{id}``.
+
+Where the directory comes from (server-26#225 S5 — the main bot is optional):
+
+  * the DRB server bot, when it is in the server: it pushes its gateway view
+    (``PUT /discord/guilds/{id}/directory``) on start, on changes and every
+    15 minutes (``directory_source: "main_bot"``);
+  * otherwise a radio bot that is in the server: C2 reads the channel list
+    over Discord REST with that bot's token (discord_rest.guild_directory —
+    the same cached snapshot the permission check reads, so no extra
+    request) and stores it (``directory_source: "radio_bot"``,
+    ``radio_bot_present``). That happens when the picker is opened and the
+    stored list is missing (inline, inside the permission-check time budget)
+    or older than ``discord_directory_refresh_seconds`` (in the background).
+
+So C2 does call Discord when a signed-in user opens the picker — with a
+radio bot's token, for a server already linked to the caller's org, through
+the rate-limited, cached client. No token string ever reaches the browser.
+
+Which servers an org sees (``_org_guilds``):
+
+  * linked to the org: shown unless the server bot said it left AND no
+    radio bot the org may use is in it (``bot_tokens.guild_ids``);
+  * acting for the org only through the unlinked-server fallback: shown only
+    while the server bot is there, exactly as before. A radio bot's
+    membership never discovers a server, and REST never creates a
+    ``discord_guilds`` doc.
 
 A pick becomes a ``discord_streams`` desired-state doc **pinned to this node**
 (streams.start_stream(node_id=...)), exactly what the old raw-ID path did, with
 the checks that path skipped: the server must belong to the node's org, the
-channel must be a voice channel the bot reported in that server, and the
+channel must be a voice channel in that server's directory, and the
 per-server and per-node stream caps apply.
 
 Auth:
@@ -29,6 +52,7 @@ Auth:
 """
 from __future__ import annotations
 
+import asyncio
 from datetime import datetime, timezone
 from typing import Optional
 
@@ -38,6 +62,7 @@ from pydantic import BaseModel, Field
 
 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.audit import write_audit
@@ -54,6 +79,14 @@ router = APIRouter(prefix="/discord", tags=["discord"])
 
 MAX_CHANNELS = 200
 _NAME_MAX = 100
+MAIN_BOT = "main_bot"
+RADIO_BOT = "radio_bot"
+# The server bot re-pushes every 15 min. A list it pushed more than three
+# pushes ago means it is down (it never said it left): a radio bot's view
+# takes over until it is back.
+MAIN_BOT_STALE_SECONDS = 45 * 60
+# How many member radio bots are asked, one after another, for the list.
+DIRECTORY_BOTS = 3
 
 
 def _now_iso() -> str:
@@ -80,16 +113,20 @@ class GuildDirectory(BaseModel):
 @router.put("/guilds/{guild_id}/directory")
 async def put_guild_directory(guild_id: str, body: GuildDirectory, _: dict = Depends(require_service_key)):
     """The server bot's view of one Discord server: its name and voice channels.
-    Merged onto ``discord_guilds/{id}`` — never touches ``org_id`` (the link)."""
+    Merged onto ``discord_guilds/{id}`` — never touches ``org_id`` (the link).
+    Cached Discord snapshots of the server are dropped (a channel was added,
+    removed or renamed), so the next permission check reads it fresh."""
     if not guild_id.isdigit():
         raise HTTPException(400, "guild_id must be a Discord snowflake.")
     data: dict = {"guild_id": guild_id, "bot_present": body.present, "directory_at": _now_iso()}
     if body.present:
         data["voice_channels"] = [c.model_dump() for c in body.voice_channels]
+        data["directory_source"] = MAIN_BOT
         if body.name:
             # Same field the /drb link flow writes (feat/213-guild-link, PR #218).
             data["guild_name"] = body.name
     await fstore.doc_set(streams.GUILDS, guild_id, data)
+    discord_rest.invalidate(guild_id=guild_id)
     return {"ok": True, "guild_id": guild_id, "voice_channels": len(body.voice_channels)}
 
 
@@ -130,24 +167,156 @@ def _require_control(can: bool) -> None:
         raise HTTPException(403, "Only the organization owner, an admin, or this node's operator can do that.")
 
 
-async def _org_guilds(org_id: str) -> list[dict]:
-    """``discord_guilds`` docs this org may stream into, bot still present.
+async def _load_tokens() -> dict[str, dict]:
+    return {t["token_id"]: t for t in await fstore.collection_list(leases.TOKENS) if t.get("token_id")}
+
+
+def _member_radio_bots(tokens: dict[str, dict], guild_id: str, org_id: Optional[str]) -> list[str]:
+    """Radio bots this org may use (pool, or the org's own) that Discord last
+    said are in the server, best first. Unknown membership (``guild_ids``
+    None) does not count: presence has to be positive."""
+    out = [tid for tid, t in tokens.items()
+           if t.get("token") and t.get("valid") is not False and t.get("org_id") in (None, org_id)
+           and isinstance(t.get("guild_ids"), list) and guild_id in t["guild_ids"]]
+    out.sort(key=lambda tid: leases._candidate_order(tid, tokens[tid], "", org_id, None))
+    return out
+
+
+async def _org_guilds(org_id: str, tokens: Optional[dict] = None) -> list[dict]:
+    """``discord_guilds`` docs this org may stream into, with a bot there.
     Mirrors streams.resolve_guild_org: explicit link, else the unlinked-server
     fallback org — except a server an org explicitly unlinked (``unlinked_at``,
-    PR #218), which gets no fallback."""
+    PR #218), which gets no fallback.
+
+    A linked server is shown while the server bot is there or a radio bot the
+    org may use is (server-26#225 S5: the main bot is optional). A server
+    that acts for the org only through the fallback still needs the server
+    bot: a radio bot's membership never discovers one."""
     fallback = settings.discord_unlinked_guild_org_id or None
+    if tokens is None:
+        tokens = await _load_tokens()
     out = []
     for g in await fstore.collection_list(streams.GUILDS):
         gid = str(g.get("guild_id") or "")
-        if not gid or g.get("bot_present") is False:
+        if not gid:
             continue
-        owner = g.get("org_id") or (None if g.get("unlinked_at") else fallback)
-        if owner and owner == org_id:
-            out.append(g)
+        if g.get("org_id"):
+            if g["org_id"] != org_id:
+                continue
+            if g.get("bot_present") is False and not _member_radio_bots(tokens, gid, org_id):
+                continue
+        else:
+            if g.get("unlinked_at") or not fallback or fallback != org_id or g.get("bot_present") is False:
+                continue
+        out.append(g)
     out.sort(key=lambda g: ((_guild_name(g) or "").lower(), g["guild_id"]))
     return out
 
 
+# ---------------------------------------------------------------------------
+# Directory from a radio bot (server-26#225 S5)
+# ---------------------------------------------------------------------------
+
+def _age_seconds(ts) -> Optional[float]:
+    when = streams._parse_ts(ts)
+    return (datetime.now(timezone.utc) - when).total_seconds() if when else None
+
+
+def _directory_due(g: dict) -> Optional[str]:
+    """None when the stored list is fine; "missing" when there is none;
+    "stale" when it is old. Only for a server linked to an org — a fallback
+    server's list is the server bot's alone."""
+    if not g.get("org_id"):
+        return None
+    if g.get("voice_channels") is None:
+        return "missing"
+    from_main = g.get("bot_present") is True and (g.get("directory_source") or MAIN_BOT) == MAIN_BOT
+    limit = MAIN_BOT_STALE_SECONDS if from_main else settings.discord_directory_refresh_seconds
+    age = _age_seconds(g.get("directory_at"))
+    return "stale" if age is None or age > limit else None
+
+
+def _main_bot_list_fresh(g: dict) -> bool:
+    """The server bot is there and its own list is current: it wins."""
+    return (g.get("bot_present") is True and (g.get("directory_source") or MAIN_BOT) == MAIN_BOT
+            and _directory_due(g) is None)
+
+
+async def _refresh_directory(g: dict, org_id: str, tokens: dict[str, dict]) -> Optional[dict]:
+    """Read one linked server's voice channels from a member radio bot and
+    store them. Returns the fields written (also merged into ``g``), or None
+    when nothing was learned. Writes only onto the existing doc, and only
+    while it is still linked to ``org_id``: never creates a doc, never
+    touches ``org_id`` or the server bot's ``bot_present``."""
+    gid = g["guild_id"]
+    members = _member_radio_bots(tokens, gid, org_id)
+    if not members:
+        return None
+    docs = [{**tokens[t], "token_id": t} for t in members]
+    res = await discord_rest.guild_directory(docs, gid, max_bots=DIRECTORY_BOTS)
+    now = _now_iso()
+    if res.status == discord_rest.OK:
+        updates: dict = {"voice_channels": res.voice_channels, "directory_at": now,
+                         "directory_source": RADIO_BOT, "radio_bot_present": True, "radio_bot_checked_at": now}
+        if res.name:
+            updates["guild_name"] = res.name
+    elif res.status == discord_rest.NOT_MEMBER:
+        updates = {"radio_bot_present": False, "radio_bot_checked_at": now}
+    else:
+        return None
+    cur = await fstore.doc_get(streams.GUILDS, gid)
+    if not cur or cur.get("org_id") != org_id:
+        return None   # unlinked / moved / removed meanwhile
+    if _main_bot_list_fresh(cur):
+        return None   # the server bot pushed a fresh list meanwhile; it wins
+    await fstore.doc_update(streams.GUILDS, gid, updates)
+    g.update(updates)
+    logger.info(f"Discord directory for {gid} from radio bot {res.token_id or '-'}: "
+                f"{res.status}, {len(res.voice_channels or [])} voice channel(s)")
+    return updates
+
+
+_directory_tasks: dict[str, asyncio.Task] = {}
+
+
+async def _refresh_quietly(g: dict, org_id: str, tokens: dict[str, dict]) -> None:
+    try:
+        await _refresh_directory(g, org_id, tokens)
+    except Exception as e:   # a background refresh must never surface anywhere
+        logger.warning(f"Discord directory refresh for {g.get('guild_id')} failed: {type(e).__name__}")
+
+
+def _refresh_in_background(g: dict, org_id: str, tokens: dict[str, dict]) -> None:
+    gid = g["guild_id"]
+    running = _directory_tasks.get(gid)
+    if running is not None and not running.done():
+        return
+    task = asyncio.create_task(_refresh_quietly(dict(g), org_id, tokens))
+    _directory_tasks[gid] = task
+    task.add_done_callback(lambda t, gid=gid: _directory_tasks.pop(gid, None)
+                           if _directory_tasks.get(gid) is t else None)
+
+
+async def _fill_directories(guilds: list[dict], org_id: str, tokens: dict[str, dict]) -> None:
+    """Missing lists are read now (all servers at once, inside the
+    permission-check budget — an answer that comes later is cached and
+    stored by the next look); stale ones are refreshed in the background and
+    the stored list is shown meanwhile."""
+    now: list[dict] = []
+    for g in guilds:
+        due = _directory_due(g)
+        if not due or not _member_radio_bots(tokens, g["guild_id"], org_id):
+            continue
+        if due == "missing":
+            now.append(g)
+        else:
+            _refresh_in_background(g, org_id, tokens)
+    if not now:
+        return
+    tasks = [asyncio.create_task(_refresh_quietly(g, org_id, tokens)) for g in now]
+    await asyncio.wait(tasks, timeout=settings.discord_perm_check_budget_seconds)
+
+
 def _bot_status(tokens: dict, *, node_id: str, guild_id: str, org_id: str) -> dict:
     res = leases.pick_token(tokens, node_id=node_id, guild_id=guild_id, org_id=org_id)
     if res.ok:
@@ -181,7 +350,10 @@ async def join_options(node_id: str, decoded: dict = Depends(require_firebase_to
     system_id = streams._effective_system(node)
     system = (await fstore.doc_get_cached("systems", system_id)) if system_id else None
 
-    guilds = await _org_guilds(org_id) if org_id else []
+    tokens = await _load_tokens()
+    guilds = await _org_guilds(org_id, tokens) if org_id else []
+    if org_id:
+        await _fill_directories(guilds, org_id, tokens)
     live = [s for s in await fstore.collection_list(streams.STREAMS, desired="on")
             if s.get("org_id") == org_id]
     nodes = {n.get("node_id"): n.get("name") for n in await fstore.collection_list("nodes", org_id=org_id)
@@ -190,7 +362,6 @@ async def join_options(node_id: str, decoded: dict = Depends(require_firebase_to
     by_channel: dict[tuple, list] = {}
     for s in sorted(live, key=lambda s: (s.get("created_at") or "", s.get("stream_id") or "")):
         by_channel.setdefault((str(s.get("guild_id")), str(s.get("channel_id"))), []).append(s)
-    tokens = {t["token_id"]: t for t in await fstore.collection_list(leases.TOKENS) if t.get("token_id")}
     token_names = {tid: t.get("name") or tid for tid, t in tokens.items()}
 
     out_guilds = []
@@ -216,6 +387,10 @@ async def join_options(node_id: str, decoded: dict = Depends(require_firebase_to
             "linked": bool(g.get("org_id")),
             "synced": g.get("voice_channels") is not None,
             "synced_at": g.get("directory_at"),
+            # Who listed the channels: "main_bot" (DRB server bot) or "radio_bot" (S5).
+            "directory_source": (g.get("directory_source") or (MAIN_BOT if g.get("voice_channels") is not None
+                                                               else None)),
+            "main_bot_present": g.get("bot_present") is True,
             "voice_channels": channels,
             "streams_on": sum(1 for s in live if s.get("guild_id") == gid),
             "bot": _bot_status(tokens, node_id=node_id, guild_id=gid, org_id=org_id),
@@ -261,10 +436,24 @@ async def start_node_stream(node_id: str, body: NodeStreamStart, decoded: dict =
         raise HTTPException(409, "This node is not listening to a system yet.")
 
     guild_id, channel_id = str(body.guild_id), str(body.channel_id)
-    guild = next((g for g in await _org_guilds(org_id) if g["guild_id"] == guild_id), None)
+    tokens = await _load_tokens()
+    guild = next((g for g in await _org_guilds(org_id, tokens) if g["guild_id"] == guild_id), None)
     if not guild:
         raise HTTPException(404, "That Discord server is not linked to this organization.")
-    channel = next((c for c in guild.get("voice_channels") or [] if str(c.get("channel_id")) == channel_id), None)
+
+    def _find(g: dict) -> Optional[dict]:
+        return next((c for c in g.get("voice_channels") or [] if str(c.get("channel_id")) == channel_id), None)
+
+    channel = _find(guild)
+    if not channel and guild.get("org_id") and not _main_bot_list_fresh(guild):
+        # No list yet, or a radio bot's list that may predate this channel:
+        # read it once more (cached snapshot) before saying no.
+        try:
+            await asyncio.wait_for(_refresh_directory(guild, org_id, tokens),
+                                   timeout=settings.discord_perm_check_budget_seconds)
+        except asyncio.TimeoutError:
+            pass
+        channel = _find(guild)
     if not channel:
         raise HTTPException(400, "That is not a voice channel DRB can see in this server. "
                                  "Refresh the list and pick again.")
@@ -287,6 +476,7 @@ async def start_node_stream(node_id: str, body: NodeStreamStart, decoded: dict =
             channel_name=channel.get("name"), guild_name=_guild_name(guild),
             created_by=f"web:{decoded.get('uid') or 'unknown'}",
             max_streams=settings.discord_guild_max_streams,
+            interactive=True,
         )
     except streams.StreamError as e:
         return JSONResponse(status_code=e.status,

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

@@ -210,7 +210,7 @@ async def send_command(
             doc = await discord_streams.start_stream(
                 org_id=node.get("org_id"), guild_id=cmd.guild_id, channel_id=cmd.channel_id,
                 system_id=system_id, node_id=node_id, preferred_token_id=cmd.preferred_token_id,
-                mode="adhoc",
+                mode="adhoc", interactive=True,
             )
         except discord_streams.StreamError as e:
             raise HTTPException(e.status, e.message)