diff --git a/reflex/environment.py b/reflex/environment.py
index fa64c87aace..b021d7eda13 100644
--- a/reflex/environment.py
+++ b/reflex/environment.py
@@ -650,9 +650,6 @@ 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 07b4acc0572..b18273342d6 100644
--- a/reflex/testing.py
+++ b/reflex/testing.py
@@ -10,6 +10,7 @@
 import os
 import platform
 import re
+import signal
 import socket
 import socketserver
 import subprocess
@@ -18,14 +19,17 @@
 import threading
 import time
 import types
-from collections.abc import Callable, Coroutine, Sequence
+from collections.abc import AsyncIterator, 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 reflex
+import reflex.environment
 import reflex.reflex
+import reflex.utils.build
 import reflex.utils.exec
 import reflex.utils.format
 import reflex.utils.prerequisites
@@ -41,7 +45,9 @@
     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
@@ -95,32 +101,6 @@ 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."""
-
-
-def _is_port_responsive(port: int) -> bool:
-    """Check if a port is responsive.
-
-    Args:
-        port: the port to check
-
-    Returns:
-        True if the port is responsive, False otherwise
-    """
-    try:
-        with contextlib.closing(
-            socket.socket(socket.AF_INET, socket.SOCK_STREAM)
-        ) as sock:
-            return sock.connect_ex(("127.0.0.1", port)) == 0
-    except (OverflowError, PermissionError, OSError):
-        return False
-
-
 @dataclasses.dataclass
 class AppHarness:
     """AppHarness executes a reflex app in-process for testing."""
@@ -133,15 +113,14 @@ class AppHarness:
     app_module_path: Path
     app_module: types.ModuleType | None = None
     app_instance: reflex.App | None = None
-    reflex_process: subprocess.Popen | None = None
-    reflex_process_log_path: Path | None = None
-    reflex_process_error_log_path: Path | None = None
+    app_asgi: ASGIApp | None = None
+    frontend_process: subprocess.Popen | None = None
     frontend_url: str | None = None
-    backend_port: int | None = None
-    frontend_port: int | None = None
+    frontend_output_thread: threading.Thread | None = None
+    backend_thread: threading.Thread | None = None
+    backend: uvicorn.Server | 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(
@@ -294,116 +273,82 @@ 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
                 )
             )
-            # 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"
+            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
             )
 
     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 _start_subprocess(
-        self, backend: bool = True, frontend: bool = True, mode: str = "dev"
-    ):
-        """Start the reflex app using subprocess instead of threads.
-
-        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)
+    def _get_backend_shutdown_handler(self):
+        if self.backend is None:
+            msg = "Backend was not initialized."
+            raise RuntimeError(msg)
 
-    def _wait_for_servers(self, backend: bool, frontend: bool):
-        """Wait for both frontend and backend servers to be ready by parsing console output.
+        original_shutdown = self.backend.shutdown
 
-        Args:
-            backend: Whether to wait for the backend server to be ready.
-            frontend: Whether to wait for the frontend server to be ready.
+        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()
 
-        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)
+            # 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()
 
-        if backend and self.backend_port is None:
-            msg = "Backend port is not set."
-            raise RuntimeError(msg)
+            # sqlalchemy async engine shutdown handler
+            try:
+                async_engine = reflex.model.get_async_engine(None)
+            except ValueError:
+                pass
+            else:
+                await async_engine.dispose()
 
-        if frontend and self.frontend_port is None:
-            msg = "Frontend port is not set."
-            raise RuntimeError(msg)
+            await original_shutdown(*args, **kwargs)
 
-        frontend_ready = False
-        backend_ready = False
+        return _shutdown
 
-        while not ((not frontend or frontend_ready) and (not backend or backend_ready)):
-            if backend and self.backend_port and _is_port_responsive(self.backend_port):
-                backend_ready = True
-            if (
-                frontend
-                and self.frontend_port
-                and _is_port_responsive(self.frontend_port)
-            ):
-                frontend_ready = True
-                self.frontend_url = f"http://localhost:{self.frontend_port}/"
-            time.sleep(POLL_INTERVAL)
+    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
 
     async def _reset_backend_state_manager(self):
         """Reset the StateManagerRedis event loop affinity.
@@ -431,14 +376,78 @@ 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 app using reflex run subprocess.
+        """Start the backend in a new thread and dev frontend as a separate process.
 
         Returns:
             self
         """
         self._initialize_app()
