diff --git a/reflex/environment.py b/reflex/environment.py
index b021d7eda13..fa64c87aace 100644
--- a/reflex/environment.py
+++ b/reflex/environment.py
@@ -650,6 +650,9 @@ class EnvironmentVariables:
     # Enable full logging of debug messages to reflex user directory.
     REFLEX_ENABLE_FULL_LOGGING: EnvVar[bool] = env_var(False)
 
+    # The path to the reflex errors log file. If not set, no separate error log will be used.
+    REFLEX_ERROR_LOG_FILE: EnvVar[Path | None] = env_var(None)
+
 
 environment = EnvironmentVariables()
 
diff --git a/reflex/testing.py b/reflex/testing.py
index b18273342d6..f3382a44453 100644
--- a/reflex/testing.py
+++ b/reflex/testing.py
@@ -10,7 +10,6 @@
 import os
 import platform
 import re
-import signal
 import socket
 import socketserver
 import subprocess
@@ -19,17 +18,16 @@
 import threading
 import time
 import types
-from collections.abc import AsyncIterator, Callable, Coroutine, Sequence
+from collections.abc import Callable, Coroutine, Sequence
 from http.server import SimpleHTTPRequestHandler
+from io import TextIOWrapper
 from pathlib import Path
 from typing import TYPE_CHECKING, Any, Literal, TypeVar
 
-import uvicorn
+import psutil
 
 import reflex
-import reflex.environment
 import reflex.reflex
-import reflex.utils.build
 import reflex.utils.exec
 import reflex.utils.format
 import reflex.utils.prerequisites
@@ -45,9 +43,7 @@
     StateManagerRedis,
     reload_state_module,
 )
-from reflex.utils import console
 from reflex.utils.export import export
-from reflex.utils.types import ASGIApp
 
 try:
     from selenium import webdriver
@@ -101,6 +97,14 @@ def __exit__(self, *excinfo):
         os.chdir(self._old_cwd.pop())
 
 
+class ReflexProcessLoggedErrorError(RuntimeError):
+    """Exception raised when the reflex process logs contain errors."""
+
+
+class ReflexProcessExitNonZeroError(RuntimeError):
+    """Exception raised when the reflex process exits with a non-zero status."""
+
+
 @dataclasses.dataclass
 class AppHarness:
     """AppHarness executes a reflex app in-process for testing."""
@@ -113,14 +117,15 @@ class AppHarness:
     app_module_path: Path
     app_module: types.ModuleType | None = None
     app_instance: reflex.App | None = None
-    app_asgi: ASGIApp | None = None
-    frontend_process: subprocess.Popen | None = None
+    reflex_process: subprocess.Popen | None = None
+    reflex_process_log_path: Path | None = None
+    reflex_process_error_log_path: Path | None = None
     frontend_url: str | None = None
-    frontend_output_thread: threading.Thread | None = None
-    backend_thread: threading.Thread | None = None
-    backend: uvicorn.Server | None = None
+    backend_port: int | None = None
+    frontend_port: int | None = None
     state_manager: StateManager | None = None
     _frontends: list[WebDriver] = dataclasses.field(default_factory=list)
