diff --git a/fastapi_startkit/src/fastapi_startkit/broadcasting/config.py b/fastapi_startkit/src/fastapi_startkit/broadcasting/config.py
index d91fc094..6667de55 100644
--- a/fastapi_startkit/src/fastapi_startkit/broadcasting/config.py
+++ b/fastapi_startkit/src/fastapi_startkit/broadcasting/config.py
@@ -8,7 +8,7 @@
 class ReverbConnectionConfig(Config):
     driver: str = "reverb"
     key: str = env("REVERB_APP_KEY", "local")
-    secret: str = env("REVERB_APP_SECRET", "secret")
+    secret: str = env("REVERB_APP_SECRET", "")
     app_id: str = env("REVERB_APP_ID", "1")
     host: str = env("REVERB_HOST", "0.0.0.0")
     port: int = env("REVERB_PORT", 8080)
diff --git a/fastapi_startkit/src/fastapi_startkit/broadcasting/provider.py b/fastapi_startkit/src/fastapi_startkit/broadcasting/provider.py
index 63f0ae25..ecc58326 100644
--- a/fastapi_startkit/src/fastapi_startkit/broadcasting/provider.py
+++ b/fastapi_startkit/src/fastapi_startkit/broadcasting/provider.py
@@ -2,13 +2,19 @@
 
 import importlib
 import importlib.util
+import json
 import sys
+from collections.abc import Mapping
 from pathlib import Path
 
 from fastapi_startkit.support import Provider
 from .config import BroadcastingConfig
 from .manager import BroadcastManager
 from .reverb.server import ReverbServer
+from .reverb.signature import is_presence, requires_auth, sign, valid_channel, valid_socket_id
+
+UNSAFE_SECRETS = ("", "secret")
+INSECURE_ENVIRONMENTS = ("local", "testing")
 
 
 class ReverbProvider(Provider):
@@ -36,7 +42,14 @@ def register(self) -> None:
         config_data = self.resolve_config(BroadcastingConfig)
         self.merge_config_from(config_data, "broadcasting")
 
-        server = ReverbServer()
+        reverb = config_data.get("connections", {}).get("reverb", {})
+        secret = reverb.get("secret") or ""
+        if secret in UNSAFE_SECRETS and self.app.env not in INSECURE_ENVIRONMENTS:
+            raise RuntimeError(
+                f"REVERB_APP_SECRET must be set to a private value when APP_ENV is {self.app.env!r}; "
+                "private and presence channel signatures would otherwise be forgeable."
+            )
+        server = ReverbServer(key=reverb.get("key", "local"), secret=secret)
         manager = BroadcastManager(config_data, server)
 
         self.app.bind("broadcasting", manager)
@@ -81,91 +94,75 @@ def _load_channels_file(self) -> None:
         spec.loader.exec_module(module)  # type: ignore[union-attr]
 
     def _mount_websocket(self) -> None:
-        """Mount the Reverb WebSocket endpoint on the FastAPI application.
-
-        The endpoint path is ``/app/{app_key}``, matching the Pusher protocol
-        expected by Laravel Echo.  The ``app_key`` is taken from the
-        ``reverb.connections.reverb.key`` config value (default: ``"local"``).
-        """
         try:
             fastapi_app = self.app.fastapi
         except RuntimeError:
-            # FastAPI not installed or app not yet booted — skip
             return
 
-        config = self.app.make("config").get("broadcasting") or {}
-        connections = config.get("connections", {}) if isinstance(config, dict) else {}
-        reverb_conn = connections.get("reverb", {}) if isinstance(connections, dict) else {}
-        app_key = reverb_conn.get("key", "local") if isinstance(reverb_conn, dict) else "local"
-
         server: ReverbServer = self.app.make("reverb.server")
 
-        from starlette.routing import WebSocketRoute
-        from starlette.applications import Starlette
-
         async def ws_endpoint(websocket):
             await server.handle(websocket)
 
-        ws_app = Starlette(routes=[WebSocketRoute(f"/app/{app_key}", ws_endpoint)])
-        fastapi_app.mount("", ws_app)
+        fastapi_app.router.add_websocket_route(f"/app/{server.key}", ws_endpoint)
 
     def _mount_auth_endpoint(self) -> None:
-        """Mount the ``/broadcasting/auth`` HTTP endpoint.
-
-        The endpoint handles the Laravel Echo authentication handshake for
-        private and presence channels.  It:
-
-        1. Reads ``channel_name`` from the POST body (form or JSON).
-        2. Resolves the current user via the container's ``auth`` binding (if
-           available), falling back to ``request.state.user``.
-        3. Calls :meth:`~.auth.ChannelAuthRegistry.authorize` with the channel
-           name and the resolved user.
-        4. Returns **200** on success or **403** on denial.
-        """
         try:
             fastapi_app = self.app.fastapi
         except RuntimeError:
             return
 
-        from fastapi import Request
-        from fastapi.responses import JSONResponse
+        from starlette.requests import Request
+        from starlette.responses import JSONResponse
 
         application = self.app
 
-        @fastapi_app.post("/broadcasting/auth")
         async def broadcasting_auth(request: Request):
-            # Resolve channel name from form data or JSON body
-            content_type = request.headers.get("content-type", "")
-            if "application/json" in content_type:
-                body = await request.json()
-                channel_name = body.get("channel_name", "")
+            if "application/json" in request.headers.get("content-type", ""):
+                try:
+                    body = await request.json()
+                except ValueError:
+                    return JSONResponse({"error": "Malformed JSON body"}, status_code=400)
             else:
-                form = await request.form()
-                channel_name = form.get("channel_name", "")
-
-            if not isinstance(channel_name, str):
-                return JSONResponse({"error": "channel_name must be a string"}, status_code=422)
+                body = await request.form()
+            if not isinstance(body, Mapping):
+                return JSONResponse({"error": "Request body must be an object"}, status_code=422)
+
+            channel_name = body.get("channel_name", "")
+            socket_id = body.get("socket_id", "")
+            if not valid_socket_id(socket_id):
+                return JSONResponse({"error": "socket_id is invalid"}, status_code=422)
+            if not valid_channel(channel_name):
+                return JSONResponse({"error": "channel_name is invalid"}, status_code=422)
+
+            server: ReverbServer = application.make("reverb.server")
+            user = self._resolve_user(application, request)
+            if not server.secret or not requires_auth(channel_name) or user is None:
+                return JSONResponse({"error": "Forbidden"}, status_code=403)
 
-            # Resolve authenticated user.
-            # Prefer request.state.user (set by auth middleware), then fall
-            # back to the container's "auth" service.
-            user = getattr(request.state, "user", None)
-            if user is None:
-                try:
-                    auth_service = application.make("auth")
-                    user_attr = getattr(auth_service, "user", None)
-                    if callable(user_attr):
-                        user = user_attr()
-                    else:
-                        user = user_attr
-                except Exception:
-                    user = None
-
-            # Authorise
             manager: BroadcastManager = application.make("broadcasting")
