feat(discord): self-serve server->org linking with /drb link (#213 item 1)

- c2-core: POST /discord/link-codes (org owner/admin) mints a single-use,
  10-minute code; minting revokes the org's older codes. The bot redeems it
  via POST /discord/guilds/{id}/link/redeem (service key) inside one
  Firestore transaction, so two servers racing for one code cannot both win
  and a server linked to org A refuses org B's code. GET/DELETE
  /discord/linked-guilds lists/unlinks the caller's org's servers; unlink
  stops that server's streams. Rate-limited (mint per user, redeem per
  server and per Discord user); every mint/link/refusal/unlink audited.
- resolve_guild_org: an explicitly unlinked server no longer falls back to
  DISCORD_UNLINKED_GUILD_ORG_ID. Linking away from the fallback org stops
  the streams it had running for that org.
- Discord bot: /drb link code: (Manage Server, enforced at runtime too);
  GatedTree answers every other command in an unlinked server with how to
  link (autocomplete returns nothing). With the default fallback org set,
  nothing is gated — unchanged behaviour.
- Frontend: "Linked Discord servers" card on Settings -> Organization —
  generate code with countdown + copy, polls for the new link, unlink.
- 16 tests: single-use, two-server race, one-server/two-org race, expiry,
  revoked/malformed codes, wrong-org takeover, idempotent relink, rate
  limits, fallback stop, scoped list, unlink without fallback.

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

drb-c2-core/app/config.py

@@ -249,7 +249,11 @@ class Settings(BaseSettings):
     discord_lease_offline_release_seconds: int = 600
     # Org a Discord server that was never linked to an org (discord_guilds/{id})
     # acts for. Default keeps today's single-org deployment working unchanged;
-    # set it to "" to require every server to be linked first.
+    # set it to "" to require every server to be linked first. Do that once
+    # every server that should keep working has been linked (/drb link, or
+    # PUT /discord/guilds/{id}/link) — unlinked servers then get only
+    # /drb link and /help (server-26#213). A server an org explicitly unlinked
+    # never falls back to this org either way.
     discord_unlinked_guild_org_id: str = "founding"
 
     class Config:

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