+    _reflex_process_log_fn: TextIOWrapper | None = None
 
     @classmethod
     def create(
@@ -273,82 +278,117 @@ def _initialize_app(self):
             # Ensure the AppHarness test does not skip State assignment due to running via pytest
             os.environ.pop(reflex.constants.PYTEST_CURRENT_TEST, None)
             os.environ[reflex.constants.APP_HARNESS_FLAG] = "true"
-            # Ensure we actually compile the app during first initialization.
             self.app_instance, self.app_module = (
                 reflex.utils.prerequisites.get_and_validate_app(
                     # Do not reload the module for pre-existing apps (only apps generated from source)
                     reload=self.app_source is not None
                 )
             )
-            self.app_asgi = self.app_instance()
-        if self.app_instance and isinstance(
-            self.app_instance._state_manager, StateManagerRedis
-        ):
-            if self.app_instance._state is None:
-                msg = "State is not set."
-                raise RuntimeError(msg)
-            # Create our own redis connection for testing.
-            self.state_manager = StateManagerRedis.create(self.app_instance._state)
-        else:
-            self.state_manager = (
-                self.app_instance._state_manager if self.app_instance else None
+            # Have to compile to ensure all state is available.
+            _ = self.app_instance()
+        self.state_manager = (
+            self.app_instance._state_manager if self.app_instance else None
+        )
+        if isinstance(self.state_manager, StateManagerDisk):
+            object.__setattr__(
+                self.state_manager, "states_directory", self.app_path / ".states"
             )
 
     def _reload_state_module(self):
         """Reload the rx.State module to avoid conflict when reloading."""
         reload_state_module(module=f"{self.app_name}.{self.app_name}")
 
-    def _get_backend_shutdown_handler(self):
-        if self.backend is None:
-            msg = "Backend was not initialized."
-            raise RuntimeError(msg)
+    def _start_subprocess(
+        self, backend: bool = True, frontend: bool = True, mode: str = "dev"
+    ):
+        """Start the reflex app using subprocess instead of threads.
 
-        original_shutdown = self.backend.shutdown
+        Args:
+            backend: Whether to start the backend server.
+            frontend: Whether to start the frontend server.
+            mode: The mode to run the app in (dev, prod, etc.).
+        """
+        self.reflex_process_log_path = self.app_path / "reflex.log"
+        self.reflex_process_error_log_path = self.app_path / "reflex_error.log"
+        self._reflex_process_log_fn = self.reflex_process_log_path.open("w")
+        command = [
+            sys.executable,
+            "-u",
+            "-m",
+            "reflex",
+            "run",
+            "--env",
+            mode,
+            "--loglevel",
+            "debug",
+        ]
+        if backend:
+            if self.backend_port is None:
+                self.backend_port = reflex.utils.processes.handle_port(
+                    "backend", 48000, auto_increment=True
+                )
+            command.extend(["--backend-port", str(self.backend_port)])
+            if not frontend:
+                command.append("--backend-only")
+        if frontend:
+            if self.frontend_port is None:
+                self.frontend_port = reflex.utils.processes.handle_port(
+                    "frontend", 43000, auto_increment=True
+                )
+            command.extend(["--frontend-port", str(self.frontend_port)])
+            if not backend:
+                command.append("--frontend-only")
+        self.reflex_process = subprocess.Popen(
+            command,
+            stdout=self._reflex_process_log_fn,
+            stderr=self._reflex_process_log_fn,
+            cwd=self.app_path,
+            env={
+                **os.environ,
+                "REFLEX_ERROR_LOG_FILE": str(self.reflex_process_error_log_path),
+                "PYTEST_CURRENT_TEST": "",
+                "APP_HARNESS_FLAG": "true",
+            },
+        )
+        self._wait_for_servers(backend=backend, frontend=frontend)
 
-        async def _shutdown(*args, **kwargs) -> None:
-            # ensure redis is closed before event loop
-            if self.app_instance is not None and isinstance(
-                self.app_instance._state_manager, StateManagerRedis
-            ):
-                with contextlib.suppress(ValueError):
-                    await self.app_instance._state_manager.close()
+    def _wait_for_servers(self, backend: bool, frontend: bool):
+        """Wait for both frontend and backend servers to be ready by parsing console output.
 
-            # socketio shutdown handler
-            if self.app_instance is not None and self.app_instance.sio is not None:
-                with contextlib.suppress(TypeError):
-                    await self.app_instance.sio.shutdown()
+        Args:
+            backend: Whether to wait for the backend server to be ready.
+            frontend: Whether to wait for the frontend server to be ready.
 
-            # sqlalchemy async engine shutdown handler
-            try:
-                async_engine = reflex.model.get_async_engine(None)
-            except ValueError:
-                pass
-            else:
-                await async_engine.dispose()
+        Raises:
+            RuntimeError: If servers did not start properly.
+        """
+        if self.reflex_process is None or self.reflex_process.pid is None:
+            msg = "Reflex process has no pid."
+            raise RuntimeError(msg)
 
-            await original_shutdown(*args, **kwargs)
+        frontend_ready = False
+        backend_ready = False
+        timeout = 30
+        start_time = time.time()
 
-        return _shutdown
+        process = psutil.Process(self.reflex_process.pid)
+        while not ((not frontend or frontend_ready) and (not backend or backend_ready)):
+            if time.time() - start_time > timeout:
+                msg = f"Timeout waiting for servers. Frontend ready: {frontend_ready}, Backend ready: {backend_ready}"
+                raise RuntimeError(msg)
 
-    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.shutdown = self._get_backend_shutdown_handler()
-        with chdir(self.app_path):
-            print(  # noqa: T201
-                "Creating backend in a new thread..."
-            )  # for pytest diagnosis
-            self.backend_thread = threading.Thread(target=self.backend.run)
-        self.backend_thread.start()
-        print("Backend started.")  # for pytest diagnosis #noqa: T201
+            for proc in process.children(recursive=True):
+                with contextlib.suppress(psutil.NoSuchProcess, psutil.AccessDenied):
+                    if ncs := proc.net_connections():
+                        for net_conn in ncs:
+                            if net_conn.status == psutil.CONN_LISTEN:
+                                if net_conn.laddr.port == self.frontend_port:
+                                    frontend_ready = True
+                                    self.frontend_url = (
+                                        f"http://localhost:{self.frontend_port}/"
+                                    )
+                                elif net_conn.laddr.port == self.backend_port:
+                                    backend_ready = True
 
     async def _reset_backend_state_manager(self):
         """Reset the StateManagerRedis event loop affinity.
@@ -376,78 +416,14 @@ async def _reset_backend_state_manager(self):
                 msg = "Failed to reset state manager."
                 raise RuntimeError(msg)
 
-    def _start_frontend(self):
-        # Set up the frontend.
-        with chdir(self.app_path):
-            config = reflex.config.get_config()
-            print("Polling for servers...")  # for pytest diagnosis #noqa: T201
-            config.api_url = "http://{}:{}".format(
-                *self._poll_for_servers(timeout=30).getsockname(),
-            )
-            print("Building frontend...")  # for pytest diagnosis #noqa: T201
-            reflex.utils.build.setup_frontend(self.app_path)
-
-        print("Frontend starting...")  # for pytest diagnosis #noqa: T201
-
-        # Start the frontend.
-        self.frontend_process = reflex.utils.processes.new_process(
-            [
-                *reflex.utils.prerequisites.get_js_package_executor(raise_on_none=True)[
-                    0
-                ],
-                "run",
-                "dev",
-            ],
-            cwd=self.app_path / reflex.utils.prerequisites.get_web_dir(),
-            env={"PORT": "0", "NO_COLOR": "1"},
-            **FRONTEND_POPEN_ARGS,
-        )
-
-    def _wait_frontend(self):
-        if self.frontend_process is None or self.frontend_process.stdout is None:
-            msg = "Frontend process has no stdout."
-            raise RuntimeError(msg)
-        while self.frontend_url is None:
-            line = self.frontend_process.stdout.readline()
-            if not line:
-                break
-            print(line)  # for pytest diagnosis #noqa: T201
-            m = re.search(reflex.constants.ReactRouter.FRONTEND_LISTENING_REGEX, line)
-            if m is not None:
-                self.frontend_url = m.group(1)
-                config = reflex.config.get_config()
-                config.deploy_url = self.frontend_url
-                break
-        if self.frontend_url is None:
-            msg = "Frontend did not start"
-            raise RuntimeError(msg)
-
-        def consume_frontend_output():
-            while True:
-                try:
-                    line = (
-                        self.frontend_process.stdout.readline()  # pyright: ignore [reportOptionalMemberAccess]
-                    )
-                # catch I/O operation on closed file.
-                except ValueError as e:
-                    console.error(str(e))
-                    break
-                if not line:
-                    break
-
-        self.frontend_output_thread = threading.Thread(target=consume_frontend_output)
-        self.frontend_output_thread.start()
-
     def start(self) -> AppHarness:
-        """Start the backend in a new thread and dev frontend as a separate process.
+        """Start the app using reflex run subprocess.
 
         Returns:
             self
         """
         self._initialize_app()
-        self._start_backend()
-        self._start_frontend()
-        self._wait_frontend()
+        self._start_subprocess()
         return self
 
     @staticmethod
@@ -476,43 +452,48 @@ def __enter__(self) -> AppHarness:
         return self.start()
 
     def stop(self) -> None:
-        """Stop the frontend and backend servers."""
-        import psutil
-
-        # Quit browsers first to avoid any lingering events being sent during shutdown.
+        """Stop the reflex subprocess."""
         for driver in self._frontends:
             driver.quit()
 
+        self._stop_reflex()
         self._reload_state_module()
 
-        if self.backend is not None:
-            self.backend.should_exit = True
-        if self.frontend_process is not None:
-            # https://stackoverflow.com/a/70565806
-            frontend_children = psutil.Process(self.frontend_process.pid).children(
-                recursive=True,
+    def _stop_reflex(self):
+        returncode = None
+        # Check if the process exited on its own or we have to kill it.
+        if (
+            self.reflex_process is not None
+            and (returncode := self.reflex_process.poll()) is None
+        ):
+            try:
+                # Kill server and children recursively.
+                reflex.utils.exec.kill(self.reflex_process.pid)
+            except (ProcessLookupError, OSError):
+                pass
+            finally:
+                self.reflex_process = None
+        if self._reflex_process_log_fn is not None:
+            with contextlib.suppress(Exception):
+                self._reflex_process_log_fn.close()
+        if self.reflex_process_log_path is not None:
+            print(self.reflex_process_log_path.read_text())  # noqa: T201  for pytest debugging
+        # If there are errors in the logs, raise an exception.
+        if (
+            self.reflex_process_error_log_path is not None
+            and self.reflex_process_error_log_path.exists()
+        ):
+            error_log_content = self.reflex_process_error_log_path.read_text()
+            if error_log_content:
+                msg = f"Reflex process error log contains errors:\n{error_log_content}"
+                raise ReflexProcessLoggedErrorError(msg)
+        # When the process exits non-zero, but wasn't killed, it is a test failure.
+        if returncode is not None and returncode != 0:
+            msg = (
+                f"Reflex process exited with code {returncode}. "
+                "Check the logs for more details."
             )
-            if sys.platform == "win32":
-                self.frontend_process.terminate()
-            else:
-                pgrp = os.getpgid(self.frontend_process.pid)
-                os.killpg(pgrp, signal.SIGTERM)
-            # kill any remaining child processes
-            for child in frontend_children:
-                # It's okay if the process is already gone.
-                with contextlib.suppress(psutil.NoSuchProcess):
-                    child.terminate()
-            _, still_alive = psutil.wait_procs(frontend_children, timeout=3)
-            for child in still_alive:
-                # It's okay if the process is already gone.
-                with contextlib.suppress(psutil.NoSuchProcess):
-                    child.kill()
-            # wait for main process to exit
-            self.frontend_process.communicate()
-        if self.backend_thread is not None:
-            self.backend_thread.join()
-        if self.frontend_output_thread is not None:
-            self.frontend_output_thread.join()
+            raise ReflexProcessExitNonZeroError(msg)
 
     def __exit__(self, *excinfo) -> None:
         """Contextmanager protocol for `stop()`.
@@ -581,39 +562,6 @@ 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.
-
-        Args:
-            timeout: how long to wait for listening socket.
-
-        Returns:
-            first active listening socket on the backend
-
-        Raises:
-            RuntimeError: when the backend hasn't started running
-            TimeoutError: when server or sockets are 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,
-        ):
-            msg = "Backend is not listening."
-            raise TimeoutError(msg)
-        return backend.servers[0].sockets[0]
-
     def frontend(
         self,
         driver_clz: type[WebDriver] | None = None,
@@ -700,71 +648,18 @@ async def get_state(self, token: str) -> BaseState:
             The state instance associated with the given token
 
         Raises:
-            RuntimeError: when the app hasn't started running
+            AssertionError: when the state manager is not initialized
         """
-        if self.state_manager is None:
-            msg = "state_manager is not set."
-            raise RuntimeError(msg)
+        assert self.state_manager is not None, "State manager is not initialized."
+        if isinstance(self.state_manager, StateManagerDisk):
+            self.state_manager.states.clear()  # always reload from disk
         try:
             return await self.state_manager.get_state(token)
         finally:
+            await self._reset_backend_state_manager()
             if isinstance(self.state_manager, StateManagerRedis):
                 await self.state_manager.close()
 
-    async def set_state(self, token: str, **kwargs) -> None:
-        """Set the state associated with the given token.
-
-        Args:
-            token: The state token to set.
-            kwargs: Attributes to set on the state.
-
-        Raises:
-            RuntimeError: when the app hasn't started running
-        """
-        if self.state_manager is None:
-            msg = "state_manager is not set."
-            raise RuntimeError(msg)
-        state = await self.get_state(token)
-        for key, value in kwargs.items():
-            setattr(state, key, value)
-        try:
-            await self.state_manager.set_state(token, state)
-        finally:
-            if isinstance(self.state_manager, StateManagerRedis):
-                await self.state_manager.close()
-
-    @contextlib.asynccontextmanager
-    async def modify_state(self, token: str) -> AsyncIterator[BaseState]:
-        """Modify the state associated with the given token and send update to frontend.
-
-        Args:
-            token: The state token to modify
-
-        Yields:
-            The state instance associated with the given token
-
-        Raises:
-            RuntimeError: when the app hasn't started running
-        """
-        if self.state_manager is None:
-            msg = "state_manager is not set."
-            raise RuntimeError(msg)
-        if self.app_instance is None:
-            msg = "App is not running."
-            raise RuntimeError(msg)
-        app_state_manager = self.app_instance.state_manager
-        if isinstance(self.state_manager, StateManagerRedis):
-            # Temporarily replace the app's state manager with our own, since
-            # the redis connection is on the backend_thread event loop
-            self.app_instance._state_manager = self.state_manager
-        try:
-            async with self.app_instance.modify_state(token) as state:
-                yield state
-        finally:
-            if isinstance(self.state_manager, StateManagerRedis):
-                self.app_instance._state_manager = app_state_manager
-                await self.state_manager.close()
-
     def poll_for_content(
         self,
         element: WebElement,
@@ -1007,7 +902,7 @@ def _run_frontend(self):
             root=web_root,
             error_page_map=error_page_map,
         ) as self.frontend_server:
-            self.frontend_url = "http://localhost:{1}".format(
+            self.frontend_url = "http://localhost:{1}/".format(
                 *self.frontend_server.socket.getsockname()
             )
             self.frontend_server.serve_forever()
@@ -1016,10 +911,7 @@ def _start_frontend(self):
         # Set up the frontend.
         with chdir(self.app_path):
             config = reflex.config.get_config()
-            print("Polling for servers...")  # for pytest diagnosis #noqa: T201
-            config.api_url = "http://{}:{}".format(
-                *self._poll_for_servers(timeout=30).getsockname(),
-            )
+            config.api_url = f"http://localhost:{self.backend_port}"
             print("Building frontend...")  # for pytest diagnosis #noqa: T201
 
             get_config().loglevel = reflex.constants.LogLevel.INFO
@@ -1048,32 +940,17 @@ def _wait_frontend(self):
             msg = "Frontend did not start"
             raise RuntimeError(msg)
 
-    def _start_backend(self):
-        if self.app_asgi is None:
-            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.shutdown = self._get_backend_shutdown_handler()
-        print(  # noqa: T201
-            "Creating backend in a new thread..."
-        )
-        self.backend_thread = threading.Thread(target=self.backend.run)
-        self.backend_thread.start()
-        print("Backend started.")  # for pytest diagnosis #noqa: T201
+    def start(self) -> AppHarness:
+        """Start the app using reflex run subprocess.
 
-    def _poll_for_servers(self, timeout: TimeoutType = None) -> socket.socket:
-        try:
-            return super()._poll_for_servers(timeout)
-        finally:
-            environment.REFLEX_SKIP_COMPILE.set(None)
+        Returns:
+            self
+        """
+        self._initialize_app()
+        self._start_subprocess(frontend=False, mode="prod")
+        self._start_frontend()
+        self._wait_frontend()
+        return self
 
     def stop(self):
         """Stop the frontend python webserver."""
diff --git a/reflex/utils/console.py b/reflex/utils/console.py
index bffeda7b977..93379e8ab67 100644
--- a/reflex/utils/console.py
+++ b/reflex/utils/console.py
@@ -128,6 +128,21 @@ def should_use_log_file_console() -> bool:
     return environment.REFLEX_ENABLE_FULL_LOGGING.get()
 
 
+@once
+def error_log_file_console():
+    """Create a console that logs errors to a file.
+
+    Returns:
+        A Console object that logs errors to a file.
+    """
+    from reflex.environment import environment
+
+    if not (env_error_log_file := environment.REFLEX_ERROR_LOG_FILE.get()):
+        return None
+    env_error_log_file.parent.mkdir(parents=True, exist_ok=True)
+    return Console(file=env_error_log_file.open("a", encoding="utf-8"))
+
+
 def print_to_log_file(msg: str, *, dedupe: bool = False, **kwargs):
     """Print a message to the log file.
 
@@ -328,6 +343,8 @@ def error(msg: str, *, dedupe: bool = False, **kwargs):
         print(f"[red]{msg}[/red]", **kwargs)
     if should_use_log_file_console():
         print_to_log_file(f"[red]{msg}[/red]", **kwargs)
+    if error_log_file := error_log_file_console():
+        error_log_file.print(f"[red]{msg}[/red]", **kwargs)
 
 
 def ask(
diff --git a/reflex/utils/exec.py b/reflex/utils/exec.py
index 15c328cb043..3a923edb932 100644
--- a/reflex/utils/exec.py
+++ b/reflex/utils/exec.py
@@ -596,6 +596,8 @@ def run_uvicorn_backend_prod(host: str, port: int, loglevel: LogLevel):
 
         # Our default args, then env args (env args win on conflicts)
         command = [
+            sys.executable,
+            "-m",
             "gunicorn",
             "--preload",
             "--worker-class",
@@ -632,6 +634,8 @@ def run_granian_backend_prod(host: str, port: int, loglevel: LogLevel):
     from reflex.utils import processes
 
     command = [
+        sys.executable,
+        "-m",
         "granian",
         *("--log-level", "critical"),
         *("--host", host),
diff --git a/tests/integration/conftest.py b/tests/integration/conftest.py
index ad1683bb5e1..07ee39ba73f 100644
--- a/tests/integration/conftest.py
+++ b/tests/integration/conftest.py
@@ -1,9 +1,7 @@
 """Shared conftest for all integration tests."""
 
 import pytest
-from pytest_mock import MockerFixture
 
-import reflex.app
 from reflex.testing import AppHarness, AppHarnessProd
 
 
@@ -20,25 +18,3 @@ def app_harness_env(request):
         The AppHarness class to use for the test.
     """
     return request.param
-
-
-@pytest.fixture(autouse=True)
-def raise_console_error(request, mocker: MockerFixture):
-    """Spy on calls to `console.error` used by the framework.
-
-    Help catch spurious error conditions that might otherwise go unnoticed.
-
-    If a test is marked with `ignore_console_error`, the spy will be ignored
-    after the test.
-
-    Args:
-        request: The pytest request object.
-        mocker: The pytest mocker object.
-
-    Yields:
-        control to the test function.
-    """
-    spy = mocker.spy(reflex.app.console, "error")
-    yield
-    if "ignore_console_error" not in request.keywords:
-        spy.assert_not_called()
diff --git a/tests/integration/test_call_script.py b/tests/integration/test_call_script.py
index f7643202f35..72ff8c5f5c8 100644
--- a/tests/integration/test_call_script.py
+++ b/tests/integration/test_call_script.py
@@ -204,7 +204,10 @@ def reset_(self):
             self.reset()
 
     app = rx.App()
-    Path("assets/external.js").write_text(external_scripts)
+    external_scripts_path = Path("assets/external.js")
+    if not external_scripts_path.exists():
+        external_scripts_path.parent.mkdir(parents=True, exist_ok=True)
+        external_scripts_path.write_text(external_scripts)
 
     @app.add_page
     def index():
diff --git a/tests/integration/test_client_storage.py b/tests/integration/test_client_storage.py
index 59b74781c4c..7e23cbc69ae 100644
--- a/tests/integration/test_client_storage.py
+++ b/tests/integration/test_client_storage.py
@@ -4,6 +4,7 @@
 
 import asyncio
 from collections.abc import Generator
+from pathlib import Path
 
 import pytest
 from selenium.webdriver import Firefox
@@ -39,6 +40,11 @@ def set_state_var(self, value: str):
         def set_input_value(self, value: str):
             self.input_value = value
 
+        @rx.event
+        async def do_reset(self):
+            root_state = await self.get_state(rx.State)
+            root_state.reset()
+
     class ClientSideSubState(ClientSideState):
         # cookies with default settings
         c1: str = rx.Cookie()
@@ -93,6 +99,9 @@ def index():
                 read_only=True,
                 id="token",
             ),
+            rx.button(
+                "State Reset", on_click=ClientSideState.do_reset, id="reset-button"
+            ),
             rx.input(
                 placeholder="state var",
                 value=ClientSideState.state_var,
@@ -238,6 +247,7 @@ async def test_client_side_state(
     driver: WebDriver,
     local_storage: utils.LocalStorage,
     session_storage: utils.SessionStorage,
+    tmp_path: Path,
 ):
     """Test client side state.
 
@@ -246,6 +256,7 @@ async def test_client_side_state(
         driver: WebDriver instance.
         local_storage: Local storage helper.
         session_storage: Session storage helper.
+        tmp_path: pytest tmp_path fixture
     """
     app = client_side.app_instance
     assert app is not None
@@ -535,8 +546,8 @@ def set_sub_sub(var: str, value: str):
     assert s1s.text == "s1s value"
 
     # reset the backend state to force refresh from client storage
-    async with client_side.modify_state(f"{token}_{state_name}") as state:
-        state.reset()
+    reset_button = driver.find_element(By.ID, "reset-button")
+    reset_button.click()
     driver.refresh()
 
     # wait for the backend connection to send the token (again)
@@ -652,7 +663,7 @@ def set_sub_sub(var: str, value: str):
             _substate_key(token, sub_sub_state_name)
         )
     elif isinstance(client_side.state_manager, (StateManagerMemory, StateManagerDisk)):
-        del client_side.state_manager.states[token]
+        client_side.state_manager.states.pop(token, None)
     if isinstance(client_side.state_manager, StateManagerDisk):
         client_side.state_manager.token_expiration = 0
         client_side.state_manager._purge_expired_states()
@@ -666,6 +677,7 @@ async def poll_for_not_hydrated():
 
     # Trigger event to get a new instance of the state since the old was expired.
     set_sub("c1", "c1 post expire")
+    set_sub_sub("s1s", "s1s post expire")
 
     # get new references to all cookie and local storage elements (again)
     c1 = driver.find_element(By.ID, "c1")
@@ -702,7 +714,7 @@ async def poll_for_not_hydrated():
     assert s3.text == "s3 value"
     assert c1s.text == "c1s value"
     assert l1s.text == "l1s value"
-    assert s1s.text == "s1s value"
+    assert s1s.text == "s1s post expire"
 
     # Get the backend state and ensure the values are still set
     async def get_sub_state():
@@ -712,11 +724,20 @@ async def get_sub_state():
         state = root_state.substates[client_side.get_state_name("_client_side_state")]
         return state.substates[client_side.get_state_name("_client_side_sub_state")]
 
-    async def poll_for_c1_set():
+    async def get_sub_sub_state():
+        sub_state = await get_sub_state()
+        return sub_state.substates[
+            client_side.get_state_name("_client_side_sub_sub_state")
+        ]
+
+    async def poll_for_post_expire_set():
         sub_state = await get_sub_state()
-        return sub_state.c1 == "c1 post expire"
+        sub_sub_state = await get_sub_sub_state()
+        return (
+            sub_state.c1 == "c1 post expire" and sub_sub_state.s1s == "s1s post expire"
+        )
 
-    assert await AppHarness._poll_for_async(poll_for_c1_set)
+    assert await AppHarness._poll_for_async(poll_for_post_expire_set)
     sub_state = await get_sub_state()
     assert sub_state.c1 == "c1 post expire"
     assert sub_state.c2 == "c2 value"
@@ -732,12 +753,10 @@ async def poll_for_c1_set():
     assert sub_state.s1 == "s1 value"
     assert sub_state.s2 == "s2 value"
     assert sub_state.s3 == "s3 value"
-    sub_sub_state = sub_state.substates[
-        client_side.get_state_name("_client_side_sub_sub_state")
-    ]
+    sub_sub_state = await get_sub_sub_state()
     assert sub_sub_state.c1s == "c1s value"
     assert sub_sub_state.l1s == "l1s value"
-    assert sub_sub_state.s1s == "s1s value"
+    assert sub_sub_state.s1s == "s1s post expire"
 
     # clear the cookie jar and local storage, ensure state reset to default
     driver.delete_all_cookies()
diff --git a/tests/integration/test_connection_banner.py b/tests/integration/test_connection_banner.py
index c50a69e9474..897caa19f15 100644
--- a/tests/integration/test_connection_banner.py
+++ b/tests/integration/test_connection_banner.py
@@ -1,5 +1,6 @@
 """Test case for displaying the connection banner when the websocket drops."""
 
+import functools
 from collections.abc import Generator
 
 import pytest
@@ -7,17 +8,25 @@
 from selenium.webdriver.common.by import By
 
 from reflex import constants
-from reflex.environment import environment
 from reflex.testing import AppHarness, WebDriver
+from reflex.utils import processes
 
 from .utils import SessionStorage
 
 
-def ConnectionBanner():
-    """App with a connection banner."""
+def ConnectionBanner(simulate_compile_context: str):
+    """App with a connection banner.
+
+    Args:
+        simulate_compile_context: The context to run the app with.
+    """
     import asyncio
 
     import reflex as rx
+    from reflex.constants import CompileContext
+    from reflex.environment import environment
+
+    environment.REFLEX_COMPILE_CONTEXT.set(CompileContext(simulate_compile_context))
 
     class State(rx.State):
         foo: int = 0
@@ -72,11 +81,11 @@ def connection_banner(
     Yields:
         running AppHarness instance
     """
-    environment.REFLEX_COMPILE_CONTEXT.set(simulate_compile_context)
-
     with AppHarness.create(
         root=tmp_path,
-        app_source=ConnectionBanner,
+        app_source=functools.partial(
+            ConnectionBanner, simulate_compile_context=simulate_compile_context.value
+        ),
         app_name=(
             "connection_banner_reflex_cloud"
             if simulate_compile_context == constants.CompileContext.DEPLOY
@@ -144,7 +153,7 @@ async def test_connection_banner(connection_banner: AppHarness):
         connection_banner: AppHarness instance.
     """
     assert connection_banner.app_instance is not None
-    assert connection_banner.backend is not None
+    assert connection_banner.reflex_process is not None
     driver = connection_banner.frontend()
 
     _assert_token(connection_banner, driver)
@@ -161,23 +170,26 @@ async def test_connection_banner(connection_banner: AppHarness):
     # Start an long event before killing the backend, to mark event_processing=true
     delay_button.click()
 
-    # Get the backend port
-    backend_port = connection_banner._poll_for_servers().getsockname()[1]
-
-    # Kill the backend
-    connection_banner.backend.should_exit = True
-    if connection_banner.backend_thread is not None:
-        connection_banner.backend_thread.join()
+    # Kill reflex
+    connection_banner._stop_reflex()
 
     # Error modal should now be displayed
-    AppHarness.expect(lambda: has_error_modal(driver))
+    AppHarness.expect(lambda: has_error_modal(driver), timeout=30, step=1)
 
     # Increment the counter with backend down
     increment_button.click()
     assert connection_banner.poll_for_value(counter_element, exp_not_equal="0") == "1"
 
-    # Bring the backend back up
-    connection_banner._start_backend(port=backend_port)
+    # Bring the backend back up once the port is free'd.
+    if result := AppHarness._poll_for(
+        lambda: processes.handle_port(
+            "backend", connection_banner.backend_port or 0, auto_increment=False
+        ),
+        timeout=120,
+    ):
+        print(f"Port {result} is now free.")
+    assert result, f"Port is not free: {connection_banner.backend_port} after timeout."
+    connection_banner._start_subprocess(frontend=False)
 
     # Create a new StateManager to avoid async loop affinity issues w/ redis
     await connection_banner._reset_backend_state_manager()
@@ -200,7 +212,7 @@ async def test_cloud_banner(
         simulate_compile_context: Which context to set for the app.
     """
     assert connection_banner.app_instance is not None
-    assert connection_banner.backend is not None
+    assert connection_banner.reflex_process is not None
     driver = connection_banner.frontend()
 
     driver.add_cookie({"name": "backend-enabled", "value": "truly"})
diff --git a/tests/integration/test_deploy_url.py b/tests/integration/test_deploy_url.py
index cd37621d659..02971c3dc1a 100644
--- a/tests/integration/test_deploy_url.py
+++ b/tests/integration/test_deploy_url.py
@@ -76,14 +76,11 @@ def test_deploy_url(deploy_url_sample: AppHarness, driver: WebDriver) -> None:
         deploy_url_sample: AppHarness fixture for testing deploy_url.
         driver: WebDriver fixture for testing deploy_url.
     """
-    import reflex as rx
-
-    deploy_url = rx.config.get_config().deploy_url
-    assert deploy_url is not None
-    assert deploy_url != "http://localhost:3000"
-    assert deploy_url == deploy_url_sample.frontend_url
-    driver.get(deploy_url)
-    assert driver.current_url.removesuffix("/") == deploy_url.removesuffix("/")
+    assert deploy_url_sample.frontend_url is not None
+    assert (
+        deploy_url_sample.frontend_url
+        in (deploy_url_sample.app_path / ".web" / "public" / "sitemap.xml").read_text()
+    )
 
 
 def test_deploy_url_in_app(deploy_url_sample: AppHarness, driver: WebDriver) -> None:
diff --git a/tests/integration/test_event_chain.py b/tests/integration/test_event_chain.py
index cb48df0f1d5..3a6db7a4e2a 100644
--- a/tests/integration/test_event_chain.py
+++ b/tests/integration/test_event_chain.py
@@ -281,6 +281,7 @@ def event_chain(tmp_path_factory) -> Generator[AppHarness, None, None]:
         app_source=EventChain,
     ) as harness:
         yield harness
+    os.environ.pop("REFLEX_REACT_STRICT_MODE", None)
 
 
 @pytest.fixture
@@ -317,6 +318,7 @@ def event_chain_strict(tmp_path_factory) -> Generator[AppHarness, None, None]:
         app_source=EventChain,
     ) as harness:
         yield harness
+    os.environ.pop("REFLEX_REACT_STRICT_MODE", None)
 
 
 @pytest.fixture
diff --git a/tests/integration/test_exception_handlers.py b/tests/integration/test_exception_handlers.py
index 6430bc746a6..e85ab315854 100644
--- a/tests/integration/test_exception_handlers.py
+++ b/tests/integration/test_exception_handlers.py
@@ -3,7 +3,7 @@
 from __future__ import annotations
 
 import time
-from collections.abc import Generator
+from collections.abc import Callable, Generator
 
 import pytest
 from selenium.webdriver.common.by import By
@@ -13,8 +13,6 @@
 
 from reflex.testing import AppHarness, AppHarnessProd
 
-pytestmark = [pytest.mark.ignore_console_error]
-
 
 def TestApp():
     """A test app for event exception handler integration."""
@@ -86,6 +84,8 @@ def test_app(
         app_name=f"testapp_{app_harness_env.__name__.lower()}",
         app_source=TestApp,
     ) as harness:
+        # disable console.error checking for this test
+        harness.reflex_process_error_log_path = None
         yield harness
 
 
@@ -108,9 +108,35 @@ def driver(test_app: AppHarness) -> Generator[WebDriver, None, None]:
         driver.quit()
 
 
+@pytest.fixture
+def get_reflex_output(test_app: AppHarness) -> Callable[[], str]:
+    """Get the output of the reflex process.
+
+    Args:
+        test_app: harness for TestApp app
+
+    Returns:
+        The output of the reflex process.
+    """
+    assert test_app.reflex_process is not None, "app is not running"
+    assert test_app.reflex_process_log_path is not None, (
+        "reflex process log path is not set"
+    )
+    initial_offset = test_app.reflex_process_log_path.stat().st_size
+
+    def f() -> str:
+        assert test_app.reflex_process_log_path is not None, (
+            "reflex process log path is not set"
+        )
+        return test_app.reflex_process_log_path.read_bytes()[initial_offset:].decode(
+            "utf-8"
+        )
+
+    return f
+
+
 def test_frontend_exception_handler_during_runtime(
-    driver: WebDriver,
-    capsys,
+    driver: WebDriver, get_reflex_output: Callable[[], str]
 ):
     """Test calling frontend exception handler during runtime.
 
@@ -119,7 +145,7 @@ def test_frontend_exception_handler_during_runtime(
 
     Args:
         driver: WebDriver instance.
-        capsys: pytest fixture for capturing stdout and stderr.
+        get_reflex_output: Function to get the reflex process output.
 
     """
     reset_button = WebDriverWait(driver, 20).until(
@@ -131,14 +157,14 @@ def test_frontend_exception_handler_during_runtime(
     # Wait for the error to be logged
     time.sleep(2)
 
-    captured_default_handler_output = capsys.readouterr()
-    assert "induce_frontend_error" in captured_default_handler_output.out
-    assert "ReferenceError" in captured_default_handler_output.out
+    captured_default_handler_output = get_reflex_output()
+    assert "induce_frontend_error" in captured_default_handler_output
+    assert "ReferenceError" in captured_default_handler_output
 
 
 def test_backend_exception_handler_during_runtime(
     driver: WebDriver,
-    capsys,
+    get_reflex_output: Callable[[], str],
 ):
     """Test calling backend exception handler during runtime.
 
@@ -147,7 +173,7 @@ def test_backend_exception_handler_during_runtime(
 
     Args:
         driver: WebDriver instance.
-        capsys: pytest fixture for capturing stdout and stderr.
+        get_reflex_output: Function to get the reflex process output.
 
     """
     reset_button = WebDriverWait(driver, 20).until(
@@ -159,15 +185,15 @@ def test_backend_exception_handler_during_runtime(
     # Wait for the error to be logged
     time.sleep(2)
 
-    captured_default_handler_output = capsys.readouterr()
-    assert "divide_by_number" in captured_default_handler_output.out
-    assert "ZeroDivisionError" in captured_default_handler_output.out
+    captured_default_handler_output = get_reflex_output()
+    assert "divide_by_number" in captured_default_handler_output
+    assert "ZeroDivisionError" in captured_default_handler_output
 
 
 def test_frontend_exception_handler_with_react(
     test_app: AppHarness,
     driver: WebDriver,
-    capsys,
+    get_reflex_output: Callable[[], str],
 ):
     """Test calling frontend exception handler during runtime.
 
@@ -176,7 +202,7 @@ def test_frontend_exception_handler_with_react(
     Args:
         test_app: harness for TestApp app
         driver: WebDriver instance.
-        capsys: pytest fixture for capturing stdout and stderr.
+        get_reflex_output: Function to get the reflex process output.
 
     """
     reset_button = WebDriverWait(driver, 20).until(
@@ -188,11 +214,11 @@ def test_frontend_exception_handler_with_react(
     # Wait for the error to be logged
     time.sleep(2)
 
-    captured_default_handler_output = capsys.readouterr()
+    captured_default_handler_output = get_reflex_output()
     if isinstance(test_app, AppHarnessProd):
-        assert "Error: Minified React error #31" in captured_default_handler_output.out
+        assert "Error: Minified React error #31" in captured_default_handler_output
     else:
         assert (
             "Error: Objects are not valid as a React child (found: object with keys \n{invalid})"
-            in captured_default_handler_output.out
+            in captured_default_handler_output
         )
diff --git a/tests/integration/test_input.py b/tests/integration/test_input.py
index 2684db8f93e..b09715c345a 100644
--- a/tests/integration/test_input.py
+++ b/tests/integration/test_input.py
@@ -52,6 +52,11 @@ def index():
             rx.button(
                 "CLEAR", on_click=rx.set_value("on_change_input", ""), id="clear"
             ),
+            rx.button(
+                "Clear Text Backend",
+                on_click=State.set_text(""),
+                id="clear-text-backend",
+            ),
         )
 
 
@@ -146,17 +151,15 @@ async def get_state_text():
     assert fully_controlled_input.poll_for_value(plain_value_input) == "ifoonitial"
 
     # clear the input on the backend
-    async with fully_controlled_input.modify_state(
-        f"{token}_{full_state_name}"
-    ) as state:
-        state.substates[state_name].text = ""
-    assert await get_state_text() == ""
+    clear_text_backend = driver.find_element(By.ID, "clear-text-backend")
+    clear_text_backend.click()
     assert (
         fully_controlled_input.poll_for_value(
             debounce_input, exp_not_equal="ifoonitial"
         )
         == ""
     )
+    assert await get_state_text() == ""
 
     # type more characters
     debounce_input.send_keys("getting testing done")
diff --git a/tests/integration/test_large_state.py b/tests/integration/test_large_state.py
index a176fa964c4..5e78a04c97e 100644
--- a/tests/integration/test_large_state.py
+++ b/tests/integration/test_large_state.py
@@ -42,7 +42,16 @@ def get_driver(large_state) -> WebDriver:
     return large_state.frontend()
 
 
-@pytest.mark.parametrize("var_count", [1, 10, 100, 1000, 10000])
+@pytest.mark.parametrize(
+    "var_count",
+    [
+        1,
+        10,
+        100,
+        1000,
+        pytest.param(10000, marks=pytest.mark.skip(reason="Invalid string length")),
+    ],
+)
 def test_large_state(var_count: int, tmp_path_factory, benchmark):
     """Measure how long it takes for button click => state update to round trip.
 
diff --git a/tests/integration/test_lifespan.py b/tests/integration/test_lifespan.py
index aa164281b8e..bb14a48fe50 100644
--- a/tests/integration/test_lifespan.py
+++ b/tests/integration/test_lifespan.py
@@ -3,6 +3,7 @@
 import functools
 from collections.abc import Generator
 
+import httpx
 import pytest
 from selenium.webdriver.common.by import By
 
@@ -23,6 +24,8 @@ def LifespanApp(
     import asyncio
     from contextlib import asynccontextmanager
 
+    from starlette.responses import JSONResponse
+
     import reflex as rx
 
     lifespan_task_global = 0
@@ -90,6 +93,17 @@ def index():
     app.register_lifespan_task(lifespan_task)
     app.register_lifespan_task(lifespan_context, inc=2)
     app.add_page(index)
+    assert app._api is not None
+    app._api.add_route(
+        "/lifespan_globals",
+        lambda req: JSONResponse(
+            {
+                "task_global": lifespan_task_global,
+                "context_global": lifespan_context_global,
+            }
+        ),
+        methods=["GET"],
+    )
 
 
 @pytest.fixture(
@@ -154,7 +168,7 @@ async def test_lifespan(lifespan_app: AppHarness):
         lifespan_app: harness for LifespanApp app
     """
     assert lifespan_app.app_module is not None, "app module is not found"
-    assert lifespan_app.app_instance is not None, "app is not running"
+    assert lifespan_app.reflex_process is not None, "app is not running"
     driver = lifespan_app.frontend()
 
     ss = SessionStorage(driver)
@@ -164,21 +178,21 @@ async def test_lifespan(lifespan_app: AppHarness):
     task_global = driver.find_element(By.ID, "task_global")
 
     assert lifespan_app.poll_for_content(context_global, exp_not_equal="0") == "2"
-    assert lifespan_app.app_module.lifespan_context_global == 2
+    async with httpx.AsyncClient() as client:
+        response = await client.get(
+            f"http://localhost:{lifespan_app.backend_port}/lifespan_globals"
+        )
+        backend_values = response.json()
+        assert backend_values["context_global"] == 2
 
     original_task_global_text = task_global.text
     original_task_global_value = int(original_task_global_text)
     lifespan_app.poll_for_content(task_global, exp_not_equal=original_task_global_text)
     driver.find_element(By.ID, "toggle-tick").click()  # avoid teardown errors
-    assert lifespan_app.app_module.lifespan_task_global > original_task_global_value
+    async with httpx.AsyncClient() as client:
+        response = await client.get(
+            f"http://localhost:{lifespan_app.backend_port}/lifespan_globals"
+        )
+        backend_values = response.json()
+        assert backend_values["task_global"] > original_task_global_value
     assert int(task_global.text) > original_task_global_value
-
-    # Kill the backend
-    assert lifespan_app.backend is not None
-    lifespan_app.backend.should_exit = True
-    if lifespan_app.backend_thread is not None:
-        lifespan_app.backend_thread.join()
-
-    # Check that the lifespan tasks have been cancelled
-    assert lifespan_app.app_module.lifespan_task_global == 0
-    assert lifespan_app.app_module.lifespan_context_global == 4
diff --git a/tests/integration/test_tailwind.py b/tests/integration/test_tailwind.py
index 5f17e04366d..2aaa73b4320 100644
--- a/tests/integration/test_tailwind.py
+++ b/tests/integration/test_tailwind.py
@@ -102,7 +102,7 @@ def test_tailwind_app(tailwind_app: AppHarness, tailwind_version: bool):
         tailwind_version: Tailwind version to use. If 0, tailwind is disabled.
     """
     assert tailwind_app.app_instance is not None
-    assert tailwind_app.backend is not None
+    assert tailwind_app.reflex_process is not None
 
     driver = tailwind_app.frontend()
 
diff --git a/tests/integration/tests_playwright/test_stateless_app.py b/tests/integration/tests_playwright/test_stateless_app.py
index 58a52f52d6b..a10846ca927 100644
--- a/tests/integration/tests_playwright/test_stateless_app.py
+++ b/tests/integration/tests_playwright/test_stateless_app.py
@@ -6,7 +6,6 @@
 import pytest
 from playwright.sync_api import Page, expect
 
-import reflex as rx
 from reflex.testing import AppHarness
 
 
@@ -46,13 +45,15 @@ def test_statelessness(stateless_app: AppHarness, page: Page):
         page: A Playwright page.
     """
     assert stateless_app.frontend_url is not None
-    assert stateless_app.backend is not None
-    assert stateless_app.backend.started
+    assert stateless_app.reflex_process is not None
+    assert (
+        stateless_app.reflex_process.poll() is None
+    )  # Ensure the process is still running
 
-    res = httpx.get(rx.config.get_config().api_url + "/_event")
+    res = httpx.get(f"http://localhost:{stateless_app.backend_port}/_event")
     assert res.status_code == 404
 
-    res2 = httpx.get(rx.config.get_config().api_url + "/ping")
+    res2 = httpx.get(f"http://localhost:{stateless_app.backend_port}/ping")
     assert res2.status_code == 200
 
     page.goto(stateless_app.frontend_url)
diff --git a/tests/units/test_testing.py b/tests/units/test_testing.py
index 8c8f1461bf0..2db0c790cf3 100644
--- a/tests/units/test_testing.py
+++ b/tests/units/test_testing.py
@@ -23,18 +23,16 @@ def BasicApp():
         class State(rx.State):
             pass
 
-        app = rx.App(_state=State)
+        app = rx.App()
         app.add_page(lambda: rx.text("Basic App"), route="/", title="index")
-        app._compile()
 
     with AppHarness.create(
         root=tmp_path,
         app_source=BasicApp,
     ) as harness:
         assert harness.app_instance is not None
-        assert harness.backend is not None
         assert harness.frontend_url is not None
-        assert harness.frontend_process is not None
-        assert harness.frontend_process.poll() is None
+        assert harness.reflex_process is not None
+        assert harness.reflex_process.poll() is None
 
-    assert harness.frontend_process.poll() is not None
+    assert harness.reflex_process.poll() is not None