-        self._start_subprocess()
+        self._start_backend()
+        self._start_frontend()
+        self._wait_frontend()
         return self
 
     @staticmethod
@@ -467,48 +476,43 @@ def __enter__(self) -> AppHarness:
         return self.start()
 
     def stop(self) -> None:
-        """Stop the reflex subprocess."""
+        """Stop the frontend and backend servers."""
+        import psutil
+
+        # Quit browsers first to avoid any lingering events being sent during shutdown.
         for driver in self._frontends:
             driver.quit()
 
-        self._stop_reflex()
         self._reload_state_module()
 
-    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 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,
             )
-            raise ReflexProcessExitNonZeroError(msg)
+            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()
 
     def __exit__(self, *excinfo) -> None:
         """Contextmanager protocol for `stop()`.
@@ -577,6 +581,39 @@ 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,
@@ -663,18 +700,71 @@ async def get_state(self, token: str) -> BaseState:
             The state instance associated with the given token
 
         Raises:
-            AssertionError: when the state manager is not initialized
+            RuntimeError: when the app hasn't started running
         """
-        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
+        if self.state_manager is None:
+            msg = "state_manager is not set."
+            raise RuntimeError(msg)
         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,
@@ -917,7 +1007,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()
@@ -926,7 +1016,10 @@ def _start_frontend(self):
         # Set up the frontend.
         with chdir(self.app_path):
             config = reflex.config.get_config()
-            config.api_url = f"http://localhost:{self.backend_port}"
+            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
 
             get_config().loglevel = reflex.constants.LogLevel.INFO
@@ -955,17 +1048,32 @@ def _wait_frontend(self):
             msg = "Frontend did not start"
             raise RuntimeError(msg)
 
-    def start(self) -> AppHarness:
-        """Start the app using reflex run subprocess.
+    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
 
-        Returns:
-            self
-        """
-        self._initialize_app()
-        self._start_subprocess(frontend=False, mode="prod")
-        self._start_frontend()
-        self._wait_frontend()
-        return self
+    def _poll_for_servers(self, timeout: TimeoutType = None) -> socket.socket:
+        try:
+            return super()._poll_for_servers(timeout)
+        finally:
+            environment.REFLEX_SKIP_COMPILE.set(None)
 
     def stop(self):
         """Stop the frontend python webserver."""
diff --git a/reflex/utils/console.py b/reflex/utils/console.py
index f6438a32179..e01101273d9 100644
--- a/reflex/utils/console.py
+++ b/reflex/utils/console.py
@@ -128,21 +128,6 @@ 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.
 
@@ -343,8 +328,6 @@ 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 c76541151d7..87803f8a153 100644
--- a/reflex/utils/exec.py
+++ b/reflex/utils/exec.py
@@ -603,8 +603,6 @@ 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", "uvicorn.workers.UvicornH11Worker"),
@@ -641,8 +639,6 @@ 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 07ee39ba73f..ad1683bb5e1 100644
--- a/tests/integration/conftest.py
+++ b/tests/integration/conftest.py
@@ -1,7 +1,9 @@
 """Shared conftest for all integration tests."""
 
 import pytest
+from pytest_mock import MockerFixture
 
+import reflex.app
 from reflex.testing import AppHarness, AppHarnessProd
 
 
@@ -18,3 +20,25 @@ 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 72ff8c5f5c8..f7643202f35 100644
--- a/tests/integration/test_call_script.py
+++ b/tests/integration/test_call_script.py
@@ -204,10 +204,7 @@ def reset_(self):
             self.reset()
 
     app = rx.App()
-    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)
+    Path("assets/external.js").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 7e23cbc69ae..59b74781c4c 100644
--- a/tests/integration/test_client_storage.py
+++ b/tests/integration/test_client_storage.py
@@ -4,7 +4,6 @@
 
 import asyncio
 from collections.abc import Generator
-from pathlib import Path
 
 import pytest
 from selenium.webdriver import Firefox
