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
+One ``discord_streams/{stream_id}`` doc per radio stream. A Discord voice
-that should carry radio. Fields:
+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,
+    guild_id: Optional[str] = None,
-    channel_id: str,
+    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
+    """Put ``system_id`` on a voice channel.
-    it for a channel that is already on reuses the node and the lease."""
+
+    * ``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)
+    async with _guild_lock(guild_id):
-    existing = await fstore.doc_get(STREAMS, stream_id) or {}
+        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"
+    if last is None:
-    raise StreamError(code, ERROR_TEXT.get(code, code), invite=last.invite if last else [])
+        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]:
+async def stop_guild(guild_id: str, reason: str, channel_id: Optional[str] = None,
-    """Stop every live stream in ONE Discord server (optionally one channel).
+                     stream_id: Optional[str] = None) -> list[str]:
-    The /leave fix for server-26#203: never touches another server's streams."""
+    """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):
+            if not isinstance(st, dict) or "/" in str(stream_id):
-                continue
+                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"],
+            await start_stream(org_id=org_id, stream_id=s["stream_id"], system_id=s["system_id"],
-                               system_id=s["system_id"], mode="persistent")
+                               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,
+# guild:channel[:suffix] (leases.STREAM_ID_PATTERN) — old one-per-channel ids
-# never a Firestore path.
+# and new per-stream ids. Anything else is a 422, never a Firestore path.
-STREAM_ID = r"^[0-9]{1,25}:[0-9]{1,25}$"
+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)
+    # Same system already on in this channel = that stream (moved here if it
-    existing = await fstore.doc_get(streams.STREAMS, stream_id) or {}
+    # 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})
+                            content={**e.extra, "detail": e.message, "code": e.code, "invite": e.invite})
-    logger.info(f"Web: radio channel {stream_id} pinned to node {node_id} ({body.mode})")
+    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
+    # guild:channel[:suffix] (leases.STREAM_ID_PATTERN). Validated so a
-    # id (e.g. one with "/") is a 422, not a Firestore path error / 500.
+    # 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}$")
+    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

drb-c2-core/tests/test_discord_leases.py

@@ -34,6 +34,7 @@ def fs(monkeypatch):
     from app.internal.mqtt_handler import mqtt_handler
     monkeypatch.setattr(mqtt_handler, "publish_discord_desired", _publish)
     streams._publish_locks.clear()
+    streams._guild_locks.clear()
 
     async def _no_discord(token_ids):
         return 0
@@ -70,6 +71,15 @@ def _lease(fake, token_id):
     return (fake.get("bot_tokens", token_id) or {}).get("lease")
 
 
+def _sid(fake, guild="g1", channel="c1"):
+    """The id of the one stream start_stream created in this channel (new
+    streams get a unique ``guild:channel:suffix`` id — server-26#213)."""
+    ids = [k for k, d in (fake.data.get("discord_streams") or {}).items()
+           if d.get("guild_id") == guild and d.get("channel_id") == channel]
+    assert len(ids) == 1, ids
+    return ids[0]
+
+
 # ---------------------------------------------------------------------------
 # pick_token — pure choice
 # ---------------------------------------------------------------------------
@@ -118,7 +128,8 @@ async def test_rejoin_reuses_the_lease_instead_of_taking_a_second_token(fs):
     assert again["token_id"] == first["token_id"] and again["reused"] is True
     leased = [t for t in ("t1", "t2") if _lease(fs, t)]
     assert leased == [first["token_id"]]                     # server-26#204 item 2
-    assert _lease(fs, first["token_id"])["streams"] == ["g1:c1"]
+    assert _lease(fs, first["token_id"])["streams"] == [first["stream_id"]]
+    assert again["stream_id"] == first["stream_id"] and again["already_on"] is True
 
 
 async def test_preferred_token_is_honoured(fs):
@@ -144,9 +155,9 @@ async def test_join_never_moves_another_guilds_stream(fs):
     _system(fs); _node(fs, "n1"); _token(fs, "t1")
     await _start("gA", "c1")
     await _start("gB", "c1")
-    a = fs.get("discord_streams", "gA:c1")
+    a = fs.get("discord_streams", _sid(fs, "gA", "c1"))
     assert a["desired"] == "on" and a["token_id"] == "t1"
-    assert sorted(_lease(fs, "t1")["streams"]) == ["gA:c1", "gB:c1"]
+    assert sorted(_lease(fs, "t1")["streams"]) == sorted([_sid(fs, "gA", "c1"), _sid(fs, "gB", "c1")])
 
 
 async def test_desired_state_never_carries_the_token_string(fs):
@@ -236,7 +247,7 @@ async def test_two_guilds_same_node_racing_share_one_session(fs):
 async def test_stop_marks_releasing_and_holds_until_the_node_acks(fs):
     _system(fs); _node(fs, "n1"); _node(fs, "n2"); _token(fs, "t1")
     await _start()
-    await streams.stop_stream("g1:c1")
+    await streams.stop_stream(_sid(fs, "g1", "c1"))
     lease = _lease(fs, "t1")
     assert lease["releasing"] is True and lease["streams"] == []
     drop_seq = lease["release_seq"]
@@ -260,25 +271,26 @@ async def test_stop_marks_releasing_and_holds_until_the_node_acks(fs):
 async def test_restart_before_release_reuses_the_logged_in_session(fs):
     _system(fs); _node(fs, "n1"); _token(fs, "t1"); _token(fs, "t2")
     first = await _start()
-    await streams.stop_stream("g1:c1")
+    await streams.stop_stream(_sid(fs, "g1", "c1"))
     again = await _start()
     assert again["token_id"] == first["token_id"]
     lease = _lease(fs, first["token_id"])
-    assert lease["releasing"] is False and lease["streams"] == ["g1:c1"]
+    assert lease["releasing"] is False and lease["streams"] == [first["stream_id"]]
+    assert again["stream_id"] == first["stream_id"]          # restarted, not a second doc
 
 
 async def test_handle_actual_settles_and_ignores_other_nodes_streams(fs):
     _system(fs); _node(fs, "n1"); _node(fs, "n2"); _token(fs, "t1"); _token(fs, "t2")
-    await _start("g1", "c1", node_id="n1")
+    a = await _start("g1", "c1", node_id="n1")
-    await _start("g2", "c1", node_id="n2")
+    b = await _start("g2", "c1", node_id="n2")
     await streams.handle_actual("n1", {
         "applied_seq": 1, "sessions": ["t1"],
-        "streams": {"g1:c1": {"state": "connected"}, "g2:c1": {"state": "error", "error": "spoof"}},
+        "streams": {a["stream_id"]: {"state": "connected"}, b["stream_id"]: {"state": "error", "error": "spoof"}},
     })
-    assert fs.get("discord_streams", "g1:c1")["actual"]["state"] == "connected"
+    assert fs.get("discord_streams", _sid(fs, "g1", "c1"))["actual"]["state"] == "connected"
-    assert fs.get("discord_streams", "g2:c1")["actual"]["state"] == "pending"   # not n1's to report
+    assert fs.get("discord_streams", _sid(fs, "g2", "c1"))["actual"]["state"] == "pending"   # not n1's to report
 
-    await streams.stop_stream("g1:c1")
+    await streams.stop_stream(_sid(fs, "g1", "c1"))
     seq = _lease(fs, "t1")["release_seq"]
     await streams.handle_actual("n1", {"applied_seq": seq, "sessions": [], "streams": {}})
     assert _lease(fs, "t1") is None
@@ -295,7 +307,7 @@ async def test_failed_publish_is_retried_by_the_reconciler(fs):
     n = len(fs.published)
     await streams.reconcile_once()
     assert len(fs.published) > n and fs.get("discord_nodes", "n1")["published"] is True
-    assert fs.published[-1][1]["streams"][0]["stream_id"] == "g1:c1"
+    assert fs.published[-1][1]["streams"][0]["stream_id"] == doc["stream_id"]
 
 
 async def test_concurrent_publishes_keep_the_retained_list_newest(fs):
@@ -315,9 +327,9 @@ async def test_leave_in_one_guild_never_touches_another(fs):
     await _start("gA", "c1")
     await _start("gB", "c1")
     stopped = await streams.stop_guild("gA", "leave")
-    assert stopped == ["gA:c1"]
+    assert stopped == [_sid(fs, "gA", "c1")]
-    assert fs.get("discord_streams", "gB:c1")["desired"] == "on"
+    assert fs.get("discord_streams", _sid(fs, "gB", "c1"))["desired"] == "on"
-    assert fs.get("discord_streams", "gA:c1")["desired"] == "off"
+    assert fs.get("discord_streams", _sid(fs, "gA", "c1"))["desired"] == "off"
 
 
 async def test_guild_org_resolution(fs, monkeypatch):
@@ -340,7 +352,7 @@ async def test_offline_node_keeps_its_lease_inside_the_grace(fs, monkeypatch):
     _node(fs, "n1", status="offline", age=120)
     await streams.reconcile_once()
     assert _lease(fs, "t1")["node_id"] == "n1"
-    assert fs.get("discord_streams", "g1:c1")["actual"]["state"] == "node_offline"
+    assert fs.get("discord_streams", _sid(fs, "g1", "c1"))["actual"]["state"] == "node_offline"
 
 
 async def test_persistent_stream_moves_after_grace_adhoc_stops(fs, monkeypatch):
@@ -351,10 +363,10 @@ async def test_persistent_stream_moves_after_grace_adhoc_stops(fs, monkeypatch):
     _node(fs, "n1", status="offline", age=900)
     _node(fs, "n2")
     await streams.reconcile_once()
-    moved = fs.get("discord_streams", "g1:c1")
+    moved = fs.get("discord_streams", _sid(fs, "g1", "c1"))
     assert moved["desired"] == "on" and moved["node_id"] == "n2" and moved["token_id"]
     assert _lease(fs, moved["token_id"])["node_id"] == "n2"
-    assert fs.get("discord_streams", "g2:c1")["desired"] == "off"
+    assert fs.get("discord_streams", _sid(fs, "g2", "c1"))["desired"] == "off"
     assert all((_lease(fs, t) or {}).get("node_id") != "n1" for t in ("t1", "t2"))
 
 
@@ -365,7 +377,7 @@ async def test_leave_while_node_offline_frees_token_after_grace(fs, monkeypatch)
     _system(fs); _node(fs, "n1"); _token(fs, "t1")
     await _start()
     _node(fs, "n1", status="offline", age=120)
-    await streams.stop_stream("g1:c1")
+    await streams.stop_stream(_sid(fs, "g1", "c1"))
     await streams.reconcile_once()
     assert _lease(fs, "t1")["releasing"] is True          # inside the grace: held
     _node(fs, "n1", status="offline", age=900)
@@ -377,7 +389,7 @@ async def test_deleted_node_lease_freed_after_grace_from_last_report(fs, monkeyp
     monkeypatch.setattr(streams.settings, "discord_lease_offline_release_seconds", 600)
     _system(fs); _node(fs, "n1"); _token(fs, "t1")
     await _start()
-    await streams.stop_stream("g1:c1")
+    await streams.stop_stream(_sid(fs, "g1", "c1"))
     fs.put("discord_nodes", "n1", {**(fs.get("discord_nodes", "n1") or {}),
                                    "actual_at": datetime.now(timezone.utc).isoformat()})
     await fs.doc_delete("nodes", "n1")
@@ -401,12 +413,12 @@ async def test_release_seq_never_names_a_list_that_still_holds_the_stream(fs, mo
         out = await real(node_id)
         if not fired:
             fired.append(1)
-            await leases.detach_stream("g1:c1", {"desired": "off"})
+            await leases.detach_stream(_sid(fs, "g1", "c1"), {"desired": "off"})
         return out
     monkeypatch.setattr(streams, "desired_for_node", _racy)
     await streams.publish_desired("n1")
     racy_seq = fs.published[-1][1]["seq"]
-    assert any(d["stream_id"] == "g1:c1" for d in fs.published[-1][1]["streams"])
+    assert any(d["stream_id"] == _sid(fs, "g1", "c1") for d in fs.published[-1][1]["streams"])
     rseq = _lease(fs, "t1")["release_seq"]
     assert rseq is None or rseq > racy_seq
     assert await leases.settle_releases("n1", set(), racy_seq) == []
@@ -417,7 +429,7 @@ async def test_retuned_node_gives_up_a_persistent_stream(fs):
     await _start(mode="persistent")
     _node(fs, "n1", system="sys-2")
     await streams.reconcile_once()
-    s = fs.get("discord_streams", "g1:c1")
+    s = fs.get("discord_streams", _sid(fs, "g1", "c1"))
     assert s["desired"] == "on" and s["token_id"] is None and s["actual"]["state"] == "error"
     assert _lease(fs, "t1")["releasing"] is True
 
@@ -432,7 +444,7 @@ async def test_force_release_detaches_streams(fs):
     old = await leases.force_release_token("t1")
     assert old["node_id"] == "n1"
     assert _lease(fs, "t1") is None
-    assert fs.get("discord_streams", "g1:c1")["token_id"] is None
+    assert fs.get("discord_streams", _sid(fs, "g1", "c1"))["token_id"] is None
 
 
 def test_node_token_pickup_only_for_the_lease_holder(fs):

drb-c2-core/tests/test_discord_multi_stream.py

@@ -0,0 +1,440 @@
+"""
+Several radio streams in ONE Discord voice channel (server-26#213 follow-up,
+owner decision 2026-10-05).
+
+Owner-reported bug on live: adding a radio channel for a server + voice
+channel that already had one — different label, system, even a different
+preferred bot — overwrote the existing stream and made its bot leave, because
+the stream id was ``guild:channel``. Now every new stream gets its own id;
+each stream has its own bot (Discord: one voice connection per bot per
+server), and a chosen bot that is already in the server is refused by name.
+"""
+import asyncio
+import threading
+
+import pytest
+from fastapi.testclient import TestClient
+
+from app.config import settings
+from app.internal import discord_leases as leases
+from app.internal import discord_streams as streams
+from app.internal.auth import require_admin_token, require_service_key_or_admin
+from app.main import app
+from app.routers import discord_web as web
+from tests.fake_firestore import FakeFirestore
+
+ORG = "org-a"
+P25, FIRE = "sys-p25", "sys-fire"
+ADMIN = {"uid": "u-admin", "role": "admin", "email": "admin@example.test"}
+OWNER = {"uid": "u-own", "org_id": ORG, "org_role": "owner"}
+client = TestClient(app)
+
+
+@pytest.fixture
+def fs(monkeypatch):
+    fake = FakeFirestore.install(monkeypatch)
+    fake.published = []
+    from app.internal.mqtt_handler import mqtt_handler
+    monkeypatch.setattr(mqtt_handler, "publish_discord_desired",
+                        lambda n, p: fake.published.append((n, p)) or True)
+    streams._publish_locks.clear()
+    streams._guild_locks.clear()
+
+    async def _no_discord(token_ids):
+        return 0
+    monkeypatch.setattr(leases, "refresh_membership", _no_discord)
+    monkeypatch.setattr(settings, "discord_unlinked_guild_org_id", "")
+    monkeypatch.setattr(settings, "discord_guild_max_streams", 4)
+    monkeypatch.setattr(settings, "discord_node_max_streams", 4)
+
+    # One node per system: a node plays the one system it is tuned to.
+    for nid, system in (("n-p25", P25), ("n-fire", FIRE)):
+        fake.put("nodes", nid, {"node_id": nid, "name": f"Node {nid}", "org_id": ORG, "status": "online",
+                                "approval_status": "approved", "assigned_system_id": system})
+    fake.put("systems", P25, {"system_id": P25, "name": "County P25", "org_id": ORG})
+    fake.put("systems", FIRE, {"system_id": FIRE, "name": "City Fire", "org_id": ORG})
+    fake.put("organizations", ORG, {"org_id": ORG, "name": "Alpha"})
+    fake.put("discord_guilds", "100", {"guild_id": "100", "org_id": ORG, "guild_name": "Fire Co. 7",
+                                       "bot_present": True,
+                                       "voice_channels": [{"channel_id": "1001", "name": "Dispatch"},
+                                                          {"channel_id": "1002", "name": "Fireground"}]})
+    for tid in ("t1", "t2", "t3"):
+        fake.put("bot_tokens", tid, {"token_id": tid, "name": f"Radio {tid}", "token": f"secret-{tid}",
+                                     "org_id": None, "guild_ids": ["100", "200"], "valid": True,
+                                     "application_id": f"app-{tid}", "lease": None})
+    yield fake
+    app.dependency_overrides.clear()
+
+
+def _lease(fs, tid):
+    return (fs.get("bot_tokens", tid) or {}).get("lease")
+
+
+def _live(fs, channel="1001"):
+    return sorted((d for d in (fs.data.get("discord_streams") or {}).values()
+                   if d.get("channel_id") == channel and d.get("desired") == "on"),
+                  key=lambda d: d.get("created_at") or "")
+
+
+def _post(body):
+    app.dependency_overrides[require_service_key_or_admin] = lambda: ADMIN
+    return client.post("/discord/streams", json={"guild_id": "100", "channel_id": "1001", **body})
+
+
+def _legacy(fs, token="t1", node="n-p25", system=P25, channel="1001"):
+    """A stream doc as it was created before this change: id guild:channel."""
+    sid = f"100:{channel}"
+    fs.put("discord_streams", sid, {
+        "stream_id": sid, "org_id": ORG, "guild_id": "100", "channel_id": channel, "channel_name": "Old label",
+        "system_id": system, "system_name": "County P25", "mode": "persistent", "desired": "on",
+        "node_id": node, "token_id": token, "lease_id": "L-old", "actual": {"state": "connected"},
+        "created_at": "2026-10-01T00:00:00+00:00",
+    })
+    fs.put("bot_tokens", token, {**fs.get("bot_tokens", token),
+                                 "lease": {"node_id": node, "lease_id": "L-old", "since": "x",
+                                           "streams": [sid], "releasing": False, "release_seq": None}})
+    return sid
+
+
+# ---------------------------------------------------------------------------
+# The reported bug
+# ---------------------------------------------------------------------------
+
+def test_adding_a_second_stream_to_a_channel_never_touches_the_first(fs):
+    """Same server + voice channel, different label, system and preferred bot:
+    a second row, each with its own bot — the first keeps playing."""
+    a = _post({"system_id": P25, "channel_name": "County", "preferred_token_id": "t1"})
+    assert a.status_code == 200, a.text
+    first = fs.get("discord_streams", a.json()["stream_id"])
+
+    b = _post({"system_id": FIRE, "channel_name": "Fire", "preferred_token_id": "t2"})
+    assert b.status_code == 200, b.text
+    assert b.json()["stream_id"] != a.json()["stream_id"]
+    assert a.json()["stream_id"].startswith("100:1001:") and b.json()["stream_id"].startswith("100:1001:")
+
+    assert fs.get("discord_streams", a.json()["stream_id"]) == first          # byte-for-byte untouched
+    live = _live(fs)
+    assert [(s["system_id"], s["channel_name"], s["token_id"], s["node_id"]) for s in live] == [
+        (P25, "County", "t1", "n-p25"), (FIRE, "Fire", "t2", "n-fire")]
+    assert _lease(fs, "t1")["streams"] == [a.json()["stream_id"]]           # t1 never left
+    assert _lease(fs, "t2")["streams"] == [b.json()["stream_id"]]
+    # Each node is told only about its own stream.
+    last = {nid: p for nid, p in fs.published}
+    assert [s["stream_id"] for s in last["n-p25"]["streams"]] == [a.json()["stream_id"]]
+    assert [s["stream_id"] for s in last["n-fire"]["streams"]] == [b.json()["stream_id"]]
+
+
+def test_any_free_bot_never_reuses_a_bot_already_in_the_server(fs):
+    """Two streams in one channel on ONE node (same token pool): different bots."""
+    fs.put("nodes", "n-fire", {**fs.get("nodes", "n-fire"), "assigned_system_id": P25})
+    fs.put("systems", "sys-3", {"system_id": "sys-3", "name": "Third", "org_id": ORG})
+    a = asyncio.run(streams.start_stream(org_id=ORG, guild_id="100", channel_id="1001", system_id=P25,
+                                         node_id="n-p25"))
+    b = asyncio.run(streams.start_stream(org_id=ORG, guild_id="100", channel_id="1001", system_id="sys-3",
+                                         node_id="n-p25"))
+    assert a["stream_id"] != b["stream_id"] and a["token_id"] != b["token_id"]
+    assert sorted(_lease(fs, a["token_id"])["streams"] + _lease(fs, b["token_id"])["streams"]) == \
+        sorted([a["stream_id"], b["stream_id"]])
+
+
+def test_preferred_bot_already_in_the_server_is_refused_by_name(fs):
+    a = _post({"system_id": P25, "channel_name": "County", "preferred_token_id": "t1"})
+    first = fs.get("discord_streams", a.json()["stream_id"])
+
+    r = _post({"system_id": FIRE, "channel_name": "Fire", "preferred_token_id": "t1"})
+    assert r.status_code == 409
+    body = r.json()
+    assert body["code"] == "preferred_busy" and body["busy_stream_id"] == a.json()["stream_id"]
+    assert "Radio t1" in body["detail"] and "County P25" in body["detail"] and "#County" in body["detail"]
+    # Not a silent swap: nothing moved, no second stream left behind half-made.
+    assert fs.get("discord_streams", a.json()["stream_id"]) == first
+    assert len(_live(fs)) == 1 and _lease(fs, "t1")["streams"] == [a.json()["stream_id"]]
+    assert all(_lease(fs, t) is None for t in ("t2", "t3"))
+
+    # Same bot in ANOTHER channel of the same server: still one voice connection per server.
+    app.dependency_overrides[require_service_key_or_admin] = lambda: ADMIN
+    r2 = client.post("/discord/streams", json={"guild_id": "100", "channel_id": "1002", "system_id": P25,
+                                               "preferred_token_id": "t1"})
+    assert r2.status_code == 409 and r2.json()["code"] == "preferred_busy"
+
+
+def test_preferred_bot_on_another_node_is_refused_not_swapped(fs):
+    # t1 is logged in on n-p25 (serving another server); a FIRE stream needs n-fire.
+    fs.put("bot_tokens", "t1", {**fs.get("bot_tokens", "t1"),
+                                "lease": {"node_id": "n-p25", "lease_id": "L", "streams": ["200:2001:aaaaaaaaaaaa"],
+                                          "releasing": False, "release_seq": None}})
+    r = _post({"system_id": FIRE, "preferred_token_id": "t1"})
+    assert r.status_code == 409 and r.json()["code"] == "preferred_other_node"
+    assert "Node n-p25" in r.json()["detail"]
+    assert _live(fs) == []
+
+
+def test_pool_exhausted_for_the_server_says_to_add_and_invite_a_bot(fs):
+    for tid in ("t2", "t3"):
+        fs.put("bot_tokens", tid, {**fs.get("bot_tokens", tid), "valid": False})
+    assert _post({"system_id": P25}).status_code == 200
+    r = _post({"system_id": FIRE})
+    assert r.status_code == 409 and r.json()["code"] == "pool_exhausted"
+    assert "one voice channel per server" in r.json()["detail"] and "invite it" in r.json()["detail"]
+
+
+def test_same_system_twice_in_a_channel_is_the_same_stream(fs):
+    a = _post({"system_id": P25, "channel_name": "County", "mode": "persistent"}).json()
+    again = _post({"system_id": P25, "channel_name": "Renamed", "mode": "adhoc"})
+    assert again.status_code == 200
+    assert again.json()["stream_id"] == a["stream_id"] and again.json()["already_on"] is True
+    doc = fs.get("discord_streams", a["stream_id"])
+    assert doc["channel_name"] == "County" and doc["mode"] == "persistent"     # a re-add is not an edit
+    assert len(_live(fs)) == 1
+    # ...and naming a DIFFERENT bot for it is a clear conflict, not a swap.
+    r = _post({"system_id": P25, "preferred_token_id": "t3" if a["token_id"] != "t3" else "t2"})
+    assert r.status_code == 409 and r.json()["code"] == "duplicate"
+
+
+def test_stop_by_stream_id_leaves_the_other_stream_in_the_channel_on(fs):
+    a = _post({"system_id": P25}).json()
+    b = _post({"system_id": FIRE}).json()
+    r = client.post("/discord/guilds/100/stop", json={"stream_id": b["stream_id"], "reason": "test"})
+    assert r.status_code == 200 and r.json()["stopped"] == [b["stream_id"]]
+    assert [s["stream_id"] for s in _live(fs)] == [a["stream_id"]]
+    # A stream id from another server matches nothing in this one.
+    assert client.post("/discord/guilds/999/stop", json={"stream_id": a["stream_id"]}).json()["stopped"] == []
+    assert client.post("/discord/guilds/100/stop", json={"stream_id": "100/1001"}).status_code == 422
+    # Channel-wide stop still stops every stream in the channel.
+    b2 = _post({"system_id": FIRE}).json()
+    assert b2["stream_id"] == b["stream_id"]                                 # restarted, not a new doc
+    r = client.post("/discord/guilds/100/stop", json={"channel_id": "1001"})
+    assert sorted(r.json()["stopped"]) == sorted([a["stream_id"], b["stream_id"]])
+
+
+def test_restart_and_edit_by_stream_id(fs):
+    a = _post({"system_id": P25, "channel_name": "County"}).json()
+    b = _post({"system_id": FIRE, "channel_name": "Fire"}).json()
+    client.post("/discord/guilds/100/stop", json={"stream_id": a["stream_id"]})
+    # The admin page's Start button names the stream: it restarts THAT one.
+    r = _post({"system_id": P25, "stream_id": a["stream_id"]})
+    assert r.status_code == 200 and r.json()["stream_id"] == a["stream_id"]
+    assert fs.get("discord_streams", b["stream_id"])["token_id"] == b["token_id"]
+    # Explicit edit to a free bot moves the stream to it (and only it).
+    free = next(t for t in ("t1", "t2", "t3") if t not in (r.json()["token_id"], b["token_id"]))
+    e = _post({"system_id": P25, "stream_id": a["stream_id"], "preferred_token_id": free})
+    assert e.status_code == 200 and e.json()["token_id"] == free
+    assert a["stream_id"] not in (_lease(fs, r.json()["token_id"]) or {}).get("streams", [])
+    # Editing to the bot the OTHER stream in this server uses is refused.
+    e2 = _post({"system_id": P25, "stream_id": a["stream_id"], "preferred_token_id": b["token_id"]})
+    assert e2.status_code == 409 and e2.json()["code"] == "preferred_busy"
+    # A stream id must belong to the server/channel it is posted with.
+    bad = client.post("/discord/streams", json={"guild_id": "100", "channel_id": "1002", "system_id": P25,
+                                                "stream_id": a["stream_id"]})
+    assert bad.status_code == 400
+    missing = _post({"system_id": P25, "stream_id": "100:1001:0123456789ab"})
+    assert missing.status_code == 404
+
+
+async def test_reconciler_replaces_each_stream_in_a_shared_channel_by_id(fs, monkeypatch):
+    monkeypatch.setattr(settings, "discord_lease_offline_release_seconds", 600)
+    fs.put("nodes", "n-p25b", {"node_id": "n-p25b", "name": "Node n-p25b", "org_id": ORG, "status": "online",
+                               "approval_status": "approved", "assigned_system_id": P25})
+    a = await streams.start_stream(org_id=ORG, guild_id="100", channel_id="1001", system_id=P25,
+                                   mode="persistent", node_id="n-p25")
+    b = await streams.start_stream(org_id=ORG, guild_id="100", channel_id="1001", system_id=FIRE,
+                                   mode="persistent")
+    from datetime import datetime, timezone, timedelta
+    old = (datetime.now(timezone.utc) - timedelta(seconds=900)).isoformat()
+    fs.put("nodes", "n-p25", {**fs.get("nodes", "n-p25"), "status": "offline", "last_seen": old})
+    await streams.reconcile_once()
+    moved = fs.get("discord_streams", a["stream_id"])
+    assert moved["desired"] == "on" and moved["node_id"] == "n-p25b" and moved["token_id"]
+    assert moved["token_id"] != fs.get("discord_streams", b["stream_id"])["token_id"]
+    assert fs.get("discord_streams", b["stream_id"])["node_id"] == "n-fire"
+    assert len(_live(fs)) == 2                                               # no third doc
+
+
+# ---------------------------------------------------------------------------
+# Races — the per-server bot rule holds under concurrency
+# ---------------------------------------------------------------------------
+
+def _hold_both_at_commit(fake):
+    barrier = threading.Barrier(2, timeout=5)
+
+    def _hook(attempt):
+        if attempt == 0:
+            barrier.wait()
+    fake.before_commit = _hook
+
+
+async def test_two_streams_one_channel_racing_never_share_a_bot(fs):
+    """Two NEW streams in one server on one node, leased concurrently: the
+    transaction re-reads the pool, so they end up on two different bots."""
+    s1, s2 = leases.new_stream_id("100", "1001"), leases.new_stream_id("100", "1001")
+    _hold_both_at_commit(fs)
+    r1, r2 = await asyncio.gather(
+        leases.lease_for_stream(s1, node_id="n-p25", org_id=ORG),
+        leases.lease_for_stream(s2, node_id="n-p25", org_id=ORG),
+    )
+    assert fs.retries >= 1                                                   # the race really happened
+    assert r1.ok and r2.ok and r1.token_id != r2.token_id
+    for tid in (r1.token_id, r2.token_id):
+        assert len([s for s in _lease(fs, tid)["streams"] if leases.guild_of(s) == "100"]) == 1
+
+
+async def test_racing_for_the_last_free_bot_in_a_server_one_loses_cleanly(fs):
+    fs.put("bot_tokens", "t3", {**fs.get("bot_tokens", "t3"), "valid": False})
+    fs.put("bot_tokens", "t2", {**fs.get("bot_tokens", "t2"),
+                                "lease": {"node_id": "n-p25", "lease_id": "L2", "streams": ["100:1002:bbbbbbbbbbbb"],
+                                          "releasing": False, "release_seq": None}})
+    s1, s2 = leases.new_stream_id("100", "1001"), leases.new_stream_id("100", "1001")
+    _hold_both_at_commit(fs)
+    r1, r2 = await asyncio.gather(
+        leases.lease_for_stream(s1, node_id="n-p25", org_id=ORG),
+        leases.lease_for_stream(s2, node_id="n-p25", org_id=ORG),
+    )
+    assert sorted([r1.ok, r2.ok]) == [False, True]
+    loser = r2 if r1.ok else r1
+    assert loser.error == "pool_exhausted"
+    assert len(_lease(fs, "t1")["streams"]) == 1
+
+
+async def test_racing_same_preferred_bot_for_two_streams_in_one_server(fs):
+    s1, s2 = leases.new_stream_id("100", "1001"), leases.new_stream_id("100", "1002")
+    _hold_both_at_commit(fs)
+    r1, r2 = await asyncio.gather(
+        leases.lease_for_stream(s1, node_id="n-p25", org_id=ORG, preferred_token_id="t1"),
+        leases.lease_for_stream(s2, node_id="n-p25", org_id=ORG, preferred_token_id="t1"),
+    )
+    assert sorted([r1.ok, r2.ok]) == [False, True]
+    loser = r2 if r1.ok else r1
+    assert loser.error == "preferred_busy" and loser.detail["busy_stream_id"] in (s1, s2)
+    assert len(_lease(fs, "t1")["streams"]) == 1
+
+
+def _slow_find(monkeypatch):
+    """Yield to the other requests right after every stream-list read (the
+    "already on?" and "server at its cap?" checks), so unserialised starts
+    would all read before any of them writes — as real Firestore reads allow."""
+    from app.internal import firestore as fstore
+    real = fstore.collection_list
+
+    async def _list(coll, **filters):
+        out = await real(coll, **filters)
+        if coll == streams.STREAMS:
+            await asyncio.sleep(0.05)
+        return out
+    monkeypatch.setattr(fstore, "collection_list", _list)
+
+
+async def test_concurrent_adds_of_one_system_to_one_channel_make_one_stream(fs, monkeypatch):
+    _slow_find(monkeypatch)
+    r1, r2 = await asyncio.gather(
+        streams.start_stream(org_id=ORG, guild_id="100", channel_id="1001", system_id=P25),
+        streams.start_stream(org_id=ORG, guild_id="100", channel_id="1001", system_id=P25),
+    )
+    assert r1["stream_id"] == r2["stream_id"]
+    assert len(_live(fs)) == 1
+    assert len([t for t in ("t1", "t2", "t3") if _lease(fs, t)]) == 1
+
+
+async def test_concurrent_adds_respect_the_server_cap(fs, monkeypatch):
+    _slow_find(monkeypatch)
+    fs.put("systems", "sys-3", {"system_id": "sys-3", "name": "Third", "org_id": ORG})
+    fs.put("nodes", "n-3", {"node_id": "n-3", "name": "Node n-3", "org_id": ORG, "status": "online",
+                            "approval_status": "approved", "assigned_system_id": "sys-3"})
+    results = await asyncio.gather(*(
+        streams.start_stream(org_id=ORG, guild_id="100", channel_id="1001", system_id=sid, max_streams=2)
+        for sid in (P25, FIRE, "sys-3")), return_exceptions=True)
+    errors = [r for r in results if isinstance(r, streams.StreamError)]
+    assert len(errors) == 1 and errors[0].code == "limit"
+    assert len(_live(fs)) == 2
+
+
+# ---------------------------------------------------------------------------
+# Old guild:channel ids keep working
+# ---------------------------------------------------------------------------
+
+def test_old_id_stop_and_admin_remove(fs):
+    sid = _legacy(fs)
+    app.dependency_overrides[require_service_key_or_admin] = lambda: ADMIN
+    app.dependency_overrides[require_admin_token] = lambda: ADMIN
+    assert client.get(f"/discord/streams/{sid}").json()["stream_id"] == sid
+    r = client.post("/discord/guilds/100/stop", json={"stream_id": sid})
+    assert r.json()["stopped"] == [sid]
+    assert _lease(fs, "t1")["releasing"] is True and _lease(fs, "t1")["streams"] == []
+    r = client.delete(f"/discord/streams/{sid}")
+    assert r.status_code == 200 and fs.get("discord_streams", sid) is None
+
+
+def test_old_id_doc_is_not_overwritten_by_an_add_and_is_restarted_by_its_own_system(fs):
+    sid = _legacy(fs)
+    b = _post({"system_id": FIRE, "channel_name": "Fire"}).json()
+    assert b["stream_id"] != sid and b["token_id"] != "t1"
+    old = fs.get("discord_streams", sid)
+    assert old["desired"] == "on" and old["token_id"] == "t1" and old["channel_name"] == "Old label"
+    # Re-adding the old stream's own system lands on the old doc (idempotent).
+    again = _post({"system_id": P25}).json()
+    assert again["stream_id"] == sid and again["already_on"] is True
+
+
+async def test_old_id_node_page_stop_and_remove(fs):
+    sid = _legacy(fs)
+    r = await web.stop_node_streams("n-p25", web.NodeStreamStop(stream_id=sid), OWNER)
+    assert r["stopped"] == [sid]
+    r = await web.remove_node_stream("n-p25", sid, OWNER)
+    assert r["ok"] and fs.get("discord_streams", sid) is None
+
+
+async def test_old_id_still_reported_on_by_the_node(fs):
+    sid = _legacy(fs)
+    await streams.handle_actual("n-p25", {"streams": {sid: {"state": "error", "error": "x", "code": "no_audio"},
+                                                      "100/evil": {"state": "connected"}}})
+    assert fs.get("discord_streams", sid)["actual"]["code"] == "no_audio"
+
+
+# ---------------------------------------------------------------------------
+# Node page + admin page flows
+# ---------------------------------------------------------------------------
+
+async def test_node_page_adds_beside_another_systems_stream(fs):
+    p25 = await web.start_node_stream("n-p25", web.NodeStreamStart(guild_id="100", channel_id="1001"), OWNER)
+    fire = await web.start_node_stream("n-fire", web.NodeStreamStart(guild_id="100", channel_id="1001"), OWNER)
+    assert p25["stream_id"] != fire["stream_id"] and p25["token_id"] != fire["token_id"]
+    assert fs.get("discord_streams", p25["stream_id"])["node_id"] == "n-p25"      # not moved
+
+    opts = await web.join_options("n-fire", OWNER)
+    ch = next(c for c in opts["guilds"][0]["voice_channels"] if c["channel_id"] == "1001")
+    assert [s["system_name"] for s in ch["streams"]] == ["County P25", "City Fire"]
+    assert {s["token_name"] for s in ch["streams"]} == {f"Radio {p25['token_id']}", f"Radio {fire['token_id']}"}
+    assert ch["stream"]["stream_id"] == fire["stream_id"]                        # this node's system's one
+    assert [s["stream_id"] for s in opts["streams"]] == [fire["stream_id"]]
+    assert opts["guilds"][0]["streams_on"] == 2
+
+    # Re-picking on the same node is idempotent (and free under the node cap).
+    again = await web.start_node_stream("n-fire", web.NodeStreamStart(guild_id="100", channel_id="1001"), OWNER)
+    assert again["stream_id"] == fire["stream_id"]
+    # Removing one leaves the other.
+    await web.remove_node_stream("n-fire", fire["stream_id"], OWNER)
+    assert [s["stream_id"] for s in _live(fs)] == [p25["stream_id"]]
+
+
+def test_admin_overview_lists_one_row_per_stream_with_its_bot(fs):
+    app.dependency_overrides[require_admin_token] = lambda: ADMIN
+    a = _post({"system_id": P25, "channel_name": "County"}).json()
+    b = _post({"system_id": FIRE, "channel_name": "Fire"}).json()
+    body = client.get("/discord/admin/overview").json()
+    rows = [(r["stream_id"], r["channel_id"], r["system_name"], r["token_name"]) for r in body["streams"]]
+    assert rows == [(a["stream_id"], "1001", "County P25", f"Radio {a['token_id']}"),
+                    (b["stream_id"], "1001", "City Fire", f"Radio {b['token_id']}")]
+    assert body["guilds"][0]["live_streams"] == 2
+    # Remove one by id: the other row stays and stays on.
+    assert client.delete(f"/discord/streams/{a['stream_id']}").status_code == 200
+    body = client.get("/discord/admin/overview").json()
+    assert [(r["stream_id"], r["desired"]) for r in body["streams"]] == [(b["stream_id"], "on")]
+
+
+def test_bot_overview_names_each_streams_bot(fs):
+    a = _post({"system_id": P25}).json()
+    b = _post({"system_id": FIRE}).json()
+    body = client.get("/discord/guilds/100/overview").json()
+    assert {(s["stream_id"], s["token_name"]) for s in body["streams"]} == {
+        (a["stream_id"], f"Radio {a['token_id']}"), (b["stream_id"], f"Radio {b['token_id']}")}
+    busy = {x["token_id"]: x["busy_here"] for x in body["bots"]}
+    assert busy[a["token_id"]] and busy[b["token_id"]]

drb-c2-core/tests/test_discord_remove_and_logging.py

@@ -53,6 +53,7 @@ def fs(monkeypatch):
     from app.internal.mqtt_handler import mqtt_handler
     monkeypatch.setattr(mqtt_handler, "publish_discord_desired", _publish)
     streams._publish_locks.clear()
+    streams._guild_locks.clear()
 
     async def _no_discord(token_ids):
         return 0
@@ -301,18 +302,18 @@ def test_remove_guild_interrupted_by_a_restart_is_still_audited(fs, monkeypatch)
 
 def test_remove_guild_route_hides_server_and_relink_restores_it(fs):
     import asyncio
-    asyncio.run(_start("1001"))
+    sid = asyncio.run(_start("1001"))["stream_id"]
     app.dependency_overrides[require_admin_token] = lambda: ADMIN
     assert client.delete("/discord/guilds/abc").status_code == 400
     assert client.delete("/discord/guilds/999").status_code == 404
 
     r = client.delete("/discord/guilds/100")
-    assert r.status_code == 200 and r.json()["streams_removed"] == ["100:1001"]
+    assert r.status_code == 200 and r.json()["streams_removed"] == [sid]
     assert r.json()["radio_bots"] == []                     # not asked to leave
     body = client.get("/discord/admin/overview").json()
     assert body["guilds"] == [] and body["streams"] == []
     (entry,) = _audit(fs, "discord.guild.remove")
-    assert entry["details"]["previous_org_id"] == ORG and entry["details"]["streams_removed"] == ["100:1001"]
+    assert entry["details"]["previous_org_id"] == ORG and entry["details"]["streams_removed"] == [sid]
 
     # Admin link brings it back.
     assert client.put("/discord/guilds/100/link", json={"org_id": ORG}).status_code == 200

drb-c2-core/tests/test_discord_web_picker.py

@@ -39,6 +39,7 @@ def fs(monkeypatch):
     from app.internal.mqtt_handler import mqtt_handler
     monkeypatch.setattr(mqtt_handler, "publish_discord_desired", lambda n, p: published.append((n, p)) or True)
     streams._publish_locks.clear()
+    streams._guild_locks.clear()
 
     async def _no_discord(token_ids):
         return 0
@@ -149,14 +150,14 @@ def test_options_reports_missing_bot_with_invite(fs):
 
 def test_start_pins_stream_to_node_with_names(fs):
     r = run(web.start_node_stream("n1", web.NodeStreamStart(guild_id="100", channel_id="1001"), OWNER))
-    assert r["ok"] and r["stream_id"] == "100:1001" and r["token_id"] == "t1"
+    assert r["ok"] and r["stream_id"].startswith("100:1001:") and r["token_id"] == "t1"
-    doc = fs.get("discord_streams", "100:1001")
+    doc = fs.get("discord_streams", r["stream_id"])
     assert doc["node_id"] == "n1" and doc["org_id"] == ORG and doc["system_id"] == SYS
     assert doc["channel_name"] == "Dispatch" and doc["guild_name"] == "Fire Co. 7"
     assert doc["created_by_discord_id"] == "web:u-own"
     assert fs.published and fs.published[-1][0] == "n1"
     opts = run(web.join_options("n1", OWNER))
-    assert [s["stream_id"] for s in opts["streams"]] == ["100:1001"]
+    assert [s["stream_id"] for s in opts["streams"]] == [r["stream_id"]]
     assert opts["guilds"][0]["voice_channels"][0]["stream"]["node_name"] == "Node n1"
 
 
@@ -191,13 +192,13 @@ def test_start_lease_refusal_carries_invite(fs):
 
 
 def test_stop_one_and_scoped_to_node(fs):
-    run(web.start_node_stream("n1", web.NodeStreamStart(guild_id="100", channel_id="1001"), OWNER))
+    sid = run(web.start_node_stream("n1", web.NodeStreamStart(guild_id="100", channel_id="1001"), OWNER))["stream_id"]
     with pytest.raises(HTTPException) as e:
-        run(web.stop_node_streams("n2", web.NodeStreamStop(stream_id="100:1001"), ADMIN))
+        run(web.stop_node_streams("n2", web.NodeStreamStop(stream_id=sid), ADMIN))
     assert e.value.status_code == 404
-    r = run(web.stop_node_streams("n1", web.NodeStreamStop(stream_id="100:1001"), OWNER))
+    r = run(web.stop_node_streams("n1", web.NodeStreamStop(stream_id=sid), OWNER))
-    assert r["stopped"] == ["100:1001"]
+    assert r["stopped"] == [sid]
-    assert fs.get("discord_streams", "100:1001")["desired"] == "off"
+    assert fs.get("discord_streams", sid)["desired"] == "off"
 
 
 def test_legacy_raw_id_join_refuses_other_orgs_server(fs):
@@ -210,8 +211,10 @@ def test_legacy_raw_id_join_refuses_other_orgs_server(fs):
 def test_stop_rejects_malformed_stream_id():
     # "a/b" would otherwise reach Firestore as a document path and 500.
     from pydantic import ValidationError
-    for bad in ("100/1001", "../x", "100:1001/extra", ""):
+    for bad in ("100/1001", "../x", "100:1001/extra", "", "100:1001:ab/cd", "100:1001:XYZ", "100:1001:"):
         with pytest.raises(ValidationError):
             web.NodeStreamStop(stream_id=bad)
-    assert web.NodeStreamStop(stream_id="100:1001").stream_id == "100:1001"
+    assert web.NodeStreamStop(stream_id="100:1001").stream_id == "100:1001"            # legacy id
+    new = leases.new_stream_id("100", "1001")
+    assert web.NodeStreamStop(stream_id=new).stream_id == new                         # per-stream id
     assert web.NodeStreamStop().stream_id is None

drb-frontend/components/NodeDiscordRadio.tsx

@@ -5,6 +5,10 @@
 // the Discord servers linked to this node's org and the voice channels the
 // DRB bot reported in them; a pick becomes a radio channel pinned to this node
 // (desired state — the node reconciles and reports back, shown per row).
+// A voice channel can carry several radio streams, each its own bot playing
+// its own system (server-26#213 follow-up): picking a channel that already
+// plays another system ADDS this node's system beside it; only a stream of
+// this node's own system there is restarted / moved here.
 // Layout follows design-canvas/discord-system-mockup.html "Radio channels".
 
 import { useCallback, useEffect, useMemo, useState } from "react";
@@ -64,6 +68,11 @@ function modeLabel(mode: DiscordStreamView["mode"]) {
   return mode === "persistent" ? "persistent" : "ad-hoc";
 }
 
+function streamLabel(s: DiscordStreamView) {
+  const sys = s.system_name ?? "radio";
+  return s.token_name ? `${sys} (${s.token_name})` : sys;
+}
+
 // ---------------------------------------------------------------------------
 // Picker
 // ---------------------------------------------------------------------------
@@ -199,6 +208,7 @@ function DiscordPicker({
                       <div className="space-y-1">
                         {grp.channels.map((c) => {
                           const here = c.stream?.node_id === nodeId;
+                          const others = (c.streams ?? []).filter((x) => x.stream_id !== c.stream?.stream_id);
                           return (
                             <label
                               key={c.channel_id}
@@ -221,8 +231,14 @@ function DiscordPicker({
                                 {c.stream && (
                                   <span className="text-xs text-ink-muted block">
                                     {here
-                                      ? "Already on this node"
+                                      ? `${options.system_name ?? "This system"} is already on this node here`
-                                      : `Carried by ${c.stream.node_name ?? "another node"} — picking it moves it to this node`}
+                                      : `${options.system_name ?? "This system"} is carried by ${c.stream.node_name ?? "another node"} — picking it moves it to this node`}
+                                  </span>
+                                )}
+                                {others.length > 0 && (
+                                  <span className="text-xs text-ink-muted block">
+                                    Also playing: {others.map(streamLabel).join(", ")}
+                                    {!c.stream && ` — this adds ${options.system_name ?? "this node's system"} beside it with another bot`}
                                   </span>
                                 )}
                               </span>
@@ -356,7 +372,7 @@ export function NodeDiscordRadio({ nodeId }: { nodeId: string }) {
 
   async function remove(s: DiscordStreamView) {
     const name = s.channel_name ?? "this channel";
-    if (!confirm(`Remove radio from ${name} permanently? It is stopped first (the node logs its radio bot out), then deleted from DRB.`)) return;
+    if (!confirm(`Remove ${streamLabel(s)} from ${name} permanently? It is stopped first (the node logs its radio bot out), then deleted from DRB. Other streams in that voice channel keep playing.`)) return;
     setBusy(`rm:${s.stream_id}`);
     setActionError(null);
     try {
@@ -429,7 +445,9 @@ export function NodeDiscordRadio({ nodeId }: { nodeId: string }) {
                 <tr key={s.stream_id} className="border-b border-line last:border-0 align-top">
                   <td className="px-4 py-2.5">
                     <div className="text-ink font-medium">🔊 {s.channel_name ?? s.channel_id}</div>
-                    <div className="text-xs text-ink-muted">{s.guild_name ?? s.guild_id}</div>
+                    <div className="text-xs text-ink-muted">
+                      {s.guild_name ?? s.guild_id} · {streamLabel(s)}
+                    </div>
                   </td>
                   <td className="px-4 py-2.5 text-ink-2">{modeLabel(s.mode)}</td>
                   <td className="px-4 py-2.5">

drb-frontend/components/discord/StreamsPanel.tsx

@@ -20,6 +20,11 @@ import { DIM, Field, INPUT, InviteLinks, Panel, Status, Table, TD } from "./ui";
 // POST /discord/guilds/{id}/stop), which accept a platform-admin token.
 // Remove is DELETE /discord/streams/{id} (admin): c2-core stops the channel
 // through desired state first, then deletes it.
+//
+// One row per STREAM, not per voice channel: a channel can carry several,
+// each with its own radio bot and system (server-26#213 follow-up). Stop,
+// Start and Remove all act on the row's stream_id, so the other streams in
+// the same channel keep playing. Add never replaces an existing stream.
 
 type Invite = { name: string; invite_url: string | null }[];
 
@@ -48,6 +53,14 @@ function AddChannelForm({
   const org = guilds.find((g) => g.guild_id === guildId)?.acting_org_id ?? null;
   const systems = overview.systems.filter((s) => s.org_id == null || s.org_id === org);
   const usableBots = bots.filter((b) => b.valid !== false && (b.org_id == null || b.org_id === org));
+  // A bot can be in one voice channel per server: name the stream holding it.
+  const busyHere = new Map<string, DiscordStream>();
+  for (const s of overview.streams) {
+    if (s.desired === "on" && s.guild_id === guildId && s.token_id) busyHere.set(s.token_id, s);
+  }
+  const inChannel = overview.streams.filter(
+    (s) => s.desired === "on" && s.guild_id === guildId && s.channel_id === channelId.trim(),
+  );
 
   async function submit(e: React.FormEvent) {
     e.preventDefault();
@@ -58,7 +71,7 @@ function AddChannelForm({
     setSaving(true);
     setError(null);
     try {
-      await c2api.startDiscordStream({
+      const res = await c2api.startDiscordStream({
         guild_id: guildId,
         channel_id: channelId.trim(),
         system_id: systemId,
@@ -66,6 +79,16 @@ function AddChannelForm({
         channel_name: channelName.trim() || undefined,
         preferred_token_id: botId || undefined,
       });
+      if (res?.already_on) {
+        // Same system re-added to a channel that already plays it: c2-core
+        // returns that stream untouched (label, mode and bot kept). Say so
+        // instead of closing as if a stream had been added.
+        setError({
+          message: `${res.system_name || "That system"} is already playing in this voice channel, so nothing was added or changed. Pick another system to add a second stream, or Remove the existing row to change its label or bot.`,
+          invite: [],
+        });
+        return;
+      }
       onDone();
     } catch (err) {
       setError(parseC2Error(err));
@@ -117,11 +140,25 @@ function AddChannelForm({
       <Field label="Radio bot">
         <select value={botId} onChange={(e) => setBotId(e.target.value)} className={INPUT}>
           <option value="">Any free bot in this server</option>
-          {usableBots.map((b) => (
+          {usableBots.map((b) => {
-            <option key={b.token_id} value={b.token_id}>Prefer {b.name}</option>
+            const busy = busyHere.get(b.token_id);
-          ))}
+            return (
+              <option key={b.token_id} value={b.token_id} disabled={!!busy}>
+                {busy
+                  ? `${b.name} — in use here (${busy.system_name || busy.system_id || "radio"}${busy.channel_name ? ` in ${busy.channel_name}` : ""})`
+                  : `Use ${b.name}`}
+              </option>
+            );
+          })}
         </select>
       </Field>
+      {inChannel.length > 0 && (
+        <div className="sm:col-span-3 text-xs text-ink-muted">
+          This voice channel already plays{" "}
+          {inChannel.map((s) => `${s.system_name || s.system_id}${s.token_name ? ` (${s.token_name})` : ""}`).join(", ")}.
+          A new system is added beside it with another bot; nothing that is playing is replaced.
+        </div>
+      )}
       {error && (
         <div className="sm:col-span-3 text-xs text-sev-major">
           {error.message}
@@ -154,6 +191,15 @@ export function StreamsPanel({
   const [busy, setBusy] = useState<string | null>(null);
   const [rowError, setRowError] = useState<{ id: string; message: string; invite: Invite } | null>(null);
   const guildsById = useMemo(() => new Map(overview.guilds.map((g) => [g.guild_id, g])), [overview.guilds]);
+  // Streams sharing a voice channel (c2-core sorts them next to each other).
+  const perChannel = useMemo(() => {
+    const m = new Map<string, number>();
+    for (const s of overview.streams) {
+      const k = `${s.guild_id}:${s.channel_id}`;
+      m.set(k, (m.get(k) ?? 0) + 1);
+    }
+    return m;
+  }, [overview.streams]);
 
   async function act(s: DiscordStream, fn: () => Promise<unknown>) {
     setBusy(s.stream_id);
@@ -171,7 +217,7 @@ export function StreamsPanel({
   return (
     <Panel
       title="Radio channels"
-      description="One row per Discord voice channel. Persistent channels come back by themselves."
+      description="One row per radio stream — a voice channel can carry several, each with its own bot and system. Persistent streams come back by themselves."
       action={
         <Button size="sm" onClick={() => setAdding((v) => !v)}>+ Add radio channel</Button>
       }
@@ -205,11 +251,19 @@ export function StreamsPanel({
             const g = guildsById.get(s.guild_id);
             const err = rowError?.id === s.stream_id ? rowError : null;
             const isOn = s.desired === "on";
+            const shared = (perChannel.get(`${s.guild_id}:${s.channel_id}`) ?? 1) - 1;
+            const name = s.channel_name || "this channel";
+            const what = `${s.system_name || s.system_id || "radio"} in ${name}`;
             return (
               <tr key={s.stream_id}>
                 <td className={TD}>
                   <b>{s.channel_name ? `🔊 ${s.channel_name}` : `channel ${shortId(s.channel_id)}`}</b>
                   <div className={DIM}>{g ? guildLabel(g) : guildLabel({ guild_id: s.guild_id, guild_name: s.guild_name })}</div>
+                  {shared > 0 && (
+                    <div className={DIM}>
+                      + {shared} other stream{shared === 1 ? "" : "s"} in this voice channel
+                    </div>
+                  )}
                 </td>
                 <td className={TD}>{source(s)}</td>
                 <td className={TD}><span className={DIM}>All talkgroups</span></td>
@@ -238,8 +292,8 @@ export function StreamsPanel({
                       variant="secondary"
                       disabled={busy !== null}
                       onClick={() => {
-                        if (!confirm(`Stop radio in ${s.channel_name || "this channel"}?`)) return;
+                        if (!confirm(`Stop ${what}?`)) return;
-                        act(s, () => c2api.stopDiscordStream(s.guild_id, s.channel_id));
+                        act(s, () => c2api.stopDiscordStream(s.guild_id, s.stream_id));
                       }}
                     >
                       {busy === s.stream_id ? "Stopping…" : "Stop"}
@@ -256,6 +310,7 @@ export function StreamsPanel({
                             channel_id: s.channel_id,
                             system_id: s.system_id as string,
                             mode: s.mode ?? "persistent",
+                            stream_id: s.stream_id,
                           }),
                         )
                       }
@@ -269,10 +324,9 @@ export function StreamsPanel({
                     className="ml-1"
                     disabled={busy !== null}
                     onClick={() => {
-                      const name = s.channel_name || "this channel";
                       const msg = isOn
-                        ? `Remove radio from ${name} permanently? It is stopped first (the node logs its radio bot out), then deleted from DRB.`
+                        ? `Remove ${what} permanently? It is stopped first (the node logs its radio bot out), then deleted from DRB.${shared > 0 ? " The other streams in this voice channel keep playing." : ""}`
-                        : `Remove ${name} from DRB permanently?`;
+                        : `Remove ${what} from DRB permanently?`;
                       if (!confirm(msg)) return;
                       act(s, () => c2api.deleteDiscordStream(s.stream_id));
                     }}

drb-frontend/lib/c2api.ts

@@ -76,6 +76,9 @@ export const c2api = {
       method: "PUT",
       body: JSON.stringify(body),
     }),
+  /** Without stream_id: ADD a stream to the voice channel (a channel can carry
+   *  several, each its own bot + system; never replaces one). With stream_id:
+   *  restart / edit that stream. */
   startDiscordStream: (body: {
     guild_id: string;
     channel_id: string;
@@ -83,6 +86,7 @@ export const c2api = {
     mode: "persistent" | "adhoc";
     channel_name?: string;
     preferred_token_id?: string;
+    stream_id?: string;
   }) => request<DiscordStream>("/discord/streams", { method: "POST", body: JSON.stringify(body) }),
   /** Permanently remove a radio channel (stopped through desired state first). Admin only. */
   deleteDiscordStream: (streamId: string) =>
@@ -99,10 +103,11 @@ export const c2api = {
     }>(`/discord/guilds/${encodeURIComponent(guildId)}?leave_radio_bots=${leaveRadioBots ? "true" : "false"}`, {
       method: "DELETE",
     }),
-  stopDiscordStream: (guildId: string, channelId: string) =>
+  /** Stop ONE stream (its voice channel may carry others, which keep playing). */
+  stopDiscordStream: (guildId: string, streamId: string) =>
     request<{ stopped: string[] }>(`/discord/guilds/${encodeURIComponent(guildId)}/stop`, {
       method: "POST",
-      body: JSON.stringify({ channel_id: channelId, reason: "stopped from web admin" }),
+      body: JSON.stringify({ stream_id: streamId, reason: "stopped from web admin" }),
     }),
 
   // Node approval

drb-frontend/lib/discord.ts

@@ -67,6 +67,8 @@ export interface DiscordStream {
   stop_reason?: string | null;
   updated_at?: string | null;
   created_at?: string | null;
+  /** POST /discord/streams: this system was already on in that channel — nothing was replaced. */
+  already_on?: boolean;
 }
 
 export interface DiscordGuild {

drb-frontend/lib/discordPicker.ts

@@ -18,7 +18,10 @@ export interface DiscordStreamView {
   channel_id: string;
   guild_name: string | null;
   channel_name: string | null;
+  system_id?: string | null;
   system_name: string | null;
+  /** The radio bot carrying it — tells apart several streams in one voice channel. */
+  token_name?: string | null;
   mode: "persistent" | "adhoc" | null;
   node_id: string | null;
   node_name: string | null;
@@ -35,7 +38,10 @@ export interface DiscordVoiceChannel {
   channel_id: string;
   name: string;
   category: string | null;
+  /** This node's system's stream in the channel, if any — what a pick restarts or moves. */
   stream: DiscordStreamView | null;
+  /** Every live stream in the channel (several systems can share it, each its own bot). */
+  streams?: DiscordStreamView[];
 }
 
 export interface DiscordGuildOption {

drb-server-discord-bot/app/commands/radio.py

@@ -23,6 +23,13 @@ STATE_TEXT = {
 }
 
 
+def _stream_name(s: dict) -> str:
+    """How one radio stream is named to a Discord user. Several streams can
+    share a voice channel, so the system and the bot are what tell them apart."""
+    bot = f" ({s['token_name']})" if s.get("token_name") else ""
+    return f"{s.get('system_name') or s.get('system_id') or 'radio'}{bot}"
+
+
 def _refusal_text(body: dict) -> str:
     text = body.get("detail") or "DRB could not start radio in that channel."
     if isinstance(text, dict):  # FastAPI HTTPException detail
@@ -40,6 +47,15 @@ class RadioCommands(commands.Cog):
     server is linked to and never returns or touches anything outside it.
     /join and /leave default to members with Manage Channels — a server admin
     can hand them to a DJ role under Server Settings > Integrations.
+
+    A voice channel can carry SEVERAL radio streams, each its own bot playing
+    its own system (server-26#213 follow-up). So:
+      * /join adds a stream for the chosen system; it never replaces another
+        system's stream in the channel. The same system again is a no-op.
+      * /leave ``stream:`` stops exactly one (autocomplete lists this server's).
+      * /leave ``channel:`` stops that channel's stream; if it has several it
+        lists them and asks which (``stream:``) — or ``all:True`` for every one.
+      * /leave with nothing stops every stream in this server (unchanged).
     """
 
     def __init__(self, bot: commands.Bot):
@@ -91,6 +107,21 @@ class RadioCommands(commands.Cog):
             out.append(app_commands.Choice(name=f"{name} ({label})", value=b["token_id"]))
         return out[:25]
 
+    async def stream_autocomplete(
+        self, interaction: discord.Interaction, current: str
+    ) -> list[app_commands.Choice[str]]:
+        if not interaction.guild_id:
+            return []
+        data = await self._guild(interaction.guild_id)
+        out = []
+        for s in data.get("streams", []):
+            ch = interaction.guild.get_channel(int(s["channel_id"])) if interaction.guild else None
+            where = f"#{ch.name}" if ch else (s.get("channel_name") or f"channel {s['channel_id']}")
+            name = f"{where} — {_stream_name(s)}"
+            if current.lower() in name.lower():
+                out.append(app_commands.Choice(name=name[:100], value=s["stream_id"]))
+        return out[:25]
+
     # ------------------------------------------------------------------
     # /join
     # ------------------------------------------------------------------
@@ -131,6 +162,8 @@ class RadioCommands(commands.Cog):
 
         system_name = body.get("system_name") or system
         head = f"**{system_name}** → {channel.mention}"
+        if body.get("already_on"):
+            head += "\nThat system is already playing in this channel — nothing was replaced."
         if body.get("delivered") is False:
             head += "\nDRB's message broker is down; the node will get this as soon as it is back."
         msg = await interaction.followup.send(f"{head}\n{STATE_TEXT['pending']}", ephemeral=True, wait=True)
@@ -169,22 +202,42 @@ class RadioCommands(commands.Cog):
     # /leave — this server only
     # ------------------------------------------------------------------
 
-    @app_commands.command(name="leave", description="Stop radio in this server (or in one channel).")
+    @app_commands.command(name="leave", description="Stop radio in this server, one channel, or one stream.")
     @app_commands.guild_only()
     @app_commands.default_permissions(manage_channels=True)
-    @app_commands.describe(channel="Only stop radio in this voice channel.")
+    @app_commands.describe(
+        channel="Only stop radio in this voice channel.",
+        stream="Stop just this radio stream (a channel can carry several).",
+        every="With channel: stop every stream in it without asking.",
+    )
+    @app_commands.rename(every="all")
+    @app_commands.autocomplete(stream=stream_autocomplete)
     async def leave(self, interaction: discord.Interaction,
-                    channel: Optional[discord.VoiceChannel] = None):
+                    channel: Optional[discord.VoiceChannel] = None,
+                    stream: Optional[str] = None,
+                    every: bool = False):
         await interaction.response.defer(ephemeral=True)
-        stopped = await c2.stop_guild(interaction.guild_id, str(channel.id) if channel else None)
+        if channel and not stream and not every:
+            # Several streams in one channel: say which, never guess.
+            data = await self._guild(interaction.guild_id, fresh=True)
+            here = [s for s in data.get("streams", []) if str(s.get("channel_id")) == str(channel.id)]
+            if len(here) > 1:
+                names = "\n".join(f"- {_stream_name(s)}" for s in here)
+                await interaction.followup.send(
+                    f"{channel.mention} has {len(here)} radio streams:\n{names}\n"
+                    "Run `/leave stream:` and pick one, or `/leave channel:` with `all:True` to stop them all.",
+                    ephemeral=True)
+                return
+        stopped = await c2.stop_guild(interaction.guild_id, str(channel.id) if channel and not stream else None,
+                                      stream_id=stream)
         self._overview.pop(interaction.guild_id, None)
         if stopped is None:
             await interaction.followup.send("Could not reach DRB. Try again in a minute.", ephemeral=True)
         elif not stopped:
-            where = channel.mention if channel else "this server"
+            where = "that stream" if stream else (channel.mention if channel else "this server")
             await interaction.followup.send(f"No radio is playing in {where}.", ephemeral=True)
         else:
-            await interaction.followup.send(f"Stopped radio in {len(stopped)} channel(s).", ephemeral=True)
+            await interaction.followup.send(f"Stopped {len(stopped)} radio stream(s).", ephemeral=True)
 
     # ------------------------------------------------------------------
     # /status — this server's org only
@@ -206,7 +259,7 @@ class RadioCommands(commands.Cog):
             for s in data.get("streams", []):
                 actual = (s.get("actual") or {})
                 state = actual.get("state") or "pending"
-                lines.append(f"<#{s['channel_id']}> — {s.get('system_name') or s.get('system_id')} · `{state}`"
+                lines.append(f"<#{s['channel_id']}> — {_stream_name(s)} · `{state}`"
                              + (" · persistent" if s.get("mode") == "persistent" else ""))
             embed.add_field(name="Radio channels", value="\n".join(lines) or "None playing.", inline=False)
             dot = {"online": "🟢", "recording": "🔴", "offline": "⚫", "unconfigured": "🟡"}
@@ -239,14 +292,17 @@ class RadioCommands(commands.Cog):
             name="/join `system` [`bot`] [`persistent`]",
             value=(
                 "Put a radio system on your current voice channel. A radio bot joins and plays it.\n"
+                "Run it again with another system to add a second stream (its own bot) to the same channel — "
+                "it never replaces one that is playing. A bot can be in one voice channel per server.\n"
                 "`persistent` keeps the channel on radio until `/leave`, reconnecting by itself.\n"
                 "Needs Manage Channels unless a server admin grants it to another role."
             ),
             inline=False,
         )
         embed.add_field(
-            name="/leave [`channel`]",
+            name="/leave [`channel`] [`stream`] [`all`]",
-            value="Stop radio in this server, or only in one voice channel. Never affects other servers.",
+            value=("Stop radio in this server, one voice channel, or one `stream`. If a channel has several "
+                   "streams, `/leave channel:` asks which (or add `all:True`). Never affects other servers."),
             inline=False,
         )
         embed.add_field(

drb-server-discord-bot/app/internal/c2_client.py

@@ -67,12 +67,14 @@ class C2Client:
             return None
 
     async def stop_guild(self, guild_id: int | str, channel_id: Optional[str] = None,
-                         reason: str = "leave") -> Optional[list]:
+                         reason: str = "leave", stream_id: Optional[str] = None) -> Optional[list]:
+        """Stop radio in one server: everything, one voice channel (every
+        stream in it), or one stream by id. C2 never reaches past the guild."""
         try:
             async with httpx.AsyncClient(timeout=30) as client:
                 r = await client.post(
                     f"{self.base}/discord/guilds/{guild_id}/stop",
-                    json={"channel_id": channel_id, "reason": reason},
+                    json={"channel_id": channel_id, "stream_id": stream_id, "reason": reason},
                     headers=self._headers(),
                 )
                 r.raise_for_status()