-            allowed = await manager.channel_registry.authorize(channel_name, user)
+            if not await manager.channel_registry.authorize(channel_name, user):
+                return JSONResponse({"error": "Forbidden"}, status_code=403)
+
+            if not is_presence(channel_name):
+                return JSONResponse({"auth": sign(server.key, server.secret, socket_id, channel_name)})
 
-            if not allowed:
+            user_id = getattr(user, "id", None)
+            if user_id is None:
                 return JSONResponse({"error": "Forbidden"}, status_code=403)
+            channel_data = json.dumps({"user_id": str(user_id)})
+            auth = sign(server.key, server.secret, socket_id, channel_name, channel_data)
+            return JSONResponse({"auth": auth, "channel_data": channel_data})
 
-            return JSONResponse({"auth": True, "channel_data": {}})
+        fastapi_app.router.add_route("/broadcasting/auth", broadcasting_auth, methods=["POST"])
+
+    @staticmethod
+    def _resolve_user(application, request):
+        user = getattr(request.state, "user", None)
+        if user is not None:
+            return user
+        try:
+            user_attr = getattr(application.make("auth"), "user", None)
+        except Exception:
+            return None
+        return user_attr() if callable(user_attr) else user_attr
diff --git a/fastapi_startkit/src/fastapi_startkit/broadcasting/reverb/channel_manager.py b/fastapi_startkit/src/fastapi_startkit/broadcasting/reverb/channel_manager.py
index d92f9446..611167c1 100644
--- a/fastapi_startkit/src/fastapi_startkit/broadcasting/reverb/channel_manager.py
+++ b/fastapi_startkit/src/fastapi_startkit/broadcasting/reverb/channel_manager.py
@@ -1,17 +1,21 @@
 class ChannelManager:
     def __init__(self):
-        self._subscriptions: dict[str, set] = {}
+        self._subscriptions: dict[str, set[str]] = {}
 
     def subscribe(self, channel: str, socket_id: str) -> None:
         self._subscriptions.setdefault(channel, set()).add(socket_id)
 
     def unsubscribe(self, channel: str, socket_id: str) -> None:
-        if channel in self._subscriptions:
-            self._subscriptions[channel].discard(socket_id)
+        subscribers = self._subscriptions.get(channel)
+        if subscribers is None:
+            return
+        subscribers.discard(socket_id)
+        if not subscribers:
+            del self._subscriptions[channel]
 
     def remove_socket(self, socket_id: str) -> None:
-        for subs in self._subscriptions.values():
-            subs.discard(socket_id)
+        for channel in list(self._subscriptions):
+            self.unsubscribe(channel, socket_id)
 
-    def subscribers(self, channel: str) -> set:
+    def subscribers(self, channel: str) -> set[str]:
         return set(self._subscriptions.get(channel, set()))
diff --git a/fastapi_startkit/src/fastapi_startkit/broadcasting/reverb/connection.py b/fastapi_startkit/src/fastapi_startkit/broadcasting/reverb/connection.py
index 945a8ea2..b25da052 100644
--- a/fastapi_startkit/src/fastapi_startkit/broadcasting/reverb/connection.py
+++ b/fastapi_startkit/src/fastapi_startkit/broadcasting/reverb/connection.py
@@ -1,4 +1,5 @@
-import uuid
+import secrets
+
 from starlette.websockets import WebSocket
 
 
@@ -7,15 +8,21 @@ def __init__(self):
         self._connections: dict[str, WebSocket] = {}
 
     def add(self, websocket: WebSocket) -> str:
-        socket_id = str(uuid.uuid4())
+        socket_id = self._new_socket_id()
+        while socket_id in self._connections:
+            socket_id = self._new_socket_id()
         self._connections[socket_id] = websocket
         return socket_id
 
+    @staticmethod
+    def _new_socket_id() -> str:
+        return f"{secrets.randbelow(10**10)}.{secrets.randbelow(10**10)}"
+
     def remove(self, socket_id: str) -> None:
         self._connections.pop(socket_id, None)
 
     def get(self, socket_id: str):
         return self._connections.get(socket_id)
 
-    def all(self) -> dict:
+    def all(self) -> dict[str, WebSocket]:
         return dict(self._connections)
diff --git a/fastapi_startkit/src/fastapi_startkit/broadcasting/reverb/server.py b/fastapi_startkit/src/fastapi_startkit/broadcasting/reverb/server.py
index aff7b65c..70e20f30 100644
--- a/fastapi_startkit/src/fastapi_startkit/broadcasting/reverb/server.py
+++ b/fastapi_startkit/src/fastapi_startkit/broadcasting/reverb/server.py
@@ -1,48 +1,108 @@
 import json
-from starlette.websockets import WebSocket, WebSocketDisconnect
-from starlette.routing import WebSocketRoute
+
 from starlette.applications import Starlette
-from .connection import ConnectionManager
+from starlette.routing import WebSocketRoute
+from starlette.websockets import WebSocket, WebSocketDisconnect
+
 from .channel_manager import ChannelManager
+from .connection import ConnectionManager
+from .signature import requires_auth, valid_channel, verify
+
+MESSAGE_TOO_BIG = 1009
 
 
 class ReverbServer:
-    def __init__(self):
+    max_message_size = 10_000
+
+    def __init__(self, key: str = "local", secret: str = ""):
+        self.key = key
+        self.secret = secret
         self.connections = ConnectionManager()
         self.channels = ChannelManager()
 
     async def handle(self, websocket: WebSocket):
         await websocket.accept()
         socket_id = self.connections.add(websocket)
-        await websocket.send_text(
-            json.dumps(
-                {
-                    "event": "pusher:connection_established",
-                    "data": json.dumps({"socket_id": socket_id, "activity_timeout": 120}),
-                }
-            )
-        )
         try:
-            while True:
-                data = await websocket.receive_text()
-                message = json.loads(data)
-                await self._handle_message(socket_id, websocket, message)
+            await self._send(
+                websocket,
+                "pusher:connection_established",
+                json.dumps({"socket_id": socket_id, "activity_timeout": 120}),
+            )
+            await self._receive_loop(socket_id, websocket)
         except WebSocketDisconnect:
+            pass
+        finally:
             self.connections.remove(socket_id)
             self.channels.remove_socket(socket_id)
 
