diff --git a/news/7065.performance.md b/news/7065.performance.md
new file mode 100644
index 00000000000..904d65d9d4f
--- /dev/null
+++ b/news/7065.performance.md
@@ -0,0 +1 @@
+Keep blocking filesystem and metadata work off the backend event loop: the disk state manager now reads state files in a worker thread, precompressed static file serving stats sidecar files off the loop, and the redis client is created with its library version pinned so redis-py no longer scans `sys.path` on every new connection.
diff --git a/reflex/istate/manager/disk.py b/reflex/istate/manager/disk.py
index b1da40ae209..6315bd722ce 100644
--- a/reflex/istate/manager/disk.py
+++ b/reflex/istate/manager/disk.py
@@ -120,6 +120,9 @@ def token_path(self, token: StateToken) -> Path:
     async def load_state(self, token: StateToken[TOKEN_TYPE]) -> TOKEN_TYPE | None:
         """Load a state object based on the provided token.
 
+        The file read and unpickle run in a worker thread so a cache miss does
+        not block the event loop.
+
         Args:
             token: The token used to identify the state object.
 
@@ -128,13 +131,15 @@ async def load_state(self, token: StateToken[TOKEN_TYPE]) -> TOKEN_TYPE | None:
         """
         token_path = self.token_path(token)
 
-        if token_path.exists():
+        def _read() -> TOKEN_TYPE | None:
             try:
                 with token_path.open(mode="rb") as file:
                     return token.deserialize(fp=file)
             except Exception:
-                pass
-        return None
+                return None
+
+        # The open and unpickle are blocking, keep them off the event loop.
+        return await asyncio.to_thread(_read)
 
     async def populate_substates(
         self, token: BaseStateToken, state: BaseState, root_state: BaseState
@@ -220,11 +225,17 @@ async def set_state_for_substate(
         if token.get_and_reset_touched_state(substate):
             pickle_state = token.serialize(substate)
             if pickle_state:
-                if not self.states_directory.exists():
-                    self.states_directory.mkdir(parents=True, exist_ok=True)
-                await run_in_thread(
-                    lambda: self.token_path(substate_token).write_bytes(pickle_state),
-                )
+                token_path = self.token_path(substate_token)
+
+                def _write() -> None:
+                    try:
+                        token_path.write_bytes(pickle_state)
+                    except FileNotFoundError:
+                        # The states directory was removed at runtime.
+                        self.states_directory.mkdir(parents=True, exist_ok=True)
+                        token_path.write_bytes(pickle_state)
+
+                await run_in_thread(_write)
 
         if isinstance(token, BaseStateToken) and isinstance(substate, BaseState):
             for substate_substate in substate.substates.values():
diff --git a/reflex/utils/precompressed_staticfiles.py b/reflex/utils/precompressed_staticfiles.py
index 1c8cd690a67..3ae291c23a9 100644
--- a/reflex/utils/precompressed_staticfiles.py
+++ b/reflex/utils/precompressed_staticfiles.py
@@ -11,6 +11,7 @@
 from os import PathLike
 from pathlib import Path
 
+import anyio.to_thread
 from starlette.datastructures import Headers
 from starlette.responses import FileResponse, Response
 from starlette.staticfiles import NotModifiedResponse, StaticFiles
@@ -129,7 +130,11 @@ def file_response(
         scope: Scope,
         status_code: int = 200,
     ) -> Response:
-        """Build a FileResponse, swapping in a precompressed sidecar when possible.
+        """Build the FileResponse for the uncompressed file.
+
+        With sidecar encodings configured this response is provisional:
+        ``get_response`` picks the sidecar off the event loop and finishes the
+        response (``Vary``/``Content-Encoding`` and the conditional check).
 
         Args:
             full_path: The resolved on-disk path to the uncompressed file.
@@ -138,37 +143,43 @@ def file_response(
             status_code: The response status code to use.
 
         Returns:
-            A file response that serves the best matching asset variant.
+            A file response for the uncompressed file.
         """
-        response_path: str | PathLike[str] = full_path
-        response_stat = stat_result
-        response_headers: dict[str, str] = {}
         media_type = (
             "text/javascript"
             if Path(full_path).suffix.lower() in {".js", ".mjs"}
             else guess_type(os.fspath(full_path))[0]
         )
-
-        if self._encodings:
-            response_headers["Vary"] = "Accept-Encoding"
-            sidecar = self._select_sidecar(full_path, scope)
-            if sidecar is not None:
-                content_encoding, response_path, response_stat = sidecar
-                response_headers["Content-Encoding"] = content_encoding
-
         response = FileResponse(
-            response_path,
+            full_path,
             status_code=status_code,
-            headers=response_headers or None,
             media_type=media_type,
-            stat_result=response_stat,
+            stat_result=stat_result,
         )
+        if self._encodings:
+            return response
+        return self._conditional(response, scope)
+
+    def _conditional(self, response: FileResponse, scope: Scope) -> Response:
+        """Turn ``response`` into a 304 when the client's cached copy is current.
+
+        Args:
+            response: The file response about to be served.
+            scope: The ASGI request scope.
+
+        Returns:
+            ``response`` or a ``NotModifiedResponse`` carrying its headers.
+        """
         if self.is_not_modified(response.headers, Headers(scope=scope)):
             return NotModifiedResponse(response.headers)
         return response
 
     async def get_response(self, path: str, scope: Scope) -> Response:
-        """Serve ``path``, re-routing the 404.html fallback through ``file_response``.
+        """Serve ``path``, swapping in a precompressed sidecar when possible.
+
+        The sidecar lookup stats files, so it runs in a worker thread like
+        Starlette's own path lookup. This also covers the 404.html fallback,
+        which Starlette builds with a bare FileResponse.
 
         Args:
             path: The requested relative file path.
@@ -178,16 +189,28 @@ async def get_response(self, path: str, scope: Scope) -> Response:
             The resolved static response for the request.
         """
         response = await super().get_response(path, scope)
-        # Starlette's get_response builds the 404.html fallback with bare FileResponse,
-        # bypassing file_response. Re-route it so the sidecar/Vary handling applies.
         if (
-            self._encodings
-            and self.html
-            and isinstance(response, FileResponse)
-            and response.status_code == 404
-            and response.stat_result is not None
+            not self._encodings
+            or not isinstance(response, FileResponse)
+            or response.stat_result is None
         ):
-            return self.file_response(
-                response.path, response.stat_result, scope, status_code=404
-            )
-        return response
+            return response
+        sidecar = await anyio.to_thread.run_sync(
+            self._select_sidecar, response.path, scope
+        )
+        headers = {"Vary": "Accept-Encoding"}
+        response_path: str | PathLike[str] = response.path
+        response_stat = response.stat_result
+        if sidecar is not None:
+            content_encoding, response_path, response_stat = sidecar
+            headers["Content-Encoding"] = content_encoding
+        return self._conditional(
+            FileResponse(
+                response_path,
+                status_code=response.status_code,
+                headers=headers,
+                media_type=response.media_type,
+                stat_result=response_stat,
+            ),
+            scope,
+        )
diff --git a/reflex/utils/prerequisites.py b/reflex/utils/prerequisites.py
index 57574d7d6bf..f4646b952cd 100644
--- a/reflex/utils/prerequisites.py
+++ b/reflex/utils/prerequisites.py
@@ -16,7 +16,7 @@
 from os import getcwd
 from pathlib import Path
 from types import ModuleType
-from typing import NamedTuple
+from typing import Any, NamedTuple
 
 from packaging import version
 from reflex_base import constants
@@ -400,10 +400,33 @@ def get_redis() -> Redis | None:
         return Redis.from_url(
             redis_url,
             retry_on_error=[RedisError],
+            **_redis_driver_kwargs(),
         )
     return None
 
 
+def _redis_driver_kwargs() -> dict[str, Any]:
+    """Pin redis-py's own version so it is not resolved per connection.
+
+    redis-py >= 7 looks its version up through ``importlib.metadata`` every
+    time a connection object is created, which scans ``sys.path`` and reads
+    package metadata on the event loop. Passing the version up front skips
+    that lookup.
+
+    Returns:
+        Keyword arguments for the redis client constructor.
+    """
+    import redis
+
+    try:
+        # Imported dynamically: the module only exists in redis-py >= 7.
+        driver_info = importlib.import_module("redis.driver_info")
+    except ImportError:
+        # redis-py < 7 takes lib_version directly.
+        return {"lib_version": redis.__version__}
+    return {"driver_info": driver_info.DriverInfo(lib_version=redis.__version__)}
+
+
 def get_redis_sync() -> RedisSync | None:
     """Get the synchronous redis client.
 
diff --git a/scripts/blockbuster_audit.py b/scripts/blockbuster_audit.py
new file mode 100644
index 00000000000..45f2434d684
--- /dev/null
+++ b/scripts/blockbuster_audit.py
@@ -0,0 +1,405 @@
+"""Audit a typical Reflex app for blocking calls on the backend event loop.
+
+Runs a small app in-process via ``AppHarness`` (dev) or ``AppHarnessProd`` with
+`blockbuster <https://github.com/cbornet/blockbuster>`_ patched into *recording*
+mode, drives it with Playwright, and writes a report listing every blocking
+call (file/socket/sqlite/lock/sleep) that ran on an asyncio loop, with the
+Python stack that reached it.
+
+Usage (blockbuster and aiosqlite are not project deps, so install them
+ad hoc; ``uv run`` would re-sync them away, hence the direct interpreter)::
+
+    uv pip install blockbuster aiosqlite
+    REFLEX_STATE_MANAGER_MODE=memory .venv/bin/python scripts/blockbuster_audit.py
+    REFLEX_STATE_MANAGER_MODE=disk .venv/bin/python scripts/blockbuster_audit.py
+    REFLEX_STATE_MANAGER_MODE=redis REFLEX_REDIS_URL=redis://localhost:6379 \\
+        .venv/bin/python scripts/blockbuster_audit.py
+    .venv/bin/python scripts/blockbuster_audit.py --prod   # backend-served static frontend
+
+The same recorder doubles as a pytest plugin that instruments the unit-test
+loops::
+
+    BB_REPORT=/tmp/bb_pytest.txt .venv/bin/python -m pytest \\
+        -p scripts.blockbuster_audit tests/units/istate
+
+Blockbuster only sees *syscall-shaped* blocking (stat/open/read/socket/
+sqlite/contended threading.Lock/time.sleep). CPU-bound work on the loop such
+as pickling, deepcopy or ``get_type_hints`` is invisible to it.
+"""
+
+# ruff: noqa: T201, ANN001, DOC201, N802, D301
+
+from __future__ import annotations
+
+import argparse
+import asyncio
+import inspect
+import os
+import sys
+import threading
+import time
+import urllib.request
+from collections import defaultdict
+from pathlib import Path
+
+REPO_ROOT = Path(__file__).resolve().parent.parent
+_lock = threading.Lock()
+_findings: dict[tuple, list[tuple[str, int, str]]] = {}
+_counts: dict[tuple, int] = defaultdict(int)
+
+
+def _record(func_name: str, frame) -> None:
+    frames = []
+    f = frame
+    while f:
+        frames.append((f.f_code.co_filename, f.f_lineno, f.f_code.co_name))
+        f = f.f_back
+    frames.reverse()
+    sig = tuple((fn, name) for fn, _, name in frames if "blockbuster" not in fn)
+    key = (func_name, sig)
+    with _lock:
+        _counts[key] += 1
+        _findings.setdefault(key, frames)
+
+
+def _wrap_blocking_recording(
+    modules, excluded_modules, func, func_name, can_block_functions, can_block_predicate
+):
+    """Drop-in for ``blockbuster._wrap_blocking`` that records instead of raising."""
+    import blockbuster.blockbuster as bbmod
+
+    def wrapper(*args, **kwargs):
+        if bbmod.blockbuster_skip.get(False):
+            return func(*args, **kwargs)
+        try:
+            asyncio.get_running_loop()
+        except RuntimeError:
+            return func(*args, **kwargs)
+        skip_token = bbmod.blockbuster_skip.set(True)
+        try:
+            if can_block_predicate(*args, **kwargs):
+                return func(*args, **kwargs)
+            frame = inspect.currentframe().f_back  # pyright: ignore[reportOptionalMemberAccess]
+            f = frame
+            while f:
+                fname = f.f_code.co_filename.replace("\\", "/")
+                for filename, functions in can_block_functions:
+                    if fname.endswith(filename) and f.f_code.co_name in functions:
+                        return func(*args, **kwargs)
+                f = f.f_back
+            _record(func_name, frame)
+            return func(*args, **kwargs)
+        finally:
+            bbmod.blockbuster_skip.reset(skip_token)
+
+    return wrapper
+
+
+def activate():
+    """Patch blockbuster into recording mode and activate all its checks.
+
+    Returns:
+        The active BlockBuster instance.
+    """
+    import blockbuster.blockbuster as bbmod
+
+    bbmod._wrap_blocking = _wrap_blocking_recording  # pyright: ignore[reportPrivateUsage]
+    bb = bbmod.BlockBuster()
+    bb.activate()
+    return bb
+
+
+def report(path: str | Path) -> int:
+    """Write the recorded findings to ``path``.
+
+    Args:
+        path: Output file.
+
+    Returns:
+        Number of distinct findings written.
+    """
+    root = str(REPO_ROOT) + "/"
+    venv = root + ".venv/"
+
+    def short(fn: str) -> str:
+        if fn.startswith(venv):
+            return "venv:" + fn.split("site-packages/", 1)[-1]
+        return fn.removeprefix(root)
+
+    def is_repo(fn: str) -> bool:
+        return fn.startswith(root) and not fn.startswith(venv)
+
+    with _lock:
+        items = sorted(_findings.items(), key=lambda kv: -_counts[kv[0]])
+    out = []
+    written = 0
+    for (func_name, sig), frames in items:
+        if any(Path(__file__).name in fn for fn, _, _ in frames):
+            continue  # driver noise: playwright's sync API runs a loop in the main thread
+        repo_frames = [fr for fr in frames if is_repo(fr[0])]
+        innermost = repo_frames[-1] if repo_frames else frames[-1]
+        if not repo_frames:
+            origin = "3RDPARTY"
+        elif "/tests/" in innermost[0]:
+            origin = "TEST"
+        else:
+            origin = "REFLEX"
+        n = _counts[func_name, sig]
+        out.append(
+            f"\n=== [{origin}] {func_name}  x{n}  innermost: "
+            f"{short(innermost[0])}:{innermost[1]} {innermost[2]}"
+        )
+        out.extend(f"    {short(fn)}:{ln}  {name}" for fn, ln, name in frames[-25:])
+        written += 1
+    Path(path).write_text("\n".join(out))
+    return written
+
+
+def pytest_configure(config):
+    """Pytest plugin hook: start recording for the whole session."""
+    activate()
+
+
+def pytest_sessionfinish(session, exitstatus):
+    """Pytest plugin hook: dump the report."""
+    n = report(os.environ.get("BB_REPORT", "/tmp/bb_pytest_report.txt"))
+    print(f"\nblockbuster distinct findings: {n}")
+
+
+def BbApp():
+    """A small app touching the common event paths."""
+    import asyncio
+
+    import reflex as rx
+
+    class Item(rx.Model, table=True):
+        __table_args__ = {"extend_existing": True}
+        name: str
+
+    class State(rx.State):
+        count: int = 0
+        loaded: str = ""
+        log: list[str] = []
+        bg_progress: int = 0
+        upload_done: bool = False
+        cookie_val: str = rx.Cookie("")
+        ls_val: str = rx.LocalStorage("")
+        db_rows: int = 0
+        script_result: str = ""
+
+        @rx.var
+        def doubled(self) -> int:
+            return self.count * 2
+
+        @rx.event
+        def on_load(self):
+            self.loaded = "loaded"
+
+        @rx.event
+        def increment(self):
+            self.count += 1
+
+        @rx.event
+        async def stream(self):
+            for i in range(3):
+                self.log.append(f"step {i}")
+                yield
+                await asyncio.sleep(0.01)
+
+        @rx.event(background=True)
+        async def bg_task(self):
+            for i in range(3):
+                await asyncio.sleep(0.01)
+                async with self:
+                    self.bg_progress = i + 1
+
+        @rx.event
+        async def handle_upload(self, files: list[rx.UploadFile]):
+            for file in files:
+                data = await file.read()
+                if not file.name:
+                    continue
+                local = rx.get_upload_dir() / file.name
+                local.parent.mkdir(parents=True, exist_ok=True)
+                local.write_bytes(data)
+            self.upload_done = True
+
+        @rx.event
+        def set_storage(self):
+            self.cookie_val = "c1"
+            self.ls_val = "l1"
+
+        @rx.event
+        def db_sync(self):
+            with rx.session() as session:
+                session.add(Item(name="x"))
+                session.commit()
+                self.db_rows = len(session.exec(Item.select()).all())
+
+        @rx.event
+        async def db_async(self):
+            async with rx.asession() as session:
+                session.add(Item(name="y"))
+                await session.commit()
+                self.db_rows = len((await session.exec(Item.select())).all())
+
+        @rx.event
+        def call_js(self):
+            return rx.call_script("1+1", callback=State.set_script_result)
+
+        @rx.event
+        def set_script_result(self, value):
+            self.script_result = str(value)
+
+        @rx.event
+        def go_other(self):
+            return rx.redirect("/other")
+
+        @rx.event
+        def boom(self):
+            msg = "boom"
+            raise ValueError(msg)
+
+    def index():
+        return rx.vstack(
+            rx.text(State.loaded, id="loaded"),
+            rx.text(State.count, id="count"),
+            rx.text(State.doubled, id="doubled"),
+            rx.button("inc", id="inc", on_click=State.increment),
+            rx.button("stream", id="stream", on_click=State.stream),
+            rx.foreach(State.log, lambda s: rx.text(s)),
+            rx.text(State.log.length(), id="loglen"),
+            rx.button("bg", id="bg", on_click=State.bg_task),
+            rx.text(State.bg_progress, id="bgp"),
+            rx.upload.root(rx.button("select"), id="upload_root"),
+            rx.button(
+                "upload",
+                id="upload",
+                on_click=State.handle_upload(rx.upload_files(upload_id="upload_root")),
+            ),
+            rx.text(State.upload_done.to_string(), id="updone"),
+            rx.button("storage", id="storage", on_click=State.set_storage),
+            rx.text(State.cookie_val, id="cookie"),
+            rx.text(State.ls_val, id="ls"),
+            rx.button("dbsync", id="dbsync", on_click=State.db_sync),
+            rx.button("dbasync", id="dbasync", on_click=State.db_async),
+            rx.text(State.db_rows, id="dbrows"),
+            rx.button("js", id="js", on_click=State.call_js),
+            rx.text(State.script_result, id="jsres"),
+            rx.button("boom", id="boom", on_click=State.boom),
+            rx.button("other", id="other", on_click=State.go_other),
+            rx.image(src="/_upload/hello.txt", id="img"),
+        )
+
+    def other():
+        return rx.vstack(
+            rx.text("other", id="otherpage"), rx.text(State.count, id="count2")
+        )
+
+    app = rx.App()
+    app.add_page(index, on_load=State.on_load)
+    app.add_page(other)
+
+
+def _drive_dev(harness, page, expect, hello: Path) -> None:
+    url = harness.frontend_url
+    page.goto(url)
+    expect(page.locator("#loaded")).to_have_text("loaded", timeout=60000)
+    page.click("#inc")
+    expect(page.locator("#count")).to_have_text("1")
+    expect(page.locator("#doubled")).to_have_text("2")
+    page.click("#stream")
+    expect(page.locator("#loglen")).to_have_text("3")
+    page.click("#bg")
+    expect(page.locator("#bgp")).to_have_text("3")
+    page.locator("#upload_root input[type=file]").set_input_files(str(hello))
+    page.click("#upload")
+    expect(page.locator("#updone")).to_have_text("true")
+    page.click("#storage")
+    expect(page.locator("#cookie")).to_have_text("c1")
+    expect(page.locator("#ls")).to_have_text("l1")
+    page.click("#dbsync")
+    expect(page.locator("#dbrows")).to_have_text("1")
+    page.click("#dbasync")
+    expect(page.locator("#dbrows")).to_have_text("2")
+    page.click("#js")
+    expect(page.locator("#jsres")).to_have_text("2")
+    page.click("#boom")
+    time.sleep(1)
+    host, port = harness._poll_for_servers().getsockname()[:2]  # pyright: ignore[reportPrivateUsage]
+    base = f"http://{host}:{port}"
+    print(
+        "GET /_upload/hello.txt ->",
+        urllib.request.urlopen(base + "/_upload/hello.txt").status,
+    )
+    print("GET /ping ->", urllib.request.urlopen(base + "/ping").status)
+    page.click("#other")
+    expect(page.locator("#otherpage")).to_have_text("other")
+    expect(page.locator("#count2")).to_have_text("1")
+    page.goto(url)  # re-hydrate an existing token
+    expect(page.locator("#count")).to_have_text("1")
+    page2 = page.context.browser.new_page()  # a second client with a fresh token
+    page2.goto(url)
+    expect(page2.locator("#loaded")).to_have_text("loaded")
+    page2.click("#inc")
+    expect(page2.locator("#count")).to_have_text("1")
+
+
+def _drive_prod(harness, page, expect) -> None:
+    url = harness.frontend_url
+    page.goto(url)
+    expect(page.locator("#loaded")).to_have_text("loaded", timeout=60000)
+    page.click("#inc")
+    expect(page.locator("#count")).to_have_text("1")
+    page.click("#other")
+    expect(page.locator("#otherpage")).to_have_text("other")
+    for enc in ("gzip", "br", "identity"):
+        req = urllib.request.Request(url + "/", headers={"Accept-Encoding": enc})
+        r = urllib.request.urlopen(req)
+        print("GET / enc", enc, "->", r.status, r.headers.get("Content-Encoding"))
+
+
+def main() -> None:
+    """Run the audit."""
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--prod", action="store_true", help="use AppHarnessProd")
+    parser.add_argument("--root", type=Path, default=Path("/tmp/bb_audit"))
+    parser.add_argument("--report", type=Path, default=None)
+    args = parser.parse_args()
+    mode = os.environ.get("REFLEX_STATE_MANAGER_MODE", "memory")
+    tag = "prod" if args.prod else mode
+    report_path = args.report or args.root / f"report_{tag}.txt"
+    args.root.mkdir(parents=True, exist_ok=True)
+    os.environ.setdefault("REFLEX_DB_URL", "sqlite:///reflex.db")
+    os.environ.setdefault("REFLEX_ASYNC_DB_URL", "sqlite+aiosqlite:///reflex.db")
+    Path("reflex.db").unlink(missing_ok=True)
+
+    activate()
+    from playwright.sync_api import expect, sync_playwright
+
+    from reflex.testing import AppHarness, AppHarnessProd
+
+    harness_cls = AppHarnessProd if args.prod else AppHarness
+    with harness_cls.create(root=args.root / f"app_{tag}", app_source=BbApp) as harness:
+        import reflex as rx
+
+        rx.Model.create_all()
+        hello = args.root / "hello.txt"
+        hello.write_text("hello")
+        with sync_playwright() as p:
+            browser = p.chromium.launch(
+                executable_path=os.environ.get("BB_CHROMIUM") or None
+            )
+            page = browser.new_page()
+            if args.prod:
+                _drive_prod(harness, page, expect)
+            else:
+                _drive_dev(harness, page, expect, hello)
+            time.sleep(3)  # let debounced state writes and expiration tasks run
+            browser.close()
+        n = report(report_path)
+        print(f"distinct findings: {n} -> {report_path}")
+
+
+if __name__ == "__main__":
+    sys.exit(main())
diff --git a/tests/units/istate/manager/test_disk.py b/tests/units/istate/manager/test_disk.py
index ac5f9fd29d6..8a7610358f7 100644
--- a/tests/units/istate/manager/test_disk.py
+++ b/tests/units/istate/manager/test_disk.py
@@ -1,9 +1,16 @@
 """Tests for the disk state manager."""
 
+import builtins
+import io
 import os
+import threading
 from pathlib import Path
 
+import pytest
+
 from reflex.istate.manager.disk import StateManagerDisk
+from reflex.istate.manager.token import BaseStateToken
+from reflex.state import BaseState
 
 
 def test_states_directory_survives_chdir(tmp_path: Path, monkeypatch):
@@ -25,3 +32,53 @@ def test_states_directory_survives_chdir(tmp_path: Path, monkeypatch):
     assert manager.states_directory == states_dir
     # Purge resolves against the original directory, not the new cwd.
     manager._purge_expired_states()
+
+
+@pytest.mark.asyncio
+async def test_state_files_are_not_touched_on_the_event_loop(
+    tmp_path: Path, monkeypatch, token: str
+):
+    """Reading and writing state files must happen in a worker thread.
+
+    Args:
+        tmp_path: A temporary directory.
+        monkeypatch: The pytest monkeypatch fixture.
+        token: A token.
+    """
+    monkeypatch.chdir(tmp_path)
+    loop_thread = threading.get_ident()
+    on_loop: list[tuple[str, str]] = []
+
+    def _watch(module, name: str):
+        original = getattr(module, name)
+
+        def wrapper(path, *args, **kwargs):
+            if threading.get_ident() == loop_thread and str(path).startswith(
+                str(tmp_path)
+            ):
+                on_loop.append((name, str(path)))
+            return original(path, *args, **kwargs)
+
+        monkeypatch.setattr(module, name, wrapper)
+
+    class Root(BaseState):
+        pass
+
+    class Child(Root):
+        num: int = 0
+
+    # Construction creates the directory synchronously; that is startup work.
+    writer, reader = StateManagerDisk(), StateManagerDisk()
+    for module, name in ((io, "open"), (builtins, "open"), (os, "stat"), (os, "mkdir")):
+        _watch(module, name)
+
+    bs_token = BaseStateToken(ident=token, cls=Root)
+    async with writer.modify_state(bs_token) as root:
+        (await root.get_state(Child)).num = 1
+    await writer.close()
+
+    root = await reader.get_state(bs_token)
+    assert (await root.get_state(Child)).num == 1
+    await reader.close()
+
+    assert on_loop == []
diff --git a/tests/units/utils/test_precompressed_staticfiles.py b/tests/units/utils/test_precompressed_staticfiles.py
index d7a7edaa007..d2e1c3633e9 100644
--- a/tests/units/utils/test_precompressed_staticfiles.py
+++ b/tests/units/utils/test_precompressed_staticfiles.py
@@ -3,6 +3,8 @@
 from __future__ import annotations
 
 import asyncio
+import os
+import threading
 from pathlib import Path
 
 import pytest
@@ -169,3 +171,32 @@ async def test_precompressed_static_files_fall_back_to_identity(tmp_path: Path):
     assert "content-encoding" not in response.headers
     assert response.headers["vary"] == "Accept-Encoding"
     assert await _collect_body(response, scope) == b"console.log('hello');"
+
+
+@pytest.mark.asyncio
+async def test_precompressed_static_files_stat_sidecars_off_the_event_loop(
+    tmp_path: Path, monkeypatch
+):
+    """The sidecar lookup must not stat files on the event loop thread."""
+    (tmp_path / "app.js").write_text("console.log('hello');")
+    (tmp_path / "app.js.gz").write_bytes(b"compressed-gzip")
+    static_files = PrecompressedStaticFiles(directory=tmp_path, encodings=["gzip"])
+
+    loop_thread = threading.get_ident()
+    on_loop: list[str] = []
+    original_stat = os.stat
+
+    def stat(path, *args, **kwargs):
+        if threading.get_ident() == loop_thread and str(path).startswith(str(tmp_path)):
+            on_loop.append(str(path))
+        return original_stat(path, *args, **kwargs)
+
+    monkeypatch.setattr(os, "stat", stat)
+
+    scope = _scope("/app.js", "gzip")
+    response = await static_files.get_response("app.js", scope)
+
+    assert isinstance(response, FileResponse)
+    assert response.headers["content-encoding"] == "gzip"
+    assert await _collect_body(response, scope) == b"compressed-gzip"
+    assert on_loop == []
diff --git a/tests/units/utils/test_prerequisites.py b/tests/units/utils/test_prerequisites.py
new file mode 100644
index 00000000000..d3ce2b29a1f
--- /dev/null
+++ b/tests/units/utils/test_prerequisites.py
@@ -0,0 +1,32 @@
+"""Tests for reflex.utils.prerequisites."""
+
+import pytest
+
+from reflex.utils import prerequisites
+
+
+def test_get_redis_pins_driver_version(monkeypatch):
+    """Creating a redis connection must not look the library version up.
+
+    redis-py resolves its version through importlib.metadata for every new
+    connection unless it is passed in, which scans sys.path on the event loop.
+
+    Args:
+        monkeypatch: The pytest monkeypatch fixture.
+    """
+    redis = pytest.importorskip("redis")
+    monkeypatch.setattr(prerequisites, "parse_redis_url", lambda: "redis://localhost")
+
+    def _lookup():
+        pytest.fail("redis-py resolved its version while creating a connection")
+
+    monkeypatch.setattr("redis.utils.get_lib_version", _lookup)
+
+    client = prerequisites.get_redis()
+    assert client is not None
+    connection = client.connection_pool.make_connection()
+    driver_info = getattr(connection, "driver_info", None)
+    lib_version = (
+        driver_info.lib_version if driver_info is not None else connection.lib_version  # pyright: ignore[reportAttributeAccessIssue]
+    )
+    assert lib_version == redis.__version__