@@ -40,11 +39,6 @@ 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()
@@ -99,9 +93,6 @@ 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,
@@ -247,7 +238,6 @@ async def test_client_side_state(
     driver: WebDriver,
     local_storage: utils.LocalStorage,
     session_storage: utils.SessionStorage,
-    tmp_path: Path,
 ):
     """Test client side state.
 
@@ -256,7 +246,6 @@ 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
@@ -546,8 +535,8 @@ def set_sub_sub(var: str, value: str):
     assert s1s.text == "s1s value"
 
     # reset the backend state to force refresh from client storage
-    reset_button = driver.find_element(By.ID, "reset-button")
-    reset_button.click()
+    async with client_side.modify_state(f"{token}_{state_name}") as state:
+        state.reset()
     driver.refresh()
 
     # wait for the backend connection to send the token (again)
@@ -663,7 +652,7 @@ def set_sub_sub(var: str, value: str):
             _substate_key(token, sub_sub_state_name)
         )
     elif isinstance(client_side.state_manager, (StateManagerMemory, StateManagerDisk)):
-        client_side.state_manager.states.pop(token, None)
+        del client_side.state_manager.states[token]
     if isinstance(client_side.state_manager, StateManagerDisk):
         client_side.state_manager.token_expiration = 0
         client_side.state_manager._purge_expired_states()
@@ -677,7 +666,6 @@ 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")
@@ -714,7 +702,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 post expire"
+    assert s1s.text == "s1s value"
 
     # Get the backend state and ensure the values are still set
     async def get_sub_state():
@@ -724,20 +712,11 @@ 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 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():
+    async def poll_for_c1_set():
         sub_state = await get_sub_state()
-        sub_sub_state = await get_sub_sub_state()
-        return (
-            sub_state.c1 == "c1 post expire" and sub_sub_state.s1s == "s1s post expire"
-        )
+        return sub_state.c1 == "c1 post expire"
 
-    assert await AppHarness._poll_for_async(poll_for_post_expire_set)
+    assert await AppHarness._poll_for_async(poll_for_c1_set)
     sub_state = await get_sub_state()
     assert sub_state.c1 == "c1 post expire"
     assert sub_state.c2 == "c2 value"
@@ -753,10 +732,12 @@ async def poll_for_post_expire_set():
     assert sub_state.s1 == "s1 value"
     assert sub_state.s2 == "s2 value"
     assert sub_state.s3 == "s3 value"
-    sub_sub_state = await get_sub_sub_state()
+    sub_sub_state = sub_state.substates[
+        client_side.get_state_name("_client_side_sub_sub_state")
+    ]
     assert sub_sub_state.c1s == "c1s value"
     assert sub_sub_state.l1s == "l1s value"
-    assert sub_sub_state.s1s == "s1s post expire"
+    assert sub_sub_state.s1s == "s1s value"
 
     # 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 897caa19f15..c50a69e9474 100644
--- a/tests/integration/test_connection_banner.py
+++ b/tests/integration/test_connection_banner.py
@@ -1,6 +1,5 @@
 """Test case for displaying the connection banner when the websocket drops."""
 
-import functools
 from collections.abc import Generator
 
 import pytest
@@ -8,25 +7,17 @@
 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(simulate_compile_context: str):
-    """App with a connection banner.
-
-    Args:
-        simulate_compile_context: The context to run the app with.
-    """
+def ConnectionBanner():
+    """App with a connection banner."""
     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
@@ -81,11 +72,11 @@ def connection_banner(
     Yields:
         running AppHarness instance
     """