-    async def _handle_message(self, socket_id: str, websocket: WebSocket, message: dict):
+    async def _receive_loop(self, socket_id: str, websocket: WebSocket):
+        while True:
+            frame = await websocket.receive()
+            if frame["type"] == "websocket.disconnect":
+                return
+            text = frame.get("text")
+            if text is not None and len(text) > self.max_message_size:
+                await self._error(websocket, "Maximum message size exceeded")
+                await websocket.close(code=MESSAGE_TOO_BIG)
+                return
+            message = self._parse(text)
+            if message is None:
+                await self._error(websocket, "Invalid message format")
+                continue
+            await self._handle_message(socket_id, websocket, message)
+
+    @staticmethod
+    def _parse(text: str | None) -> dict[str, object] | None:
+        if text is None:
+            return None
+        try:
+            message = json.loads(text)
+        except ValueError:
+            return None
+        if not isinstance(message, dict) or not isinstance(message.get("event", ""), str):
+            return None
+        return message
+
+    async def _handle_message(self, socket_id: str, websocket: WebSocket, message: dict[str, object]):
         event = message.get("event", "")
         if event == "pusher:ping":
-            await websocket.send_text(json.dumps({"event": "pusher:pong", "data": {}}))
+            await self._send(websocket, "pusher:pong", {})
         elif event == "pusher:subscribe":
-            channel = message.get("data", {}).get("channel", "")
-            self.channels.subscribe(channel, socket_id)
-            await websocket.send_text(
-                json.dumps({"event": "pusher_internal:subscription_succeeded", "channel": channel, "data": {}})
-            )
+            await self._subscribe(socket_id, websocket, message.get("data"))
+
+    async def _subscribe(self, socket_id: str, websocket: WebSocket, data: object):
+        if not isinstance(data, dict) or not isinstance(data.get("channel"), str) or not data["channel"]:
+            await self._error(websocket, "Invalid subscription")
+            return
+        channel = data["channel"]
+        if not valid_channel(channel):
+            await self._subscription_error(websocket, channel, "InvalidChannel", "Invalid channel name", 400)
+            return
+        if requires_auth(channel) and not verify(
+            self.key, self.secret, socket_id, channel, data.get("auth"), data.get("channel_data")
+        ):
+            await self._subscription_error(websocket, channel, "AuthError", "Invalid signature", 401)
+            return
+        self.channels.subscribe(channel, socket_id)
+        await self._send(websocket, "pusher_internal:subscription_succeeded", {}, channel)
+
+    async def _subscription_error(self, websocket: WebSocket, channel: str, kind: str, error: str, status: int):
+        await self._send(
+            websocket, "pusher:subscription_error", {"type": kind, "error": error, "status": status}, channel
+        )
+
+    async def _error(self, websocket: WebSocket, message: str):
+        await self._send(websocket, "pusher:error", {"code": 4200, "message": message})
+
+    @staticmethod
+    async def _send(websocket: WebSocket, event: str, data: object, channel: str | None = None):
+        payload: dict[str, object] = {"event": event, "data": data}
+        if channel is not None:
+            payload["channel"] = channel
+        await websocket.send_text(json.dumps(payload))
 
-    async def broadcast_to_channel(self, channel_name: str, event_name: str, data: dict):
+    async def broadcast_to_channel(self, channel_name: str, event_name: str, data: dict[str, object]):
         payload = json.dumps({"event": event_name, "channel": channel_name, "data": data})
         for socket_id in self.channels.subscribers(channel_name):
             ws = self.connections.get(socket_id)
