Merge pull request 'Discord: several radio streams per voice channel; adding never overwrites (#213)' (#227) from fix/213-multi-stream-per-channel into main

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

@@ -32,7 +32,15 @@ Firestore layout:
     in_use / assigned_node_id / assigned_at — legacy mirrors of ``lease``,
     still written so the old /tokens page and any reader of them stay right.
 
-  discord_streams/{guild_id}:{channel_id} — see discord_streams.py.
+  discord_streams/{stream_id} — see discord_streams.py. ``stream_id`` is
+    ``{guild_id}:{channel_id}:{suffix}`` (server-26#213 follow-up: several
+    streams may share one voice channel, each with its own bot); docs created
+    before that are plain ``{guild_id}:{channel_id}`` and keep working.
+
+Several streams in ONE guild therefore always hold DIFFERENT tokens: a token's
+lease lists its streams and ``pick_token`` skips any token that already has a
+stream in the guild. Both the read and the write happen inside the lease
+transaction, so two concurrent starts in one guild can never share a token.
 """
 from __future__ import annotations
 
@@ -63,10 +71,23 @@ def invite_url(application_id: Optional[str], guild_id: Optional[str] = None) ->
     return url
 
 
+# Old ids are ``guild:channel``; new ones add ``:<hex>``. Safe as a Firestore
+# doc id (no "/") and as a URL path segment. Every route validates against it.
+STREAM_ID_PATTERN = r"^[0-9]{1,25}:[0-9]{1,25}(?::[0-9a-f]{8,32})?$"
+
+
 def stream_id_for(guild_id: str, channel_id: str) -> str:
+    """The LEGACY one-stream-per-channel id. Only for reading docs created
+    before several streams could share a channel — never for a new stream."""
     return f"{guild_id}:{channel_id}"
 
 
+def new_stream_id(guild_id: str, channel_id: str) -> str:
+    """A fresh id for a new stream. Unique per stream, so adding a radio
+    channel can never land on (and overwrite) an existing one."""
+    return f"{guild_id}:{channel_id}:{uuid.uuid4().hex[:12]}"
+
+
 def guild_of(stream_id: str) -> str:
     return stream_id.split(":", 1)[0]
 
@@ -81,10 +102,15 @@ class LeaseResult:
     token_id: Optional[str] = None
     lease_id: Optional[str] = None
     reused: bool = False
-    # error codes: pool_empty | no_bot_in_guild | pool_exhausted
+    # 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)
     error: Optional[str] = None
     # bots that would work if invited to this guild: [{name, invite_url}]
     invite: list = field(default_factory=list)
+    # preferred_* errors: {token_name, busy_stream_id, node_id}
+    detail: dict = field(default_factory=dict)
 
 
 def _lease_streams(tok: dict) -> list:
@@ -124,9 +150,15 @@ def pick_token(
     preferred_token_id: Optional[str] = None,
     exclude_stream_id: Optional[str] = None,
 ) -> LeaseResult:
-    """Pure choice over a snapshot of bot_tokens. No I/O — unit tested directly."""
+    """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."""
     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)
     usable: list[tuple] = []
     not_member: list[dict] = []
     for tok_id, tok in tokens.items():
@@ -155,6 +187,31 @@ def pick_token(
     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:
+    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):
+        return LeaseResult(ok=False, error="preferred_unavailable",
+                           detail={"token_name": (tok or {}).get("name") or tok_id})
+    name = tok.get("name") or tok_id
+    busy = [s for s in _lease_streams(tok) if s != exclude_stream_id and guild_of(s) == guild_id]
+    if busy:
+        # One voice connection per bot per guild (Discord): never move the
+        # bot off another stream to serve this one.
+        return LeaseResult(ok=False, error="preferred_busy",
+                           detail={"token_name": name, "busy_stream_id": busy[0]})
+    lease = tok.get("lease") or {}
+    if lease.get("node_id") and lease["node_id"] != node_id:
+        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:
+        return LeaseResult(ok=False, error="no_bot_in_guild",
+                           invite=[{"name": name, "invite_url": invite_url(tok.get("application_id"), guild_id)}])
+    return LeaseResult(ok=True, token_id=tok_id)
+
+
 async def lease_for_stream(
     stream_id: str,
     *,
@@ -186,6 +243,8 @@ 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
+            # An explicit edit to a different bot moves the stream to it.
+            and (not preferred_token_id or preferred_token_id == prev_tok_id)
         ):
             lease = prev_tok["lease"]
             if lease.get("releasing"):

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

@@ -1,8 +1,14 @@
 """
 Discord radio channels as desired state (server-26#203, #204; node-26#25).
 
-One ``discord_streams/{guild_id}:{channel_id}`` doc per Discord voice channel
-that should carry radio. Fields:
+One ``discord_streams/{stream_id}`` doc per radio stream. A Discord voice
+channel may carry SEVERAL streams (server-26#213 follow-up, owner decision
+2026-10-05): each has its own radio bot and plays its own system, so a channel
+can hold e.g. County P25 on one bot and City Fire on another. ``stream_id`` is
+``{guild_id}:{channel_id}:{suffix}`` (``discord_leases.new_stream_id``); docs
+from before are ``{guild_id}:{channel_id}`` and are read, stopped and removed
+exactly the same way. Adding a stream never reuses another stream's doc except
+the same system in the same channel (idempotent re-add / restart). Fields:
 
   org_id, guild_id, channel_id, channel_name, guild_name,
   system_id, system_name,
@@ -201,12 +207,14 @@ async def publish_desired(node_id: str) -> bool:
 # ---------------------------------------------------------------------------
 
 class StreamError(Exception):
-    def __init__(self, code: str, message: str, invite: Optional[list] = None, status: int = 409):
+    def __init__(self, code: str, message: str, invite: Optional[list] = None, status: int = 409,
+                 extra: Optional[dict] = None):
         super().__init__(message)
         self.code = code
         self.message = message
         self.invite = invite or []
         self.status = status
+        self.extra = extra or {}
 
 
 ERROR_TEXT = {
@@ -214,16 +222,69 @@ ERROR_TEXT = {
     "no_node": "No online node in your organization is listening to that system right now.",
     "pool_empty": "No radio bots have been added to DRB yet. A platform admin adds them under Radio bots.",
     "no_bot_in_guild": "No free radio bot is a member of this server. Invite one with the link below, then try again.",
-    "pool_exhausted": "Every radio bot that can join this server is already busy. Stop another radio channel or add a bot.",
+    "pool_exhausted": ("Every radio bot that can join this server is already busy — a bot can be in only one voice "
+                       "channel per server, and a bot carrying radio on another node can't be shared. Add another "
+                       "radio bot to DRB and invite it to this server, or stop a radio channel here."),
     "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.",
+    "not_found": "That radio channel no longer exists.",
 }
 
+# One c2-core process. Serialising start_stream per guild makes "is this
+# system already playing in this channel?" + "is the server at its cap?" +
+# create a real check-then-write: two concurrent adds of the same system to
+# one channel land on ONE stream, and two adds can't both take the last slot.
+_guild_locks: dict[str, asyncio.Lock] = {}
+
+
+def _guild_lock(guild_id: str) -> asyncio.Lock:
+    lock = _guild_locks.get(guild_id)
+    if lock is None:
+        lock = _guild_locks[guild_id] = asyncio.Lock()
+    return lock
+
+
+def _stream_label(s: dict) -> str:
+    where = f"#{s['channel_name']}" if s.get("channel_name") else f"channel {s.get('channel_id')}"
+    return f"{s.get('system_name') or s.get('system_id') or 'radio'} in {where}"
+
+
+async def find_channel_stream(guild_id: str, channel_id: str, system_id: str) -> Optional[dict]:
+    """The stream already carrying ``system_id`` in this voice channel: a live
+    one first, else the most recently updated stopped one (restarted instead
+    of piling up a new doc per /join + /leave). None if there is neither."""
+    same = [s for s in await fstore.collection_list(STREAMS, guild_id=str(guild_id))
+            if str(s.get("channel_id")) == str(channel_id) and s.get("system_id") == system_id and s.get("stream_id")]
+    live = sorted((s for s in same if s.get("desired") == "on"), key=lambda s: s.get("created_at") or "")
+    if live:
+        return live[0]
+    same.sort(key=lambda s: s.get("updated_at") or "", reverse=True)
+    return same[0] if same else None
+
+
+async def _lease_error(res: leases.LeaseResult) -> "StreamError":
+    code = res.error or "no_node"
+    d = res.detail or {}
+    bot = d.get("token_name") or "That radio bot"
+    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"
+        return StreamError(code, f"{bot} is already playing {what} in this server. A bot can be in only one voice "
+                                 "channel per server — pick another bot, or let DRB choose any free one.",
+                           extra={"busy_stream_id": d.get("busy_stream_id")})
+    if code == "preferred_other_node":
+        node = (await fstore.doc_get("nodes", d["node_id"]) if d.get("node_id") else None) or {}
+        return StreamError(code, f"{bot} is carrying radio on another node ({node.get('name') or d.get('node_id')}) "
+                                 "that can't serve this channel. Pick another bot, or let DRB choose any free one.")
+    return StreamError(code, ERROR_TEXT.get(code, code), invite=res.invite)
+
 
 async def start_stream(
     *,
     org_id: str,
-    guild_id: str,
-    channel_id: str,
+    guild_id: Optional[str] = None,
+    channel_id: Optional[str] = None,
     system_id: str,
     mode: str = "adhoc",
     channel_name: Optional[str] = None,
@@ -232,16 +293,67 @@ async def start_stream(
     node_id: Optional[str] = None,
     preferred_token_id: Optional[str] = None,
     max_streams: Optional[int] = None,
+    stream_id: Optional[str] = None,
 ) -> dict:
-    """Make ``guild_id:channel_id`` carry ``system_id``. Idempotent: re-running
-    it for a channel that is already on reuses the node and the lease."""
+    """Put ``system_id`` on a voice channel.
+
+    * ``stream_id`` given — restart / edit THAT stream (the reconciler, the
+      admin page's Start button). Its guild and channel come from the doc.
+    * otherwise — ADD a stream to ``guild_id:channel_id``. If this system is
+      already on in that channel the existing stream is returned (idempotent
+      re-add, ``already_on``); a stopped one with this system is restarted;
+      anything else gets a brand-new stream id, so another system's stream in
+      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.
+    """
     if mode not in ("persistent", "adhoc"):
         raise StreamError("bad_mode", "mode must be persistent or adhoc", status=400)
     if not org_id:
         raise StreamError("no_org", ERROR_TEXT["no_org"], status=403)
+    existing: dict = {}
+    if stream_id:
+        existing = await fstore.doc_get(STREAMS, stream_id) or {}
+        if not existing:
+            raise StreamError("not_found", ERROR_TEXT["not_found"], status=404)
+        doc_guild, doc_channel = str(existing.get("guild_id")), str(existing.get("channel_id"))
+        if (guild_id and str(guild_id) != doc_guild) or (channel_id and str(channel_id) != doc_channel):
+            raise StreamError("bad_stream", "That radio channel belongs to a different server or voice channel.",
+                              status=400)
+        guild_id, channel_id = doc_guild, doc_channel
+    if not guild_id or not channel_id:
+        raise StreamError("bad_stream", "guild_id and channel_id are required.", status=400)
     guild_id, channel_id = str(guild_id), str(channel_id)
-    stream_id = leases.stream_id_for(guild_id, channel_id)
-    existing = await fstore.doc_get(STREAMS, stream_id) or {}
+    async with _guild_lock(guild_id):
+        return await _start_locked(
+            org_id=org_id, guild_id=guild_id, channel_id=channel_id, system_id=system_id, mode=mode,
+            channel_name=channel_name, guild_name=guild_name, created_by=created_by, node_id=node_id,
+            preferred_token_id=preferred_token_id, max_streams=max_streams, stream_id=stream_id,
+            existing=existing,
+        )
+
+
+async def _start_locked(*, org_id: str, guild_id: str, channel_id: str, system_id: str, mode: str,
+                        channel_name: Optional[str], guild_name: Optional[str], created_by: Optional[str],
+                        node_id: Optional[str], preferred_token_id: Optional[str], max_streams: Optional[int],
+                        stream_id: Optional[str], existing: dict) -> dict:
+    already_on = False
+    if stream_id:
+        existing = await fstore.doc_get(STREAMS, stream_id) or existing   # re-read under the lock
+    else:
+        existing = await find_channel_stream(guild_id, channel_id, system_id) or {}
+        if existing.get("desired") == "on":
+            already_on = True
+            if preferred_token_id and existing.get("token_id") and existing["token_id"] != preferred_token_id:
+                tok = await fstore.doc_get(leases.TOKENS, existing["token_id"]) or {}
+                raise StreamError("duplicate", f"{_stream_label(existing)} is already playing on radio bot "
+                                               f"{tok.get('name') or existing['token_id']}. Remove that one "
+                                               "first, or pick another system.",
+                                  extra={"busy_stream_id": existing["stream_id"]})
+            # A re-add is not an edit: keep the running stream's label and mode.
+            channel_name, mode = None, existing.get("mode") or mode
+        stream_id = existing.get("stream_id") or leases.new_stream_id(guild_id, channel_id)
 
     if max_streams is not None and existing.get("desired") != "on":
         live = [s for s in await fstore.collection_list(STREAMS, guild_id=guild_id) if s.get("desired") == "on"]
@@ -263,6 +375,12 @@ async def start_stream(
         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 {}
+            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)
     if not candidates:
         raise StreamError("no_node", ERROR_TEXT["no_node"])
 
@@ -306,13 +424,16 @@ async def start_stream(
             if prev_node and prev_node != node["node_id"]:
                 await publish_desired(prev_node)
             logger.info(f"Discord stream {stream_id} -> node {node['node_id']} token {res.token_id}"
-                        f" ({'reused' if res.reused else 'new'} lease)")
+                        f" ({'reused' if res.reused else 'new'} lease{', already on' if already_on else ''})")
             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}
+                    "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
         last = res
-    code = last.error if last and last.error else "no_node"
-    raise StreamError(code, ERROR_TEXT.get(code, code), invite=last.invite if last else [])
+    if last is None:
+        raise StreamError("no_node", ERROR_TEXT["no_node"])
+    raise await _lease_error(last)
 
 
 async def stop_stream(stream_id: str, reason: str = "stopped") -> bool:
@@ -395,14 +516,19 @@ async def remove_guild(guild_id: str, by: str) -> dict:
             "guild_name": link.get("guild_name")}
 
 
-async def stop_guild(guild_id: str, reason: str, channel_id: Optional[str] = None) -> list[str]:
-    """Stop every live stream in ONE Discord server (optionally one channel).
-    The /leave fix for server-26#203: never touches another server's streams."""
+async def stop_guild(guild_id: str, reason: str, channel_id: Optional[str] = None,
+                     stream_id: Optional[str] = None) -> list[str]:
+    """Stop live streams in ONE Discord server: all of them, every stream in
+    one voice channel, or one stream by id. The /leave fix for server-26#203:
+    never touches another server's streams (a ``stream_id`` from another
+    server simply matches nothing)."""
     stopped = []
     for s in await fstore.collection_list(STREAMS, guild_id=str(guild_id)):
         if s.get("desired") != "on":
             continue
-        if channel_id and s.get("channel_id") != str(channel_id):
+        if channel_id and str(s.get("channel_id")) != str(channel_id):
+            continue
+        if stream_id and s.get("stream_id") != stream_id:
             continue
         await stop_stream(s["stream_id"], reason)
         stopped.append(s["stream_id"])
@@ -456,8 +582,8 @@ async def handle_actual(node_id: str, payload: dict) -> None:
     recheck: set = set()
     if isinstance(reported, dict):
         for stream_id, st in list(reported.items())[:50]:
-            if not isinstance(st, dict):
-                continue
+            if not isinstance(st, dict) or "/" in str(stream_id):
+                continue  # "/" would be a Firestore sub-path, never a stream id
             doc = await fstore.doc_get(STREAMS, str(stream_id))
             if not doc or doc.get("node_id") != node_id or doc.get("desired") != "on":
                 continue
@@ -552,8 +678,8 @@ async def reconcile_once() -> None:
             continue
         org_id = s.get("org_id")
         try:
-            await start_stream(org_id=org_id, guild_id=s["guild_id"], channel_id=s["channel_id"],
-                               system_id=s["system_id"], mode="persistent")
+            await start_stream(org_id=org_id, stream_id=s["stream_id"], system_id=s["system_id"],
+                               mode="persistent")
         except StreamError as e:
             await _mark(s, "error", e.message)
 

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

@@ -23,7 +23,7 @@ from typing import Optional
 from fastapi import APIRouter, Depends, HTTPException, Path, Security
 from fastapi.responses import JSONResponse
 from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
-from pydantic import BaseModel
+from pydantic import BaseModel, Field
 
 from app.config import settings
 from app.internal import discord_leases as leases
@@ -35,14 +35,14 @@ from app.internal.logger import logger
 
 router = APIRouter(prefix="/discord", tags=["discord"])
 _bearer = HTTPBearer(auto_error=False)
-# guild:channel snowflakes (leases.stream_id_for) — anything else is a 422,
-# never a Firestore path.
-STREAM_ID = r"^[0-9]{1,25}:[0-9]{1,25}$"
+# guild:channel[:suffix] (leases.STREAM_ID_PATTERN) — old one-per-channel ids
+# and new per-stream ids. Anything else is a 422, never a Firestore path.
+STREAM_ID = leases.STREAM_ID_PATTERN
 
 
 def _error(e: streams.StreamError) -> JSONResponse:
     return JSONResponse(status_code=e.status,
-                        content={"detail": e.message, "code": e.code, "invite": e.invite})
+                        content={**e.extra, "detail": e.message, "code": e.code, "invite": e.invite})
 
 
 async def _org_or_403(guild_id: str) -> str:
@@ -54,7 +54,8 @@ async def _org_or_403(guild_id: str) -> str:
 
 def _public_stream(s: dict) -> dict:
     keep = ("stream_id", "guild_id", "channel_id", "channel_name", "system_id", "system_name", "mode",
-            "desired", "node_id", "token_id", "actual", "stop_reason", "updated_at", "delivered", "reused")
+            "desired", "node_id", "token_id", "token_name", "actual", "stop_reason", "updated_at", "delivered",
+            "reused", "already_on")
     return {k: s.get(k) for k in keep if k in s}
 
 
