diff --git a/news/6932.feature.md b/news/6932.feature.md
new file mode 100644
index 00000000000..e2c80c4affb
--- /dev/null
+++ b/news/6932.feature.md
@@ -0,0 +1 @@
+The default client-server transport is now a plain WebSocket speaking a lightweight JSON event protocol, replacing Socket.IO. `python-socketio` moved to the optional `reflex[socketio]` extra and `socket.io-client` is only loaded by the frontend when configured. The Socket.IO transport remains available via `transport="socketio"` (websocket) or `transport="polling"` in `rxconfig.py`; apps passing a custom `sio` server to `rx.App` must set one of these and install the extra. Uvicorn is now fully optional: `AppHarness` serves tests with Granian's embedded server (native websocket support), and the uvicorn backend fallback requires the new `reflex[uvicorn]` extra, which brings the `websockets` protocol library uvicorn needs for the WebSocket transport.
diff --git a/packages/reflex-base/news/6932.bugfix.md b/packages/reflex-base/news/6932.bugfix.md
new file mode 100644
index 00000000000..b37302c397d
--- /dev/null
+++ b/packages/reflex-base/news/6932.bugfix.md
@@ -0,0 +1 @@
+`enqueue_stream_delta` (used by streaming uploads) no longer registers its event future under the root context's txid. Previously, any unrelated event enqueued while a streamed upload's future still lingered was spuriously attached to it as a child, failing with "Cannot add a child to an EventFuture that is already done" once the upload finished — a latent race that Socket.IO's extra latency usually hid.
diff --git a/packages/reflex-base/news/6932.feature.md b/packages/reflex-base/news/6932.feature.md
new file mode 100644
index 00000000000..4f4ed3e1661
--- /dev/null
+++ b/packages/reflex-base/news/6932.feature.md
@@ -0,0 +1 @@
+The default client transport is a plain WebSocket speaking JSON `[event_name, payload]` frames. `config.transport` gains a `"socketio"` value; socket.io-client is only loaded in the browser when `"socketio"` or `"polling"` is configured.
diff --git a/packages/reflex-base/src/reflex_base/.templates/web/utils/helpers/websocket.js b/packages/reflex-base/src/reflex_base/.templates/web/utils/helpers/websocket.js
new file mode 100644
index 00000000000..dc09fa8f517
--- /dev/null
+++ b/packages/reflex-base/src/reflex_base/.templates/web/utils/helpers/websocket.js
@@ -0,0 +1,332 @@
+// Plain WebSocket transport speaking the Reflex JSON event protocol: each
+// frame is a JSON array `[event_name, payload]`. Mirrors the socket.io-client
+// surface that state.js and upload.js rely on: connected, connect(),
+// disconnect(), emit(), on(), io.opts.query, and _callbacks.
+
+// Protocol-level message names (must match reflex/event_namespace.py).
+const HANDSHAKE_MESSAGE = "_handshake";
+const PING_MESSAGE = "_ping";
+const PONG_MESSAGE = "_pong";
+
+// Python's json.dumps emits bare Infinity/-Infinity/NaN tokens (invalid JSON).
+// Rewrite them outside string literals so JSON.parse accepts the payload.
+// 1e999 / -1e999 overflow to ±Infinity; NaN has no JSON literal, so it is
+// swapped for a sentinel string and revived back to NaN after parsing.
+// The alternation matches whole string literals first (passed through unchanged),
+// guaranteeing bare-token matches only land in numeric positions.
+const NAN_SENTINEL = "__reflex_nan__";
+const NON_FINITE_FLOAT_RE = /"(?:[^"\\]|\\.)*"|-?\bInfinity\b|\bNaN\b/g;
+const NON_FINITE_REPLACEMENTS = {
+  Infinity: "1e999",
+  "-Infinity": "-1e999",
+  NaN: `"${NAN_SENTINEL}"`,
+};
+const rewriteBareNonFiniteFloats = (str) =>
+  str.replace(NON_FINITE_FLOAT_RE, (match) =>
+    match[0] === '"' ? match : NON_FINITE_REPLACEMENTS[match],
+  );
+const reviveNonFiniteFloats = (_k, v) => (v === NAN_SENTINEL ? NaN : v);
+
+/**
+ * JSON.stringify replacer that sends undefined fields as null instead of
+ * removing them. Also assigned as the socket.io encoder replacer.
+ * @param _k The key being serialized.
+ * @param v The value being serialized.
+ * @returns The value to serialize.
+ */
+export const undefinedToNull = (_k, v) => (v === undefined ? null : v);
+
+/**
+ * Parse JSON, tolerating bare non-finite float tokens.
+ * @param text The text to parse.
+ * @param fallback The value to return if the text is unparsable.
+ * @returns The parsed value, or the fallback.
+ */
+export const parseJsonLenient = (text, fallback) => {
+  try {
+    return JSON.parse(text);
+  } catch (e) {
+    try {
+      return JSON.parse(
+        rewriteBareNonFiniteFloats(text),
+        reviveNonFiniteFloats,
+      );
+    } catch (e2) {
+      return fallback;
+    }
+  }
+};
+
+/**
+ * Serialize an outgoing frame.
+ * @param frame The frame array to serialize.
+ * @returns The JSON string.
+ */
+const stringifyFrame = (frame) => JSON.stringify(frame, undefinedToNull);
+
+export class ReflexWebSocket {
+  /**
+   * Create the transport and start connecting.
+   * @param url The http(s) endpoint URL of the backend event route.
+   * @param opts Options: `query` (object) and `protocols` (subprotocol list).
+   */
+  constructor(url, opts) {
+    this._url = new URL(url);
+    // Exposed as io.opts for socket.io API compatibility: state.js refreshes
+    // io.opts.query before reconnecting.
+    this.io = { opts };
+    this.connected = false;
+    // upload.js reads socket._callbacks.$event directly.
+    this._callbacks = {};
+    this._ws = null;
+    // Frames emitted while disconnected, flushed on (re)connect.
+    this._sendQueue = [];
+    this._watchdogTimer = null;
+    // Heartbeat window: 145 seconds (25s ping interval + 120s ping timeout)
+    // in ms; refined by the server handshake.
+    this._watchdogMs = (25 + 120) * 1000;
+    // Give up after 20 seconds on a dial that neither opens nor errors, so
+    // a connect_error always fires and retries proceed.
+    this._connectTimeoutMs = 20 * 1000;
+    this._connectTimer = null;
+    this._closeReason = null;
+    // Network emulation and OS offline do not interrupt established
+    // websockets, so treat the browser's offline event as a disconnect.
+    // Localhost connections keep working offline.
+    this._offlineListener = null;
+    if (
+      typeof addEventListener === "function" &&
+      this._url.hostname !== "localhost"
+    ) {
+      this._offlineListener = () => this._onOffline();
+      addEventListener("offline", this._offlineListener, false);
+    }
+    this.connect();
+  }
+
+  /**
+   * Remove registered handlers. With no arguments, also releases the global
+   * offline listener (transport disposal).
+   * @param event The event name; omit to remove all handlers.
+   * @param fn The handler to remove; omit to remove all handlers for event.
+   */
+  off(event, fn) {
+    if (event === undefined) {
+      this._callbacks = {};
+      if (this._offlineListener) {
+        removeEventListener("offline", this._offlineListener, false);
+        this._offlineListener = null;
+      }
+      return;
+    }
+    if (fn === undefined) {
+      delete this._callbacks["$" + event];
+      return;
+    }
+    const handlers = this._callbacks["$" + event];
+    const ix = handlers ? handlers.indexOf(fn) : -1;
+    if (ix !== -1) {
+      handlers.splice(ix, 1);
+    }
+  }
+
+  /**
+   * Register a handler for an event.
+   * @param event The event name.
+   * @param fn The handler function.
+   */
+  on(event, fn) {
+    (this._callbacks["$" + event] ??= []).push(fn);
+  }
+
+  /**
+   * Invoke the registered handlers for a local event.
+   * @param event The event name.
+   * @param args The handler arguments.
+   */
+  _emitLocal(event, ...args) {
+    for (const fn of this._callbacks["$" + event] ?? []) {
+      fn(...args);
+    }
+  }
+
+  /**
+   * Open the websocket connection if not already open or connecting.
+   */
+  connect() {
+    if (this._ws && this._ws.readyState <= WebSocket.OPEN) {
+      // CONNECTING (0) or OPEN (1): already dialing or connected.
+      return;
+    }
+    const url = new URL(this._url);
+    // Secure endpoints (https or already-wss) stay secure.
+    url.protocol =
+      url.protocol === "https:" || url.protocol === "wss:" ? "wss:" : "ws:";
+    url.search = new URLSearchParams(this.io.opts.query ?? {}).toString();
+    this._closeReason = null;
+    const ws = new WebSocket(url, this.io.opts.protocols);
+    this._ws = ws;
+    this._connectTimer = setTimeout(() => {
+      if (this._ws === ws && !this.connected) {
+        ws.close();
+      }
+    }, this._connectTimeoutMs);
+    ws.onmessage = (msg) => {
+      if (this._ws === ws) {
+        // Ignore stragglers from a superseded connection.
+        this._onMessage(msg.data);
+      }
+    };
+    ws.onclose = (event) => {
+      if (this._ws !== ws) {
+        // A newer connection or an explicit disconnect() superseded this one.
+        return;
+      }
+      this._clearConnectTimer();
+      this._clearWatchdog();
+      const wasConnected = this.connected;
+      this.connected = false;
+      if (!wasConnected) {
+        // Never handshaked: this was a failed connection attempt.
+        this._emitLocal(
+          "connect_error",
+          new Error("websocket connection failed"),
+        );
+      } else {
+        this._emitLocal("disconnect", this._closeReason ?? "transport close", {
+          code: event.code,
+          reason: event.reason,
+        });
+      }
+    };
+  }
+
+  /**
+   * Close the connection deliberately (reason "io client disconnect").
+   */
+  disconnect() {
+    this._teardown("io client disconnect", undefined);
+  }
+
+  /**
+   * Report the disconnect immediately when the browser goes offline.
+   */
+  _onOffline() {
+    if (this.connected) {
+      this._teardown("transport close", {
+        description: "network connection lost",
+      });
+    }
+  }
+
+  /**
+   * Tear down the current connection, reporting the disconnect synchronously
+   * (onclose may never fire during page unload or while offline).
+   * @param reason The disconnect reason to report.
+   * @param details The disconnect details to report.
+   */
+  _teardown(reason, details) {
+    this._clearConnectTimer();
+    this._clearWatchdog();
+    const ws = this._ws;
+    if (!ws) {
+      return;
+    }
+    // Detach so the onclose handler does not double-report.
+    this._ws = null;
+    if (this.connected) {
+      this.connected = false;
+      this._emitLocal("disconnect", reason, details);
+    }
+    if (ws.readyState <= WebSocket.OPEN) {
+      ws.onclose = null;
+      ws.onmessage = null;
+      ws.close(1000);
+    }
+  }
+
+  /**
+   * Send an event to the backend, buffering while disconnected.
+   * @param event The event name.
+   * @param data The event payload.
+   */
+  emit(event, data) {
+    const frame = stringifyFrame([event, data]);
+    if (this.connected && this._ws?.readyState === WebSocket.OPEN) {
+      this._ws.send(frame);
+    } else {
+      this._sendQueue.push(frame);
+    }
+  }
+
+  /**
+   * Handle one incoming frame.
+   * @param text The raw frame text.
+   */
+  _onMessage(text) {
+    const message = parseJsonLenient(text, undefined);
+    if (!Array.isArray(message)) {
+      console.error("Failed to parse websocket message", text);
+      return;
+    }
+    const [event, payload] = message;
+    if (event === PING_MESSAGE) {
+      // The server pings every interval regardless of traffic, so resetting
+      // the watchdog only here avoids timer churn per data message.
+      this._resetWatchdog();
+      this._ws?.send(stringifyFrame([PONG_MESSAGE]));
+      return;
+    }
+    if (event === HANDSHAKE_MESSAGE) {
+      // Application-level liveness confirmed; adopt the server's heartbeat
+      // settings (sent in seconds, converted to ms) for the connection
+      // watchdog.
+      this._clearConnectTimer();
+      this._watchdogMs = (payload.ping_interval + payload.ping_timeout) * 1000;
+      this._resetWatchdog();
+      this.connected = true;
+      const queue = this._sendQueue;
+      this._sendQueue = [];
+      for (const frame of queue) {
+        this._ws.send(frame);
+      }
+      this._emitLocal("connect");
+      return;
+    }
+    this._emitLocal(event, payload);
+  }
+
+  /**
+   * (Re)arm the dead-connection watchdog; fires when no message (heartbeat
+   * included) arrives within the server's ping interval + timeout.
+   */
+  _resetWatchdog() {
+    this._clearWatchdog();
+    this._watchdogTimer = setTimeout(() => {
+      if (this._ws && this.connected) {
+        this._closeReason = "ping timeout";
+        this._ws.close();
+      }
+    }, this._watchdogMs);
+  }
+
+  /**
+   * Cancel the dead-connection watchdog.
+   */
+  _clearWatchdog() {
+    if (this._watchdogTimer) {
+      clearTimeout(this._watchdogTimer);
+      this._watchdogTimer = null;
+    }
+  }
+
+  /**
+   * Cancel the connect timeout.
+   */
+  _clearConnectTimer() {
+    if (this._connectTimer) {
+      clearTimeout(this._connectTimer);
+      this._connectTimer = null;
+    }
+  }
+}
diff --git a/packages/reflex-base/src/reflex_base/.templates/web/utils/state.js b/packages/reflex-base/src/reflex_base/.templates/web/utils/state.js
index 8ba6d00509c..1d89d06df65 100644
--- a/packages/reflex-base/src/reflex_base/.templates/web/utils/state.js
+++ b/packages/reflex-base/src/reflex_base/.templates/web/utils/state.js
@@ -1,5 +1,4 @@
 // State management for Reflex web apps.
-import io from "socket.io-client";
 import env from "$/env.json";
 import reflexEnvironment from "$/reflex.json";
 import Cookies from "universal-cookie";