@@ -0,0 +1,169 @@
+"""
+Self-serve Discord server -> DRB org linking (server-26#213 item 1).
+
+Same shape as the account-link codes in routers/links.py (web mints a short
+code, the Discord bot redeems it), with the gaps that pattern has closed:
+
+* **Single-use under a race.** Redeem runs in one Firestore transaction that
+  reads the code doc and the guild doc before writing either, so two servers
+  redeeming one code at once cannot both win — the loser's transaction is
+  re-run, sees ``redeemed_at`` set, and is refused.
+* **Short TTL** (``CODE_TTL_MINUTES``) and **one live code per org**: minting
+  a new code revokes the org's older unredeemed ones.
+* **Never moves a server between orgs.** A server already linked to org A
+  refuses org B's code (409) until org A unlinks it from its own settings.
+* An explicitly unlinked server does **not** fall back to
+  ``DISCORD_UNLINKED_GUILD_ORG_ID`` (see ``discord_streams.resolve_guild_org``)
+  — otherwise unlinking org B's server would hand it to the founding org.
+
+Firestore layout:
+
+  discord_link_codes/{CODE}   CODE = 10 chars, no dashes, upper-case
+    org_id, created_by_uid, created_at, expires_at,
+    redeemed_at|None, redeemed_guild_id|None, revoked: bool
+
+  discord_guilds/{guild_id}   (shared with discord_streams.py)
+    guild_id, org_id|None, guild_name, linked_by, linked_via, linked_at,
+    unlinked_at|None, unlinked_by|None
+
+Person names are never stored: the redeemer is kept as a Discord user id.
+"""
+from __future__ import annotations
+
+import re
+import secrets
+from datetime import datetime, timedelta, timezone
+from typing import Optional
+
+from app.internal import firestore as fstore
+
+CODES = "discord_link_codes"
+GUILDS = "discord_guilds"
+CODE_TTL_MINUTES = 10
+# Unambiguous alphabet: no 0/O, 1/I/L. 31^10 ~= 8e14 codes.
+_ALPHABET = "23456789ABCDEFGHJKMNPQRSTUVWXYZ"
+_CODE_LEN = 10
+
+
+class LinkError(Exception):
+    def __init__(self, status: int, code: str, message: str):
+        super().__init__(message)
+        self.status, self.code, self.message = status, code, message
+
+
+def _now() -> datetime:
+    return datetime.now(timezone.utc)
+
+
+def new_code() -> str:
+    return "".join(secrets.choice(_ALPHABET) for _ in range(_CODE_LEN))
+
+
+def normalize(code: str) -> str:
+    """Accept what a person pastes: any case, with or without dashes/spaces."""
+    return re.sub(r"[^A-Za-z0-9]", "", code or "").upper()
+
+
+def display(code: str) -> str:
+    """RVR7Q4K22X -> RVR-7Q4K-22X (the mockup's grouping, for reading aloud)."""
+    return f"{code[:3]}-{code[3:7]}-{code[7:]}"
+
+
+async def mint_code(org_id: str, uid: str) -> dict:
+    """Create a code for ``org_id`` and revoke the org's older unredeemed ones."""
+    for old in await fstore.collection_list(CODES, org_id=org_id):
+        if not old.get("redeemed_at") and not old.get("revoked") and old.get("code"):
+            await fstore.doc_update(CODES, old["code"], {"revoked": True})
+    code = new_code()
+    now = _now()
+    expires_at = now + timedelta(minutes=CODE_TTL_MINUTES)
+    await fstore.doc_set(CODES, code, {
+        "code": code,
+        "org_id": org_id,
+        "created_by_uid": uid,
+        "created_at": now.isoformat(),
+        "expires_at": expires_at.isoformat(),
+        "redeemed_at": None,
+        "redeemed_guild_id": None,
+        "revoked": False,
+    }, merge=False)
+    return {"code": display(code), "expires_at": expires_at.isoformat(),
+            "expires_minutes": CODE_TTL_MINUTES}
+
+
+def _redeem_txn(code: str, guild_id: str, guild_name: Optional[str], discord_user_id: str, now: datetime):
+    """Build the transaction body. Every read happens before any write."""
+    db = fstore.db
+
+    def _txn(txn):
+        code_ref = db.collection(CODES).document(code)
+        guild_ref = db.collection(GUILDS).document(guild_id)
+        code_snap = code_ref.get(transaction=txn)
+        guild_snap = guild_ref.get(transaction=txn)
+        doc = code_snap.to_dict() if code_snap.exists else None
+        guild = guild_snap.to_dict() if guild_snap.exists else None
+
+        if not doc or doc.get("revoked"):
+            return {"error": LinkError(404, "invalid_code",
+                                       "That link code is not valid. Generate a new one in DRB → "
+                                       "Settings → Organization.")}
+        if doc.get("redeemed_at"):
+            return {"error": LinkError(409, "code_used", "That link code has already been used.")}
+        if now > datetime.fromisoformat(doc["expires_at"]):
+            return {"error": LinkError(410, "code_expired",
+                                       "That link code has expired. Generate a new one in DRB → "
+                                       "Settings → Organization.")}
+        org_id = doc["org_id"]
+        current = (guild or {}).get("org_id")
+        if current and current != org_id:
+            return {"error": LinkError(409, "linked_elsewhere",
+                                       "This server is already linked to a different DRB organization. "
+                                       "That organization has to unlink it first.")}
+
+        iso = now.isoformat()
+        txn.update(code_ref, {"redeemed_at": iso, "redeemed_guild_id": guild_id,
+                              "redeemed_by_discord_id": discord_user_id})
+        txn.set(guild_ref, {
+            "guild_id": guild_id,
+            "org_id": org_id,
+            "guild_name": (guild_name or "")[:100] or (guild or {}).get("guild_name"),
+            "linked_by": f"discord:{discord_user_id}",
+            "linked_via": "code",
+            "linked_at": ((guild or {}).get("linked_at") if current == org_id else None) or iso,
+            "unlinked_at": None,
+            "unlinked_by": None,
+        }, merge=True)
+        return {"org_id": org_id, "already_linked": current == org_id,
+                "created_by_uid": doc.get("created_by_uid")}
+
+    return _txn
+
+
+async def redeem_code(raw_code: str, guild_id: str, guild_name: Optional[str], discord_user_id: str) -> dict:
+    """Link ``guild_id`` to the code's org. Raises LinkError on refusal."""
+    code = normalize(raw_code)
+    if len(code) != _CODE_LEN:
+        raise LinkError(404, "invalid_code",
+                        "That link code is not valid. Generate a new one in DRB → Settings → Organization.")
+    result = await fstore.run_transaction(
+        _redeem_txn(code, str(guild_id), guild_name, str(discord_user_id), _now()))
+    if "error" in result:
+        raise result["error"]
+    return result
+
+
+async def list_org_guilds(org_id: str) -> list[dict]:
+    return [
+        {"guild_id": g.get("guild_id"), "guild_name": g.get("guild_name"),
+         "linked_at": g.get("linked_at"), "linked_via": g.get("linked_via") or "admin"}
+        for g in await fstore.collection_list(GUILDS, org_id=org_id)
+    ]
+
+
+async def unlink(org_id: str, guild_id: str, uid: str) -> None:
+    guild = await fstore.doc_get(GUILDS, guild_id)
+    if not guild or guild.get("org_id") != org_id:
+        raise LinkError(404, "not_linked", "That Discord server is not linked to this organization.")
+    await fstore.doc_set(GUILDS, guild_id, {
+        "org_id": None, "unlinked_at": _now().isoformat(), "unlinked_by": uid,
+    }, merge=True)

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