diff --git a/fastapi_startkit/src/fastapi_startkit/broadcasting/reverb/signature.py b/fastapi_startkit/src/fastapi_startkit/broadcasting/reverb/signature.py
new file mode 100644
index 00000000..dcf71ef9
--- /dev/null
+++ b/fastapi_startkit/src/fastapi_startkit/broadcasting/reverb/signature.py
@@ -0,0 +1,43 @@
+import hashlib
+import hmac
+import re
+
+from typing_extensions import TypeIs
+
+SOCKET_ID = re.compile(r"[0-9]+\.[0-9]+", re.ASCII)
+CHANNEL_NAME = re.compile(r"[A-Za-z0-9_\-=@,.;]{1,200}", re.ASCII)
+
+
+def valid_socket_id(socket_id: object) -> TypeIs[str]:
+    return isinstance(socket_id, str) and SOCKET_ID.fullmatch(socket_id) is not None
+
+
+def valid_channel(channel: object) -> TypeIs[str]:
+    return isinstance(channel, str) and CHANNEL_NAME.fullmatch(channel) is not None
+
+
+def requires_auth(channel: object) -> bool:
+    return isinstance(channel, str) and channel.startswith(("private-", "presence-"))
+
+
+def is_presence(channel: str) -> bool:
+    return channel.startswith("presence-")
+
+
+def sign(key: str, secret: str, socket_id: str, channel: str, channel_data: str | None = None) -> str:
+    parts = [socket_id, channel] if channel_data is None else [socket_id, channel, channel_data]
+    digest = hmac.new(secret.encode(), ":".join(parts).encode(), hashlib.sha256).hexdigest()
+    return f"{key}:{digest}"
+
+
+def verify(key: str, secret: str, socket_id: str, channel: str, auth: object, channel_data: object = None) -> bool:
+    if not secret or not isinstance(auth, str):
+        return False
+    if not valid_socket_id(socket_id) or not valid_channel(channel) or not requires_auth(channel):
+        return False
+    if not is_presence(channel):
+        channel_data = None
+    elif not isinstance(channel_data, str):
+        return False
+    expected = sign(key, secret, socket_id, channel, channel_data)
+    return hmac.compare_digest(auth.encode(), expected.encode())
diff --git a/fastapi_startkit/tests/broadcasting/conftest.py b/fastapi_startkit/tests/broadcasting/conftest.py
new file mode 100644
index 00000000..fb1da947
--- /dev/null
+++ b/fastapi_startkit/tests/broadcasting/conftest.py
@@ -0,0 +1,9 @@
+import pytest
+
+SECRET = "test-reverb-secret"
+
+
+@pytest.fixture(autouse=True)
+def reverb_secret(monkeypatch):
+    monkeypatch.setenv("REVERB_APP_SECRET", SECRET)
+    return SECRET
diff --git a/fastapi_startkit/tests/broadcasting/test_private_channel_auth.py b/fastapi_startkit/tests/broadcasting/test_private_channel_auth.py
new file mode 100644
index 00000000..13833b94
--- /dev/null
+++ b/fastapi_startkit/tests/broadcasting/test_private_channel_auth.py
@@ -0,0 +1,268 @@
+import asyncio
+import json
+import re
+
+import pytest
+from starlette.testclient import TestClient
+
+from fastapi_startkit.application import Application
+from fastapi_startkit.broadcasting.provider import ReverbProvider
+from fastapi_startkit.broadcasting.reverb.signature import sign
+from fastapi_startkit.support import Provider
+
+KEY = "local"
+
+
+class RouteProvider(Provider):
+    def register(self) -> None:
+        pass
+
+    def boot(self) -> None:
+        @self.app.fastapi.get("/ping")
+        async def ping():
+            return {"pong": True}
+
+
+class User:
+    def __init__(self, id):
+        self.id = id
+
+
+@pytest.fixture
+def base_path(tmp_path):
+    routes = tmp_path / "routes"
+    routes.mkdir()
+    (routes / "channels.py").write_text(
+        "from fastapi_startkit.broadcasting import channel\n"
+        "\n"
+        "@channel('private-orders.{order_id}')\n"
+        "async def authorize_orders(user, order_id: int):\n"
+        "    return user is not None and user.id == order_id\n"
+        "\n"
+        "@channel('presence-room.{room}')\n"
+        "async def authorize_room(user, room):\n"
+        "    return user is not None\n"
+    )
+    return tmp_path
+
+
+def make_client(base_path, providers, user=None):
+    app = Application(base_path=base_path, env="testing", providers=providers)
+
+    @app.fastapi.middleware("http")
+    async def attach_user(request, call_next):
+        request.state.user = user
+        return await call_next(request)
+
+    @app.fastapi.get("/after")
+    async def after():
+        return {"after": True}
+
+    return app, TestClient(app.fastapi)
+
+
+def connect(ws):
+    return json.loads(json.loads(ws.receive_text())["data"])["socket_id"]
+
+
+def subscribe(ws, channel, auth=None, channel_data=None):
+    data = {"channel": channel}
+    if auth is not None:
+        data["auth"] = auth
+    if channel_data is not None:
+        data["channel_data"] = channel_data
+    ws.send_text(json.dumps({"event": "pusher:subscribe", "data": data}))
+    return json.loads(ws.receive_text())
+
+
+def ping(ws):
+    ws.send_text(json.dumps({"event": "pusher:ping", "data": {}}))
+    return json.loads(ws.receive_text())
+
+
+class TestRouteOrder:
+    @pytest.mark.parametrize("reverb_first", [True, False])
+    def test_routes_and_auth_endpoint_reachable(self, base_path, reverb_first):
+        providers = [ReverbProvider, RouteProvider] if reverb_first else [RouteProvider, ReverbProvider]
+        _, client = make_client(base_path, providers, user=User(7))
+
+        assert client.get("/ping").json() == {"pong": True}
+        assert client.get("/after").json() == {"after": True}
+        response = client.post("/broadcasting/auth", data={"socket_id": "1.1", "channel_name": "private-orders.7"})
+        assert response.status_code == 200
+
+        with client.websocket_connect(f"/app/{KEY}") as ws:
+            assert connect(ws)
+
+
+class TestPrivateSubscription:
+    def test_unauthenticated_subscribe_is_rejected_and_receives_nothing(self, base_path):
+        app, client = make_client(base_path, [ReverbProvider])
+        server = app.make("reverb.server")
+
+        with client.websocket_connect(f"/app/{KEY}") as ws:
+            socket_id = connect(ws)
+            reply = subscribe(ws, "private-orders.7")
+            assert reply["event"] == "pusher:subscription_error"
+            assert reply["channel"] == "private-orders.7"
+            assert reply["data"]["status"] == 401
+            assert socket_id not in server.channels.subscribers("private-orders.7")
+
+            asyncio.run(server.broadcast_to_channel("private-orders.7", "OrderShipped", {"secret": 1}))
+            assert ping(ws)["event"] == "pusher:pong"
+
+    def test_forged_signature_is_rejected(self, base_path):
+        _, client = make_client(base_path, [ReverbProvider])
+        with client.websocket_connect(f"/app/{KEY}") as ws:
+            connect(ws)
+            assert subscribe(ws, "private-orders.7", auth=f"{KEY}:{'0' * 64}")["event"] == "pusher:subscription_error"
+            assert subscribe(ws, "private-orders.7", auth="ünïcode")["event"] == "pusher:subscription_error"
+            assert subscribe(ws, "private-orders.7", auth=123)["event"] == "pusher:subscription_error"
+
+    def test_signature_for_another_socket_is_rejected(self, base_path):
+        _, client = make_client(base_path, [ReverbProvider], user=User(7))
+        auth = client.post(
+            "/broadcasting/auth", data={"socket_id": "999.999", "channel_name": "private-orders.7"}
+        ).json()["auth"]
+        with client.websocket_connect(f"/app/{KEY}") as ws:
+            connect(ws)
+            assert subscribe(ws, "private-orders.7", auth=auth)["event"] == "pusher:subscription_error"
+
+    def test_unauthorized_user_cannot_obtain_signature(self, base_path):
+        _, client = make_client(base_path, [ReverbProvider], user=User(8))
+        response = client.post("/broadcasting/auth", data={"socket_id": "1.1", "channel_name": "private-orders.7"})
+        assert response.status_code == 403
+
+    def test_authorized_subscribe_receives_broadcast(self, base_path):
+        app, client = make_client(base_path, [ReverbProvider], user=User(7))
+        server = app.make("reverb.server")
+        with client.websocket_connect(f"/app/{KEY}") as ws:
+            socket_id = connect(ws)
+            auth = client.post(
+                "/broadcasting/auth", json={"socket_id": socket_id, "channel_name": "private-orders.7"}
+            ).json()["auth"]
+            assert subscribe(ws, "private-orders.7", auth=auth)["event"] == "pusher_internal:subscription_succeeded"
+
+            asyncio.run(server.broadcast_to_channel("private-orders.7", "OrderShipped", {"order_id": 7}))
+            message = json.loads(ws.receive_text())
+            assert message["event"] == "OrderShipped"
+            assert message["data"] == {"order_id": 7}
+
+
+class TestPresenceSubscription:
+    def test_presence_requires_signed_channel_data(self, base_path):
+        _, client = make_client(base_path, [ReverbProvider], user=User(7))
+        with client.websocket_connect(f"/app/{KEY}") as ws:
+            socket_id = connect(ws)
+            body = client.post(
+                "/broadcasting/auth", json={"socket_id": socket_id, "channel_name": "presence-room.1"}
+            ).json()
+            assert json.loads(body["channel_data"])["user_id"] == "7"
+
+            tampered = json.dumps({"user_id": "1", "user_info": {}})
+            assert subscribe(ws, "presence-room.1", body["auth"], tampered)["event"] == "pusher:subscription_error"
+            reply = subscribe(ws, "presence-room.1", body["auth"], body["channel_data"])
+            assert reply["event"] == "pusher_internal:subscription_succeeded"
+
+    def test_presence_requires_channel_data(self, base_path):
+        _, client = make_client(base_path, [ReverbProvider], user=User(7))
+        with client.websocket_connect(f"/app/{KEY}") as ws:
+            socket_id = connect(ws)
+            auth = client.post(
+                "/broadcasting/auth", data={"socket_id": socket_id, "channel_name": "presence-room.1"}
+            ).json()["auth"]
+            assert subscribe(ws, "presence-room.1", auth=auth)["event"] == "pusher:subscription_error"
+
+
+class TestSignatureConfusion:
+    @pytest.mark.parametrize("target", ["private-orders.1", "presence-room.vip"])
+    def test_anonymous_client_cannot_reuse_public_signature_for_protected_channel(self, base_path, target):
+        app, client = make_client(base_path, [ReverbProvider])
+        server = app.make("reverb.server")
+        with client.websocket_connect(f"/app/{KEY}") as ws:
+            socket_id = connect(ws)
+            response = client.post(
+                "/broadcasting/auth", data={"channel_name": "news", "socket_id": f"{socket_id}:{target}"}
+            )
+            auth = response.json().get("auth", f"{KEY}:{'0' * 64}")
+
+            assert subscribe(ws, target, auth=auth, channel_data="news")["event"] == "pusher:subscription_error"
+            assert socket_id not in server.channels.subscribers(target)
+            assert response.status_code in (403, 422)
+
+    def test_private_verification_ignores_client_channel_data(self, base_path, reverb_secret):
+        _, client = make_client(base_path, [ReverbProvider])
+        with client.websocket_connect(f"/app/{KEY}") as ws:
+            socket_id = connect(ws)
+            auth = sign(KEY, reverb_secret, socket_id, "private-orders.1", "news")
+            reply = subscribe(ws, "private-orders.1", auth=auth, channel_data="news")
+            assert reply["event"] == "pusher:subscription_error"
+
+    def test_private_signature_accepts_and_ignores_extra_channel_data(self, base_path):
+        _, client = make_client(base_path, [ReverbProvider], user=User(7))
+        with client.websocket_connect(f"/app/{KEY}") as ws:
+            socket_id = connect(ws)
+            body = client.post(
+                "/broadcasting/auth", data={"socket_id": socket_id, "channel_name": "private-orders.7"}
+            ).json()
+            assert "channel_data" not in body
+            reply = subscribe(ws, "private-orders.7", auth=body["auth"], channel_data="anything")
+            assert reply["event"] == "pusher_internal:subscription_succeeded"
+
+    def test_socket_ids_use_pusher_format(self, base_path):
+        _, client = make_client(base_path, [ReverbProvider])
+        with client.websocket_connect(f"/app/{KEY}") as ws:
+            assert re.fullmatch(r"\d+\.\d+", connect(ws))
+
+    @pytest.mark.parametrize("socket_id", ["1.1:private-orders.7", "abc", "1", "1.1\n", "1.1.1", " 1.1", "١.١"])
+    def test_auth_rejects_malformed_socket_id(self, base_path, socket_id):
+        _, client = make_client(base_path, [ReverbProvider], user=User(7))
+        response = client.post("/broadcasting/auth", data={"socket_id": socket_id, "channel_name": "private-orders.7"})
+        assert response.status_code == 422
+
+    @pytest.mark.parametrize("channel", ["private-orders.7:x", "private-orders 7", "private-" + "a" * 200, ""])
+    def test_auth_rejects_invalid_channel_name(self, base_path, channel):
+        _, client = make_client(base_path, [ReverbProvider], user=User(7))
+        response = client.post("/broadcasting/auth", data={"socket_id": "1.1", "channel_name": channel})
+        assert response.status_code == 422
+
+    @pytest.mark.parametrize("channel", ["news:private-orders.1", "private-orders.1:news", "a b"])
+    def test_subscribe_rejects_invalid_channel_name(self, base_path, channel):
+        app, client = make_client(base_path, [ReverbProvider])
+        server = app.make("reverb.server")
+        with client.websocket_connect(f"/app/{KEY}") as ws:
+            socket_id = connect(ws)
+            reply = subscribe(ws, channel)
+            assert reply["event"] == "pusher:subscription_error"
+            assert reply["data"]["status"] == 400
+            assert socket_id not in server.channels.subscribers(channel)
+
+    def test_auth_refuses_public_channels(self, base_path):
+        _, client = make_client(base_path, [ReverbProvider], user=User(7))
+        response = client.post("/broadcasting/auth", data={"socket_id": "1.1", "channel_name": "news"})
+        assert response.status_code == 403
+
+    @pytest.mark.parametrize("channel", ["private-open", "presence-open"])
+    def test_auth_requires_user_even_when_callback_allows(self, base_path, channel):
+        app, client = make_client(base_path, [ReverbProvider])
+        app.make("broadcasting").channel_registry.register(channel, lambda user: True)
+        response = client.post("/broadcasting/auth", data={"socket_id": "1.1", "channel_name": channel})
+        assert response.status_code == 403
+
+    def test_presence_refuses_user_without_id(self, base_path):
+        _, client = make_client(base_path, [ReverbProvider], user=User(None))
+        response = client.post("/broadcasting/auth", data={"socket_id": "1.1", "channel_name": "presence-room.1"})
+        assert response.status_code == 403
+
+
+class TestAuthRequestBody:
+    @pytest.mark.parametrize("body", [[1], "x", 1, None])
+    def test_rejects_non_object_json_body(self, base_path, body):
+        _, client = make_client(base_path, [ReverbProvider], user=User(7))
+        response = client.post("/broadcasting/auth", json=body)
+        assert response.status_code == 422
+
+    def test_rejects_invalid_json(self, base_path):
+        _, client = make_client(base_path, [ReverbProvider], user=User(7))
+        response = client.post("/broadcasting/auth", content=b"{not json", headers={"content-type": "application/json"})
+        assert response.status_code == 400
diff --git a/fastapi_startkit/tests/broadcasting/test_provider.py b/fastapi_startkit/tests/broadcasting/test_provider.py
index 8b3f0c10..d685f7fa 100644
--- a/fastapi_startkit/tests/broadcasting/test_provider.py
+++ b/fastapi_startkit/tests/broadcasting/test_provider.py
@@ -1,60 +1,75 @@
+import json
 from types import SimpleNamespace
 
