diff --git a/news/+prod-fork-preload.performance.md b/news/+prod-fork-preload.performance.md
new file mode 100644
index 00000000000..f0bfcd0171f
--- /dev/null
+++ b/news/+prod-fork-preload.performance.md
@@ -0,0 +1 @@
+Production backend workers on Linux are now forked from a supervisor that has already imported the app, so they share its memory instead of each importing the app again, which cuts memory use and startup time. Set `REFLEX_BACKEND_START_METHOD=spawn` for apps that are not fork-safe.
diff --git a/packages/reflex-base/news/+prod-fork-preload.performance.md b/packages/reflex-base/news/+prod-fork-preload.performance.md
new file mode 100644
index 00000000000..53d78489de8
--- /dev/null
+++ b/packages/reflex-base/news/+prod-fork-preload.performance.md
@@ -0,0 +1 @@
+Add the `REFLEX_BACKEND_START_METHOD` environment variable to choose how production backend workers are started (`fork`, `spawn`, or `forkserver`).
diff --git a/packages/reflex-base/src/reflex_base/environment.py b/packages/reflex-base/src/reflex_base/environment.py
index 867e2dcfbc0..00e38b9e882 100644
--- a/packages/reflex-base/src/reflex_base/environment.py
+++ b/packages/reflex-base/src/reflex_base/environment.py
@@ -815,6 +815,13 @@ class EnvironmentVariables:
     # Whether to run Granian in a spawn process. This enables Reflex to pick up on environment variable changes between hot reloads.
     REFLEX_STRICT_HOT_RELOAD: EnvVar[bool] = env_var(False)
 
+    # The multiprocessing start method for production backend workers. Unset means "fork" wherever
+    # the interpreter itself defaults to a fork-based method (Linux), so workers share the app
+    # preloaded by the supervisor. Set to "spawn" for apps that are not fork-safe.
+    REFLEX_BACKEND_START_METHOD: EnvVar[
+        Literal["fork", "spawn", "forkserver"] | None
+    ] = env_var(None)
+
     # The path to the reflex log file. If not set, the log file will be stored in the reflex user directory.
     REFLEX_LOG_FILE: EnvVar[Path | None] = env_var(None)
 