@@ -75,10 +75,16 @@ def _parse_ts(value) -> Optional[datetime]:
 
 async def resolve_guild_org(guild_id: str) -> Optional[str]:
     """The org a Discord server acts for: its ``discord_guilds`` link, else
-    ``settings.discord_unlinked_guild_org_id`` (None when that is empty)."""
+    ``settings.discord_unlinked_guild_org_id`` (None when that is empty).
+
+    A server an org explicitly unlinked (``unlinked_at`` set, server-26#213)
+    gets no fallback: otherwise unlinking org B's server would quietly hand it
+    to the founding org."""
     link = await fstore.doc_get(GUILDS, str(guild_id))
     if link and link.get("org_id"):
         return link["org_id"]
+    if link and link.get("unlinked_at"):
+        return None
     return settings.discord_unlinked_guild_org_id or None
 
 

drb-c2-core/app/main.py

@@ -20,6 +20,7 @@ from app.internal.auth import (
 from app.routers import nodes, systems, calls, upload, tokens, incidents, alerts, admin, trips, places, links, users
 from app.routers import enrollment, media, org, waitlist, telemetry, replay, traffic, cameras
 from app.routers import discord as discord_router
+from app.routers import discord_link as discord_link_router
 from app.internal import dynsec
 from app.internal import firestore as fstore
 
@@ -158,6 +159,7 @@ app.include_router(media.router)
 # auth is per-endpoint: bot routes take the service key (or a platform admin);
 # the token-fetch route takes the calling node's own API key.
 app.include_router(discord_router.router)
+app.include_router(discord_link_router.router)  # auth per-endpoint (org owner / bot service key) — server-26#213
 # NOTE: there used to be an app.routers.mqtt_auth router here (an HTTP
 # backend for the mosquitto-go-auth plugin). That plugin's upstream project
 # is archived (no CVE patches) and was rejected for an internet-facing

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

@@ -0,0 +1,125 @@
+"""
+Self-serve Discord server -> org linking (server-26#213 item 1).
+
+  POST   /discord/link-codes                    org owner / platform admin — mint a code
+  GET    /discord/linked-guilds                 org owner / platform admin — this org's servers
+  DELETE /discord/linked-guilds/{guild_id}      org owner / platform admin — unlink one
+  POST   /discord/guilds/{guild_id}/link/redeem Discord bot (service key) — /drb link code:
+
+Platform admins may pass ?org_id= to act on another org (resolve_org_scope),
+same as the rest of the org-scoped API. The admin-only
+``PUT /discord/guilds/{id}/link`` in routers/discord.py stays for support.
+
+Every mint, link, refusal and unlink is written to audit_log. Rate limits are
+the in-process sliding-window limiter from internal/auth.py (single c2-core
+instance; not shared across replicas).
+"""
+from __future__ import annotations
+
+from typing import Optional
+
+from fastapi import APIRouter, Depends, HTTPException
+from fastapi.responses import JSONResponse
+from pydantic import BaseModel
+
+from app.config import settings
+from app.internal import discord_guild_link as glink
+from app.internal import discord_streams as streams
+from app.internal import firestore as fstore
+from app.internal.audit import write_audit
+from app.internal.auth import (
+    _RateLimiter, describe_actor, require_org_owner_token, require_service_key, resolve_org_scope,
+)
+from app.internal.logger import logger
+
+router = APIRouter(prefix="/discord", tags=["discord"])
+
+# Minting: 5 codes per user per 10 min — each mint revokes the previous one,
+# so this only stops a runaway client.
+mint_limiter = _RateLimiter(max_calls=5, window_seconds=600)
+# Redeeming: 5 attempts per server and 10 per Discord user per 10 min. With a
+# 10-character code over 31 symbols and a 10-minute TTL, guessing is hopeless
+# at this rate even across many throwaway servers.
+redeem_guild_limiter = _RateLimiter(max_calls=5, window_seconds=600)
+redeem_user_limiter = _RateLimiter(max_calls=10, window_seconds=600)
+
+
+def _caller_org(decoded: dict, org_id: Optional[str]) -> str:
+    return resolve_org_scope(decoded, org_id)
+
+
+@router.post("/link-codes")
+async def create_link_code(org_id: Optional[str] = None, decoded: dict = Depends(require_org_owner_token)):
+    """Mint a single-use code an org owner pastes into ``/drb link code:``."""
+    scope = _caller_org(decoded, org_id)
+    uid, email = describe_actor(decoded)
+    mint_limiter.check(f"{uid}:discord_link_code")
+    out = await glink.mint_code(scope, uid)
+    await write_audit(uid, email, "discord.link_code.create",
+                      details={"org_id": scope, "expires_at": out["expires_at"]})
+    logger.info(f"Discord link code minted for org={scope!r} by uid={uid}")
+    return {**out, "command": f"/drb link code:{out['code']}"}
+
+
+@router.get("/linked-guilds")
+async def linked_guilds(org_id: Optional[str] = None, decoded: dict = Depends(require_org_owner_token)):
+    scope = _caller_org(decoded, org_id)
+    return {
+        "guilds": await glink.list_org_guilds(scope),
+        # False while DISCORD_UNLINKED_GUILD_ORG_ID is set: unlinked servers
+        # then still act for that org instead of seeing only /drb link + /help.
+        "unlinked_guilds_restricted": not settings.discord_unlinked_guild_org_id,
+    }
+
+
+@router.delete("/linked-guilds/{guild_id}")
+async def unlink_guild(guild_id: str, org_id: Optional[str] = None,
+                       decoded: dict = Depends(require_org_owner_token)):
+    scope = _caller_org(decoded, org_id)
+    uid, email = describe_actor(decoded)
+    try:
+        await glink.unlink(scope, guild_id, uid)
+    except glink.LinkError as e:
+        raise HTTPException(e.status, e.message)
+    # Its radio channels were playing for this org; they go with the link.
+    stopped = await streams.stop_guild(guild_id, "unlinked")
+    await write_audit(uid, email, "discord.guild.unlink",
+                      details={"org_id": scope, "guild_id": guild_id, "streams_stopped": len(stopped)})
+    return {"ok": True, "guild_id": guild_id, "streams_stopped": len(stopped)}
+
+
+class RedeemBody(BaseModel):
+    code: str
+    discord_user_id: str
+    guild_name: Optional[str] = None
+
+
+@router.post("/guilds/{guild_id}/link/redeem")
+async def redeem_link_code(guild_id: str, body: RedeemBody, principal: dict = Depends(require_service_key)):
+    """The Discord bot redeems a code for the server it was typed in. The bot
+    has already checked the member holds Manage Server."""
+    redeem_guild_limiter.check(f"guild:{guild_id}")
+    redeem_user_limiter.check(f"user:{body.discord_user_id}")
+    actor = f"discord:{body.discord_user_id}"
+    before = await streams.resolve_guild_org(guild_id)
+    try:
+        result = await glink.redeem_code(body.code, guild_id, body.guild_name, body.discord_user_id)
+    except glink.LinkError as e:
+        await write_audit(actor, "", "discord.guild.link_refused",
+                          details={"guild_id": guild_id, "reason": e.code})
+        return JSONResponse(status_code=e.status, content={"detail": e.message, "code": e.code})
+
+    org_id = result["org_id"]
+    stopped: list = []
+    if before and before != org_id:
+        # The server was acting for DISCORD_UNLINKED_GUILD_ORG_ID; anything it
+        # had playing belongs to that org, not the one it just linked to.
+        stopped = await streams.stop_guild(guild_id, "relinked")
+    await write_audit(actor, "", "discord.guild.link", details={
+        "guild_id": guild_id, "org_id": org_id, "already_linked": result["already_linked"],
+        "code_created_by_uid": result.get("created_by_uid"), "streams_stopped": len(stopped),
+    })
+    logger.info(f"Discord guild {guild_id} linked to org={org_id!r} by {actor}")
+    org = await fstore.doc_get("organizations", org_id)
+    return {"ok": True, "guild_id": guild_id, "org_id": org_id,
+            "org_name": (org or {}).get("name"), "already_linked": result["already_linked"]}

drb-c2-core/tests/test_discord_guild_link.py

@@ -0,0 +1,251 @@
+"""
+Self-serve Discord server -> org linking (server-26#213 item 1).
+
+Race tests run two real transactions against tests/fake_firestore.py, held
+together at the commit point, so the interleaving the single-use guarantee
+has to survive actually happens.
+"""
+import asyncio
+import threading
+from datetime import datetime, timedelta, timezone
+
+import pytest
+from fastapi import HTTPException
+
+from app.config import settings
+from app.internal import auth as auth_mod
+from app.internal import discord_guild_link as glink
+from app.internal import discord_streams as streams
+from app.routers import discord_link as router
+from tests.fake_firestore import FakeFirestore
+
+OWNER_A = {"uid": "owner-a", "email": "a@example.test", "org_id": "org-a", "org_role": "owner"}
+OWNER_B = {"uid": "owner-b", "email": "b@example.test", "org_id": "org-b", "org_role": "owner"}
+
+
+@pytest.fixture
+def fs(monkeypatch):
+    fake = FakeFirestore.install(monkeypatch)
+    fake.put("organizations", "org-a", {"org_id": "org-a", "name": "Riverton Fire Co. 7"})
+    fake.put("organizations", "org-b", {"org_id": "org-b", "name": "Harbor Towing"})
+    for lim in (router.mint_limiter, router.redeem_guild_limiter, router.redeem_user_limiter):
+        lim._log.clear()
+    stopped: list = []
+
+    async def _stop_guild(guild_id, reason, channel_id=None):
+        stopped.append((guild_id, reason))
+        return [f"{guild_id}:c1"]
+    monkeypatch.setattr(streams, "stop_guild", _stop_guild)
+    fake.stopped = stopped
+    monkeypatch.setattr(settings, "discord_unlinked_guild_org_id", "")
+    return fake
+
+
+def _audit(fake, action):
+    return [e for e in (fake.data.get("audit_log") or {}).values() if e["action"] == action]
+
+
+async def _mint(decoded=OWNER_A):
+    return await router.create_link_code(org_id=None, decoded=decoded)
+
+
+async def _redeem(code, guild="g1", user="u1", name="Station 7"):
+    return await router.redeem_link_code(
+        guild, router.RedeemBody(code=code, discord_user_id=user, guild_name=name), principal={"service": True})
+
+
+def _status(resp):
+    return getattr(resp, "status_code", 200)
+
+
+def _hold_both_at_commit(fake):
+    barrier = threading.Barrier(2, timeout=5)
+
+    def _hook(attempt):
+        if attempt == 0:
+            barrier.wait()
+    fake.before_commit = _hook
+
+
+# ---------------------------------------------------------------------------
+# Mint
+# ---------------------------------------------------------------------------
+
+async def test_mint_returns_display_code_and_audits(fs):
+    out = await _mint()
+    assert len(out["code"]) == 12 and out["code"][3] == "-" and out["code"][8] == "-"
+    assert out["expires_minutes"] == glink.CODE_TTL_MINUTES
+    assert out["command"] == f"/drb link code:{out['code']}"
+    doc = fs.get(glink.CODES, glink.normalize(out["code"]))
+    assert doc["org_id"] == "org-a" and doc["redeemed_at"] is None and doc["revoked"] is False
+    assert len(_audit(fs, "discord.link_code.create")) == 1
+
+
+async def test_minting_again_revokes_the_previous_code(fs):
+    first = await _mint()
+    second = await _mint()
+    assert fs.get(glink.CODES, glink.normalize(first["code"]))["revoked"] is True
+    resp = await _redeem(first["code"])
+    assert _status(resp) == 404
+    assert _status(await _redeem(second["code"])) == 200
+
+
+async def test_mint_rate_limited_per_user(fs):
+    for _ in range(5):
+        await _mint()
+    with pytest.raises(HTTPException) as e:
+        await _mint()
+    assert e.value.status_code == 429
+
+
+async def test_member_cannot_mint(fs, monkeypatch):
+    member = {**OWNER_A, "org_role": "member"}
+    monkeypatch.setattr(auth_mod.firebase_auth, "verify_id_token", lambda _t: member)
+
+    class _Cred:
+        credentials = "tok"
+    with pytest.raises(HTTPException) as e:
+        await auth_mod.require_org_owner_token(_Cred())
+    assert e.value.status_code == 403
+
+
+# ---------------------------------------------------------------------------
+# Redeem
+# ---------------------------------------------------------------------------
+
+async def test_redeem_links_the_guild_and_is_single_use(fs):
+    out = await _mint()
+    resp = await _redeem(out["code"].lower())          # pasted in lower case, dashes kept
+    assert resp["org_id"] == "org-a" and resp["org_name"] == "Riverton Fire Co. 7"
+    assert await streams.resolve_guild_org("g1") == "org-a"
+    g = fs.get(streams.GUILDS, "g1")
+    assert g["linked_by"] == "discord:u1" and g["linked_via"] == "code" and g["guild_name"] == "Station 7"
+    code_doc = fs.get(glink.CODES, glink.normalize(out["code"]))
+    assert code_doc["redeemed_guild_id"] == "g1"
+    assert len(_audit(fs, "discord.guild.link")) == 1
+
+    again = await _redeem(out["code"], guild="g2", user="u2")
+    assert _status(again) == 409
+    assert fs.get(streams.GUILDS, "g2") is None
+    assert [e["details"]["reason"] for e in _audit(fs, "discord.guild.link_refused")] == ["code_used"]
+
+
+async def test_two_servers_racing_for_one_code_only_one_wins(fs):
+    out = await _mint()
+    _hold_both_at_commit(fs)
+    r1, r2 = await asyncio.gather(_redeem(out["code"], guild="g1", user="u1"),
+                                  _redeem(out["code"], guild="g2", user="u2"))
+    assert fs.retries >= 1                                   # the race really happened
+    assert sorted([_status(r1), _status(r2)]) == [200, 409]
+    linked = [g for g in ("g1", "g2") if (fs.get(streams.GUILDS, g) or {}).get("org_id")]
+    assert len(linked) == 1
+    assert fs.get(glink.CODES, glink.normalize(out["code"]))["redeemed_guild_id"] == linked[0]
+
+
+async def test_one_server_racing_two_orgs_codes_ends_on_exactly_one(fs):
+    a = await _mint(OWNER_A)
+    b = await _mint(OWNER_B)
+    _hold_both_at_commit(fs)
+    r1, r2 = await asyncio.gather(_redeem(a["code"], user="u1"), _redeem(b["code"], user="u2"))
+    assert sorted([_status(r1), _status(r2)]) == [200, 409]
+    winner = "org-a" if _status(r1) == 200 else "org-b"
+    loser_code = b["code"] if winner == "org-a" else a["code"]
+    assert fs.get(streams.GUILDS, "g1")["org_id"] == winner
+    # The refused code was not consumed — its org can still use it elsewhere.
+    assert fs.get(glink.CODES, glink.normalize(loser_code))["redeemed_at"] is None
+
+
+async def test_expired_code_is_refused(fs):
+    out = await _mint()
+    key = glink.normalize(out["code"])
+    past = (datetime.now(timezone.utc) - timedelta(seconds=1)).isoformat()
+    fs.put(glink.CODES, key, {**fs.get(glink.CODES, key), "expires_at": past})
+    resp = await _redeem(out["code"])
+    assert _status(resp) == 410
+    assert fs.get(streams.GUILDS, "g1") is None
+    assert fs.get(glink.CODES, key)["redeemed_at"] is None
+
+
+async def test_unknown_or_malformed_code_is_refused(fs):
+    assert _status(await _redeem("NOPE")) == 404
+    assert _status(await _redeem("ABC-DEFG-HJK")) == 404
+
+
+async def test_wrong_org_cannot_take_over_a_linked_server(fs):
+    a = await _mint(OWNER_A)
+    assert _status(await _redeem(a["code"])) == 200
+    b = await _mint(OWNER_B)
+    resp = await _redeem(b["code"], user="u9")
+    assert _status(resp) == 409
+    assert fs.get(streams.GUILDS, "g1")["org_id"] == "org-a"
+    assert fs.get(glink.CODES, glink.normalize(b["code"]))["redeemed_at"] is None
+    assert fs.stopped == []
+
+
+async def test_same_org_relink_is_idempotent(fs):
+    first = await _mint()
+    await _redeem(first["code"])
+    linked_at = fs.get(streams.GUILDS, "g1")["linked_at"]
+    second = await _mint()
+    resp = await _redeem(second["code"])
+    assert resp["already_linked"] is True
+    assert fs.get(streams.GUILDS, "g1")["linked_at"] == linked_at
+
+
+async def test_redeem_rate_limited_per_server(fs):
+    for _ in range(5):
+        await _redeem("WRONGCODE1", user="u-many")
+    with pytest.raises(HTTPException) as e:
+        await _redeem("WRONGCODE1", user="u-other")
+    assert e.value.status_code == 429
+
+
+# ---------------------------------------------------------------------------
+# DISCORD_UNLINKED_GUILD_ORG_ID back-compat
+# ---------------------------------------------------------------------------
+
+async def test_linking_away_from_the_fallback_org_stops_its_streams(fs, monkeypatch):
+    monkeypatch.setattr(settings, "discord_unlinked_guild_org_id", "founding")
+    assert await streams.resolve_guild_org("g1") == "founding"
+    out = await _mint()
+    await _redeem(out["code"])
+    assert fs.stopped == [("g1", "relinked")]
+    assert await streams.resolve_guild_org("g1") == "org-a"
+
+
+async def test_linking_to_the_fallback_org_itself_stops_nothing(fs, monkeypatch):
+    monkeypatch.setattr(settings, "discord_unlinked_guild_org_id", "org-a")
+    out = await _mint()
+    await _redeem(out["code"])
+    assert fs.stopped == []
+
+
+# ---------------------------------------------------------------------------
+# List / unlink
+# ---------------------------------------------------------------------------
+
+async def test_list_is_scoped_to_the_callers_org(fs, monkeypatch):
+    await _redeem((await _mint(OWNER_A))["code"], guild="g1")
+    await _redeem((await _mint(OWNER_B))["code"], guild="g2", user="u2")
+    out = await router.linked_guilds(org_id="org-b", decoded=OWNER_A)   # non-admin override ignored
+    assert [g["guild_id"] for g in out["guilds"]] == ["g1"]
+    assert out["unlinked_guilds_restricted"] is True
+    monkeypatch.setattr(settings, "discord_unlinked_guild_org_id", "founding")
+    assert (await router.linked_guilds(org_id=None, decoded=OWNER_A))["unlinked_guilds_restricted"] is False
+
+
+async def test_unlink_stops_streams_and_never_falls_back(fs, monkeypatch):
+    monkeypatch.setattr(settings, "discord_unlinked_guild_org_id", "founding")
+    await _redeem((await _mint())["code"])
+    fs.stopped.clear()
+    with pytest.raises(HTTPException) as e:
+        await router.unlink_guild("g1", org_id=None, decoded=OWNER_B)       # not B's server
+    assert e.value.status_code == 404
+    out = await router.unlink_guild("g1", org_id=None, decoded=OWNER_A)
+    assert out["ok"] and fs.stopped == [("g1", "unlinked")]
+    assert await streams.resolve_guild_org("g1") is None                # NOT "founding"
+    assert len(_audit(fs, "discord.guild.unlink")) == 1
+    # ...and it can be linked again, by any org, with a fresh code.
+    resp = await _redeem((await _mint(OWNER_B))["code"])
+    assert resp["org_id"] == "org-b"
+    assert await streams.resolve_guild_org("g1") == "org-b"