+import pytest
+from starlette.testclient import TestClient
+
 from fastapi_startkit.application import Application
+from fastapi_startkit.broadcasting.config import ReverbConnectionConfig
 from fastapi_startkit.broadcasting.manager import BroadcastManager
 from fastapi_startkit.broadcasting.provider import ReverbProvider
 from fastapi_startkit.broadcasting.reverb.server import ReverbServer
+from fastapi_startkit.broadcasting.reverb.signature import sign
+
+from .conftest import SECRET
+
+
+class TestSecret:
+    def test_has_no_default_secret(self, monkeypatch):
+        monkeypatch.delenv("REVERB_APP_SECRET")
+        assert ReverbConnectionConfig().secret == ""
+
+    @pytest.mark.parametrize("secret", [None, "", "secret"])
+    def test_refuses_to_boot_outside_local_and_testing_without_real_secret(self, monkeypatch, tmp_path, secret):
+        monkeypatch.delenv("PYTEST_CURRENT_TEST")
+        if secret is None:
+            monkeypatch.delenv("REVERB_APP_SECRET")
+        else:
+            monkeypatch.setenv("REVERB_APP_SECRET", secret)
+        with pytest.raises(RuntimeError, match="REVERB_APP_SECRET"):
+            Application(base_path=tmp_path, env="production", providers=[ReverbProvider])
+
+    def test_boots_in_production_with_real_secret(self, monkeypatch, tmp_path):
+        monkeypatch.delenv("PYTEST_CURRENT_TEST")
+        app = Application(base_path=tmp_path, env="production", providers=[ReverbProvider])
+        assert app.make("reverb.server").secret == SECRET
+
+    @pytest.mark.parametrize("env", ["local", "testing"])
+    def test_boots_in_local_and_testing_without_secret(self, monkeypatch, tmp_path, env):
+        monkeypatch.delenv("PYTEST_CURRENT_TEST")
+        monkeypatch.delenv("REVERB_APP_SECRET")
+        app = Application(base_path=tmp_path, env=env, providers=[ReverbProvider])
+        assert app.make("reverb.server").secret == ""
+
+    def test_empty_secret_rejects_every_protected_subscription(self, monkeypatch, tmp_path):
+        monkeypatch.delenv("REVERB_APP_SECRET")
+        app = make_app(tmp_path)
+        register_rule(app, "private-room.{room}", lambda user, room: True)
+        app.bind("auth", SimpleNamespace(user=SimpleNamespace(id=1)))
+        client = client_for(app)
+        with client.websocket_connect("/app/local") as ws:
+            socket_id = json.loads(json.loads(ws.receive_text())["data"])["socket_id"]
+            assert post_auth(app, channel_name="private-room.7", socket_id=socket_id).status_code == 403
 