@@ -75,8 +76,13 @@ async def guild_overview(guild_id: str, _: dict = Depends(require_service_key_or
     nodes = await fstore.collection_list("nodes", org_id=org_id)
     gstreams = [s for s in await fstore.collection_list(streams.STREAMS, guild_id=guild_id)
                 if s.get("desired") == "on"]
+    all_tokens = await fstore.collection_list(leases.TOKENS)
+    names = {t.get("token_id"): t.get("name") for t in all_tokens}
+    # Several streams can share a voice channel; the bot names each by its bot.
+    gstreams = [{**s, "token_name": names.get(s.get("token_id"))} for s in gstreams]
+    gstreams.sort(key=lambda s: (s.get("channel_name") or "", s.get("channel_id") or "", s.get("created_at") or ""))
     bots = []
-    for t in await fstore.collection_list(leases.TOKENS):
+    for t in all_tokens:
         if t.get("org_id") not in (None, org_id) or t.get("valid") is False:
             continue
         lease = t.get("lease") or {}
@@ -104,6 +110,9 @@ class StreamStart(BaseModel):
     guild_id: str
     channel_id: str
     system_id: str
+    # Set = restart / edit THAT stream. Unset = add a stream to the channel
+    # (never overwrites another system's stream there — several may share it).
+    stream_id: Optional[str] = Field(None, pattern=STREAM_ID)
     mode: str = "adhoc"
     preferred_token_id: Optional[str] = None
     channel_name: Optional[str] = None
@@ -125,7 +134,7 @@ async def start_stream(body: StreamStart, _: dict = Depends(require_service_key_
             org_id=org_id, guild_id=body.guild_id, channel_id=body.channel_id, system_id=body.system_id,
             mode=body.mode, channel_name=body.channel_name, guild_name=body.guild_name,
             created_by=body.created_by, preferred_token_id=body.preferred_token_id,
-            max_streams=settings.discord_guild_max_streams,
+            max_streams=settings.discord_guild_max_streams, stream_id=body.stream_id,
         )
     except streams.StreamError as e:
         return _error(e)
@@ -133,7 +142,8 @@ async def start_stream(body: StreamStart, _: dict = Depends(require_service_key_
 
 
 @router.get("/streams/{stream_id}")
-async def get_stream(stream_id: str, _: dict = Depends(require_service_key_or_admin)):
+async def get_stream(stream_id: str = Path(..., pattern=STREAM_ID),
+                     _: dict = Depends(require_service_key_or_admin)):
     doc = await fstore.doc_get(streams.STREAMS, stream_id)
     if not doc:
         raise HTTPException(404, "Stream not found.")
@@ -141,13 +151,17 @@ async def get_stream(stream_id: str, _: dict = Depends(require_service_key_or_ad
 
 
 class GuildStop(BaseModel):
+    # Neither: every live stream in the server. channel_id: every stream in
+    # that voice channel. stream_id: just that one (it must be this server's).
     channel_id: Optional[str] = None
+    stream_id: Optional[str] = Field(None, pattern=STREAM_ID)
     reason: str = "stopped"
 
 
 @router.post("/guilds/{guild_id}/stop")
 async def stop_guild(guild_id: str, body: GuildStop, _: dict = Depends(require_service_key_or_admin)):
-    stopped = await streams.stop_guild(guild_id, body.reason[:60], channel_id=body.channel_id)
+    stopped = await streams.stop_guild(guild_id, body.reason[:60], channel_id=body.channel_id,
+                                       stream_id=body.stream_id)
     return {"stopped": stopped}
 
 
@@ -293,8 +307,9 @@ async def admin_overview(_: dict = Depends(require_admin_token)):
         g["acting_org_name"] = (orgs.get(acting) or {}).get("name") if acting else None
         g["live_streams"] = sum(1 for r in rows if r.get("guild_id") == g["guild_id"] and r.get("desired") == "on")
 
+    # Streams sharing a voice channel sit together, oldest first.
     rows.sort(key=lambda r: (0 if r.get("desired") == "on" else 1, r.get("guild_name") or r.get("guild_id") or "",
-                             r.get("channel_name") or r.get("channel_id") or ""))
+                             r.get("channel_id") or "", r.get("created_at") or "", r.get("stream_id") or ""))
     systems = await fstore.collection_list("systems")
     # Names for every node a stream or a radio-bot lease points at.
     referenced = {r.get("node_id") for r in rows} | {(t.get("lease") or {}).get("node_id") for t in tokens.values()}

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

@@ -156,14 +156,16 @@ def _bot_status(tokens: dict, *, node_id: str, guild_id: str, org_id: str) -> di
             "invite": [i for i in res.invite if i.get("invite_url")][:5]}
 
 
-def _stream_view(s: dict, node_names: dict) -> dict:
+def _stream_view(s: dict, node_names: dict, token_names: Optional[dict] = None) -> dict:
     return {
         "stream_id": s.get("stream_id"),
         "guild_id": s.get("guild_id"),
         "channel_id": s.get("channel_id"),
         "guild_name": s.get("guild_name"),
         "channel_name": s.get("channel_name"),
+        "system_id": s.get("system_id"),
         "system_name": s.get("system_name"),
+        "token_name": (token_names or {}).get(s.get("token_id") or ""),
         "mode": s.get("mode"),
         "node_id": s.get("node_id"),
         "node_name": node_names.get(s.get("node_id") or "") or s.get("node_id"),
@@ -184,8 +186,12 @@ async def join_options(node_id: str, decoded: dict = Depends(require_firebase_to
             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)
              if n.get("node_id")} if org_id else {}
-    by_channel = {s.get("stream_id"): s for s in live}
+    # Several streams can share one voice channel (each its own bot + system).
+    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 = []
     for g in guilds:
@@ -193,12 +199,16 @@ async def join_options(node_id: str, decoded: dict = Depends(require_firebase_to
         channels = []
         for c in sorted(g.get("voice_channels") or [],
                         key=lambda c: (c.get("position") or 0, (c.get("name") or "").lower())):
-            s = by_channel.get(leases.stream_id_for(gid, str(c.get("channel_id"))))
+            on = by_channel.get((gid, str(c.get("channel_id"))), [])
+            # ``stream``: the one a pick here would touch — this system's (a
+            # pick restarts / moves it). Every other stream is left alone.
+            mine = next((s for s in on if s.get("system_id") == system_id), None)
             channels.append({
                 "channel_id": str(c.get("channel_id")),
                 "name": c.get("name"),
                 "category": c.get("category"),
-                "stream": _stream_view(s, nodes) if s else None,
+                "stream": _stream_view(mine, nodes, token_names) if mine else None,
+                "streams": [_stream_view(s, nodes, token_names) for s in on],
             })
         out_guilds.append({
             "guild_id": gid,
@@ -227,7 +237,7 @@ async def join_options(node_id: str, decoded: dict = Depends(require_firebase_to
             "guild_max": settings.discord_guild_max_streams,
         },
         "guilds": out_guilds,
-        "streams": [_stream_view(s, nodes) for s in here],
+        "streams": [_stream_view(s, nodes, token_names) for s in here],
     }
 
 
@@ -259,8 +269,9 @@ async def start_node_stream(node_id: str, body: NodeStreamStart, decoded: dict =
         raise HTTPException(400, "That is not a voice channel DRB can see in this server. "
                                  "Refresh the list and pick again.")
 
-    stream_id = leases.stream_id_for(guild_id, channel_id)
-    existing = await fstore.doc_get(streams.STREAMS, stream_id) or {}
+    # Same system already on in this channel = that stream (moved here if it
+    # is on another node); anything else is a NEW stream beside the others.
+    existing = await streams.find_channel_stream(guild_id, channel_id, system_id) or {}
     already_here = existing.get("desired") == "on" and existing.get("node_id") == node_id
     if not already_here:
         used = sum(1 for s in await fstore.collection_list(streams.STREAMS, node_id=node_id)
@@ -279,16 +290,16 @@ async def start_node_stream(node_id: str, body: NodeStreamStart, decoded: dict =
         )
     except streams.StreamError as e:
         return JSONResponse(status_code=e.status,
-                            content={"detail": e.message, "code": e.code, "invite": e.invite})
-    logger.info(f"Web: radio channel {stream_id} pinned to node {node_id} ({body.mode})")
+                            content={**e.extra, "detail": e.message, "code": e.code, "invite": e.invite})
+    logger.info(f"Web: radio channel {doc['stream_id']} pinned to node {node_id} ({body.mode})")
     return {"ok": True, "stream_id": doc["stream_id"], "token_id": doc.get("token_id"),
             "delivered": doc.get("delivered"), "reused": doc.get("reused"), "actual": doc.get("actual")}
 
 
 class NodeStreamStop(BaseModel):
-    # guild:channel snowflakes (leases.stream_id_for). Validated so a malformed
-    # id (e.g. one with "/") is a 422, not a Firestore path error / 500.
-    stream_id: Optional[str] = Field(None, pattern=r"^[0-9]{1,25}:[0-9]{1,25}$")
+    # guild:channel[:suffix] (leases.STREAM_ID_PATTERN). Validated so a
+    # malformed id (e.g. one with "/") is a 422, not a Firestore path error / 500.
+    stream_id: Optional[str] = Field(None, pattern=leases.STREAM_ID_PATTERN)
 
 
 @router.post("/nodes/{node_id}/streams/stop")
@@ -307,7 +318,7 @@ async def stop_node_streams(node_id: str, body: NodeStreamStop, decoded: dict =
 
 @router.delete("/nodes/{node_id}/streams/{stream_id}")
 async def remove_node_stream(node_id: str,
-                             stream_id: str = Path(..., pattern=r"^[0-9]{1,25}:[0-9]{1,25}$"),
+                             stream_id: str = Path(..., pattern=leases.STREAM_ID_PATTERN),
                              decoded: dict = Depends(require_firebase_token)):
     """Permanently remove a radio channel this node carries (or last carried):
     stopped through desired state first so the node logs the bot out and the