+    environment.REFLEX_COMPILE_CONTEXT.set(simulate_compile_context)
+
     with AppHarness.create(
         root=tmp_path,
-        app_source=functools.partial(
-            ConnectionBanner, simulate_compile_context=simulate_compile_context.value
-        ),
+        app_source=ConnectionBanner,
         app_name=(
             "connection_banner_reflex_cloud"
             if simulate_compile_context == constants.CompileContext.DEPLOY
@@ -153,7 +144,7 @@ async def test_connection_banner(connection_banner: AppHarness):
         connection_banner: AppHarness instance.
     """
     assert connection_banner.app_instance is not None
-    assert connection_banner.reflex_process is not None
+    assert connection_banner.backend is not None
     driver = connection_banner.frontend()
 
     _assert_token(connection_banner, driver)
@@ -170,26 +161,23 @@ async def test_connection_banner(connection_banner: AppHarness):
     # Start an long event before killing the backend, to mark event_processing=true
     delay_button.click()
 
-    # Kill reflex
-    connection_banner._stop_reflex()
+    # 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()
 
     # Error modal should now be displayed
-    AppHarness.expect(lambda: has_error_modal(driver), timeout=30, step=1)
+    AppHarness.expect(lambda: has_error_modal(driver))
 
     # 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 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)
+    # Bring the backend back up
+    connection_banner._start_backend(port=backend_port)
 
     # Create a new StateManager to avoid async loop affinity issues w/ redis
     await connection_banner._reset_backend_state_manager()
@@ -212,7 +200,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.reflex_process is not None
+    assert connection_banner.backend 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 02971c3dc1a..cd37621d659 100644
--- a/tests/integration/test_deploy_url.py
+++ b/tests/integration/test_deploy_url.py
@@ -76,11 +76,14 @@ 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.
     """
-    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()
-    )
+    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("/")
 
 
 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 3a6db7a4e2a..cb48df0f1d5 100644
--- a/tests/integration/test_event_chain.py
+++ b/tests/integration/test_event_chain.py
@@ -281,7 +281,6 @@ 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
@@ -318,7 +317,6 @@ 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 e85ab315854..6430bc746a6 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 Callable, Generator
+from collections.abc import Generator
 
 import pytest
 from selenium.webdriver.common.by import By
@@ -13,6 +13,8 @@
 
 from reflex.testing import AppHarness, AppHarnessProd
 
+pytestmark = [pytest.mark.ignore_console_error]
+
 
 def TestApp():
     """A test app for event exception handler integration."""
@@ -84,8 +86,6 @@ 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,35 +108,9 @@ 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, get_reflex_output: Callable[[], str]
+    driver: WebDriver,
+    capsys,
 ):
     """Test calling frontend exception handler during runtime.
 
@@ -145,7 +119,7 @@ def test_frontend_exception_handler_during_runtime(
 
     Args:
         driver: WebDriver instance.
-        get_reflex_output: Function to get the reflex process output.
+        capsys: pytest fixture for capturing stdout and stderr.
 
     """
     reset_button = WebDriverWait(driver, 20).until(
@@ -157,14 +131,14 @@ def test_frontend_exception_handler_during_runtime(
     # Wait for the error to be logged
     time.sleep(2)
 
-    captured_default_handler_output = get_reflex_output()
-    assert "induce_frontend_error" in captured_default_handler_output
-    assert "ReferenceError" in captured_default_handler_output
+    captured_default_handler_output = capsys.readouterr()
+    assert "induce_frontend_error" in captured_default_handler_output.out
+    assert "ReferenceError" in captured_default_handler_output.out
 
 
 def test_backend_exception_handler_during_runtime(
     driver: WebDriver,
-    get_reflex_output: Callable[[], str],
+    capsys,
 ):
     """Test calling backend exception handler during runtime.
 
@@ -173,7 +147,7 @@ def test_backend_exception_handler_during_runtime(
 
     Args:
         driver: WebDriver instance.
-        get_reflex_output: Function to get the reflex process output.
+        capsys: pytest fixture for capturing stdout and stderr.
 
     """
     reset_button = WebDriverWait(driver, 20).until(
@@ -185,15 +159,15 @@ def test_backend_exception_handler_during_runtime(
     # Wait for the error to be logged
     time.sleep(2)
 
-    captured_default_handler_output = get_reflex_output()
-    assert "divide_by_number" in captured_default_handler_output
-    assert "ZeroDivisionError" in captured_default_handler_output
+    captured_default_handler_output = capsys.readouterr()
+    assert "divide_by_number" in captured_default_handler_output.out
+    assert "ZeroDivisionError" in captured_default_handler_output.out
 
 
 def test_frontend_exception_handler_with_react(
     test_app: AppHarness,
     driver: WebDriver,
-    get_reflex_output: Callable[[], str],
+    capsys,
 ):
     """Test calling frontend exception handler during runtime.
 
@@ -202,7 +176,7 @@ def test_frontend_exception_handler_with_react(
     Args:
         test_app: harness for TestApp app
         driver: WebDriver instance.
-        get_reflex_output: Function to get the reflex process output.
+        capsys: pytest fixture for capturing stdout and stderr.
 
     """
     reset_button = WebDriverWait(driver, 20).until(
@@ -214,11 +188,11 @@ def test_frontend_exception_handler_with_react(
     # Wait for the error to be logged
     time.sleep(2)
 
-    captured_default_handler_output = get_reflex_output()
+    captured_default_handler_output = capsys.readouterr()
     if isinstance(test_app, AppHarnessProd):
-        assert "Error: Minified React error #31" in captured_default_handler_output
+        assert "Error: Minified React error #31" in captured_default_handler_output.out
     else:
         assert (
             "Error: Objects are not valid as a React child (found: object with keys \n{invalid})"
-            in captured_default_handler_output
+            in captured_default_handler_output.out
         )
diff --git a/tests/integration/test_input.py b/tests/integration/test_input.py
index b09715c345a..2684db8f93e 100644
--- a/tests/integration/test_input.py
+++ b/tests/integration/test_input.py
@@ -52,11 +52,6 @@ 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",
-            ),
         )
 
 
@@ -151,15 +146,17 @@ async def get_state_text():
     assert fully_controlled_input.poll_for_value(plain_value_input) == "ifoonitial"
 
     # clear the input on the backend
-    clear_text_backend = driver.find_element(By.ID, "clear-text-backend")
-    clear_text_backend.click()
+    async with fully_controlled_input.modify_state(
+        f"{token}_{full_state_name}"
+    ) as state:
+        state.substates[state_name].text = ""
+    assert await get_state_text() == ""
     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 5e78a04c97e..a176fa964c4 100644
--- a/tests/integration/test_large_state.py
+++ b/tests/integration/test_large_state.py
@@ -42,16 +42,7 @@ def get_driver(large_state) -> WebDriver:
     return large_state.frontend()
 
 
-@pytest.mark.parametrize(
-    "var_count",
-    [
-        1,
-        10,
-        100,
-        1000,
-        pytest.param(10000, marks=pytest.mark.skip(reason="Invalid string length")),
-    ],
-)
+@pytest.mark.parametrize("var_count", [1, 10, 100, 1000, 10000])
 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 bb14a48fe50..aa164281b8e 100644
--- a/tests/integration/test_lifespan.py
+++ b/tests/integration/test_lifespan.py
@@ -3,7 +3,6 @@
 import functools
 from collections.abc import Generator
 
-import httpx
 import pytest
 from selenium.webdriver.common.by import By
 
@@ -24,8 +23,6 @@ def LifespanApp(
     import asyncio
     from contextlib import asynccontextmanager
 
-    from starlette.responses import JSONResponse
-
     import reflex as rx
 
     lifespan_task_global = 0
@@ -93,17 +90,6 @@ 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(
@@ -168,7 +154,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.reflex_process is not None, "app is not running"
+    assert lifespan_app.app_instance is not None, "app is not running"
     driver = lifespan_app.frontend()
 
     ss = SessionStorage(driver)
@@ -178,21 +164,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"
-    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
+    assert lifespan_app.app_module.lifespan_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
-    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 lifespan_app.app_module.lifespan_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 beb957e4957..071570ce44a 100644
--- a/tests/integration/test_tailwind.py
+++ b/tests/integration/test_tailwind.py
@@ -106,7 +106,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.reflex_process is not None
+    assert tailwind_app.backend 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 33d390169ab..9d3942fccc7 100644
--- a/tests/integration/tests_playwright/test_stateless_app.py
+++ b/tests/integration/tests_playwright/test_stateless_app.py
@@ -6,6 +6,7 @@
 import pytest
 from playwright.sync_api import Page, expect
 
+import reflex as rx
 from reflex.testing import AppHarness
 
 
@@ -45,15 +46,13 @@ def test_statelessness(stateless_app: AppHarness, page: Page):
         page: A Playwright page.
     """
     assert stateless_app.frontend_url is not None
-    assert stateless_app.reflex_process is not None
-    assert (
-        stateless_app.reflex_process.poll() is None
-    )  # Ensure the process is still running
+    assert stateless_app.backend is not None
+    assert stateless_app.backend.started
 
-    res = httpx.get(f"http://localhost:{stateless_app.backend_port}/_event")
+    res = httpx.get(rx.config.get_config().api_url + "/_event")
     assert res.status_code == 404
 
-    res2 = httpx.get(f"http://localhost:{stateless_app.backend_port}/ping")
+    res2 = httpx.get(rx.config.get_config().api_url + "/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 2db0c790cf3..8c8f1461bf0 100644
--- a/tests/units/test_testing.py
+++ b/tests/units/test_testing.py
@@ -23,16 +23,18 @@ def BasicApp():
         class State(rx.State):
             pass
 
-        app = rx.App()
+        app = rx.App(_state=State)
         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.reflex_process is not None
-        assert harness.reflex_process.poll() is None
+        assert harness.frontend_process is not None
+        assert harness.frontend_process.poll() is None
 
-    assert harness.reflex_process.poll() is not None
+    assert harness.frontend_process.poll() is not None