-
-_MISSING = object()
+            auth = sign("local", "", socket_id, "private-room.7")
+            ws.send_text(json.dumps({"event": "pusher:subscribe", "data": {"channel": "private-room.7", "auth": auth}}))
+            assert json.loads(ws.receive_text())["event"] == "pusher:subscription_error"
 
 
 def make_app(tmp_path):
     return Application(base_path=tmp_path, env="testing", providers=[ReverbProvider])
 
 
-def mount_auth(tmp_path):
-    """Register broadcasting and mount only the auth endpoint.
-
-    The full ``boot()`` mounts a catch-all WebSocket sub-app at ``""`` which
-    shadows the HTTP auth route, so exercise the auth handler on its own.
-    """
-    app = Application(base_path=tmp_path, env="testing", providers=[])
-    provider = ReverbProvider(app)
-    provider.register()
-    provider._mount_auth_endpoint()
-    return app
-
+def client_for(app):
+    return TestClient(app.fastapi)
 
-def auth_handler(app):
-    for route in app.fastapi.routes:
-        if getattr(route, "path", None) == "/broadcasting/auth":
-            return route.endpoint
-    raise AssertionError("auth route not mounted")
 
+def register_rule(app, pattern, callback):
+    app.make("broadcasting").channel_registry.register(pattern, callback)
 
-class FakeRequest:
-    """Stand-in for a Starlette Request.
 
-    ``from __future__ import annotations`` in the provider turns the handler's
-    ``Request`` annotation into an unresolved forward-ref, so FastAPI cannot
-    inject a real request through the test client. Calling the handler with
-    this fake exercises the same body directly.
-    """
-
-    def __init__(self, content_type, *, json_body=None, form_body=None, user=_MISSING):
-        self.headers = {"content-type": content_type}
-        self._json = json_body or {}
-        self._form = form_body or {}
-        self.state = SimpleNamespace()
-        if user is not _MISSING:
-            self.state.user = user
-
-    async def json(self):
-        return self._json
-
-    async def form(self):
-        return self._form
+def post_auth(app, **body):
+    return client_for(app).post("/broadcasting/auth", data=body)
 
 
 class TestRegister:
@@ -63,6 +78,10 @@ def test_binds_manager_and_server(self, tmp_path):
         assert isinstance(app.make("broadcasting"), BroadcastManager)
         assert isinstance(app.make("reverb.server"), ReverbServer)
 
+    def test_server_uses_configured_credentials(self, tmp_path, reverb_secret):
+        server = make_app(tmp_path).make("reverb.server")
+        assert (server.key, server.secret) == ("local", reverb_secret)
+
     def test_publishes_channels_stub(self, tmp_path):
         app = make_app(tmp_path)
         published = app.published_resources.get("broadcasting", {})
@@ -70,53 +89,99 @@ def test_publishes_channels_stub(self, tmp_path):
 
 
 class TestAuthEndpoint:
-    async def test_public_channel_allowed_json(self, tmp_path):
-        handler = auth_handler(mount_auth(tmp_path))
-        response = await handler(FakeRequest("application/json", json_body={"channel_name": "orders"}))
-        assert response.status_code == 200
-
-    async def test_private_channel_denied_without_rule(self, tmp_path):
-        handler = auth_handler(mount_auth(tmp_path))
-        response = await handler(FakeRequest("application/json", json_body={"channel_name": "private-secret"}))
+    def test_auth_route_is_reachable_after_full_boot(self, tmp_path):
+        response = post_auth(make_app(tmp_path), channel_name="private-secret", socket_id="1.1")
         assert response.status_code == 403
 
-    async def test_accepts_form_encoded_body(self, tmp_path):
-        handler = auth_handler(mount_auth(tmp_path))
-        response = await handler(
-            FakeRequest("application/x-www-form-urlencoded", form_body={"channel_name": "public-room"})
-        )
+    def test_routes_added_after_boot_are_not_shadowed(self, tmp_path):
+        app = make_app(tmp_path)
+        app.fastapi.get("/ping")(lambda: {"ok": True})
+        assert client_for(app).get("/ping").status_code == 200
+
+    def test_private_channel_returns_signed_auth(self, tmp_path):
+        app = make_app(tmp_path)
+        register_rule(app, "private-room.{room}", lambda user, room: True)
+        app.bind("auth", SimpleNamespace(user=SimpleNamespace(id=1)))
+        response = post_auth(app, channel_name="private-room.7", socket_id="123.456")
         assert response.status_code == 200