@@ -20,6 +19,11 @@ import {
 import debounce from "$/utils/helpers/debounce";
 import throttle from "$/utils/helpers/throttle";
 import { uploadFiles } from "$/utils/helpers/upload";
+import {
+  ReflexWebSocket,
+  parseJsonLenient,
+  undefinedToNull,
+} from "$/utils/helpers/websocket";
 
 // Endpoint URLs.
 const EVENTURL = env.EVENT;
@@ -471,25 +475,6 @@ const resolveSocket = (socket) => {
   return socket?.current ?? socket;
 };
 
-// Python's json.dumps emits bare Infinity/-Infinity/NaN tokens (invalid JSON).
-// Rewrite them outside string literals so JSON.parse accepts the payload.
-// 1e999 / -1e999 overflow to ±Infinity; NaN has no JSON literal, so it is
-// swapped for a sentinel string and revived back to NaN after parsing.
-// The alternation matches whole string literals first (passed through unchanged),
-// guaranteeing bare-token matches only land in numeric positions.
-const NAN_SENTINEL = "__reflex_nan__";
-const NON_FINITE_FLOAT_RE = /"(?:[^"\\]|\\.)*"|-?\bInfinity\b|\bNaN\b/g;
-const NON_FINITE_REPLACEMENTS = {
-  Infinity: "1e999",
-  "-Infinity": "-1e999",
-  NaN: `"${NAN_SENTINEL}"`,
-};
-const rewriteBareNonFiniteFloats = (str) =>
-  str.replace(NON_FINITE_FLOAT_RE, (match) =>
-    match[0] === '"' ? match : NON_FINITE_REPLACEMENTS[match],
-  );
-const reviveNonFiniteFloats = (_k, v) => (v === NAN_SENTINEL ? NaN : v);
-
 /**
  * Queue events to be processed and trigger processing of queue.
  * @param events Array of events to queue.
@@ -577,6 +562,8 @@ export const connect = async (
   navigate,
   params,
 ) => {
+  // Connecting (again) revokes a pending unmount cancellation.
+  socket.cancelConnect = false;
   // Socket already allocated, just reconnect it if needed.
   if (socket.current) {
     if (!socket.current.connected) {
@@ -584,37 +571,52 @@ export const connect = async (
     }
     return;
   }
+  // Another connect() call may be awaiting the socket.io-client import;
+  // don't create a second transport.
+  if (socket.connecting) {
+    return;
+  }
+  socket.connecting = true;
 
   // Get backend URL object from the endpoint.
   const endpoint = getBackendURL(EVENTURL);
   const on_hydrated_queue = [];
 
   // Create the socket.
-  socket.current = io(endpoint.href, {
-    path: endpoint["pathname"],
-    transports: transports,
-    protocols: [reflexEnvironment.version],
-    autoUnref: false,
-    query: { token: getToken() },
-    reconnection: false, // Reconnection will be handled manually.
-  });
-  socket.current.wait_connect = !socket.current.connected;
-  // Ensure undefined fields in events are sent as null instead of removed
-  socket.current.io.encoder.replacer = (k, v) => (v === undefined ? null : v);
-  socket.current.io.decoder.tryParse = (str) => {
-    try {
-      return JSON.parse(str);
-    } catch (e) {
-      try {
-        return JSON.parse(
-          rewriteBareNonFiniteFloats(str),
-          reviveNonFiniteFloats,
-        );
-      } catch (e2) {
-        return false;
+  const transport = transports[0];
+  try {
+    if (transport === "websocket") {
+      // Default transport: plain WebSocket speaking the Reflex event protocol.
+      socket.current = new ReflexWebSocket(endpoint.href, {
+        query: { token: getToken() },
+        protocols: [reflexEnvironment.version],
+      });
+    } else {
+      // Socket.IO transport ("socketio" over websocket, or "polling"); the
+      // client library is only loaded when this transport is configured.
+      const { default: io } = await import("socket.io-client");
+      if (socket.cancelConnect) {
+        // The event loop unmounted while the import was pending.
+        return;
       }
+      socket.current = io(endpoint.href, {
+        path: endpoint["pathname"],
+        transports: [transport === "socketio" ? "websocket" : transport],
+        protocols: [reflexEnvironment.version],
+        autoUnref: false,
+        query: { token: getToken() },
+        reconnection: false, // Reconnection will be handled manually.
+      });
+      // Ensure undefined fields in events are sent as null instead of removed
+      socket.current.io.encoder.replacer = undefinedToNull;
+      // The decoder API expects false (not undefined) for unparsable input.
+      socket.current.io.decoder.tryParse = (str) =>
+        parseJsonLenient(str, false);
     }
-  };
+  } finally {
+    socket.connecting = false;
+  }
+  socket.current.wait_connect = !socket.current.connected;
   // Set up a reconnect helper function
   socket.current.reconnect = () => {
     if (
@@ -789,6 +791,9 @@ export const connect = async (
     window.sessionStorage.setItem(TOKEN_KEY, new_token);
   });
 
+  // Track the handler on the ref so unmount cleanup can remove it; a
+  // surviving listener would resurrect a transport for the unmounted hook.
+  socket.visibilityHandler = checkVisibility;
   document.addEventListener("visibilitychange", checkVisibility);
 };
 
@@ -1113,6 +1118,15 @@ export const useEventLoop = (
     // Cleanup function.
     return () => {
       mounted.current = false;
+      // Abort a connect() that is still awaiting the socket.io-client import.
+      socket.cancelConnect = true;
+      if (socket.visibilityHandler) {
+        document.removeEventListener(
+          "visibilitychange",
+          socket.visibilityHandler,
+        );
+        socket.visibilityHandler = null;
+      }
       if (socket.current) {
         socket.current.disconnect();
         socket.current.off();
diff --git a/packages/reflex-base/src/reflex_base/config.py b/packages/reflex-base/src/reflex_base/config.py
index b15533da731..c2f62c830a7 100644
--- a/packages/reflex-base/src/reflex_base/config.py
+++ b/packages/reflex-base/src/reflex_base/config.py
@@ -189,7 +189,7 @@ class BaseConfig:
         hydrate_fallback: Function returning the component shown while the page is hydrating (React Router's HydrateFallback), used when App.hydrate_fallback is not set. Formatted such that `from path_0.path_1... import path[-1]`, and calling it with no arguments would work. For example, "my_app.components.loading".
         plugins: List of plugins to use in the app.
         disable_plugins: List of plugin types to disable in the app.
-        transport: The transport method for client-server communication.
+        transport: The transport for client-server communication: "websocket" (plain WebSocket, default), or "socketio"/"polling" (Socket.IO; requires the reflex[socketio] extra).
     """
 
     app_name: str
@@ -273,7 +273,7 @@ class BaseConfig:
 
     disable_plugins: list[type[Plugin]] = dataclasses.field(default_factory=list)
 
-    transport: Literal["websocket", "polling"] = "websocket"
+    transport: Literal["websocket", "socketio", "polling"] = "websocket"
 
     # Whether to skip plugin checks.
     _skip_plugins_checks: bool = dataclasses.field(default=False, repr=False)
diff --git a/packages/reflex-base/src/reflex_base/event/processor/event_processor.py b/packages/reflex-base/src/reflex_base/event/processor/event_processor.py
index 71625ebc427..702ee075ca6 100644
--- a/packages/reflex-base/src/reflex_base/event/processor/event_processor.py
+++ b/packages/reflex-base/src/reflex_base/event/processor/event_processor.py
@@ -27,8 +27,8 @@
 logger = logging.getLogger(__name__)
 
 if TYPE_CHECKING:
-    from reflex.app import EventNamespace
     from reflex.event import Event, EventSpec
+    from reflex.event_namespace import BaseEventNamespace
 
 if hasattr(asyncio, "QueueShutDown"):
 
@@ -133,7 +133,7 @@ def configure(
         self,
         *,
         state_manager: StateManager | None = None,
-        event_namespace: EventNamespace | None = None,
+        event_namespace: BaseEventNamespace | None = None,
     ) -> Self:
         """Set up the event processor.
 
@@ -473,9 +473,10 @@ async def _emit_delta_impl(
         task_future = await self.enqueue(
             token,
             event,
+            # Fork for a fresh txid: reusing the root txid would register this
+            # future under it, attaching unrelated events as children.
             ev_ctx=dataclasses.replace(
-                self._root_context,
-                token=token,
+                self._root_context.fork(token=token),
                 emit_delta_impl=_emit_delta_impl,
             ),
         )
diff --git a/packages/reflex-base/src/reflex_base/utils/console.py b/packages/reflex-base/src/reflex_base/utils/console.py
index b8a5985322f..ea22b5069f7 100644
--- a/packages/reflex-base/src/reflex_base/utils/console.py
+++ b/packages/reflex-base/src/reflex_base/utils/console.py
@@ -291,11 +291,15 @@ def _exclude_paths_from_frame_info() -> list[Path]:
 
     import click
     import granian
-    import socketio
     import typing_extensions
 
     import reflex_base
 
+    try:
+        import socketio
+    except ImportError:
+        socketio = None
+
     try:
         import reflex as rx
     except ImportError:
diff --git a/packages/reflex-base/src/reflex_base/utils/log.py b/packages/reflex-base/src/reflex_base/utils/log.py
index c7c17e10bf7..95a26a54f54 100644
--- a/packages/reflex-base/src/reflex_base/utils/log.py
+++ b/packages/reflex-base/src/reflex_base/utils/log.py
@@ -641,11 +641,15 @@ def _exclude_paths_from_frame_info() -> list[Path]:
 
     import click
     import granian
-    import socketio
     import typing_extensions
 
     import reflex_base
 
+    try:
+        import socketio
+    except ImportError:
+        socketio = None
+
     try:
         import reflex as rx
     except ImportError:
diff --git a/pyproject.toml b/pyproject.toml
index 2638de9c41b..45f1b602b51 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -21,12 +21,11 @@ keywords = ["web", "framework"]
 requires-python = ">=3.10,<4.0"
 dependencies = [
   "click >=8.2",
-  "granian[reload] >=2.7.4",
+  "granian[reload] >=2.8.1",
   "httpx >=0.26,<1.0",
   "packaging >=24.2,<27",
   "psutil >=7.0.0,<8.0; sys_platform == 'win32'",
   "python-multipart >=0.0.32,<1.0",
-  "python-socketio >=5.12.0,<6.0",
   "redis >=6.4,<8.0",
   "rich >=13,<16",
   "starlette >=1.3.1",
@@ -66,6 +65,9 @@ db = [
   "sqlmodel >=0.0.24,<0.1",
 ]
 pydantic = ["reflex-base[pydantic]"]
+socketio = ["python-socketio >=5.12.0,<6.0"]
+# The uvicorn backend: server, websocket support, and the posix prod runner.
+uvicorn = ["uvicorn >=0.20.0", "websockets >=13.0", "gunicorn >=23.0"]
 
 [project.urls]
 homepage = "https://reflex.dev"
@@ -105,6 +107,7 @@ dev = [
   "pytest-split",
   "pytest",
   "python-dotenv",
+  "python-socketio",
   "pyyaml",
   "reflex-docgen",
   "reflex-release",
diff --git a/reflex/app.py b/reflex/app.py
index 63ec53a4c75..a1e9b659f7a 100644
--- a/reflex/app.py
+++ b/reflex/app.py
@@ -13,9 +13,7 @@
 import logging
 import operator
 import sys
-import time
 import traceback
-import urllib.parse
 from collections.abc import (
     AsyncIterator,
     Callable,
@@ -25,7 +23,6 @@
     Sequence,
 )
 from contextvars import Token
-from types import SimpleNamespace
 from typing import TYPE_CHECKING, Any, overload
 
 from reflex_base import constants
@@ -33,21 +30,14 @@
 from reflex_base.config import get_config, reload_config
 from reflex_base.context.base import BaseContext
 from reflex_base.environment import environment
-from reflex_base.event import (
-    _EVENT_FIELDS,
-    Event,
-    EventSpec,
-    EventType,
-    IndividualEventType,
-    noop,
-)
+from reflex_base.event import Event, EventSpec, EventType, IndividualEventType, noop
 from reflex_base.event.context import EventContext
 from reflex_base.event.processor import BaseStateEventProcessor, EventProcessor
 from reflex_base.registry import RegistrationContext
 from reflex_base.telemetry_context import CompileTrigger, TelemetryContext
 from reflex_base.utils import memo_paths
 from reflex_base.utils.imports import ImportVar
-from reflex_base.utils.types import ASGIApp, Message, Receive, Scope, Send
+from reflex_base.utils.types import ASGIApp, Receive, Scope, Send
 from reflex_components_core.base.error_boundary import ErrorBoundary
 from reflex_components_core.base.fragment import Fragment
 from reflex_components_core.core.banner import (
@@ -58,12 +48,11 @@
 from reflex_components_core.core.breakpoints import set_breakpoints
 from reflex_components_core.core.sticky import sticky
 from reflex_components_sonner.toast import toast
-from socketio import ASGIApp as EngineIOApp
-from socketio import AsyncNamespace, AsyncServer
 from starlette.applications import Starlette
 from starlette.middleware import cors
 from starlette.requests import Request
 from starlette.responses import JSONResponse, Response
+from starlette.routing import WebSocketRoute
 from starlette.staticfiles import StaticFiles
 from typing_extensions import Unpack
 
@@ -73,7 +62,7 @@
 from reflex.app_mixins import AppMixin, LifespanMixin, MiddlewareMixin
 from reflex.compiler import compiler
 from reflex.compiler.compiler import readable_name_from_component
-from reflex.istate.data import RouterData
+from reflex.event_namespace import BaseEventNamespace, WebsocketEventNamespace
 from reflex.istate.manager import StateManager, StateModificationContext
 from reflex.istate.manager.token import BaseStateToken
 from reflex.route import (
@@ -98,7 +87,6 @@
     should_prerender_routes,
 )
 from reflex.utils.misc import run_in_thread
-from reflex.utils.token_manager import RedisTokenManager, TokenManager
 
 logger = logging.getLogger(__name__)
 
@@ -111,11 +99,15 @@
     from reflex_base.plugins import Plugin
     from reflex_base.plugins.base import AddPageProtocol
     from reflex_base.vars import Var
+    from socketio import AsyncServer
 
     # Define custom types.
     ComponentCallable = Callable[[], Component | tuple[Component, ...] | str | Var]
 else:
     ComponentCallable = Callable[[], Component | tuple[Component, ...] | str]
+    # Runtime placeholder so annotations resolve without the optional
+    # python-socketio dependency installed.
+    AsyncServer = Any
 
 Reducer = Callable[[Event], Coroutine[Any, Any, StateUpdate]]
 
@@ -448,7 +440,7 @@ class App(MiddlewareMixin, LifespanMixin):
     admin_dash: AdminDash | None = None
 
     # The async server name space.
-    _event_namespace: EventNamespace | None = None
+    _event_namespace: BaseEventNamespace | None = None
 
     # The processor queue for handling events.
     _event_processor: EventProcessor | None = None
@@ -478,7 +470,7 @@ class App(MiddlewareMixin, LifespanMixin):
     ) = None
 
     @property
-    def event_namespace(self) -> EventNamespace | None:
+    def event_namespace(self) -> BaseEventNamespace | None:
         """Get the event namespace.
 
         Returns:
@@ -552,7 +544,8 @@ def _setup_state(self) -> None:
         """Set up the state for the app.
 
         Raises:
-            RuntimeError: If the socket server is invalid.
+            RuntimeError: If the socket server is invalid, or the Socket.IO
+                transport is requested without python-socketio installed.
         """
         if not self._state:
             return
@@ -562,76 +555,46 @@ def _setup_state(self) -> None:
         # Set up the state manager.
         self._state_manager = StateManager.create()
 
-        # Set up the Socket.IO AsyncServer.
-        if not self.sio:
-            self.sio = AsyncServer(
-                async_mode="asgi",
-                cors_allowed_origins=(
-                    (
-                        "*"
-                        if config.cors_allowed_origins == ("*",)
-                        else list(config.cors_allowed_origins)
-                    )
-                    if config.transport == "websocket"
-                    else []
-                ),
-                cors_credentials=config.transport == "websocket",
-                max_http_buffer_size=environment.REFLEX_SOCKET_MAX_HTTP_BUFFER_SIZE.get(),
-                ping_interval=environment.REFLEX_SOCKET_INTERVAL.get(),
-                ping_timeout=environment.REFLEX_SOCKET_TIMEOUT.get(),
-                json=SimpleNamespace(
-                    dumps=staticmethod(format.json_dumps),
-                    loads=staticmethod(json.loads),
-                ),
-                allow_upgrades=False,
-                transports=[config.transport],
-            )
-        elif getattr(self.sio, "async_mode", "") != "asgi":
-            msg = f"Custom `sio` must use `async_mode='asgi'`, not '{self.sio.async_mode}'."
-            raise RuntimeError(msg)
-
-        # Create the socket app. Note event endpoint constant replaces the default 'socket.io' path.
-        socket_app = EngineIOApp(self.sio, socketio_path="")
         namespace = config.get_event_namespace()
+        event_path = config.prepend_backend_path(str(constants.Endpoint.EVENT))
 
-        # Create the event namespace and attach the main app. Not related to any paths.
-        self._event_namespace = EventNamespace(namespace, self)
-
-        # Register the event namespace with the socket.
-        self.sio.register_namespace(self.event_namespace)
-        # Mount the socket app with the API.
-        if self._api:
-
-            class HeaderMiddleware:
-                def __init__(self, app: ASGIApp):
-                    self.app = app
-
-                async def __call__(self, scope: Scope, receive: Receive, send: Send):
-                    original_send = send
-
-                    async def modified_send(message: Message):
-                        if message["type"] == "websocket.accept":
-                            if scope.get("subprotocols"):
-                                # The following *does* say "subprotocol" instead of "subprotocols", intentionally.
-                                message["subprotocol"] = scope["subprotocols"][0]
-
-                            headers = dict(message.get("headers", []))
-                            header_key = b"sec-websocket-protocol"
-                            if subprotocol := headers.get(header_key):
-                                message["headers"] = [
-                                    *message.get("headers", []),
-                                    (header_key, subprotocol),
-                                ]
+        if self.sio is not None or config.transport in ("socketio", "polling"):
+            # Legacy Socket.IO transport, kept behind the optional dependency.
+            if self.sio is not None and config.transport == "websocket":
+                msg = (
+                    "A custom `sio` server requires the Socket.IO transport; "
+                    'set transport="socketio" (or "polling") in rxconfig.py.'
+                )
+                raise RuntimeError(msg)
+            try:
+                from reflex.socketio_namespace import (
+                    EventNamespace,
+                    create_socketio_app,
+                )
+            except ImportError as ex:
+                msg = (
+                    f"transport={config.transport!r} requires the python-socketio "
+                    "package. Install it with: pip install 'reflex[socketio]'"
+                )
+                raise RuntimeError(msg) from ex
 
-                        return await original_send(message)
+            socket_app = create_socketio_app(self, config)
 
-                    return await self.app(scope, receive, modified_send)
+            # Create the event namespace and attach the main app. Not related to any paths.
+            self._event_namespace = EventNamespace(namespace, self)
 
-            socket_app_with_headers = HeaderMiddleware(socket_app)
-            self._api.mount(
-                config.prepend_backend_path(str(constants.Endpoint.EVENT)),
-                socket_app_with_headers,
-            )
+            # Register the event namespace with the socket.
+            self.sio.register_namespace(self._event_namespace)  # pyright: ignore[reportOptionalMemberAccess]
+            # Mount the socket app with the API.
+            if self._api:
+                self._api.mount(event_path, socket_app)
+        else:
+            # Default transport: plain WebSocket served by the API itself.
+            self._event_namespace = WebsocketEventNamespace(namespace, self)
+            if self._api:
+                self._api.router.routes.append(
+                    WebSocketRoute(event_path, self._event_namespace.handle_websocket)
+                )
 
         # Check the exception handlers
         self._validate_exception_handlers()
@@ -1920,336 +1883,21 @@ async def health(_request: Request) -> JSONResponse:
     return JSONResponse(content=health_status, status_code=status_code)
 
 
-class EventNamespace(AsyncNamespace):
-    """The event namespace."""
-
-    # The application object.
-    app: App
-
-    # Maximum error-level log entries a single session may produce via the
-    # client_error event before further reports from it are dropped.
-    _MAX_CLIENT_ERRORS_PER_SID = 5
-
-    # Process-wide bound on error-level client_error log entries per time
-    # window; per-SID budgets alone reset on reconnect, so scripted
-    # reconnects could otherwise flood the logs.
-    _CLIENT_ERROR_WINDOW_SECONDS = 60.0
-    _MAX_CLIENT_ERRORS_PER_WINDOW = 20
-
-    def __init__(self, namespace: str, app: App):
-        """Initialize the event namespace.
-
-        Args:
-            namespace: The namespace.
-            app: The application object.
-        """
-        super().__init__(namespace)
-        self.app = app
-
-        # Use TokenManager for distributed duplicate tab prevention
-        self._token_manager = TokenManager.create()
-
-        # Number of client_error reports logged per SID, for rate limiting.
-        self._client_error_counts: dict[str, int] = {}
-
-        # Start time and count of the current process-wide client_error window.
-        self._client_error_window_start = 0.0
-        self._client_error_window_count = 0
-
-    @property
-    def token_to_sid(self) -> Mapping[str, str]:
-        """Get token to SID mapping for backward compatibility.
-
-        Note: this mapping is read-only.
-
-        Returns:
-            The token to SID mapping.
-        """
-        # For backward compatibility, expose the underlying dict
-        return self._token_manager.token_to_sid
-
-    @property
-    def sid_to_token(self) -> dict[str, str]:
-        """Get SID to token mapping for backward compatibility.
-
-        Returns:
-            The SID to token mapping dict.
-        """
-        # For backward compatibility, expose the underlying dict
-        return self._token_manager.sid_to_token
-
-    async def on_connect(self, sid: str, environ: dict):
-        """Event for when the websocket is connected.
-
-        Args:
-            sid: The Socket.IO session id.
-            environ: The request information, including HTTP headers.
-        """
-        if isinstance(self._token_manager, RedisTokenManager):
-            # Make sure this instance is watching for updates from other instances.
-            self._token_manager.ensure_lost_and_found_task(self.emit_update)
-        query_params = urllib.parse.parse_qs(environ.get("QUERY_STRING", ""))
-        token_list = query_params.get("token", [])
-        if token_list:
-            await self.link_token_to_sid(sid, token_list[0])
-        else:
-            logger.warning(f"No token provided in connection for session {sid}")
-
-        subprotocol = environ.get("HTTP_SEC_WEBSOCKET_PROTOCOL")
-        if subprotocol and subprotocol != constants.Reflex.VERSION:
-            logger.warning(
-                f"Frontend version {subprotocol} for session {sid} does not match the backend version {constants.Reflex.VERSION}."
-            )
-
-    def on_disconnect(self, sid: str) -> asyncio.Task | None:
-        """Event for when the websocket disconnects.
-
-        Args:
-            sid: The Socket.IO session id.
-
-        Returns:
-            An asyncio Task for cleaning up the token, or None.
-        """
-        self._client_error_counts.pop(sid, None)
-        # Get token before cleaning up
-        disconnect_token = self.sid_to_token.get(sid)
-        if disconnect_token:
-            # Use async cleanup through token manager
-            task = asyncio.create_task(
-                self._token_manager.disconnect_token(disconnect_token, sid),
-                name=f"reflex_disconnect_token|{disconnect_token}|{time.time()}",
-            )
-            # Don't await to avoid blocking disconnect, but handle potential errors
-            task.add_done_callback(
-                lambda t: (
-                    t.exception()
-                    and logger.error(f"Token cleanup error: {t.exception()}")
-                )
-            )
-            return task
-        return None
-
-    async def emit_update(self, update: StateUpdate, token: str) -> None:
-        """Emit an update to the client.
-
-        Args:
-            update: The state update to send.
-            token: The client token (tab) associated with the event.
-        """
-        socket_record = self._token_manager.token_to_socket.get(token)
-        if (
-            socket_record is None
-            or socket_record.instance_id != self._token_manager.instance_id
-        ):
-            if isinstance(self._token_manager, RedisTokenManager):
-                # The socket belongs to another instance of the app, send it to the lost and found.
-                await self._token_manager.emit_lost_and_found(token, update)
-            else:
-                # If the socket record is None, we are not connected to a client. Prevent sending
-                # updates to all clients.
-                logger.warning(
-                    f"Attempting to send delta to disconnected client {token!r}"
-                )
-            return
-        # Creating a task prevents the update from being blocked behind other coroutines.
-        await asyncio.create_task(
-            self.emit(str(constants.SocketEvent.EVENT), update, to=socket_record.sid),
-            name=f"reflex_emit_event|{token}|{socket_record.sid}|{time.time()}",
-        )
-
-    async def on_event(self, sid: str, data: Any):
-        """Event for receiving front-end websocket events.
-
-        Args:
-            sid: The Socket.IO session id.
-            data: The event data.
+def __getattr__(name: str) -> Any:
+    """Resolve the optional Socket.IO EventNamespace export lazily.
 
-        Raises:
-            RuntimeError: If the Socket.IO is badly initialized.
-            EventDeserializationError: If the event data is not a dictionary.
-        """
-        # Determine the token for this SID
-        if (token := self.sid_to_token.get(sid)) is None:
-            logger.warning(
-                f"Received event from session {sid} with no associated token. This may indicate a bug. Event data: {data}"
-            )
-            return
-
-        fields = data
-
-        if isinstance(fields, str):
-            logger.warning(
-                "Received event data as a string. This generally should not happen and may indicate a bug."
-                f" Event data: {fields}"
-            )
-            try:
-                fields = json.loads(fields)
-            except json.JSONDecodeError as ex:
-                msg = f"Failed to deserialize event data: {fields}."
-                raise exceptions.EventDeserializationError(msg) from ex
-
-        if not isinstance(fields, dict):
-            msg = f"Event data must be a dictionary, but received {fields} of type {type(fields)}."
-            raise exceptions.EventDeserializationError(msg)
-
-        try:
-            # Get the event.
-            event = Event(**{k: v for k, v in fields.items() if k in _EVENT_FIELDS})
-        except (TypeError, ValueError) as ex:
-            msg = f"Failed to deserialize event data: {fields}."
-            raise exceptions.EventDeserializationError(msg) from ex
-
-        # Get the event environment.
-        if self.app.sio is None:
-            msg = "Socket.IO is not initialized."
-            raise RuntimeError(msg)
-        environ = self.app.sio.get_environ(sid, self.namespace)
-        if environ is None:
-            msg = "Socket.IO environ is not initialized."
-            raise RuntimeError(msg)
-
-        # Get the client headers.
-        headers = {
-            k.decode("utf-8"): v.decode("utf-8")
-            for (k, v) in environ["asgi.scope"]["headers"]
-        }
-
-        # Get the client IP
-        try:
-            client_ip = environ["asgi.scope"]["client"][0]
-            headers["asgi-scope-client"] = client_ip
-        except (KeyError, IndexError):
-            client_ip = environ.get("REMOTE_ADDR", "0.0.0.0")
-
-        # Unroll reverse proxy forwarded headers.
-        client_ip = (
-            headers
-            .get(
-                "x-forwarded-for",
-                client_ip,
-            )
-            .partition(",")[0]
-            .strip()
-        )
-        router_data = event.router_data
-        router_data.update({
-            constants.RouteVar.QUERY: format.format_query_params(event.router_data),
-            constants.RouteVar.CLIENT_TOKEN: token,
-            constants.RouteVar.SESSION_ID: sid,
-            constants.RouteVar.HEADERS: headers,
-            constants.RouteVar.CLIENT_IP: client_ip,
-        })
-        router_data[constants.RouteVar.PATH] = "/" + (
-            self.app.router(path) or "404"
-            if (path := router_data.get(constants.RouteVar.PATH))
-            else "404"
-        ).removeprefix("/")
-        await self.app.event_processor.enqueue(token, event)
-
-    async def on_ping(self, sid: str):
-        """Event for testing the API endpoint.
-
-        Args:
-            sid: The Socket.IO session id.
-        """
-        # Emit the test event.
-        await self.emit(str(constants.SocketEvent.PING), "pong", to=sid)
-
-    async def on_client_error(self, sid: str, data: Any):
-        """Handle errors reported by the frontend.
-
-        This is a dedicated socket event rather than a state event
-        (``FrontendEventExceptionState.handle_frontend_exception``) because a
-        state event is addressed by a handler name the frontend derives from
-        its own state definitions. When those definitions are what disagree
-        with the backend -- the case this handler exists to report -- the name
-        may not resolve and the report is lost. A fixed socket event name
-        cannot drift, and it still gets through after the frontend has stopped
-        sending events on detecting the mismatch.
-
-        Reports are routed through the app's ``frontend_exception_handler``,
-        so frontend errors (especially state update processing errors) are
-        visible in backend logs and reach custom exception handlers.
-
-        Args:
-            sid: The Socket.IO session id.
-            data: The error data from the client.
-        """
-        if not isinstance(data, dict):
-            logger.debug(f"Ignoring malformed client_error payload from SID {sid}.")
-            return
-
-        # Check the sender and the rate limits before sanitizing: sanitizing is
-        # linear in the size of the client-supplied values, and reports that are
-        # dropped here must not cost more than the check itself.
-        if sid not in self.sid_to_token:
-            # Sockets without a linked token are not known clients; don't let
-            # them write error-level entries into the backend logs.
-            logger.debug(f"Ignoring client_error report from unknown SID {sid}.")
-            return
-
-        # Rate limit per session so a client cannot flood the backend logs.
-        error_count = self._client_error_counts.get(sid, 0)
-        if error_count >= self._MAX_CLIENT_ERRORS_PER_SID:
-            return
-
-        # Also bound total entries per time window: per-SID budgets reset on
-        # reconnect, so they alone do not stop scripted reconnect loops.
-        now = time.monotonic()
-        if now - self._client_error_window_start > self._CLIENT_ERROR_WINDOW_SECONDS:
-            self._client_error_window_start = now
-            self._client_error_window_count = 0
-        if self._client_error_window_count >= self._MAX_CLIENT_ERRORS_PER_WINDOW:
-            if self._client_error_window_count == self._MAX_CLIENT_ERRORS_PER_WINDOW:
-                # Warn once per window so suppression is visible in the logs
-                # and a flooding client cannot silently starve reports from
-                # other sessions.
-                self._client_error_window_count += 1
-                logger.warning(
-                    f"Received more than {self._MAX_CLIENT_ERRORS_PER_WINDOW} "
-                    f"client_error reports in {self._CLIENT_ERROR_WINDOW_SECONDS:.0f}s; "
-                    "suppressing further reports for this window."
-                )
-            return
-        self._client_error_window_count += 1
-        self._client_error_counts[sid] = error_count + 1
-
-        error_type = format.sanitize_client_log_value(data.get("error_type", "unknown"))
-        if error_type == constants.ClientErrorType.DISPATCH_MISSING:
-            substate = format.sanitize_client_log_value(data.get("substate", ""))
-            report = (
-                f"[SID: {sid}] State update failed: "
-                f"no dispatch function for substate(s) '{substate}'. "
-                "This indicates a frontend/backend state mismatch. "
-                "Rebuild the frontend or check that api_url points to the matching backend."
-            )
-        else:
-            message = format.sanitize_client_log_value(
-                data.get("message", "No error message provided")
-            )
-            report = f"[SID: {sid}] {error_type}: {message}"
-        # Route through the app's frontend exception handler so custom
-        # handlers (e.g. error trackers) receive client errors too.
-        self.app.frontend_exception_handler(Exception(report))
-
-    async def link_token_to_sid(self, sid: str, token: str):
-        """Link a token to a session id.
+    Args:
+        name: The attribute name.
 
-        Args:
-            sid: The Socket.IO session id.
-            token: The client token.
-        """
-        # Use TokenManager for duplicate detection and Redis support
-        new_token = await self._token_manager.link_token_to_sid(token, sid)
+    Returns:
+        The resolved attribute.
 
-        if new_token:
-            # Duplicate detected, emit new token to client
-            await self.emit("new_token", new_token, to=sid)
+    Raises:
+        AttributeError: If the attribute is unknown.
+    """
+    if name == "EventNamespace":
+        from reflex.socketio_namespace import EventNamespace
 
-        # Update client state to apply new sid/token for running background tasks.
-        if self.app._state is not None:
-            async with self.app.state_manager.modify_state(
-                BaseStateToken(ident=new_token or token, cls=self.app._state)
-            ) as state:
-                state.router_data[constants.RouteVar.SESSION_ID] = sid
-                state.router = RouterData.from_router_data(state.router_data)
+        return EventNamespace
+    msg = f"module {__name__!r} has no attribute {name!r}"
+    raise AttributeError(msg)
diff --git a/reflex/event_namespace.py b/reflex/event_namespace.py
new file mode 100644
index 00000000000..0c25f462219
--- /dev/null
+++ b/reflex/event_namespace.py
@@ -0,0 +1,601 @@
+"""Event namespaces bridging client sessions to the Reflex event loop."""
+
+from __future__ import annotations
+
+import asyncio
+import contextlib
+import json
+import logging
+import time
+import urllib.parse
+import uuid
+from abc import ABC, abstractmethod
+from collections.abc import Mapping, MutableMapping
+from typing import TYPE_CHECKING, Any
+
+from reflex_base import constants
+from reflex_base.config import get_config
+from reflex_base.environment import environment
+from reflex_base.event import _EVENT_FIELDS, Event
+from starlette.websockets import WebSocket, WebSocketDisconnect
+
+from reflex.istate.data import RouterData
+from reflex.istate.manager.token import BaseStateToken
+from reflex.state import StateUpdate
+from reflex.utils import exceptions, format
+from reflex.utils.token_manager import RedisTokenManager, TokenManager
+
+if TYPE_CHECKING:
+    from reflex.app import App
+
+logger = logging.getLogger(__name__)
+
+# Protocol-level message names for the plain WebSocket transport. These are
+# reserved (underscore-prefixed) and never dispatched as application events.
+# They must match the names in .templates/web/utils/helpers/websocket.js.
+HANDSHAKE_MESSAGE = "_handshake"
+PING_MESSAGE = "_ping"
+PONG_MESSAGE = "_pong"
+
+# Application-level socket event names, resolved once for the hot paths.
+_EVENT = str(constants.SocketEvent.EVENT)
+_PING = str(constants.SocketEvent.PING)
+_CLIENT_ERROR = str(constants.SocketEvent.CLIENT_ERROR)
+
+# The heartbeat frame is static; serialize it once.
+_PING_FRAME = json.dumps([PING_MESSAGE])
+
+
+class BaseEventNamespace(ABC):
+    """Transport-agnostic handler for client event sessions."""
+
+    # The application object.
+    app: App
+
+    # Maximum error-level log entries a single session may produce via the
+    # client_error event before further reports from it are dropped.
+    _MAX_CLIENT_ERRORS_PER_SID = 5
+
+    # Process-wide bound on error-level client_error log entries per time
+    # window; per-SID budgets alone reset on reconnect, so scripted
+    # reconnects could otherwise flood the logs.
+    _CLIENT_ERROR_WINDOW_SECONDS = 60.0
+    _MAX_CLIENT_ERRORS_PER_WINDOW = 20
+
+    def __init__(self, namespace: str, app: App):
+        """Initialize the event namespace.
+
+        Args:
+            namespace: The namespace.
+            app: The application object.
+        """
+        self.namespace = namespace
+        self.app = app
+
+        # Use TokenManager for distributed duplicate tab prevention
+        self._token_manager = TokenManager.create()
+
+        # Number of client_error reports logged per SID, for rate limiting.
+        self._client_error_counts: dict[str, int] = {}
+
+        # Start time and count of the current process-wide client_error window.
+        self._client_error_window_start = 0.0
+        self._client_error_window_count = 0
+
+    @property
+    def token_to_sid(self) -> Mapping[str, str]:
+        """Get token to SID mapping for backward compatibility.
+
+        Note: this mapping is read-only.
+
+        Returns:
+            The token to SID mapping.
+        """
+        # For backward compatibility, expose the underlying dict
+        return self._token_manager.token_to_sid
+
+    @property
+    def sid_to_token(self) -> dict[str, str]:
+        """Get SID to token mapping for backward compatibility.
+
+        Returns:
+            The SID to token mapping dict.
+        """
+        # For backward compatibility, expose the underlying dict
+        return self._token_manager.sid_to_token
+
+    @abstractmethod
+    async def emit(self, event: str, data: Any = None, to: str | None = None) -> None:
+        """Emit an event to a connected client session.
+
+        Args:
+            event: The event name.
+            data: The event payload.
+            to: The session id to emit to.
+        """
+
+    async def handle_connect(
+        self, sid: str, query_string: str, subprotocol: str | None
+    ) -> None:
+        """Handle a new client session connecting.
+
+        Args:
+            sid: The session id.
+            query_string: The raw query string of the connection request.
+            subprotocol: The websocket subprotocol offered by the client.
+        """
+        if isinstance(self._token_manager, RedisTokenManager):
+            # Make sure this instance is watching for updates from other instances.
+            self._token_manager.ensure_lost_and_found_task(self.emit_update)
+        query_params = urllib.parse.parse_qs(query_string)
+        token_list = query_params.get("token", [])
+        if token_list:
+            await self.link_token_to_sid(sid, token_list[0])
+        else:
+            logger.warning(f"No token provided in connection for session {sid}")
+
+        if subprotocol and subprotocol != constants.Reflex.VERSION:
+            logger.warning(
+                f"Frontend version {subprotocol} for session {sid} does not match the backend version {constants.Reflex.VERSION}."
+            )
+
+    def handle_disconnect(self, sid: str) -> asyncio.Task | None:
+        """Handle a client session disconnecting.
+
+        Args:
+            sid: The session id.
+
+        Returns:
+            An asyncio Task for cleaning up the token, or None.
+        """
+        self._client_error_counts.pop(sid, None)
+        # Get token before cleaning up
+        disconnect_token = self.sid_to_token.get(sid)
+        if disconnect_token:
+            # Use async cleanup through token manager
+            task = asyncio.create_task(
+                self._token_manager.disconnect_token(disconnect_token, sid),
+                name=f"reflex_disconnect_token|{disconnect_token}|{time.time()}",
+            )
+            # Don't await to avoid blocking disconnect, but handle potential errors
+            task.add_done_callback(
+                lambda t: (
+                    t.exception()
+                    and logger.error(f"Token cleanup error: {t.exception()}")
+                )
+            )
+            return task
+        return None
+
+    async def emit_update(self, update: StateUpdate, token: str) -> None:
+        """Emit an update to the client.
+
+        Args:
+            update: The state update to send.
+            token: The client token (tab) associated with the event.
+        """
+        socket_record = self._token_manager.token_to_socket.get(token)
+        if (
+            socket_record is None
+            or socket_record.instance_id != self._token_manager.instance_id
+        ):
+            if isinstance(self._token_manager, RedisTokenManager):
+                # The socket belongs to another instance of the app, send it to the lost and found.
+                await self._token_manager.emit_lost_and_found(token, update)
+            else:
+                # If the socket record is None, we are not connected to a client. Prevent sending
+                # updates to all clients.
+                logger.warning(
+                    f"Attempting to send delta to disconnected client {token!r}"
+                )
+            return
+        # Creating a task prevents the update from being blocked behind other coroutines.
+        await asyncio.create_task(
+            self.emit(_EVENT, update, to=socket_record.sid),
+            name=f"reflex_emit_event|{token}|{socket_record.sid}|{time.time()}",
+        )
+
+    async def handle_event(
+        self, sid: str, data: Any, asgi_scope: MutableMapping[str, Any]
+    ) -> None:
+        """Handle an incoming front-end event.
+
+        Args:
+            sid: The session id.
+            data: The event data.
+            asgi_scope: The ASGI scope of the client connection.
+
+        Raises:
+            EventDeserializationError: If the event data is malformed.
+        """
+        # Determine the token for this SID
+        if (token := self.sid_to_token.get(sid)) is None:
+            logger.warning(
+                f"Received event from session {sid} with no associated token. This may indicate a bug. Event data: {data}"
+            )
+            return
+
+        fields = data
+
+        if isinstance(fields, str):
+            logger.warning(
+                "Received event data as a string. This generally should not happen and may indicate a bug."
+                f" Event data: {fields}"
+            )
+            try:
+                fields = json.loads(fields)
+            except json.JSONDecodeError as ex:
+                msg = f"Failed to deserialize event data: {fields}."
+                raise exceptions.EventDeserializationError(msg) from ex
+
+        if not isinstance(fields, dict):
+            msg = f"Event data must be a dictionary, but received {fields} of type {type(fields)}."
+            raise exceptions.EventDeserializationError(msg)
+
+        try:
+            # Get the event.
+            event = Event(**{k: v for k, v in fields.items() if k in _EVENT_FIELDS})
+        except (TypeError, ValueError) as ex:
+            msg = f"Failed to deserialize event data: {fields}."
+            raise exceptions.EventDeserializationError(msg) from ex
+
+        # The dataclass does not validate field types.
+        if (
+            not isinstance(event.name, str)
+            or not isinstance(event.payload, dict)
+            or not isinstance(event.router_data, dict)
+        ):
+            msg = "Event fields have invalid types."
+            raise exceptions.EventDeserializationError(msg)
+
+        # Decode the connection headers once: the scope is per-connection
+        # state, so cache the decoded mapping in it and copy per event (the
+        # copy is mutated below and ends up in the event's router_data).
+        base_headers = asgi_scope.get("_reflex_headers")
+        if base_headers is None:
+            base_headers = {
+                k.decode("utf-8"): v.decode("utf-8") for (k, v) in asgi_scope["headers"]
+            }
+            asgi_scope["_reflex_headers"] = base_headers
+        headers = dict(base_headers)
+
+        # Get the client IP
+        client = asgi_scope.get("client")
+        if client:
+            client_ip = client[0]
+            headers["asgi-scope-client"] = client_ip
+        else:
+            client_ip = "0.0.0.0"
+
+        # Unroll reverse proxy forwarded headers.
+        client_ip = (
+            headers
+            .get(
+                "x-forwarded-for",
+                client_ip,
+            )
+            .partition(",")[0]
+            .strip()
+        )
+        router_data = event.router_data
+        try:
+            # The nested values are still client-controlled.
+            router_data.update({
+                constants.RouteVar.QUERY: format.format_query_params(event.router_data),
+                constants.RouteVar.CLIENT_TOKEN: token,
+                constants.RouteVar.SESSION_ID: sid,
+                constants.RouteVar.HEADERS: headers,
+                constants.RouteVar.CLIENT_IP: client_ip,
+            })
+            router_data[constants.RouteVar.PATH] = "/" + (
+                self.app.router(path) or "404"
+                if (path := router_data.get(constants.RouteVar.PATH))
+                else "404"
+            ).removeprefix("/")
+        except (AttributeError, LookupError, TypeError, ValueError) as ex:
+            msg = "Failed to normalize event router_data."
+            raise exceptions.EventDeserializationError(msg) from ex
+        await self.app.event_processor.enqueue(token, event)
+
+    async def handle_ping(self, sid: str) -> None:
+        """Handle an application-level ping test event.
+
+        Args:
+            sid: The session id.
+        """
+        # Emit the test event.
+        await self.emit(_PING, "pong", to=sid)
+
+    async def handle_client_error(self, sid: str, data: Any) -> None:
+        """Handle errors reported by the frontend.
+
+        This is a dedicated socket event rather than a state event
+        (``FrontendEventExceptionState.handle_frontend_exception``) because a
+        state event is addressed by a handler name the frontend derives from
+        its own state definitions. When those definitions are what disagree
+        with the backend -- the case this handler exists to report -- the name
+        may not resolve and the report is lost. A fixed socket event name
+        cannot drift, and it still gets through after the frontend has stopped
+        sending events on detecting the mismatch.
+
+        Reports are routed through the app's ``frontend_exception_handler``,
+        so frontend errors (especially state update processing errors) are
+        visible in backend logs and reach custom exception handlers.
+
+        Args:
+            sid: The session id.
+            data: The error data from the client.
+        """
+        if not isinstance(data, dict):
+            logger.debug(f"Ignoring malformed client_error payload from SID {sid}.")
+            return
+
+        # Check the sender and the rate limits before sanitizing: sanitizing is
+        # linear in the size of the client-supplied values, and reports that are
+        # dropped here must not cost more than the check itself.
+        if sid not in self.sid_to_token:
+            # Sockets without a linked token are not known clients; don't let
+            # them write error-level entries into the backend logs.
+            logger.debug(f"Ignoring client_error report from unknown SID {sid}.")
+            return
+
+        # Rate limit per session so a client cannot flood the backend logs.
+        error_count = self._client_error_counts.get(sid, 0)
+        if error_count >= self._MAX_CLIENT_ERRORS_PER_SID:
+            return
+
+        # Also bound total entries per time window: per-SID budgets reset on
+        # reconnect, so they alone do not stop scripted reconnect loops.
+        now = time.monotonic()
+        if now - self._client_error_window_start > self._CLIENT_ERROR_WINDOW_SECONDS:
+            self._client_error_window_start = now
+            self._client_error_window_count = 0
+        if self._client_error_window_count >= self._MAX_CLIENT_ERRORS_PER_WINDOW:
+            if self._client_error_window_count == self._MAX_CLIENT_ERRORS_PER_WINDOW:
+                # Warn once per window so suppression is visible in the logs
+                # and a flooding client cannot silently starve reports from
+                # other sessions.
+                self._client_error_window_count += 1
+                logger.warning(
+                    f"Received more than {self._MAX_CLIENT_ERRORS_PER_WINDOW} "
+                    f"client_error reports in {self._CLIENT_ERROR_WINDOW_SECONDS:.0f}s; "
+                    "suppressing further reports for this window."
+                )
+            return
+        self._client_error_window_count += 1
+        self._client_error_counts[sid] = error_count + 1
+
+        error_type = format.sanitize_client_log_value(data.get("error_type", "unknown"))
+        if error_type == constants.ClientErrorType.DISPATCH_MISSING:
+            substate = format.sanitize_client_log_value(data.get("substate", ""))
+            report = (
+                f"[SID: {sid}] State update failed: "
+                f"no dispatch function for substate(s) '{substate}'. "
+                "This indicates a frontend/backend state mismatch. "
+                "Rebuild the frontend or check that api_url points to the matching backend."
+            )
+        else:
+            message = format.sanitize_client_log_value(
+                data.get("message", "No error message provided")
+            )
+            report = f"[SID: {sid}] {error_type}: {message}"
+        # Route through the app's frontend exception handler so custom
+        # handlers (e.g. error trackers) receive client errors too.
+        self.app.frontend_exception_handler(Exception(report))
+
+    async def link_token_to_sid(self, sid: str, token: str):
+        """Link a token to a session id.
+
+        Args:
+            sid: The session id.
+            token: The client token.
+        """
+        # Use TokenManager for duplicate detection and Redis support
+        new_token = await self._token_manager.link_token_to_sid(token, sid)
+
+        if new_token:
+            # Duplicate detected, emit new token to client
+            await self.emit("new_token", new_token, to=sid)
+
+        # Update client state to apply new sid/token for running background tasks.
+        if self.app._state is not None:
+            async with self.app.state_manager.modify_state(
+                BaseStateToken(ident=new_token or token, cls=self.app._state)
+            ) as state:
+                state.router_data[constants.RouteVar.SESSION_ID] = sid
+                state.router = RouterData.from_router_data(state.router_data)
+
+
+class WebsocketEventNamespace(BaseEventNamespace):
+    """Default event transport over a plain WebSocket.
+
+    Frames are JSON arrays ``[event_name, payload]``.
+    """
+
+    def __init__(self, namespace: str, app: App):
+        """Initialize the websocket event namespace.
+
+        Args:
+            namespace: The namespace.
+            app: The application object.
+        """
+        super().__init__(namespace, app)
+        self._sockets: dict[str, WebSocket] = {}
+
+    async def emit(self, event: str, data: Any = None, to: str | None = None) -> None:
+        """Emit an event to a connected client session.
+
+        Args:
+            event: The event name.
+            data: The event payload.
+            to: The session id to emit to.
+        """
+        websocket = self._sockets.get(to) if to is not None else None
+        if websocket is None:
+            # Routine race: the client disconnected while an event was still
+            # being processed, so its remaining updates have nowhere to go.
+            logger.debug(f"Attempted to emit {event!r} to unknown session {to!r}.")
+            return
+        try:
+            await websocket.send_text(format.json_dumps([event, data]))
+        except Exception:
+            # The connection went away mid-send; the receive loop cleans up.
+            logger.debug(f"Failed to emit {event!r} to session {to!r}.", exc_info=True)
+
+    @staticmethod
+    def _origin_allowed(origin: str | None) -> bool:
+        """Check a connection's Origin header against the CORS config.
+
+        Args:
+            origin: The Origin header value, if any.
+
+        Returns:
+            Whether the connection is allowed.
+        """
+        if origin is None:
+            # Non-browser clients don't send an Origin header.
+            return True
+        allowed_origins = get_config().cors_allowed_origins
+        return "*" in allowed_origins or origin in allowed_origins
+
+    async def handle_websocket(self, websocket: WebSocket) -> None:
+        """Serve one client websocket connection for its full lifetime.
+
+        Args:
+            websocket: The client websocket connection.
+        """
+        if not self._origin_allowed(websocket.headers.get("origin")):
+            # Reject cross-origin connections before accepting.
+            await websocket.close(code=1008)
+            return
+        subprotocols = websocket.scope.get("subprotocols") or []
+        # Echo the client's offered subprotocol (the Reflex version); browsers
+        # abort the connection if the server selects none.
+        await websocket.accept(subprotocol=subprotocols[0] if subprotocols else None)
+
+        sid = str(uuid.uuid4())
+        ping_interval = environment.REFLEX_SOCKET_INTERVAL.get()
+        ping_timeout = environment.REFLEX_SOCKET_TIMEOUT.get()
+        max_message_size = environment.REFLEX_SOCKET_MAX_HTTP_BUFFER_SIZE.get()
+        self._sockets[sid] = websocket
+        last_received = time.monotonic()
+
+        async def heartbeat() -> None:
+            try:
+                while True:
+                    await asyncio.sleep(ping_interval)
+                    if time.monotonic() - last_received > ping_interval + ping_timeout:
+                        await websocket.close(code=1001)
+                        return
+                    await websocket.send_text(_PING_FRAME)
+            except Exception:
+                # Socket went away; the receive loop handles cleanup.
+                return
+
+        heartbeat_task = asyncio.create_task(
+            heartbeat(), name=f"reflex_heartbeat|{sid}"
+        )
+        try:
+            # The handshake confirms application-level liveness and carries the
+            # heartbeat settings for the client's connection watchdog.
+            await websocket.send_text(
+                format.json_dumps([
+                    HANDSHAKE_MESSAGE,
+                    {"ping_interval": ping_interval, "ping_timeout": ping_timeout},
+                ])
+            )
+            await self.handle_connect(
+                sid,
+                websocket.scope.get("query_string", b"").decode(),
+                subprotocols[0] if subprotocols else None,
+            )
+            if sid not in self._token_manager.sid_to_token:
+                # No token was linked; not a Reflex client.
+                await websocket.close(code=1008)
+                return
+            while True:
+                received = await websocket.receive()
+                if received["type"] == "websocket.disconnect":
+                    break
+                last_received = time.monotonic()
+                text = received.get("text")
+                if text is None:
+                    # Binary frame; not part of the protocol.
+                    logger.debug(f"Closing session {sid}: received a binary frame.")
+                    await websocket.close(code=1003)
+                    break
+                # ASGI delivers complete messages, so the server has already
+                # buffered the frame; its protocol-level caps (enforced during
+                # frame reassembly) bound that allocation. This check applies
+                # the Reflex policy limit on top.
+                # The limit is in bytes; UTF-8 encodes 1-4 bytes per character,
+                # so more characters than the limit is certainly over, and a
+                # quarter or fewer certainly under -- only encode to count the
+                # exact bytes in between (bounding the copy to 4x the limit).
+                text_length = len(text)
+                if text_length > max_message_size or (
+                    text_length * 4 > max_message_size
+                    and len(text.encode("utf-8")) > max_message_size
+                ):
+                    logger.debug(
+                        f"Closing session {sid}: message over {max_message_size} bytes."
+                    )
+                    await websocket.close(code=1009)
+                    break
+                try:
+                    message = json.loads(text)
+                except json.JSONDecodeError:
+                    message = None
+                if (
+                    not isinstance(message, list)
+                    or not message
+                    or not isinstance(message[0], str)
+                ):
+                    # A Reflex client never sends malformed frames; close
+                    # instead of logging per frame, which a hostile client
+                    # could use to flood the logs.
+                    logger.debug(f"Closing session {sid}: malformed frame.")
+                    await websocket.close(code=1002)
+                    break
+                event = message[0]
+                data = message[1] if len(message) > 1 else None
+                try:
+                    # Ordered by frequency: events are the hot path, heartbeat
+                    # pongs arrive once per ping interval.
+                    if event == _EVENT:
+                        await self.handle_event(sid, data, websocket.scope)
+                    elif event == PONG_MESSAGE:
+                        continue
+                    elif event == _PING:
+                        await self.handle_ping(sid)
+                    elif event == _CLIENT_ERROR:
+                        await self.handle_client_error(sid, data)
+                    else:
+                        logger.debug(
+                            f"Ignoring unknown socket event {event!r} from session {sid}."
+                        )
+                except exceptions.EventDeserializationError:
+                    # Client-controlled input a Reflex client never sends;
+                    # close instead of logging per frame.
+                    logger.debug(f"Closing session {sid}: undeserializable event.")
+                    await websocket.close(code=1002)
+                    break
+                except Exception:
+                    # A failing handler is a server-side bug: log it loudly;
+                    # the connection survives.
+                    logger.exception(
+                        f"Error handling socket event {event!r} for session {sid}."
+                    )
+        except WebSocketDisconnect:
+            pass
+        finally:
+            heartbeat_task.cancel()
+            self._sockets.pop(sid, None)
+            cleanup_task = self.handle_disconnect(sid)
+            if cleanup_task is not None:
+                # Await the token cleanup so an immediate reconnect is not
+                # treated as a duplicate tab; shielded so cancellation (e.g.
+                # server shutdown) cannot abort it. Errors are logged by the
+                # task's done callback.
+                with contextlib.suppress(Exception):
+                    await asyncio.shield(cleanup_task)
diff --git a/reflex/socketio_namespace.py b/reflex/socketio_namespace.py
new file mode 100644
index 00000000000..9ba9337d374
--- /dev/null
+++ b/reflex/socketio_namespace.py
@@ -0,0 +1,185 @@
+"""Socket.IO event transport (requires the optional python-socketio dependency)."""
+
+from __future__ import annotations
+
+import json
+from types import SimpleNamespace
+from typing import TYPE_CHECKING, Any
+
+from reflex_base.environment import environment
+from reflex_base.utils.types import ASGIApp, Message, Receive, Scope, Send
+from socketio import ASGIApp as EngineIOApp
+from socketio import AsyncNamespace, AsyncServer
+
+from reflex.event_namespace import BaseEventNamespace
+from reflex.utils import format
+
+if TYPE_CHECKING:
+    import asyncio
+
+    from reflex_base.config import Config
+
+    from reflex.app import App
+
+
+class EventNamespace(AsyncNamespace, BaseEventNamespace):
+    """The Socket.IO event namespace."""
+
+    def __init__(self, namespace: str, app: App):
+        """Initialize the event namespace.
+
+        Args:
+            namespace: The namespace.
+            app: The application object.
+        """
+        AsyncNamespace.__init__(self, namespace)
+        BaseEventNamespace.__init__(self, namespace, app)
+
+    async def on_connect(self, sid: str, environ: dict):
+        """Event for when the websocket is connected.
+
+        Args:
+            sid: The Socket.IO session id.
+            environ: The request information, including HTTP headers.
+        """
+        await self.handle_connect(
+            sid,
+            environ.get("QUERY_STRING", ""),
+            environ.get("HTTP_SEC_WEBSOCKET_PROTOCOL"),
+        )
+
+    def on_disconnect(self, sid: str) -> asyncio.Task | None:
+        """Event for when the websocket disconnects.
+
+        Args:
+            sid: The Socket.IO session id.
+
+        Returns:
+            An asyncio Task for cleaning up the token, or None.
+        """
+        return self.handle_disconnect(sid)
+
+    async def on_event(self, sid: str, data: Any):
+        """Event for receiving front-end websocket events.
+
+        Args:
+            sid: The Socket.IO session id.
+            data: The event data.
+
+        Raises:
+            RuntimeError: If the Socket.IO is badly initialized.
+        """
+        if self.app.sio is None:
+            msg = "Socket.IO is not initialized."
+            raise RuntimeError(msg)
+        environ = self.app.sio.get_environ(sid, self.namespace)
+        if environ is None:
+            msg = "Socket.IO environ is not initialized."
+            raise RuntimeError(msg)
+        await self.handle_event(sid, data, environ["asgi.scope"])
+
+    async def on_ping(self, sid: str):
+        """Event for testing the API endpoint.
+
+        Args:
+            sid: The Socket.IO session id.
+        """
+        await self.handle_ping(sid)
+
+    async def on_client_error(self, sid: str, data: Any):
+        """Handle errors reported by the frontend.
+
+        Args:
+            sid: The Socket.IO session id.
+            data: The error data from the client.
+        """
+        await self.handle_client_error(sid, data)
+
+
+class _HeaderMiddleware:
+    """Echo the websocket subprotocol on accept, which engineio does not."""
+
+    def __init__(self, app: ASGIApp):
+        """Initialize the middleware.
+
+        Args:
+            app: The ASGI app to wrap.
+        """
+        self.app = app
+
+    async def __call__(self, scope: Scope, receive: Receive, send: Send):
+        """Handle an ASGI connection.
+
+        Args:
+            scope: The ASGI scope.
+            receive: The ASGI receive callable.
+            send: The ASGI send callable.
+
+        Returns:
+            The result of the wrapped app.
+        """
+        original_send = send
+
+        async def modified_send(message: Message):
+            if message["type"] == "websocket.accept":
+                if scope.get("subprotocols"):
+                    # The following *does* say "subprotocol" instead of "subprotocols", intentionally.
+                    message["subprotocol"] = scope["subprotocols"][0]
+
+                headers = dict(message.get("headers", []))
+                header_key = b"sec-websocket-protocol"
+                if subprotocol := headers.get(header_key):
+                    message["headers"] = [
+                        *message.get("headers", []),
+                        (header_key, subprotocol),
+                    ]
+
+            return await original_send(message)
+
+        return await self.app(scope, receive, modified_send)
+
+
+def create_socketio_app(app: App, config: Config) -> ASGIApp:
+    """Create the Socket.IO server for an app and return its ASGI app.
+
+    Creates ``app.sio`` if the user did not supply their own server.
+
+    Args:
+        app: The Reflex app.
+        config: The app configuration.
+
+    Returns:
+        The ASGI app serving the Socket.IO server.
+
+    Raises:
+        RuntimeError: If a custom ``sio`` server does not use asgi mode.
+    """
+    if not app.sio:
+        app.sio = AsyncServer(
+            async_mode="asgi",
+            cors_allowed_origins=(
+                (
+                    "*"
+                    if config.cors_allowed_origins == ("*",)
+                    else list(config.cors_allowed_origins)
+                )
+                if config.transport != "polling"
+                else []
+            ),
+            cors_credentials=config.transport != "polling",
+            max_http_buffer_size=environment.REFLEX_SOCKET_MAX_HTTP_BUFFER_SIZE.get(),
+            ping_interval=environment.REFLEX_SOCKET_INTERVAL.get(),
+            ping_timeout=environment.REFLEX_SOCKET_TIMEOUT.get(),
+            json=SimpleNamespace(
+                dumps=staticmethod(format.json_dumps),
+                loads=staticmethod(json.loads),
+            ),
+            allow_upgrades=False,
+            transports=["polling" if config.transport == "polling" else "websocket"],
+        )
+    elif getattr(app.sio, "async_mode", "") != "asgi":
+        msg = f"Custom `sio` must use `async_mode='asgi'`, not '{app.sio.async_mode}'."
+        raise RuntimeError(msg)
+
+    # Create the socket app. Note event endpoint constant replaces the default 'socket.io' path.
+    return _HeaderMiddleware(EngineIOApp(app.sio, socketio_path=""))
diff --git a/reflex/testing.py b/reflex/testing.py
index 51c950e637a..672a94373c8 100644
--- a/reflex/testing.py
+++ b/reflex/testing.py
@@ -25,7 +25,7 @@
 from pathlib import Path
 from typing import TYPE_CHECKING, Any, ClassVar, Literal, TypeVar
 
-import uvicorn
+from granian.constants import Interfaces
 from reflex_base.components.memo import MEMOS
 from reflex_base.config import get_config, reload_config
 from reflex_base.environment import environment
@@ -74,6 +74,147 @@
     FRONTEND_POPEN_ARGS["start_new_session"] = True
 
 
+class _EmbeddedServer:
+    """In-process granian server with a uvicorn-like control surface.
+
+    Serves the given ASGI app object directly, so the harness shares the app
+    and state instances with the running server, and granian's native
+    websocket support means no separate websocket library is required. The
+    port is resolved up front because granian cannot report an OS-assigned
+    port back to Python.
+    """
+
+    def __init__(self, app: ASGIApp, host: str = "127.0.0.1", port: int = 0) -> None:
+        """Prepare the server without starting it.
+
+        Args:
+            app: the ASGI app object to serve.
+            host: the address to bind to.
+            port: the port to bind to; 0 picks a free port immediately.
+        """
+        if port == 0:
+            with socket.socket() as probe:
+                probe.bind((host, 0))
+                port = probe.getsockname()[1]
+        self.app = app
+        self.host = host
+        self.port = port
+        # Monkeypatchable async shutdown hook, mirroring uvicorn.Server.shutdown.
+        self.shutdown: Callable[..., Coroutine[Any, Any, None]] = self._noop_shutdown
+        self._should_exit = threading.Event()
+        self._loop: asyncio.AbstractEventLoop | None = None
+        self._server: Any = None
+
+    @staticmethod
+    async def _noop_shutdown(*args, **kwargs) -> None:
+        """Default shutdown hook.
+
+        Args:
+            *args: ignored.
+            **kwargs: ignored.
+        """
+
+    def getsockname(self) -> tuple[str, int]:
+        """The address the server is bound to.
+
+        Returns:
+            The (host, port) tuple the server serves on.
+        """
+        return (self.host, self.port)
+
+    def is_listening(self) -> bool:
+        """Whether the server accepts connections.
+
+        Returns:
+            True if a TCP connection to the bound address succeeds.
+        """
+        try:
+            socket.create_connection((self.host, self.port), timeout=0.1).close()
+        except OSError:
+            return False
+        return True
+
+    @property
+    def should_exit(self) -> bool:
+        """Whether the server was asked to stop.
+
+        Returns:
+            True after `should_exit` has been set.
+        """
+        return self._should_exit.is_set()
+
+    @should_exit.setter
+    def should_exit(self, value: bool) -> None:
+        if not value:
+            return
+        self._should_exit.set()
+        loop, server = self._loop, self._server
+        if loop is not None and server is not None:
+
+            def _interrupt() -> None:
+                server.interrupt_signal = True
+                server.main_loop_interrupt.set()
+
+            # A closed loop means the server is already down.
+            with contextlib.suppress(RuntimeError):
+                loop.call_soon_threadsafe(_interrupt)
+
+    def run(self) -> None:
+        """Serve the app until `should_exit` is set; used as a thread target."""
+        asyncio.run(self._serve())
+
+    async def _serve(self) -> None:
+        from granian.server.embed import Server
+
+        try:
+            # Another process can claim the probed port before granian binds
+            # it; retry the bind on a fresh port.
+            for attempts_left in reversed(range(10)):
+                server = Server(
+                    self.app,
+                    address=self.host,
+                    port=self.port,
+                    interface=Interfaces.ASGI,
+                    log_enabled=False,
+                )
+                self._server = server
+                self._loop = asyncio.get_running_loop()
+                if self._should_exit.is_set():
+                    # Stopped before startup: let serve() exit right after binding.
+                    server.interrupt_signal = True
+                    server.main_loop_interrupt.set()
+                try:
+                    await server.serve()
+                except (OSError, RuntimeError) as ex:
+                    # Granian surfaces bind failures as RuntimeError; the
+                    # message is platform-specific: os error 98 (posix), or
+                    # 10048/10013 (windows; exclusively-held ports fail with
+                    # WSAEACCES rather than WSAEADDRINUSE).
+                    message = str(ex).lower()
+                    if (
+                        "address already in use" not in message
+                        and "os error 10048" not in message
+                        and "os error 10013" not in message
+                    ):
+                        raise
+                    if self._should_exit.is_set():
+                        break
+                    if not attempts_left:
+                        raise
+                    logger.warning(
+                        f"Port {self.port} unavailable ({ex}); retrying on a fresh port."
+                    )
+                    with socket.socket() as probe:
+                        probe.bind((self.host, 0))
+                        self.port = probe.getsockname()[1]
+                    continue
+                # serve() returned: shutdown, or a server failure after
+                # startup -- never restart on a different port.
+                break
+        finally:
+            await self.shutdown()
+
+
 # borrowed from py3.11
 class chdir(contextlib.AbstractContextManager):  # noqa: N801
     """Non thread-safe context manager to change the current working directory."""
@@ -118,7 +259,7 @@ class AppHarness:
     frontend_url: str | None = None
     frontend_output_thread: threading.Thread | None = None
     backend_thread: threading.Thread | None = None
-    backend: uvicorn.Server | None = None
+    backend: _EmbeddedServer | None = None
     _frontends: list[WebDriver] = dataclasses.field(default_factory=list)
     _registry_token: contextvars.Token[RegistrationContext] | None = None
     _base_registration_context: ClassVar[RegistrationContext] | None = None
@@ -341,13 +482,7 @@ def _start_backend(self, port: int = 0):
         if self.app_asgi is None:
             msg = "App was not initialized."
             raise RuntimeError(msg)
-        self.backend = uvicorn.Server(
-            uvicorn.Config(
-                app=self.app_asgi,
-                host="127.0.0.1",
-                port=port,
-            )
-        )
+        self.backend = _EmbeddedServer(self.app_asgi, port=port)
         self.backend.shutdown = self._get_backend_shutdown_handler()
 
         def _run_backend(context: contextvars.Context) -> None:
@@ -576,38 +711,27 @@ async def _poll_for_async(
             await asyncio.sleep(step)
         return False
 
-    def _poll_for_servers(self, timeout: TimeoutType = None) -> socket.socket:
-        """Poll backend server for listening sockets.
+    def _poll_for_servers(self, timeout: TimeoutType = None) -> _EmbeddedServer:
+        """Poll the backend server until it is listening.
 
         Args:
-            timeout: how long to wait for listening socket.
+            timeout: how long to wait for the listening server.
 
         Returns:
-            first active listening socket on the backend
+            the backend server, exposing `getsockname()` for its bound address
 
         Raises:
             RuntimeError: when the backend hasn't started running
-            TimeoutError: when server or sockets are not ready
+            TimeoutError: when the server is not ready
         """
         if self.backend is None:
             msg = "Backend is not running."
             raise RuntimeError(msg)
         backend = self.backend
-        # check for servers to be initialized
-        if not self._poll_for(
-            target=lambda: getattr(backend, "servers", False),
-            timeout=timeout,
-        ):
-            msg = "Backend servers are not initialized."
-            raise TimeoutError(msg)
-        # check for sockets to be listening
-        if not self._poll_for(
-            target=lambda: getattr(backend.servers[0], "sockets", False),
-            timeout=timeout,
-        ):
+        if not self._poll_for(target=backend.is_listening, timeout=timeout):
             msg = "Backend is not listening."
             raise TimeoutError(msg)
-        return backend.servers[0].sockets[0]
+        return backend
 
     def frontend(
         self,
@@ -817,23 +941,16 @@ class AppHarnessProd(AppHarness):
     """AppHarnessProd executes a reflex app in-process for testing.
 
     In prod mode, instead of running `react-router dev` the app is exported as static
-    files and served via Starlette StaticFiles in a dedicated Uvicorn server.
-    Additionally, the backend runs in multi-worker mode.
+    files and served via Starlette StaticFiles on a dedicated embedded server.
     """
 
     frontend_thread: threading.Thread | None = None
-    frontend_server: uvicorn.Server | None = None
+    frontend_server: _EmbeddedServer | None = None
 
     def _run_frontend(self):
         with chdir(self.app_path):
             frontend_app = reflex.utils.exec._frontend_prod_app()
-        self.frontend_server = uvicorn.Server(
-            uvicorn.Config(
-                app=frontend_app,
-                host="127.0.0.1",
-                port=0,
-            )
-        )
+        self.frontend_server = _EmbeddedServer(frontend_app)
         self.frontend_server.run()
 
     def _start_frontend(self):
@@ -869,22 +986,15 @@ def _start_frontend(self):
     def _wait_frontend(self):
         self._poll_for(
             lambda: (
-                self.frontend_server is not None
-                and getattr(self.frontend_server, "servers", [])
-                and self.frontend_server.servers[0].sockets
+                self.frontend_server is not None and self.frontend_server.is_listening()
             )
         )
-        if (
-            self.frontend_server is None
-            or not self.frontend_server.servers[0].sockets
-            or not self.frontend_server.servers[0].sockets[0].fileno()
-        ):
+        if self.frontend_server is None or not self.frontend_server.is_listening():
             msg = "Frontend did not start"
             raise RuntimeError(msg)
-        frontend_socket = self.frontend_server.servers[0].sockets[0]
         config = get_config()
         self.frontend_url = "http://{}:{}".format(
-            *frontend_socket.getsockname()
+            *self.frontend_server.getsockname()
         ) + config.prepend_frontend_path("/")
         config.deploy_url = self.frontend_url
 
@@ -893,14 +1003,7 @@ def _start_backend(self):
             msg = "App was not initialized."
             raise RuntimeError(msg)
         environment.REFLEX_SKIP_COMPILE.set(True)
-        self.backend = uvicorn.Server(
-            uvicorn.Config(
-                app=self.app_asgi,
-                host="127.0.0.1",
-                port=0,
-                workers=reflex.utils.processes.get_num_workers(),
-            ),
-        )
+        self.backend = _EmbeddedServer(self.app_asgi)
         self.backend.shutdown = self._get_backend_shutdown_handler()
 
         def _run_backend(context: contextvars.Context) -> None:
@@ -916,7 +1019,7 @@ def _run_backend(context: contextvars.Context) -> None:
         self.backend_thread.start()
         print("Backend started.")  # for pytest diagnosis #noqa: T201
 
-    def _poll_for_servers(self, timeout: TimeoutType = None) -> socket.socket:
+    def _poll_for_servers(self, timeout: TimeoutType = None) -> _EmbeddedServer:
         try:
             return super()._poll_for_servers(timeout)
         finally:
diff --git a/reflex/utils/exec.py b/reflex/utils/exec.py
index 780e86d7473..0e3e8142a89 100644
--- a/reflex/utils/exec.py
+++ b/reflex/utils/exec.py
@@ -407,6 +407,15 @@ def _warn_user_about_uvicorn():
     logger.warning(
         "Using Uvicorn for backend as it is installed. This behavior will change in 0.8.0 to use Granian by default."
     )
+    if (
+        importlib.util.find_spec("websockets") is None
+        and importlib.util.find_spec("wsproto") is None
+    ):
+        logger.warning(
+            "Uvicorn has no websocket protocol library installed, so the default "
+            "WebSocket transport will not connect. Install `reflex[uvicorn]` or "
+            "use Granian (REFLEX_USE_GRANIAN=1)."
+        )
 
 
 def should_use_granian():
@@ -630,9 +639,22 @@ def run_uvicorn_backend(host: str, port: int, loglevel: LogLevel):
         reload=True,
         reload_dirs=list(map(str, get_reload_paths())),
         reload_delay=0.1,
+        ws_max_size=_uvicorn_ws_max_size(),
     )
 
 
+def _uvicorn_ws_max_size() -> int:
+    """Websocket message size limit for uvicorn.
+
+    Never below uvicorn's 16 MiB default, so unrelated websocket endpoints
+    keep working; raised when the Reflex policy limit needs more.
+
+    Returns:
+        The message size limit in bytes.
+    """
+    return max(environment.REFLEX_SOCKET_MAX_HTTP_BUFFER_SIZE.get(), 16 * 1024 * 1024)
+
+
 HOTRELOAD_IGNORE_EXTENSIONS = (
     "txt",
     "toml",
@@ -749,6 +771,7 @@ def run_uvicorn_backend_prod(
             *("--host", host),
             *("--port", str(port)),
             *("--workers", str(_get_backend_workers())),
+            *("--ws-max-size", str(_uvicorn_ws_max_size())),
             "--factory",
             app_module,
         ]
diff --git a/tests/benchmarks/test_event_transport.py b/tests/benchmarks/test_event_transport.py
new file mode 100644
index 00000000000..4c9292ab084
--- /dev/null
+++ b/tests/benchmarks/test_event_transport.py
@@ -0,0 +1,285 @@
+"""Benchmarks comparing the plain WebSocket transport with Socket.IO.
+
+Measures the server-side transport layer in isolation: inbound event frames
+from an established connection to the (mocked) event processor, and outbound
+state updates to the (mocked) wire. Both transports share BaseEventNamespace,
+so the difference is the framing and dispatch layer. Socket.IO runs with
+``async_handlers=False`` (inline dispatch), its cheapest configuration.
+"""
+
+import asyncio
+import json
+from types import SimpleNamespace
+from typing import Any
+from unittest import mock
+
+import pytest
+import pytest_asyncio
+from pytest_codspeed import BenchmarkFixture
+from reflex_base.utils import format
+
+from reflex.event_namespace import WebsocketEventNamespace
+from reflex.state import StateUpdate
+
+NUM_MESSAGES = 100
+NAMESPACE = "/_event"
+TOKEN = "benchmark-token"
+
+_EVENT_FIELDS = {
+    "name": "benchmark___state.increment",
+    "router_data": {
+        "pathname": "/benchmark",
+        "asPath": "/benchmark?tab=2",
+        "query": {"tab": "2"},
+    },
+    "payload": {"value": 42, "label": "increment", "flag": True},
+}
+
+_UPDATE = StateUpdate(
+    delta={
+        "benchmark___state": {f"var_{i}": f"value_{i}" for i in range(15)}
+        | {"counter": 42, "flag": True, "items": list(range(10))},
+    }
+)
+
+_DISCONNECT = object()
+
+_ASGI_SCOPE = {
+    "type": "websocket",
+    "headers": [(b"host", b"localhost")],
+    "client": ("127.0.0.1", 1234),
+}
+
+
+def _make_app(sio: Any = None) -> SimpleNamespace:
+    """Build a minimal app double for the event namespace.
+
+    Returns:
+        The app double.
+    """
+    enqueued: list[Any] = []
+
+    async def enqueue(token: str, event: Any) -> None:  # noqa: RUF029
+        enqueued.append((token, event))
+
+    return SimpleNamespace(
+        _state=None,
+        sio=sio,
+        router=lambda _path: None,
+        event_processor=SimpleNamespace(enqueue=enqueue),
+        enqueued=enqueued,
+    )
+
+
+class FakeWebSocket:
+    """Minimal stand-in for a starlette WebSocket."""
+
+    def __init__(self, frames: list[str]):
+        """Initialize with the inbound frames to deliver."""
+        self.scope: dict[str, Any] = {
+            "type": "websocket",
+            "query_string": f"token={TOKEN}".encode(),
+            "subprotocols": [],
+            "headers": [(b"host", b"localhost")],
+            "client": ("127.0.0.1", 1234),
+        }
+        self.headers: dict[str, str] = {}
+        self.sent: list[str] = []
+        self._incoming = [*frames, _DISCONNECT]
+        self._pos = 0
+
+    async def accept(self, subprotocol: str | None = None):
+        """Accept the connection."""
+
+    async def send_text(self, text: str):
+        """Record an outgoing frame."""
+        self.sent.append(text)
+
+    async def receive(self) -> dict[str, Any]:
+        """Return the next queued frame as an ASGI message.
+
+        Returns:
+            The ASGI websocket message.
+        """
+        item = self._incoming[self._pos]
+        self._pos += 1
+        if item is _DISCONNECT:
+            return {"type": "websocket.disconnect", "code": 1000}
+        return {"type": "websocket.receive", "text": item}
+
+    async def close(self, code: int = 1000):
+        """Close the connection."""
+
+
+@pytest_asyncio.fixture
+async def websocket_inbound():  # noqa: RUF029 - async so it runs on the benchmark loop
+    """Runner delivering NUM_MESSAGES event frames over the plain transport.
+
+    Yields:
+        An async callable running one full connection lifecycle.
+    """
+    with mock.patch("reflex.utils.prerequisites.check_redis_used", return_value=False):
+        app = _make_app()
+        namespace = WebsocketEventNamespace(NAMESPACE, app)  # pyright: ignore[reportArgumentType]
+        frames = [json.dumps(["event", _EVENT_FIELDS])] * NUM_MESSAGES
+
+        async def run() -> None:
+            websocket = FakeWebSocket(frames)
+            await namespace.handle_websocket(websocket)  # pyright: ignore[reportArgumentType]
+            assert len(app.enqueued) >= NUM_MESSAGES
+
+        yield run
+
+
+@pytest_asyncio.fixture
+async def socketio_inbound():  # noqa: RUF029 - async so it runs on the benchmark loop
+    """Runner delivering NUM_MESSAGES event packets over Socket.IO.
+
+    Yields:
+        An async callable running one full connection lifecycle.
+    """
+    pytest.importorskip("socketio")
+    from socketio import AsyncServer
+
+    from reflex.socketio_namespace import EventNamespace
+
+    sio = AsyncServer(
+        async_mode="asgi",
+        async_handlers=False,
+        json=SimpleNamespace(
+            dumps=staticmethod(format.json_dumps),
+            loads=staticmethod(json.loads),
+        ),
+    )
+    with mock.patch("reflex.utils.prerequisites.check_redis_used", return_value=False):
+        app = _make_app(sio=sio)
+        namespace = EventNamespace(NAMESPACE, app)  # pyright: ignore[reportArgumentType]
+        sio.register_namespace(namespace)
+
+    async def eio_send(_eio_sid: str, _data: str) -> None:
+        pass
+
+    sio.eio.send = eio_send
+    event_packet = "2" + NAMESPACE + "," + json.dumps(["event", _EVENT_FIELDS])
+    counter = 0
+
+    async def run() -> None:
+        nonlocal counter
+        counter += 1
+        eio_sid = f"eio-{counter}"
+        await sio._handle_eio_connect(
+            eio_sid,
+            {"QUERY_STRING": f"token={TOKEN}-{counter}", "asgi.scope": _ASGI_SCOPE},
+        )
+        await sio._handle_eio_message(eio_sid, "0" + NAMESPACE + ",")
+        for _ in range(NUM_MESSAGES):
+            await sio._handle_eio_message(eio_sid, event_packet)
+        await sio._handle_eio_message(eio_sid, "1" + NAMESPACE + ",")
+        # Let the disconnect cleanup task run.
+        for _ in range(3):
+            await asyncio.sleep(0)
+        assert len(app.enqueued) >= NUM_MESSAGES
+
+    yield run
+
+
+@pytest_asyncio.fixture
+async def websocket_outbound():
+    """Runner emitting NUM_MESSAGES state updates over the plain transport.
+
+    Yields:
+        An async callable emitting the updates.
+    """
+    with mock.patch("reflex.utils.prerequisites.check_redis_used", return_value=False):
+        app = _make_app()
+        namespace = WebsocketEventNamespace(NAMESPACE, app)  # pyright: ignore[reportArgumentType]
+        websocket = FakeWebSocket([])
+        namespace._sockets["sid-1"] = websocket  # pyright: ignore[reportArgumentType]
+        await namespace.link_token_to_sid("sid-1", TOKEN)
+
+        async def run() -> None:
+            for _ in range(NUM_MESSAGES):
+                await namespace.emit_update(_UPDATE, TOKEN)
+
+        yield run
+
+
+@pytest_asyncio.fixture
+async def socketio_outbound():
+    """Runner emitting NUM_MESSAGES state updates over Socket.IO.
+
+    Yields:
+        An async callable emitting the updates.
+    """
+    pytest.importorskip("socketio")
+    from socketio import AsyncServer
+
+    from reflex.socketio_namespace import EventNamespace
+
+    sio = AsyncServer(
+        async_mode="asgi",
+        async_handlers=False,
+        json=SimpleNamespace(
+            dumps=staticmethod(format.json_dumps),
+            loads=staticmethod(json.loads),
+        ),
+    )
+    with mock.patch("reflex.utils.prerequisites.check_redis_used", return_value=False):
+        app = _make_app(sio=sio)
+        namespace = EventNamespace(NAMESPACE, app)  # pyright: ignore[reportArgumentType]
+        sio.register_namespace(namespace)
+
+    async def eio_send(_eio_sid: str, _data: str) -> None:
+        pass
+
+    sio.eio.send = eio_send
+
+    async def run() -> None:
+        for _ in range(NUM_MESSAGES):
+            await namespace.emit_update(_UPDATE, TOKEN)
+
+    # Connect a socket.io session and link the token to its sid.
+    eio_sid = "eio-emit"
+    await sio._handle_eio_connect(
+        eio_sid, {"QUERY_STRING": f"token={TOKEN}", "asgi.scope": _ASGI_SCOPE}
+    )
+    await sio._handle_eio_message(eio_sid, "0" + NAMESPACE + ",")
+    assert TOKEN in namespace.token_to_sid
+
+    yield run
+
+
+def test_transport_inbound_websocket(websocket_inbound, benchmark: BenchmarkFixture):
+    """Benchmark inbound event handling on the plain WebSocket transport."""
+    loop = asyncio.get_event_loop()
+
+    @benchmark
+    def _():
+        loop.run_until_complete(websocket_inbound())
+
+
+def test_transport_inbound_socketio(socketio_inbound, benchmark: BenchmarkFixture):
+    """Benchmark inbound event handling on the Socket.IO transport."""
+    loop = asyncio.get_event_loop()
+
+    @benchmark
+    def _():
+        loop.run_until_complete(socketio_inbound())
+
+
+def test_transport_outbound_websocket(websocket_outbound, benchmark: BenchmarkFixture):
+    """Benchmark emitting state updates on the plain WebSocket transport."""
+    loop = asyncio.get_event_loop()
+
+    @benchmark
+    def _():
+        loop.run_until_complete(websocket_outbound())
+
+
+def test_transport_outbound_socketio(socketio_outbound, benchmark: BenchmarkFixture):
+    """Benchmark emitting state updates on the Socket.IO transport."""
+    loop = asyncio.get_event_loop()
+
+    @benchmark
+    def _():
+        loop.run_until_complete(socketio_outbound())
diff --git a/tests/integration/tests_playwright/test_stateless_app.py b/tests/integration/tests_playwright/test_stateless_app.py
index 0451230a556..de88ef12f53 100644
--- a/tests/integration/tests_playwright/test_stateless_app.py
+++ b/tests/integration/tests_playwright/test_stateless_app.py
@@ -48,7 +48,7 @@ def test_statelessness(stateless_app: AppHarness, page: Page):
     """
     assert stateless_app.frontend_url is not None
     assert stateless_app.backend is not None
-    assert stateless_app.backend.started
+    assert stateless_app.backend.is_listening()
 
     config = get_config()
     res = httpx.get(config.api_url + config.prepend_backend_path(str(Endpoint.EVENT)))
diff --git a/tests/units/reflex_base/event/processor/test_event_processor.py b/tests/units/reflex_base/event/processor/test_event_processor.py
index d5dda19dca3..65b469c6689 100644
--- a/tests/units/reflex_base/event/processor/test_event_processor.py
+++ b/tests/units/reflex_base/event/processor/test_event_processor.py
@@ -611,6 +611,34 @@ async def test_stream_delta_noop_handler_yields_nothing(token: str):
     assert collected == []
 
 
+async def test_stream_delta_future_does_not_claim_root_txid(token: str):
+    """Regression: a streamed event must not reuse the root context's txid.
+
+    Otherwise unrelated events forking from the root context attach to the
+    stream's future as children (#6932).
+
+    Args:
+        token: The client token.
+    """
+    ep = EventProcessor(graceful_shutdown_timeout=2)
+    ep.configure()
+    assert ep._root_context is not None
+    root_txid = ep._root_context.txid
+    async with ep:
+        event = Event.from_event_type(delta_event())[0]
+        root_txid_futures = []
+        parents = []
+        async for _ in ep.enqueue_stream_delta(token, event):
+            root_txid_futures.append(ep._futures.get(root_txid))
+            unrelated = await ep.enqueue(token, Event.from_event_type(noop_event())[0])
+            parents.append(unrelated.parent)
+        assert root_txid_futures
+        assert all(f is None for f in root_txid_futures)
+        assert parents
+        assert all(parent is None for parent in parents)
+        await ep.join(timeout=5)
+
+
 async def test_stream_delta_not_configured_raises():
     """enqueue_stream_delta raises RuntimeError if processor is not configured."""
     ep = EventProcessor()
diff --git a/tests/units/test_app.py b/tests/units/test_app.py
index 484796337a5..7b46e069d79 100644
--- a/tests/units/test_app.py
+++ b/tests/units/test_app.py
@@ -4069,7 +4069,7 @@ def emit(self, record: logging.LogRecord):
             if key is not None:
                 captured[key].append(record.getMessage())
 
-    app_logger = logging.getLogger("reflex.app")
+    app_logger = logging.getLogger("reflex.event_namespace")
     handler = _CaptureHandler(level=logging.DEBUG)
     previous_level = app_logger.level
     app_logger.addHandler(handler)
diff --git a/tests/units/test_event_namespace.py b/tests/units/test_event_namespace.py
new file mode 100644
index 00000000000..b4b1daec5cb
--- /dev/null
+++ b/tests/units/test_event_namespace.py
@@ -0,0 +1,444 @@
+"""Tests for the plain WebSocket event transport in reflex/event_namespace.py."""
+
+import asyncio
+import json
+from typing import Any
+from unittest.mock import AsyncMock, Mock
+
+import pytest
+from starlette.routing import WebSocketRoute
+
+from reflex.app import App
+from reflex.event_namespace import (
+    HANDSHAKE_MESSAGE,
+    PONG_MESSAGE,
+    WebsocketEventNamespace,
+)
+
+_DISCONNECT = object()
+
+
+class FakeWebSocket:
+    """Minimal stand-in for a starlette WebSocket."""
+
+    def __init__(
+        self,
+        query_string: bytes = b"token=tok1",
+        origin: str | None = None,
+        subprotocols: list[str] | None = None,
+    ):
+        """Initialize the fake websocket."""
+        self.scope: dict[str, Any] = {
+            "type": "websocket",
+            "query_string": query_string,
+            "subprotocols": subprotocols or [],
+            "headers": [(b"host", b"localhost")],
+            "client": ("127.0.0.1", 1234),
+        }
+        self.headers = {"origin": origin} if origin is not None else {}
+        self.sent: list[Any] = []
+        self.accepted_subprotocol: str | None = None
+        self.accepted = False
+        self.close_code: int | None = None
+        self._incoming: asyncio.Queue = asyncio.Queue()
+
+    async def accept(self, subprotocol: str | None = None):
+        """Record the accept call."""
+        self.accepted = True
+        self.accepted_subprotocol = subprotocol
+
+    async def send_text(self, text: str):
+        """Record an outgoing frame."""
+        self.sent.append(json.loads(text))
+
+    async def close(self, code: int = 1000):
+        """Record the close call."""
+        self.close_code = code
+
+    async def receive(self) -> dict[str, Any]:
+        """Return the next queued frame as an ASGI message.
+
+        Returns:
+            The ASGI websocket message.
+        """
+        item = await self._incoming.get()
+        if item is _DISCONNECT:
+            return {"type": "websocket.disconnect", "code": 1000}
+        if isinstance(item, bytes):
+            return {"type": "websocket.receive", "bytes": item}
+        return {"type": "websocket.receive", "text": item}
+
+    def feed(self, *frames: Any):
+        """Queue incoming frames (lists are JSON-encoded) and a disconnect."""
+        for frame in frames:
+            self._incoming.put_nowait(
+                frame if isinstance(frame, (str, bytes)) else json.dumps(frame)
+            )
+        self._incoming.put_nowait(_DISCONNECT)
+
+
+@pytest.fixture
+def mock_app() -> Mock:
+    """A mock app for the event namespace.
+
+    Returns:
+        The mock app.
+    """
+    app = Mock()
+    app._state = None
+    app.router = Mock(return_value=None)
+    app.event_processor.enqueue = AsyncMock()
+    return app
+
+
+@pytest.fixture
+def namespace(mock_app: Mock, mocker) -> WebsocketEventNamespace:
+    """A websocket event namespace with a mock app and a local token manager.
+
+    Redis is disabled so token linking cannot leak into a shared Redis.
+
+    Returns:
+        The namespace.
+    """
+    mocker.patch("reflex.utils.prerequisites.check_redis_used", return_value=False)
+    return WebsocketEventNamespace("/_event", mock_app)
+
+
+async def _drain_tasks():
+    """Let pending disconnect-cleanup tasks run to completion."""
+    for _ in range(3):
+        await asyncio.sleep(0)
+
+
+@pytest.mark.asyncio
+async def test_handshake_and_token_link(namespace: WebsocketEventNamespace):
+    """The server sends the handshake first and links the token from the query."""
+    websocket = FakeWebSocket(subprotocols=["0.0.1"])
+    websocket.feed()
+    await namespace.handle_websocket(websocket)  # pyright: ignore[reportArgumentType]
+
+    assert websocket.accepted
+    assert websocket.accepted_subprotocol == "0.0.1"
+    assert websocket.sent[0][0] == HANDSHAKE_MESSAGE
+    assert set(websocket.sent[0][1]) == {"ping_interval", "ping_timeout"}
+    await _drain_tasks()
+    # The session was linked and unlinked again on disconnect.
+    assert "tok1" not in namespace.token_to_sid
+
+
+@pytest.mark.asyncio
+async def test_event_is_enqueued(namespace: WebsocketEventNamespace, mock_app: Mock):
+    """An incoming event frame reaches the app's event processor."""
+    websocket = FakeWebSocket()
+    websocket.feed([
+        "event",
+        {"token": "tok1", "name": "state.on_click", "payload": {}, "router_data": {}},
+    ])
+    await namespace.handle_websocket(websocket)  # pyright: ignore[reportArgumentType]
+    await _drain_tasks()
+
+    mock_app.event_processor.enqueue.assert_awaited_once()
+    token, event = mock_app.event_processor.enqueue.await_args.args
+    assert token == "tok1"
+    assert event.name == "state.on_click"
+    assert event.router_data["headers"]["host"] == "localhost"
+    assert event.router_data["ip"] == "127.0.0.1"
+
+
+@pytest.mark.asyncio
+async def test_ping_pong(namespace: WebsocketEventNamespace):
+    """An application-level ping event gets a pong reply."""
+    websocket = FakeWebSocket()
+    websocket.feed(["ping"], [PONG_MESSAGE])
+    await namespace.handle_websocket(websocket)  # pyright: ignore[reportArgumentType]
+    await _drain_tasks()
+
+    assert ["ping", "pong"] in websocket.sent
+
+
+@pytest.mark.asyncio
+async def test_client_error_reaches_exception_handler(
+    namespace: WebsocketEventNamespace, mock_app: Mock
+):
+    """A client_error frame is routed to the frontend exception handler."""
+    errors: list[str] = []
+    mock_app.frontend_exception_handler = lambda exc: errors.append(str(exc))
+    websocket = FakeWebSocket()
+    websocket.feed(["client_error", {"error_type": "boom", "message": "it broke"}])
+    await namespace.handle_websocket(websocket)  # pyright: ignore[reportArgumentType]
+    await _drain_tasks()
+
+    assert len(errors) == 1
+    assert "it broke" in errors[0]
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("frame", ["not json", '{"an": "object"}', "[42]"])
+async def test_malformed_frame_closes_connection(
+    namespace: WebsocketEventNamespace, frame: str
+):
+    """A malformed frame closes the connection with 1002 (protocol error)."""
+    websocket = FakeWebSocket()
+    websocket.feed(frame, ["ping"])
+    await namespace.handle_websocket(websocket)  # pyright: ignore[reportArgumentType]
+    await _drain_tasks()
+
+    assert websocket.close_code == 1002
+    # Nothing after the malformed frame is processed.
+    assert ["ping", "pong"] not in websocket.sent
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "payload",
+    [
+        None,
+        "not an event",
+        42,
+        {"name": 123, "payload": {}, "router_data": {}},
+        {"name": "x", "payload": "nope", "router_data": {}},
+        {"name": "x", "payload": {}, "router_data": {"query": "not-a-dict"}},
+    ],
+)
+async def test_undeserializable_event_closes_connection(
+    namespace: WebsocketEventNamespace, payload: object
+):
+    """An event frame that fails deserialization closes with 1002."""
+    websocket = FakeWebSocket()
+    websocket.feed(["event", payload], ["ping"])
+    await namespace.handle_websocket(websocket)  # pyright: ignore[reportArgumentType]
+    await _drain_tasks()
+
+    assert websocket.close_code == 1002
+    assert ["ping", "pong"] not in websocket.sent
+
+
+@pytest.mark.asyncio
+async def test_handler_error_keeps_connection(
+    namespace: WebsocketEventNamespace,
+    mock_app: Mock,
+    caplog: pytest.LogCaptureFixture,
+):
+    """A server-side handler failure is logged and the connection survives."""
+    import logging
+
+    mock_app.event_processor.enqueue.side_effect = RuntimeError("server bug")
+    websocket = FakeWebSocket()
+    websocket.feed(
+        ["event", {"name": "state.on_click", "payload": {}, "router_data": {}}],
+        ["ping"],
+    )
+    with caplog.at_level(logging.ERROR, logger="reflex.event_namespace"):
+        await namespace.handle_websocket(websocket)  # pyright: ignore[reportArgumentType]
+    await _drain_tasks()
+
+    assert websocket.close_code is None
+    assert ["ping", "pong"] in websocket.sent
+    assert any(
+        record.levelno == logging.ERROR
+        and "Error handling socket event" in record.getMessage()
+        for record in caplog.records
+    )
+
+
+@pytest.mark.asyncio
+async def test_tokenless_connection_rejected(namespace: WebsocketEventNamespace):
+    """A connection without a token closes with 1008 (policy violation)."""
+    websocket = FakeWebSocket(query_string=b"")
+    websocket.feed(["ping"])
+    await namespace.handle_websocket(websocket)  # pyright: ignore[reportArgumentType]
+    await _drain_tasks()
+
+    assert websocket.close_code == 1008
+    assert ["ping", "pong"] not in websocket.sent
+
+
+@pytest.mark.asyncio
+async def test_oversize_message_closes_connection(
+    namespace: WebsocketEventNamespace, monkeypatch: pytest.MonkeyPatch
+):
+    """A frame over the size limit closes the connection with 1009."""
+    monkeypatch.setenv("REFLEX_SOCKET_MAX_HTTP_BUFFER_SIZE", "10")
+    websocket = FakeWebSocket()
+    websocket.feed(["event", {"payload": "x" * 100}])
+    await namespace.handle_websocket(websocket)  # pyright: ignore[reportArgumentType]
+    await _drain_tasks()
+
+    assert websocket.close_code == 1009
+
+
+@pytest.mark.asyncio
+async def test_oversize_multibyte_message_closes_connection(
+    namespace: WebsocketEventNamespace, monkeypatch: pytest.MonkeyPatch
+):
+    """The size limit counts bytes, so multibyte text cannot sneak past it."""
+    monkeypatch.setenv("REFLEX_SOCKET_MAX_HTTP_BUFFER_SIZE", "25")
+    # 15 characters (under the limit) but 29 UTF-8 bytes (over it).
+    frame = '["x","€€€€€€€"]'
+    assert len(frame) <= 25 < len(frame.encode("utf-8"))
+    websocket = FakeWebSocket()
+    websocket.feed(frame)
+    await namespace.handle_websocket(websocket)  # pyright: ignore[reportArgumentType]
+    await _drain_tasks()
+
+    assert websocket.close_code == 1009
+
+
+@pytest.mark.asyncio
+async def test_multibyte_message_within_limit_is_processed(
+    namespace: WebsocketEventNamespace, monkeypatch: pytest.MonkeyPatch
+):
+    """Multibyte frames within the byte limit pass through the exact check."""
+    # 12 characters, 14 bytes: over limit/4 (triggers the exact byte count)
+    # but within the limit itself.
+    monkeypatch.setenv("REFLEX_SOCKET_MAX_HTTP_BUFFER_SIZE", "14")
+    websocket = FakeWebSocket()
+    websocket.feed('["ping","€"]')
+    await namespace.handle_websocket(websocket)  # pyright: ignore[reportArgumentType]
+    await _drain_tasks()
+
+    assert websocket.close_code is None
+    assert ["ping", "pong"] in websocket.sent
+
+
+@pytest.mark.asyncio
+async def test_binary_frame_closes_connection(namespace: WebsocketEventNamespace):
+    """A binary frame closes the connection with 1003 (unsupported data)."""
+    websocket = FakeWebSocket()
+    websocket.feed(b"\x00\x01")
+    await namespace.handle_websocket(websocket)  # pyright: ignore[reportArgumentType]
+    await _drain_tasks()
+
+    assert websocket.close_code == 1003
+
+
+@pytest.mark.asyncio
+async def test_disallowed_origin_is_rejected(
+    namespace: WebsocketEventNamespace, mocker
+):
+    """A cross-origin connection is closed before being accepted."""
+    from reflex_base.config import get_config
+
+    mocker.patch.object(
+        get_config(), "cors_allowed_origins", ("https://allowed.example",)
+    )
+    websocket = FakeWebSocket(origin="https://evil.example")
+    await namespace.handle_websocket(websocket)  # pyright: ignore[reportArgumentType]
+
+    assert not websocket.accepted
+    assert websocket.close_code == 1008
+
+
+@pytest.mark.asyncio
+async def test_allowed_origin_is_accepted(namespace: WebsocketEventNamespace, mocker):
+    """A connection from an allowed origin is accepted."""
+    from reflex_base.config import get_config
+
+    mocker.patch.object(
+        get_config(), "cors_allowed_origins", ("https://allowed.example",)
+    )
+    websocket = FakeWebSocket(origin="https://allowed.example")
+    websocket.feed()
+    await namespace.handle_websocket(websocket)  # pyright: ignore[reportArgumentType]
+    await _drain_tasks()
+
+    assert websocket.accepted
+
+
+@pytest.mark.asyncio
+async def test_duplicate_token_gets_new_token(namespace: WebsocketEventNamespace):
+    """A second tab connecting with the same token receives a new_token frame."""
+    first = FakeWebSocket()
+    second = FakeWebSocket()
+    namespace._sockets["sid1"] = first  # pyright: ignore[reportArgumentType]
+    namespace._sockets["sid2"] = second  # pyright: ignore[reportArgumentType]
+    await namespace.link_token_to_sid("sid1", "tok1")
+    await namespace.link_token_to_sid("sid2", "tok1")
+
+    new_token_frames = [frame for frame in second.sent if frame[0] == "new_token"]
+    assert len(new_token_frames) == 1
+    assert new_token_frames[0][1] != "tok1"
+
+
+@pytest.mark.asyncio
+async def test_emit_to_unknown_sid_does_not_raise(
+    namespace: WebsocketEventNamespace,
+    caplog: pytest.LogCaptureFixture,
+):
+    """Emitting to a session that went away is a silent no-op.
+
+    A client disconnecting mid-event is routine, so nothing above DEBUG may be
+    logged.
+    """
+    import logging
+
+    with caplog.at_level(logging.DEBUG, logger="reflex.event_namespace"):
+        await namespace.emit("event", {"delta": {}}, to="gone")
+    assert all(record.levelno <= logging.DEBUG for record in caplog.records)
+
+
+def test_default_transport_uses_websocket_namespace():
+    """The default transport sets up the plain websocket namespace."""
+    app = App(enable_state=True)
+    assert isinstance(app.event_namespace, WebsocketEventNamespace)
+    assert app.sio is None
+    assert app._api is not None
+    websocket_routes = [
+        route for route in app._api.router.routes if isinstance(route, WebSocketRoute)
+    ]
+    assert [route.path for route in websocket_routes] == ["/_event"]
+
+
+def test_socketio_transport_uses_socketio_namespace(
+    monkeypatch: pytest.MonkeyPatch,
+):
+    """transport="socketio" sets up the Socket.IO server and namespace."""
+    from reflex.socketio_namespace import EventNamespace
+
+    monkeypatch.setenv("REFLEX_TRANSPORT", "socketio")
+    app = App(enable_state=True)
+    assert isinstance(app.event_namespace, EventNamespace)
+    assert app.sio is not None
+    # Plain websocket transport under the hood.
+    assert app.sio.eio.transports == ["websocket"]
+
+
+def test_polling_transport_uses_socketio_namespace(
+    monkeypatch: pytest.MonkeyPatch,
+):
+    """transport="polling" sets up the Socket.IO server with polling only."""
+    from reflex.socketio_namespace import EventNamespace
+
+    monkeypatch.setenv("REFLEX_TRANSPORT", "polling")
+    app = App(enable_state=True)
+    assert isinstance(app.event_namespace, EventNamespace)
+    assert app.sio is not None
+    assert app.sio.eio.transports == ["polling"]
+
+
+def test_custom_sio_requires_socketio_transport():
+    """A custom sio server with the default transport raises a clear error."""
+    from socketio import AsyncServer
+
+    with pytest.raises(RuntimeError, match=r"requires the Socket\.IO transport"):
+        App(sio=AsyncServer(async_mode="asgi"))
+
+
+def test_custom_sio_with_socketio_transport(monkeypatch: pytest.MonkeyPatch):
+    """A custom sio server works with the Socket.IO transport."""
+    from socketio import AsyncServer
+
+    monkeypatch.setenv("REFLEX_TRANSPORT", "socketio")
+    sio = AsyncServer(async_mode="asgi")
+    app = App(sio=sio)
+    assert app.sio is sio
+
+
+def test_app_event_namespace_reexport():
+    """reflex.app.EventNamespace still resolves to the Socket.IO namespace."""
+    import reflex.app
+    from reflex.socketio_namespace import EventNamespace
+
+    assert reflex.app.EventNamespace is EventNamespace
+    with pytest.raises(AttributeError):
+        _ = reflex.app.DoesNotExist
diff --git a/tests/units/test_testing.py b/tests/units/test_testing.py
index a38682c06af..e77e4874aff 100644
--- a/tests/units/test_testing.py
+++ b/tests/units/test_testing.py
@@ -189,6 +189,116 @@ def test_app_harness_initialize_reloads_existing_imported_app(
     harness_mocks.get_and_validate_app.assert_called_once_with(reload=True)
 
 
+def test_embedded_server_retries_taken_port():
+    """The embedded server rebinds to a fresh port when its probed port is taken."""
+    import socket
+    import threading
+    import time
+
+    async def app(scope, receive, send):
+        if scope["type"] == "lifespan":
+            while True:
+                message = await receive()
+                if message["type"] == "lifespan.startup":
+                    await send({"type": "lifespan.startup.complete"})
+                elif message["type"] == "lifespan.shutdown":
+                    await send({"type": "lifespan.shutdown.complete"})
+                    return
+
+    server = reflex_testing._EmbeddedServer(app)
+    probed_port = server.port
+    # Steal the probed port before the server binds it. On Windows only
+    # SO_EXCLUSIVEADDRUSE makes the port unavailable to other binders.
+    blocker = socket.socket()
+    if exclusive := getattr(socket, "SO_EXCLUSIVEADDRUSE", None):
+        blocker.setsockopt(socket.SOL_SOCKET, exclusive, 1)
+    blocker.bind((server.host, probed_port))
+    blocker.listen(1)
+    thread = threading.Thread(target=server.run)
+    thread.start()
+    try:
+        deadline = time.monotonic() + 15
+        while time.monotonic() < deadline:
+            if server.port != probed_port and server.is_listening():
+                break
+            time.sleep(0.05)
+        assert server.port != probed_port
+        assert server.is_listening()
+    finally:
+        blocker.close()
+        server.should_exit = True
+        thread.join(timeout=15)
+        assert not thread.is_alive()
+
+
+def test_embedded_server_stops_after_unexpected_serve_return(monkeypatch):
+    """A serve() return that was not requested stops the server without rebinding."""
+    import granian.server.embed
+
+    class FakeServer:
+        def __init__(self, *args, **kwargs):
+            pass
+
+        async def serve(self):
+            return
+
+    monkeypatch.setattr(granian.server.embed, "Server", FakeServer)
+    server = reflex_testing._EmbeddedServer(app=FakeServer)  # pyright: ignore[reportArgumentType]
+    port_before = server.port
+    server.run()
+    assert server.port == port_before
+
+
+def test_embedded_server_raises_after_retries_exhausted(monkeypatch):
+    """Exhausted bind retries re-raise the error instead of returning silently."""
+    import granian.server.embed
+
+    calls = []
+
+    class FakeServer:
+        def __init__(self, *args, **kwargs):
+            pass
+
+        async def serve(self):
+            calls.append(1)
+            bind_error = "Address already in use (os error 98)"
+            raise RuntimeError(bind_error)
+
+    monkeypatch.setattr(granian.server.embed, "Server", FakeServer)
+    server = reflex_testing._EmbeddedServer(app=FakeServer)  # pyright: ignore[reportArgumentType]
+    with pytest.raises(RuntimeError, match="in use"):
+        server.run()
+    assert len(calls) == 10
+
+
+def test_embedded_server_shutdown_wins_over_exhausted_retries(monkeypatch):
+    """A stop requested during the last failed bind ends the server cleanly."""
+    import threading
+
+    import granian.server.embed
+
+    calls = []
+    holder = {}
+
+    class FakeServer:
+        def __init__(self, *args, **kwargs):
+            self.interrupt_signal = False
+            self.main_loop_interrupt = threading.Event()
+
+        async def serve(self):
+            calls.append(1)
+            if len(calls) == 10:
+                holder["server"].should_exit = True
+            bind_error = "Address already in use (os error 98)"
+            raise RuntimeError(bind_error)
+
+    monkeypatch.setattr(granian.server.embed, "Server", FakeServer)
+    server = reflex_testing._EmbeddedServer(app=FakeServer)  # pyright: ignore[reportArgumentType]
+    holder["server"] = server
+    server.run()
+    assert len(calls) == 10
+
+
 def test_app_harness_frontend_env_has_development_condition(
     tmp_path, monkeypatch: pytest.MonkeyPatch, harness_mocks
 ) -> None:
diff --git a/uv.lock b/uv.lock
index ab4dfd64387..aee828ef476 100644
--- a/uv.lock
+++ b/uv.lock
@@ -585,7 +585,7 @@ resolution-markers = [
     "python_full_version < '3.11'",
 ]
 dependencies = [
-    { name = "numpy", version = "2.2.6", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.11'" },
+    { name = "numpy", version = "2.2.6", source = { registry = "https://pypi.org/simple" } },
 ]
 sdist = { url = "https://files.pythonhosted.org/packages/66/54/eb9bfc647b19f2009dd5c7f5ec51c4e6ca831725f1aea7a993034f483147/contourpy-1.3.2.tar.gz", hash = "sha256:b6945942715a034c671b7fc54f9588126b0b8bf23db2696e3ca8328f3ff0ab54", size = 13466130, upload-time = "2025-04-15T17:47:53.79Z" }
 wheels = [
@@ -663,7 +663,7 @@ resolution-markers = [
     "python_full_version == '3.11.*' and sys_platform != 'emscripten' and sys_platform != 'win32'",
 ]
 dependencies = [
-    { name = "numpy", version = "2.4.6", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version == '3.11.*'" },
+    { name = "numpy", version = "2.4.6", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.12'" },
     { name = "numpy", version = "2.5.1", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.12'" },
 ]
 sdist = { url = "https://files.pythonhosted.org/packages/58/01/1253e6698a07380cd31a736d248a3f2a50a7c88779a1813da27503cadc2a/contourpy-1.3.3.tar.gz", hash = "sha256:083e12155b210502d0bca491432bb04d56dc3432f95a979b429f2848c3dbe880", size = 13466174, upload-time = "2025-07-26T12:03:12.549Z" }
@@ -1003,7 +1003,7 @@ name = "exceptiongroup"
 version = "1.3.1"
 source = { registry = "https://pypi.org/simple" }
 dependencies = [
-    { name = "typing-extensions", marker = "python_full_version < '3.11'" },
+    { name = "typing-extensions" },
 ]
 sdist = { url = "https://files.pythonhosted.org/packages/50/79/66800aadf48771f6b62f7eb014e352e5d06856655206165d775e675a02c9/exceptiongroup-1.3.1.tar.gz", hash = "sha256:8b412432c6055b0b7d14c310000ae93352ed6754f70fa8f7c34141f91c4e3219", size = 30371, upload-time = "2025-11-21T23:01:54.787Z" }
 wheels = [
@@ -1412,6 +1412,18 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/e4/d3/5268aeabf2ad82658c4e2ff3a060648d0f02f3926cb53247c0e4d0dab49e/griffelib-2.1.0-py3-none-any.whl", hash = "sha256:cc7b3d2d2865ad0b909fcc38086e3f554b5ea7acbaa7bbb7ecaa3f5dfb7d9f00", size = 142560, upload-time = "2026-06-19T12:05:38.742Z" },
 ]
 
+[[package]]
+name = "gunicorn"
+version = "26.0.0"
+source = { registry = "https://pypi.org/simple" }
+dependencies = [
+    { name = "packaging" },
+]
+sdist = { url = "https://files.pythonhosted.org/packages/6d/b7/a4a3f632f823e432ce6bc65f62961b7980c898c77f075a2f7118cb3846fe/gunicorn-26.0.0.tar.gz", hash = "sha256:ca9346f85e3a4aeeb64d491045c16b9a35647abd37ea15efe53080eb8b090baf", size = 727286, upload-time = "2026-05-05T06:38:25.529Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/e6/40/9c2384fc2be4ad25dd4a49decd5ad9ea5a3639814c11bd40ab77cb9f0a14/gunicorn-26.0.0-py3-none-any.whl", hash = "sha256:40233d26a5f0d1872916188c276e21641155111c2853f0c2cd55260aec0d24fc", size = 212009, upload-time = "2026-05-05T06:38:23.007Z" },
+]
+
 [[package]]
 name = "h11"
 version = "0.16.0"
@@ -1879,15 +1891,15 @@ resolution-markers = [
     "python_full_version < '3.11'",
 ]
 dependencies = [
-    { name = "contourpy", version = "1.3.2", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.11'" },
-    { name = "cycler", marker = "python_full_version < '3.11'" },
-    { name = "fonttools", marker = "python_full_version < '3.11'" },
-    { name = "kiwisolver", marker = "python_full_version < '3.11'" },
-    { name = "numpy", version = "2.2.6", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.11'" },
-    { name = "packaging", marker = "python_full_version < '3.11'" },
-    { name = "pillow", marker = "python_full_version < '3.11'" },
-    { name = "pyparsing", marker = "python_full_version < '3.11'" },
-    { name = "python-dateutil", marker = "python_full_version < '3.11'" },
+    { name = "contourpy", version = "1.3.2", source = { registry = "https://pypi.org/simple" } },
+    { name = "cycler" },
+    { name = "fonttools" },
+    { name = "kiwisolver" },
+    { name = "numpy", version = "2.2.6", source = { registry = "https://pypi.org/simple" } },
+    { name = "packaging" },
+    { name = "pillow" },
+    { name = "pyparsing" },
+    { name = "python-dateutil" },
 ]
 sdist = { url = "https://files.pythonhosted.org/packages/63/1b/4be5be87d43d327a0cf4de1a56e86f7f84c89312452406cf122efe2839e6/matplotlib-3.10.9.tar.gz", hash = "sha256:fd66508e8c6877d98e586654b608a0456db8d7e8a546eb1e2600efd957302358", size = 34811233, upload-time = "2026-04-24T00:14:13.539Z" }
 wheels = [
@@ -1963,16 +1975,16 @@ resolution-markers = [
     "python_full_version == '3.11.*' and sys_platform != 'emscripten' and sys_platform != 'win32'",
 ]
 dependencies = [
-    { name = "contourpy", version = "1.3.3", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.11'" },
-    { name = "cycler", marker = "python_full_version >= '3.11'" },
-    { name = "fonttools", marker = "python_full_version >= '3.11'" },
-    { name = "kiwisolver", marker = "python_full_version >= '3.11'" },
-    { name = "numpy", version = "2.4.6", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version == '3.11.*'" },
+    { name = "contourpy", version = "1.3.3", source = { registry = "https://pypi.org/simple" } },
+    { name = "cycler" },
+    { name = "fonttools" },
+    { name = "kiwisolver" },
+    { name = "numpy", version = "2.4.6", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.12'" },
     { name = "numpy", version = "2.5.1", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.12'" },
-    { name = "packaging", marker = "python_full_version >= '3.11'" },
-    { name = "pillow", marker = "python_full_version >= '3.11'" },
-    { name = "pyparsing", marker = "python_full_version >= '3.11'" },
-    { name = "python-dateutil", marker = "python_full_version >= '3.11'" },
+    { name = "packaging" },
+    { name = "pillow" },
+    { name = "pyparsing" },
+    { name = "python-dateutil" },
 ]
 sdist = { url = "https://files.pythonhosted.org/packages/49/64/f9a391af28f518b11ad45a8a712353c94a0aefce09d3703200e5c54b610a/matplotlib-3.11.1.tar.gz", hash = "sha256:69647db5746941c793d6e445a4cd349323ffb87d9cc958c2ad84a659b4832d30", size = 32612045, upload-time = "2026-07-18T03:39:46.63Z" }
 wheels = [
@@ -2534,10 +2546,10 @@ resolution-markers = [
     "python_full_version < '3.11'",
 ]
 dependencies = [
-    { name = "numpy", version = "2.2.6", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.11'" },
-    { name = "python-dateutil", marker = "python_full_version < '3.11'" },
-    { name = "pytz", marker = "python_full_version < '3.11'" },
-    { name = "tzdata", marker = "python_full_version < '3.11'" },
+    { name = "numpy", version = "2.2.6", source = { registry = "https://pypi.org/simple" } },
+    { name = "python-dateutil" },
+    { name = "pytz" },
+    { name = "tzdata" },
 ]
 sdist = { url = "https://files.pythonhosted.org/packages/33/01/d40b85317f86cf08d853a4f495195c73815fdf205eef3993821720274518/pandas-2.3.3.tar.gz", hash = "sha256:e05e1af93b977f7eafa636d043f9f94c7ee3ac81af99c13508215942e64c993b", size = 4495223, upload-time = "2025-09-29T23:34:51.853Z" }
 wheels = [
@@ -2606,10 +2618,10 @@ resolution-markers = [
     "python_full_version == '3.11.*' and sys_platform != 'emscripten' and sys_platform != 'win32'",
 ]
 dependencies = [
-    { name = "numpy", version = "2.4.6", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version == '3.11.*'" },
+    { name = "numpy", version = "2.4.6", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.12'" },
     { name = "numpy", version = "2.5.1", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.12'" },
-    { name = "python-dateutil", marker = "python_full_version >= '3.11'" },
-    { name = "tzdata", marker = "(python_full_version >= '3.11' and sys_platform == 'emscripten') or (python_full_version >= '3.11' and sys_platform == 'win32')" },
+    { name = "python-dateutil" },
+    { name = "tzdata", marker = "sys_platform == 'emscripten' or sys_platform == 'win32'" },
 ]
 sdist = { url = "https://files.pythonhosted.org/packages/be/4f/5f3422a2afec5ffc46308b79e53291365a93748b498ac2e58bead0197916/pandas-3.0.5.tar.gz", hash = "sha256:dca3734d6ab7c906e6730f0788b0a1dbb9f2467731f9711f77995c8e9d62d712", size = 4658219, upload-time = "2026-07-22T22:19:28.819Z" }
 wheels = [
@@ -3645,7 +3657,6 @@ dependencies = [
     { name = "packaging" },
     { name = "psutil", marker = "sys_platform == 'win32'" },
     { name = "python-multipart" },
-    { name = "python-socketio" },
     { name = "redis" },
     { name = "reflex-base" },
     { name = "reflex-components-code" },
@@ -3676,6 +3687,15 @@ db = [
 pydantic = [
     { name = "reflex-base", extra = ["pydantic"] },
 ]
+socketio = [
+    { name = "python-socketio" },
+]
+uvicorn = [
+    { name = "gunicorn" },
+    { name = "uvicorn" },
+    { name = "websockets", version = "16.1.1", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.11'" },
+    { name = "websockets", version = "17.0.1", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.11'" },
+]
 
 [package.dev-dependencies]
 dev = [
@@ -3710,6 +3730,7 @@ dev = [
     { name = "pytest-rerunfailures" },
     { name = "pytest-split" },
     { name = "python-dotenv" },
+    { name = "python-socketio" },
     { name = "pyyaml" },
     { name = "reflex-docgen" },
     { name = "reflex-release" },
@@ -3730,13 +3751,14 @@ dev = [
 requires-dist = [
     { name = "alembic", marker = "extra == 'db'", specifier = ">=1.15.2,<2.0" },
     { name = "click", specifier = ">=8.2" },
-    { name = "granian", extras = ["reload"], specifier = ">=2.7.4" },
+    { name = "granian", extras = ["reload"], specifier = ">=2.8.1" },
+    { name = "gunicorn", marker = "extra == 'uvicorn'", specifier = ">=23.0" },
     { name = "httpx", specifier = ">=0.26,<1.0" },
     { name = "packaging", specifier = ">=24.2,<27" },
     { name = "psutil", marker = "sys_platform == 'win32'", specifier = ">=7.0.0,<8.0" },
     { name = "pydantic", marker = "extra == 'db'", specifier = ">=2.12.0,<3.0" },
     { name = "python-multipart", specifier = ">=0.0.32,<1.0" },
-    { name = "python-socketio", specifier = ">=5.12.0,<6.0" },
+    { name = "python-socketio", marker = "extra == 'socketio'", specifier = ">=5.12.0,<6.0" },
     { name = "redis", specifier = ">=6.4,<8.0" },
     { name = "reflex-base", editable = "packages/reflex-base" },
     { name = "reflex-base", extras = ["pydantic"], marker = "extra == 'pydantic'", editable = "packages/reflex-base" },
@@ -3757,9 +3779,11 @@ requires-dist = [
     { name = "sqlmodel", marker = "extra == 'db'", specifier = ">=0.0.24,<0.1" },
     { name = "starlette", specifier = ">=1.3.1" },
     { name = "typing-extensions", specifier = ">=4.13.0" },
+    { name = "uvicorn", marker = "extra == 'uvicorn'", specifier = ">=0.20.0" },
+    { name = "websockets", marker = "extra == 'uvicorn'", specifier = ">=13.0" },
     { name = "wrapt", specifier = ">=1.17.0,<2.2" },
 ]
-provides-extras = ["db", "pydantic"]
+provides-extras = ["db", "pydantic", "socketio", "uvicorn"]
 
 [package.metadata.requires-dev]
 dev = [
@@ -3791,6 +3815,7 @@ dev = [
     { name = "pytest-rerunfailures" },
     { name = "pytest-split" },
     { name = "python-dotenv" },
+    { name = "python-socketio" },
     { name = "pyyaml" },
     { name = "reflex-docgen", editable = "packages/reflex-docgen" },
     { name = "reflex-release", editable = "packages/reflex-release" },
@@ -4306,7 +4331,7 @@ resolution-markers = [
     "python_full_version < '3.11'",
 ]
 dependencies = [
-    { name = "numpy", version = "2.2.6", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.11'" },
+    { name = "numpy", version = "2.2.6", source = { registry = "https://pypi.org/simple" } },
 ]
 sdist = { url = "https://files.pythonhosted.org/packages/0f/37/6964b830433e654ec7485e45a00fc9a27cf868d622838f6b6d9c5ec0d532/scipy-1.15.3.tar.gz", hash = "sha256:eae3cf522bc7df64b42cad3925c876e1b0b6c35c1337c93e12c0f366f55b0eaf", size = 59419214, upload-time = "2025-05-08T16:13:05.955Z" }
 wheels = [
@@ -4367,7 +4392,7 @@ resolution-markers = [
     "python_full_version == '3.11.*' and sys_platform != 'emscripten' and sys_platform != 'win32'",
 ]
 dependencies = [
-    { name = "numpy", version = "2.4.6", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version == '3.11.*'" },
+    { name = "numpy", version = "2.4.6", source = { registry = "https://pypi.org/simple" } },
 ]
 sdist = { url = "https://files.pythonhosted.org/packages/7a/97/5a3609c4f8d58b039179648e62dd220f89864f56f7357f5d4f45c29eb2cc/scipy-1.17.1.tar.gz", hash = "sha256:95d8e012d8cb8816c226aef832200b1d45109ed4464303e997c5b13122b297c0", size = 30573822, upload-time = "2026-02-23T00:26:24.851Z" }
 wheels = [
@@ -4446,7 +4471,7 @@ resolution-markers = [
     "python_full_version >= '3.12' and python_full_version < '3.14' and sys_platform != 'emscripten' and sys_platform != 'win32'",
 ]
 dependencies = [
-    { name = "numpy", version = "2.5.1", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.12'" },
+    { name = "numpy", version = "2.5.1", source = { registry = "https://pypi.org/simple" } },
 ]
 sdist = { url = "https://files.pythonhosted.org/packages/a7/25/c2700dfaf6442b4effaa91af24ebce5dc9d31bb4a69706313aae70d72cd0/scipy-1.18.0.tar.gz", hash = "sha256:67b2ad2ad54c72ca6d04975a9b2df8c3638c34ddd5b28738e94fc2b57929d378", size = 30774447, upload-time = "2026-06-19T15:01:43.456Z" }
 wheels = [