diff --git a/pyproject.toml b/pyproject.toml
index e46f43cbd28..71671f49578 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -32,7 +32,7 @@ dependencies = [
   "starlette >=1.3.1",
   "typing_extensions >=4.13.0",
   "wrapt >=1.17.0,<2.6",
-  "reflex-base >= 0.10.0a2.dev0",
+  "reflex-base >= 0.10.1.dev0",
   "reflex-components-code >= 0.10.0",
   "reflex-components-core >= 0.10.0",
   "reflex-components-dataeditor >= 0.10.0",
diff --git a/reflex/reflex.py b/reflex/reflex.py
index f22dd4c7d32..facbd1fa24d 100644
--- a/reflex/reflex.py
+++ b/reflex/reflex.py
@@ -565,8 +565,9 @@ def _run_prod(running_mode: constants.RunningMode, port: int, host: str):
 
     _skip_compile()
 
-    # Post a telemetry event.
-    telemetry.send("run-prod")
+    # Post a telemetry event. Sent synchronously: on Linux the backend forks its
+    # workers from this process, which must hold no telemetry thread by then.
+    telemetry._process_event("run-prod", config.telemetry_enabled)
 
     # Display custom message when there is a keyboard interrupt.
     atexit.register(processes.atexit_handler)
diff --git a/reflex/utils/exec.py b/reflex/utils/exec.py
index ad56c97abad..cea18a0ff11 100644
--- a/reflex/utils/exec.py
+++ b/reflex/utils/exec.py
@@ -1028,6 +1028,55 @@ def run_uvicorn_backend_prod(
     )
 
 
+def _backend_start_method() -> str | None:
+    """Resolve the multiprocessing start method for production backend workers.
+
+    Returns:
+        The start method to force, or None to keep the interpreter default.
+    """
+    if (method := environment.REFLEX_BACKEND_START_METHOD.get()) is not None:
+        return method
+    import multiprocessing
+
+    # Python defaults to fork (<3.14) or forkserver (3.14+) on Linux and to
+    # spawn elsewhere; only where fork is already the platform norm do we rely
+    # on it so workers can share the supervisor's pages.
+    if multiprocessing.get_start_method() in ("fork", "forkserver"):
+        return "fork"
+    return None
+
+
+def _preload_for_fork(app_target: str | None) -> bool:
+    """Import the app in the supervisor so forked workers share its pages.
+
+    Args:
+        app_target: The ASGI app target; None means the reflex app, which is
+            imported here. Any other target lives in an already-loaded module.
+
+    Returns:
+        Whether forking is safe; False when a telemetry thread is still alive.
+    """
+    import gc
+
+    from reflex_base.utils import serializers
+
+    from reflex.utils import prerequisites, telemetry
+
+    if app_target is None:
+        prerequisites.get_app()
+    serializers._prepare_serializers_for_fork()
+    # Forking while the telemetry thread may hold a lock can deadlock the child.
+    # Nothing in the supervisor sends telemetry after this point, so no thread
+    # exists at any later fork either (e.g. a worker respawn).
+    if not telemetry._shutdown_executor():
+        return False
+    # Freezing keeps worker GC passes from writing to the preloaded objects'
+    # headers, which would copy-on-write the shared pages private again.
+    gc.collect()
+    gc.freeze()
+    return True
+
+
 def run_granian_backend_prod(
     host: str, port: int, loglevel: LogLevel, app_target: str | None = None
 ):
@@ -1039,12 +1088,22 @@ def run_granian_backend_prod(
         loglevel: The log level.
         app_target: The ASGI app target to run. Defaults to the reflex app instance.
     """
+    import multiprocessing
+
     from granian.constants import Interfaces
     from granian.log import LogLevels
     from granian.server import Server as Granian
 
     logger.debug("Using Granian for backend")
 
+    if (start_method := _backend_start_method()) is not None:
+        # Set before the preload so the app's import-time multiprocessing
+        # objects use the requested context.
+        multiprocessing.set_start_method(start_method, force=True)
+        if start_method == "fork" and not _preload_for_fork(app_target):
+            logger.debug("A telemetry send is still running; spawning workers.")
+            multiprocessing.set_start_method("spawn", force=True)
+
     granian_app = Granian(
         target=app_target or get_app_instance_from_file(),
         factory=True,
diff --git a/reflex/utils/telemetry.py b/reflex/utils/telemetry.py
index af954ccc1e3..e669935681c 100644
--- a/reflex/utils/telemetry.py
+++ b/reflex/utils/telemetry.py
@@ -501,6 +501,46 @@ def _get_telemetry_executor() -> ThreadPoolExecutor:
     return _executor
 
 
+def _shutdown_executor(timeout: float = 2) -> bool:
+    """Deliver queued telemetry and stop the worker thread.
+
+    Called by a supervisor right before it forks workers, so no telemetry
+    thread is alive at the fork. A send that stalls past the timeout (e.g. on
+    a DNS lookup) is abandoned so it cannot hold up startup, and the caller
+    must then not fork.
+
+    Args:
+        timeout: Maximum number of seconds to wait for queued telemetry.
+
+    Returns:
+        Whether the worker thread has stopped, i.e. whether forking is safe.
+    """
+    global _executor
+    if (executor := _executor) is None:
+        return True
+    drained = _flush(timeout)
+    if not drained:
+        logger.debug(f"Telemetry did not drain within {timeout}s before forking.")
+    # Stay published while stopping: a racing send() then fails to submit (and
+    # is suppressed) instead of starting a second pool before the fork.
+    executor.shutdown(wait=drained, cancel_futures=True)
+    with _executor_lock:
+        if _executor is executor:
+            _executor = None
+    return not any(thread.is_alive() for thread in executor._threads)
+
+
+def _reset_executor_after_fork() -> None:
+    """Drop the inherited executor and lock; the child owns neither's thread."""
+    global _executor, _executor_lock
+    _executor_lock = threading.Lock()
+    _executor = None
+
+
+if hasattr(os, "register_at_fork"):
+    os.register_at_fork(after_in_child=_reset_executor_after_fork)
+
+
 def _current_registration_context() -> RegistrationContext | None:
     """Return the caller's RegistrationContext, or None if none is attached.
 
@@ -634,8 +674,7 @@ def _maybe_alias_legacy_distinct_id(telemetry_enabled: bool | None) -> None:
             # Don't latch: a later enabled send in this process should retry.
             return
 
-        # Latch before the alias send below (which re-enters send()) so it cannot
-        # recurse and so the attempt happens at most once per process.
+        # Latch first so the attempt happens at most once per process.
         _legacy_alias_attempted = True
 
         # Resolve the installation id first: a brand-new install is created and
@@ -647,8 +686,10 @@ def _maybe_alias_legacy_distinct_id(telemetry_enabled: bool | None) -> None:
 
         # distinct_id is the UUID form (set by get_event_defaults); send the
         # legacy integer as ``alias`` so PostHog coerces it to the same float as
-        # the historic events and merges the two persons.
-        send("$create_alias", telemetry_enabled, properties={"alias": installation_id})
+        # the historic events and merges the two persons. Delivered on the
+        # calling thread: this runs from the worker, or from a caller that must
+        # not start one.
+        _send("$create_alias", telemetry_enabled, properties={"alias": installation_id})
 
         # Record the new semantics regardless of outcome; we must not retry every
         # run even if the lossy legacy id failed to match.
@@ -668,7 +709,9 @@ def send(
     pool, so neither the data gathering (blocking syscalls and subprocess calls)
     nor the synchronous HTTP request blocks the caller — in particular the
     asyncio event loop serving a running app. Delivery is best-effort: any
-    failure is suppressed and never surfaces to the caller.
+    failure is suppressed and never surfaces to the caller. A caller that must
+    not start a thread (a process about to fork) uses ``_process_event``
+    directly instead.
 
     Args:
         event: The event name.
diff --git a/tests/units/test_reflex.py b/tests/units/test_reflex.py
index 7865009e5ff..b51580a7717 100644
--- a/tests/units/test_reflex.py
+++ b/tests/units/test_reflex.py
@@ -173,6 +173,45 @@ def test_compile_app_worker_flushes_telemetry_on_failure(mocker):
     flush.assert_called_once_with()
 
 
+@pytest.mark.parametrize(
+    ("running_mode", "launcher"),
+    [
+        (constants.RunningMode.BACKEND_ONLY, "run_backend_prod"),
+        (constants.RunningMode.FRONTEND_ONLY, "run_frontend_prod"),
+    ],
+)
+def test_run_prod_sends_telemetry_synchronously(
+    mocker, running_mode: constants.RunningMode, launcher: str
+):
+    """The run-prod event is delivered on the calling thread, before the server.
+
+    The server forks its workers from this process, so the event must not be
+    queued on the telemetry thread pool. Going through ``_process_event`` keeps
+    the one-time alias migration, which a backend-only run has no other event
+    to ride on.
+    """
+    from reflex.utils import build, exec, processes, telemetry
+
+    config = mocker.patch.object(reflex, "get_config").return_value
+    mocker.patch.object(reflex, "_compile_app")
+    mocker.patch.object(reflex, "_skip_compile")
+    mocker.patch.object(build, "setup_frontend_prod")
+    mocker.patch.object(processes, "atexit_handler")
+    mocker.patch("atexit.register")
+    mocker.patch.object(exec, "notify_app_running")
+    mocker.patch.object(exec, "notify_frontend")
+    queued = mocker.patch.object(telemetry, "send")
+    calls = mocker.Mock()
+    calls.attach_mock(mocker.patch.object(telemetry, "_process_event"), "send")
+    calls.attach_mock(mocker.patch.object(exec, launcher), "serve")
+
+    reflex._run_prod(running_mode, 8000, "127.0.0.1")
+
+    queued.assert_not_called()
+    calls.send.assert_called_once_with("run-prod", config.telemetry_enabled)
+    assert [name for name, _args, _kwargs in calls.mock_calls] == ["send", "serve"]
+
+
 def test_cloud_commands_registered():
     """The hosting CLI commands import, resolve, and dispatch only on demand."""
     probe = """
diff --git a/tests/units/test_telemetry.py b/tests/units/test_telemetry.py
index 4fc3a125514..db2cc600e3a 100644
--- a/tests/units/test_telemetry.py
+++ b/tests/units/test_telemetry.py
@@ -1,9 +1,12 @@
 import asyncio
 import importlib.metadata
 import json
+import os
 import sys
 import threading
+import time
 import uuid
+import warnings
 from types import SimpleNamespace
 
 import pytest
@@ -549,13 +552,16 @@ def test_maybe_alias_sends_create_alias_for_legacy_install(mocker: MockerFixture
     mocker.patch.object(telemetry, "ensure_reflex_installation_id", return_value=12345)
     mocker.patch.object(telemetry, "has_uuid_distinct_id_semantics", return_value=False)
     mark = mocker.patch.object(telemetry, "mark_uuid_distinct_id_semantics")
-    send_mock = mocker.patch.object(telemetry, "send")
+    send_mock = mocker.patch.object(telemetry, "_send")
+    queued = mocker.patch.object(telemetry, "send")
 
     telemetry._maybe_alias_legacy_distinct_id(telemetry_enabled=True)
 
     send_mock.assert_called_once_with(
         "$create_alias", True, properties={"alias": 12345}
     )
+    # The alias goes out on the calling thread, never through the pool.
+    queued.assert_not_called()
     mark.assert_called_once()
 
 
@@ -564,7 +570,7 @@ def test_maybe_alias_skips_for_uuid_native_install(mocker: MockerFixture):
     mocker.patch.object(telemetry, "ensure_reflex_installation_id", return_value=12345)
     mocker.patch.object(telemetry, "has_uuid_distinct_id_semantics", return_value=True)
     mark = mocker.patch.object(telemetry, "mark_uuid_distinct_id_semantics")
-    send_mock = mocker.patch.object(telemetry, "send")
+    send_mock = mocker.patch.object(telemetry, "_send")
 
     telemetry._maybe_alias_legacy_distinct_id(telemetry_enabled=True)
 
@@ -577,7 +583,7 @@ def test_maybe_alias_skips_without_installation_id(mocker: MockerFixture):
     mocker.patch.object(telemetry, "ensure_reflex_installation_id", return_value=None)
     has = mocker.patch.object(telemetry, "has_uuid_distinct_id_semantics")
     mark = mocker.patch.object(telemetry, "mark_uuid_distinct_id_semantics")
-    send_mock = mocker.patch.object(telemetry, "send")
+    send_mock = mocker.patch.object(telemetry, "_send")
 
     telemetry._maybe_alias_legacy_distinct_id(telemetry_enabled=True)
 
@@ -589,7 +595,7 @@ def test_maybe_alias_skips_without_installation_id(mocker: MockerFixture):
 def test_maybe_alias_skips_when_telemetry_disabled(mocker: MockerFixture):
     """Disabled telemetry does no work and leaves the marker unwritten."""
     ensure = mocker.patch.object(telemetry, "ensure_reflex_installation_id")
-    send_mock = mocker.patch.object(telemetry, "send")
+    send_mock = mocker.patch.object(telemetry, "_send")
 
     telemetry._maybe_alias_legacy_distinct_id(telemetry_enabled=False)
 
@@ -602,7 +608,7 @@ def test_maybe_alias_runs_at_most_once_per_process(mocker: MockerFixture):
     mocker.patch.object(telemetry, "ensure_reflex_installation_id", return_value=7)
     mocker.patch.object(telemetry, "has_uuid_distinct_id_semantics", return_value=False)
     mocker.patch.object(telemetry, "mark_uuid_distinct_id_semantics")
-    send_mock = mocker.patch.object(telemetry, "send")
+    send_mock = mocker.patch.object(telemetry, "_send")
 
     telemetry._maybe_alias_legacy_distinct_id(telemetry_enabled=True)
     telemetry._maybe_alias_legacy_distinct_id(telemetry_enabled=True)
@@ -622,8 +628,6 @@ def test_maybe_alias_create_alias_payload(
     )
 
     telemetry._maybe_alias_legacy_distinct_id(telemetry_enabled=True)
-    # The $create_alias is now sent on the telemetry worker thread; wait for it.
-    telemetry._flush()
 
     urlopen.assert_called_once()
     payload = posted_json(urlopen.call_args)
@@ -817,6 +821,128 @@ def test_flush_returns_false_when_worker_does_not_drain_in_time():
         blocker.result(timeout=5)
 
 
+def test_executor_is_recreated_after_fork():
+    """A forked child drops the inherited pool, whose thread it does not own."""
+    inherited = telemetry._get_telemetry_executor()
+
+    telemetry._reset_executor_after_fork()
+
+    fresh = telemetry._get_telemetry_executor()
+    assert fresh is not inherited
+    assert fresh.submit(lambda: 1).result(timeout=5) == 1
+    # In this process the orphaned pool's thread is real; stop it.
+    inherited.shutdown()
+
+
+def test_executor_lock_is_recreated_after_fork():
+    """A lock held by another thread at fork time does not block the child."""
+    orphaned = telemetry._get_telemetry_executor()
+    held = telemetry._executor_lock
+    held.acquire()
+    try:
+        telemetry._reset_executor_after_fork()
+    finally:
+        held.release()
+        orphaned.shutdown()
+
+    assert telemetry._executor_lock is not held
+    assert not telemetry._executor_lock.locked()
+
+
+@pytest.mark.skipif(not hasattr(os, "fork"), reason="os.fork is POSIX-only")
+def test_forked_child_starts_with_fresh_executor_state():
+    """The at-fork hook is registered and runs in a real forked child."""
+    telemetry._get_telemetry_executor()
+    with warnings.catch_warnings():
+        # The telemetry worker thread is alive, which is the point of the test.
+        warnings.simplefilter("ignore", DeprecationWarning)
+        with telemetry._executor_lock:
+            pid = os.fork()
+            if pid == 0:
+                clean = (
+                    telemetry._executor is None
+                    and not telemetry._executor_lock.locked()
+                )
+                os._exit(0 if clean else 1)
+    _, status = os.waitpid(pid, 0)
+    assert os.waitstatus_to_exitcode(status) == 0
+
+
+def test_shutdown_executor_drains_queued_work_and_stops_the_thread():
+    """Queued events are delivered and no telemetry thread survives the call."""
+    done = threading.Event()
+    executor = telemetry._get_telemetry_executor()
+    executor.submit(done.set)
+    workers = list(executor._threads)
+
+    assert telemetry._shutdown_executor()
+
+    assert done.is_set()
+    assert telemetry._executor is None
+    assert workers
+    assert not any(worker.is_alive() for worker in workers)
+
+
+def test_shutdown_executor_keeps_the_pool_published_until_it_stops(
+    mocker: MockerFixture,
+):
+    """A send racing the shutdown cannot start a second pool before the fork."""
+    executor = telemetry._get_telemetry_executor()
+    seen = []
+    stop = executor.shutdown
+
+    def shutdown(*args, **kwargs):
+        # What a concurrent send() would submit to while the pool stops.
+        seen.append(telemetry._get_telemetry_executor())
+        stop(*args, **kwargs)
+
+    mocker.patch.object(executor, "shutdown", side_effect=shutdown)
+
+    telemetry._shutdown_executor()
+
+    assert seen == [executor]
+    assert telemetry._executor is None
+
+
+def test_shutdown_executor_gives_up_on_a_stalled_task():
+    """A send that hangs (e.g. on DNS) cannot block the fork past the timeout."""
+    running = threading.Event()
+    release = threading.Event()
+
+    def stall():
+        running.set()
+        release.wait(10)
+
+    executor = telemetry._get_telemetry_executor()
+    workers = list(executor._threads)
+    executor.submit(stall)
+    queued = executor.submit(lambda: None)
+    # Only a task the worker has picked up keeps it alive; a queued one would
+    # just be cancelled.
+    assert running.wait(5)
+    started = time.monotonic()
+    try:
+        # The stalled send's thread is still alive, so forking is unsafe.
+        assert not telemetry._shutdown_executor(timeout=0.05)
+        elapsed = time.monotonic() - started
+    finally:
+        release.set()
+        for worker in workers:
+            worker.join(5)
+
+    assert elapsed < 2
+    assert telemetry._executor is None
+    assert queued.cancelled()
+
+
+def test_shutdown_executor_without_executor_is_a_noop():
+    """Nothing to drain when no telemetry was ever sent."""
+    assert telemetry._shutdown_executor()
+    assert telemetry._shutdown_executor()
+
+    assert telemetry._executor is None
+
+
 def test_send_event_posts_json_without_httpx(mocker: MockerFixture):
     """Delivery goes through urllib, so backend workers never import httpx."""
     urlopen = mocker.patch("reflex.utils.telemetry.urllib.request.urlopen")
diff --git a/tests/units/utils/test_exec.py b/tests/units/utils/test_exec.py
index 57a687abf66..2bc85ee2098 100644
--- a/tests/units/utils/test_exec.py
+++ b/tests/units/utils/test_exec.py
@@ -1,6 +1,7 @@
 """Tests for development backend launchers in ``reflex.utils.exec``."""
 
 import builtins
+import gc
 import logging
 import multiprocessing
 import os
@@ -20,6 +21,7 @@
 from reflex_base.utils import serializers
 
 from reflex.utils import exec as exec_utils
+from reflex.utils import prerequisites, telemetry
 from reflex.utils.precompressed_staticfiles import PrecompressedStaticFiles
 
 DEV_BACKEND_RELOAD_ENV_NAME = environment.REFLEX_DEV_BACKEND_RELOAD_ACTIVE.name
@@ -397,6 +399,151 @@ def test_arbitrate_ssr_env_var_wins(monkeypatch: pytest.MonkeyPatch):
     assert exec_utils.arbitrate_ssr(True) is False
 
 
+def _fake_granian_prod(
+    mocker: MockerFixture, calls: list[str], telemetry_stopped: bool = True
+):
+    """Patch granian and the prod launcher's collaborators, recording call order."""
+    granian_server = pytest.importorskip("granian.server")
+
+    class FakeGranian:
+        def __init__(self, *_args, **_kwargs):
+            pass
+
+        def startup(self, *_args):
+            calls.append("workers")
+
+        def serve(self):
+            calls.append("serve")
+            self.startup(None, None)
+
+    mocker.patch.object(granian_server, "Server", FakeGranian)
+    mocker.patch.object(
+        exec_utils, "get_app_instance_from_file", return_value="app:app"
+    )
+    mocker.patch.object(exec_utils, "_get_backend_workers", return_value=1)
+    mocker.patch.object(
+        multiprocessing,
+        "set_start_method",
+        side_effect=lambda method, force=False: calls.append(f"start:{method}"),
+    )
+    mocker.patch.object(
+        prerequisites, "get_app", side_effect=lambda: calls.append("preload")
+    )
+    mocker.patch.object(
+        serializers,
+        "_prepare_serializers_for_fork",
+        side_effect=lambda: calls.append("serializers"),
+    )
+    mocker.patch.object(
+        telemetry,
+        "_shutdown_executor",
+        side_effect=lambda: calls.append("drain") or telemetry_stopped,
+    )
+    mocker.patch.object(gc, "freeze", side_effect=lambda: calls.append("freeze"))
+
+
+def test_run_granian_backend_prod_preloads_app_before_forking(
+    mocker: MockerFixture, monkeypatch: pytest.MonkeyPatch
+):
+    """With fork, the app is imported, telemetry drained and the heap frozen first."""
+    monkeypatch.setenv(environment.REFLEX_BACKEND_START_METHOD.name, "fork")
+    calls: list[str] = []
+    _fake_granian_prod(mocker, calls)
+
+    exec_utils.run_granian_backend_prod(
+        host="0.0.0.0", port=8000, loglevel=exec_utils.LogLevel.INFO
+    )
+
+    assert calls == [
+        "start:fork",
+        "preload",
+        "serializers",
+        "drain",
+        "freeze",
+        "serve",
+        "workers",
+    ]
+
+
+def test_run_granian_backend_prod_spawns_when_telemetry_is_stuck(
+    mocker: MockerFixture, monkeypatch: pytest.MonkeyPatch
+):
+    """A telemetry thread that outlives the drain makes forking unsafe.
+
+    The app is still imported under the requested start method, so anything it
+    creates at import time uses the same context as before the fallback.
+    """
+    monkeypatch.setenv(environment.REFLEX_BACKEND_START_METHOD.name, "fork")
+    calls: list[str] = []
+    _fake_granian_prod(mocker, calls, telemetry_stopped=False)
+
+    exec_utils.run_granian_backend_prod(
+        host="0.0.0.0", port=8000, loglevel=exec_utils.LogLevel.INFO
+    )
+
+    assert calls == [
+        "start:fork",
+        "preload",
+        "serializers",
+        "drain",
+        "start:spawn",
+        "serve",
+        "workers",
+    ]
+
+
+def test_run_granian_backend_prod_spawn_skips_preload(
+    mocker: MockerFixture, monkeypatch: pytest.MonkeyPatch
+):
+    """Spawned workers re-import the app, so the supervisor does not load it."""
+    monkeypatch.setenv(environment.REFLEX_BACKEND_START_METHOD.name, "spawn")
+    calls: list[str] = []
+    _fake_granian_prod(mocker, calls)
+
+    exec_utils.run_granian_backend_prod(
+        host="0.0.0.0", port=8000, loglevel=exec_utils.LogLevel.INFO
+    )
+
+    assert calls == ["start:spawn", "serve", "workers"]
+
+
+def test_run_granian_backend_prod_custom_target_only_freezes(
+    mocker: MockerFixture, monkeypatch: pytest.MonkeyPatch
+):
+    """A non-reflex target lives in an already-imported module."""
+    monkeypatch.setenv(environment.REFLEX_BACKEND_START_METHOD.name, "fork")
+    calls: list[str] = []
+    _fake_granian_prod(mocker, calls)
+
+    exec_utils.run_granian_backend_prod(
+        host="0.0.0.0",
+        port=8000,
+        loglevel=exec_utils.LogLevel.INFO,
+        app_target="reflex.utils.exec:_frontend_prod_app",
+    )
+
+    assert calls == ["start:fork", "serializers", "drain", "freeze", "serve", "workers"]
+
+
+@pytest.mark.parametrize(
+    ("default_method", "expected"),
+    [("fork", "fork"), ("forkserver", "fork"), ("spawn", None)],
+)
+def test_backend_start_method_follows_platform_default(
+    mocker: MockerFixture,
+    monkeypatch: pytest.MonkeyPatch,
+    default_method: str,
+    expected: str | None,
+):
+    """Fork is forced only where the interpreter already defaults to forking."""
+    monkeypatch.delenv(environment.REFLEX_BACKEND_START_METHOD.name, raising=False)
+    mocker.patch.object(
+        multiprocessing, "get_start_method", return_value=default_method
+    )
+
+    assert exec_utils._backend_start_method() == expected
+
+
 def test_get_routes_manifest_router_missing_manifest(
     tmp_path: Path, monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
 ):
@@ -597,6 +744,7 @@ def test_run_granian_backend_prod_json_logs_in_json_mode(
         exec_utils, "get_app_instance_from_file", return_value="app:app"
     )
     mocker.patch.object(exec_utils, "_get_backend_workers", return_value=1)
+    mocker.patch.object(exec_utils, "_backend_start_method", return_value=None)
     granian_server = pytest.importorskip("granian.server")
     options: dict[str, object] = {}