+        assert response.json() == {"auth": sign("local", SECRET, "123.456", "private-room.7")}
+
+    def test_accepts_json_body(self, tmp_path):
+        app = make_app(tmp_path)
+        register_rule(app, "private-room.{room}", lambda user, room: True)
+        app.bind("auth", SimpleNamespace(user=SimpleNamespace(id=1)))
+        response = client_for(app).post(
+            "/broadcasting/auth", json={"channel_name": "private-room.7", "socket_id": "123.456"}
+        )
+        assert response.json()["auth"] == sign("local", SECRET, "123.456", "private-room.7")
+
+    def test_presence_channel_signs_channel_data(self, tmp_path):
+        app = make_app(tmp_path)
+        register_rule(app, "presence-room", lambda user: True)
+        app.bind("auth", SimpleNamespace(user=SimpleNamespace(id=9)))
+        body = post_auth(app, channel_name="presence-room", socket_id="1.2").json()
+        assert body["channel_data"] == '{"user_id": "9"}'
+        assert body["auth"] == sign("local", SECRET, "1.2", "presence-room", body["channel_data"])
+
+    def test_denied_callback_returns_403(self, tmp_path):
+        app = make_app(tmp_path)
+        register_rule(app, "private-room.{room}", lambda user, room: False)
+        assert post_auth(app, channel_name="private-room.7", socket_id="1.1").status_code == 403
 
-    async def test_rejects_non_string_json_channel_name(self, tmp_path):
-        handler = auth_handler(mount_auth(tmp_path))
-        response = await handler(FakeRequest("application/json", json_body={"channel_name": 42}))
+    def test_requires_socket_id(self, tmp_path):
+        assert post_auth(make_app(tmp_path), channel_name="orders").status_code == 422
+
+    def test_rejects_non_string_json_channel_name(self, tmp_path):
+        response = client_for(make_app(tmp_path)).post(
+            "/broadcasting/auth", json={"channel_name": 42, "socket_id": "1.1"}
+        )
         assert response.status_code == 422
 
-    async def test_rejects_file_upload_as_channel_name(self, tmp_path):
-        handler = auth_handler(mount_auth(tmp_path))
-        response = await handler(
-            FakeRequest("multipart/form-data", form_body={"channel_name": SimpleNamespace(filename="x.txt")})
+    def test_rejects_file_upload_as_channel_name(self, tmp_path):
+        response = client_for(make_app(tmp_path)).post(
+            "/broadcasting/auth", data={"socket_id": "1.1"}, files={"channel_name": ("x.txt", b"x")}
         )
         assert response.status_code == 422
 
-    async def test_uses_request_state_user(self, tmp_path):
-        handler = auth_handler(mount_auth(tmp_path))
-        response = await handler(FakeRequest("application/json", json_body={"channel_name": "orders"}, user=object()))
-        assert response.status_code == 200
+    def test_uses_request_state_user(self, tmp_path):
+        app = make_app(tmp_path)
+        register_rule(app, "private-me", lambda user: user.id == 3)
+
+        @app.fastapi.middleware("http")
+        async def authenticate(request, call_next):
+            request.state.user = SimpleNamespace(id=3)
+            return await call_next(request)
 
-    async def test_resolves_user_from_callable_auth_service(self, tmp_path):
-        app = mount_auth(tmp_path)
+        assert post_auth(app, channel_name="private-me", socket_id="1.1").status_code == 200
+
+    def test_resolves_user_from_callable_auth_service(self, tmp_path):
+        app = make_app(tmp_path)
+        register_rule(app, "private-me", lambda user: user.id == 1)
         app.bind("auth", SimpleNamespace(user=lambda: SimpleNamespace(id=1)))
-        handler = auth_handler(app)
-        response = await handler(FakeRequest("application/json", json_body={"channel_name": "orders"}))
-        assert response.status_code == 200
+        assert post_auth(app, channel_name="private-me", socket_id="1.1").status_code == 200
 
-    async def test_resolves_user_from_attribute_auth_service(self, tmp_path):
-        app = mount_auth(tmp_path)
+    def test_resolves_user_from_attribute_auth_service(self, tmp_path):
+        app = make_app(tmp_path)
+        register_rule(app, "private-me", lambda user: user.id == 2)
         app.bind("auth", SimpleNamespace(user=SimpleNamespace(id=2)))
-        handler = auth_handler(app)
-        response = await handler(FakeRequest("application/json", json_body={"channel_name": "orders"}))
-        assert response.status_code == 200
+        assert post_auth(app, channel_name="private-me", socket_id="1.1").status_code == 200
+
+
+class TestWebsocketSubscription:
+    def test_private_subscribe_requires_signature_from_auth_endpoint(self, tmp_path):
+        app = make_app(tmp_path)
+        register_rule(app, "private-room.{room}", lambda user, room: True)
+        app.bind("auth", SimpleNamespace(user=SimpleNamespace(id=1)))
+        client = client_for(app)
+        with client.websocket_connect("/app/local") as ws:
+            socket_id = json.loads(json.loads(ws.receive_text())["data"])["socket_id"]
+
+            ws.send_text(json.dumps({"event": "pusher:subscribe", "data": {"channel": "private-room.7"}}))
+            assert json.loads(ws.receive_text())["event"] == "pusher:subscription_error"
+
+            auth = post_auth(app, channel_name="private-room.7", socket_id=socket_id).json()["auth"]
+            ws.send_text(json.dumps({"event": "pusher:subscribe", "data": {"channel": "private-room.7", "auth": auth}}))
+            assert json.loads(ws.receive_text())["event"] == "pusher_internal:subscription_succeeded"
 
 
 class TestChannelsFileAutoloading:
@@ -136,14 +201,11 @@ async def test_registered_channel_authorizer_is_used(self, tmp_path):
         assert await registry.authorize("private-room.7", None) is True
 
     def test_missing_channels_file_is_ignored(self, tmp_path):
-        # No routes/channels.py — boot must still succeed.
         app = make_app(tmp_path)
         assert isinstance(app.make("broadcasting"), BroadcastManager)
 
 
 class _RaisingFastapiApp:
-    """Minimal app whose ``.fastapi`` raises, mimicking FastAPI not installed."""
-
     base_path = "/nonexistent-base-path"
 
     @property
@@ -153,13 +215,10 @@ def fastapi(self):
 
 class TestMountingWithoutFastapi:
     def test_mount_websocket_skips_when_fastapi_unavailable(self):
-        provider = ReverbProvider(_RaisingFastapiApp())
-        provider._mount_websocket()  # returns without raising
+        ReverbProvider(_RaisingFastapiApp())._mount_websocket()
 
     def test_mount_auth_endpoint_skips_when_fastapi_unavailable(self):
-        provider = ReverbProvider(_RaisingFastapiApp())
-        provider._mount_auth_endpoint()  # returns without raising
+        ReverbProvider(_RaisingFastapiApp())._mount_auth_endpoint()
 
     def test_load_channels_file_skips_when_absent(self):
-        provider = ReverbProvider(_RaisingFastapiApp())
-        provider._load_channels_file()  # base_path has no routes/channels.py
+        ReverbProvider(_RaisingFastapiApp())._load_channels_file()
diff --git a/fastapi_startkit/tests/broadcasting/test_reverb_internals.py b/fastapi_startkit/tests/broadcasting/test_reverb_internals.py
new file mode 100644
index 00000000..2d559c63
--- /dev/null
+++ b/fastapi_startkit/tests/broadcasting/test_reverb_internals.py
@@ -0,0 +1,84 @@
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+from starlette.websockets import WebSocketDisconnect
+
+from fastapi_startkit.broadcasting.reverb.channel_manager import ChannelManager
+from fastapi_startkit.broadcasting.reverb.connection import ConnectionManager
+from fastapi_startkit.broadcasting.reverb.server import ReverbServer
+from fastapi_startkit.broadcasting.reverb.signature import sign, verify
+
+KEY = "local"
+SECRET = "unit-secret"
+SOCKET_ID = "123.456"
+
+
+def test_unsubscribe_from_unknown_channel_keeps_other_subscriptions():
+    channels = ChannelManager()
+    channels.subscribe("orders.1", "1.1")
+
+    channels.unsubscribe("missing", "1.1")
+
+    assert channels.subscribers("orders.1") == {"1.1"}
+    assert channels.subscribers("missing") == set()
+
+
+def test_connection_manager_retries_when_socket_id_is_taken(monkeypatch):
+    connections = ConnectionManager()
+    first, second = MagicMock(), MagicMock()
+    ids = iter(["1.1", "1.1", "2.2"])
+    monkeypatch.setattr(ConnectionManager, "_new_socket_id", staticmethod(lambda: next(ids)))
+
+    first_id = connections.add(first)
+    second_id = connections.add(second)
+
+    assert (first_id, second_id) == ("1.1", "2.2")
+    assert connections.get("1.1") is first
+    assert connections.get("2.2") is second
+
+
+@pytest.mark.parametrize(
+    ("socket_id", "channel"),
+    [
+        ("not-a-socket-id", "private-orders.1"),
+        (SOCKET_ID, "private-orders:1"),
+        (SOCKET_ID, "orders.1"),
+    ],
+)
+def test_verify_rejects_invalid_socket_id_or_unprotected_channel(socket_id, channel):
+    auth = sign(KEY, SECRET, socket_id, channel)
+
+    assert verify(KEY, SECRET, socket_id, channel, auth) is False
+
+
+def test_verify_accepts_valid_signature_for_private_channel():
+    channel = "private-orders.1"
+    auth = sign(KEY, SECRET, SOCKET_ID, channel)
+
+    assert verify(KEY, SECRET, SOCKET_ID, channel, auth) is True
+
+
+async def test_disconnect_while_handshaking_releases_connection():
+    server = ReverbServer(KEY, SECRET)
+    websocket = MagicMock()
+    websocket.accept = AsyncMock()
+    websocket.send_text = AsyncMock(side_effect=WebSocketDisconnect(1006))
+
+    await server.handle(websocket)
+
+    assert server.connections.all() == {}
+
+
+async def test_broadcast_skips_failing_socket_and_reaches_the_rest():
+    server = ReverbServer(KEY, SECRET)
+    broken, healthy = MagicMock(), MagicMock()
+    broken.send_text = AsyncMock(side_effect=RuntimeError("closed"))
+    healthy.send_text = AsyncMock()
+    for websocket in (broken, healthy):
+        socket_id = server.connections.add(websocket)
+        server.channels.subscribe("orders.1", socket_id)
+
+    await server.broadcast_to_channel("orders.1", "OrderShipped", {"order_id": 1})
+
+    broken.send_text.assert_awaited_once()
+    healthy.send_text.assert_awaited_once()
diff --git a/fastapi_startkit/tests/broadcasting/test_reverb_server.py b/fastapi_startkit/tests/broadcasting/test_reverb_server.py
index c0c95377..6d8f1b50 100644
--- a/fastapi_startkit/tests/broadcasting/test_reverb_server.py
+++ b/fastapi_startkit/tests/broadcasting/test_reverb_server.py
@@ -1,5 +1,6 @@
 import json
 import pytest
+from starlette.websockets import WebSocketDisconnect
 from starlette.testclient import TestClient
 from fastapi_startkit.broadcasting.reverb.server import ReverbServer
 
@@ -71,3 +72,74 @@ def test_socket_id_is_unique(client):
             data = json.loads(msg["data"])
             socket_ids.append(data["socket_id"])
     assert len(set(socket_ids)) == 3
+
+
+def _connect(ws):
+    return json.loads(json.loads(ws.receive_text())["data"])["socket_id"]
+
+
+def _ping(ws):
+    ws.send_text(json.dumps({"event": "pusher:ping", "data": {}}))
+    return json.loads(ws.receive_text())["event"]
+
+
+@pytest.mark.parametrize(
+    "frame",
+    [
+        "not json",
+        "[1, 2]",
+        "42",
+        json.dumps({"event": "pusher:subscribe", "data": "x"}),
+        json.dumps({"event": "pusher:subscribe", "data": {"channel": 1}}),
+        json.dumps({"event": 5}),
+    ],
+)
+def test_malformed_frame_replies_with_error_and_keeps_connection(client, frame):
+    with client.websocket_connect("/app/local") as ws:
+        _connect(ws)
+        ws.send_text(frame)
+        assert json.loads(ws.receive_text())["event"] == "pusher:error"
+        assert _ping(ws) == "pusher:pong"
+
+
+def test_binary_frame_replies_with_error(client):
+    with client.websocket_connect("/app/local") as ws:
+        _connect(ws)
+        ws.send_bytes(b"\x00\x01")
+        assert json.loads(ws.receive_text())["event"] == "pusher:error"
+        assert _ping(ws) == "pusher:pong"
+
+
+def test_oversized_frame_closes_connection(server, client):
+    with client.websocket_connect("/app/local") as ws:
+        _connect(ws)
+        ws.send_text("x" * (server.max_message_size + 1))
+        assert json.loads(ws.receive_text())["event"] == "pusher:error"
+        with pytest.raises(WebSocketDisconnect) as closed:
+            ws.receive_text()
+        assert closed.value.code == 1009
+    assert server.connections.all() == {}
+
+
+def test_connection_is_cleaned_up_after_malformed_frames(server, client):
+    with client.websocket_connect("/app/local") as ws:
+        socket_id = _connect(ws)
+        ws.send_text(json.dumps({"event": "pusher:subscribe", "data": {"channel": "orders.1"}}))
+        ws.receive_text()
+        ws.send_text("not json")
+        ws.receive_text()
+    assert socket_id not in server.connections.all()
+    assert socket_id not in server.channels.subscribers("orders.1")
+
+
+def test_connection_is_cleaned_up_when_handler_fails(server, client, monkeypatch):
+    async def explode(*args):
+        raise RuntimeError("boom")
+
+    monkeypatch.setattr(server, "_handle_message", explode)
+    with pytest.raises(RuntimeError):
+        with client.websocket_connect("/app/local") as ws:
+            _connect(ws)
+            ws.send_text(json.dumps({"event": "pusher:ping"}))
+            ws.receive_text()
+    assert server.connections.all() == {}