diff --git a/pyproject.toml b/pyproject.toml
index 7c1ec4677c1..641fd0a4b98 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -28,6 +28,7 @@ dependencies = [
   "gunicorn >=20.1.0,<24.0",
   "httpx >=0.25.1,<1.0",
   "jinja2 >=3.1.2,<4.0",
+  "jsonpatch >=1.33,<2.0",
   "lazy_loader >=0.4",
   "packaging >=23.1,<25.0",
   "platformdirs >=3.10.0,<5.0",
diff --git a/reflex/.templates/web/utils/state.js b/reflex/.templates/web/utils/state.js
index 04f424c2454..f7e15e837fd 100644
--- a/reflex/.templates/web/utils/state.js
+++ b/reflex/.templates/web/utils/state.js
@@ -16,6 +16,7 @@ import {
 } from "$/utils/context.js";
 import debounce from "$/utils/helpers/debounce";
 import throttle from "$/utils/helpers/throttle";
+import { applyPatch } from "fast-json-patch/index.mjs";
 
 // Endpoint URLs.
 const EVENTURL = env.EVENT;
@@ -127,7 +128,7 @@ export const isStateful = () => {
   if (event_queue.length === 0) {
     return false;
   }
-  return event_queue.some((event) => event.name.startsWith("reflex___state"));
+  return event_queue.some((event) => event.name.startsWith(state_name));
 };
 
 /**
@@ -485,13 +486,48 @@ export const connect = async (
     window.removeEventListener("pagehide", pagehideHandler);
   });
 
+  const last_substate_info = {};
+  const last_substate_hash = {};
+
+  const getSubstateFromUpdate = (update, substate_name) => {
+    if (update.__patch) {
+      if (last_substate_hash[substate_name] !== update.__previous_hash) {
+        return null;
+      }
+      last_substate_hash[substate_name] = update.__hash;
+      return applyPatch(last_substate_info[substate_name], update.__patch)
+        .newDocument;
+    } else {
+      last_substate_hash[substate_name] = update.__hash;
+      return update.__full;
+    }
+  };
+
   // On each received message, queue the updates and events.
   socket.current.on("event", async (update) => {
+    const failed_substates = [];
     for (const substate in update.delta) {
-      dispatch[substate](update.delta[substate]);
+      const new_substate_info = getSubstateFromUpdate(
+        update.delta[substate],
+        substate,
+      );
+      if (new_substate_info === null) {
+        console.error("Received patch out of order", update.delta[substate]);
+        failed_substates.push(substate);
+        delete update.delta[substate];
+        continue;
+      }
+      last_substate_info[substate] = new_substate_info;
+      update.delta[substate] = new_substate_info;
+      dispatch[substate](new_substate_info);
     }
     applyClientStorageDelta(client_storage, update.delta);
     event_processing = !update.final;
+    if (failed_substates.length > 0) {
+      update.events.push(
+        Event(state_name + ".partial_hydrate", { states: failed_substates }),
+      );
+    }
     if (update.events) {
       queueEvents(update.events, socket);
     }
@@ -872,6 +908,7 @@ export const useEventLoop = (
     (async () => {
       // Process all outstanding events.
       while (event_queue.length > 0 && !event_processing) {
+        await new Promise((resolve) => setTimeout(resolve, 0));
         await processEvent(socket.current);
       }
     })();
@@ -911,7 +948,7 @@ export const useEventLoop = (
   // Route after the initial page hydration.
   useEffect(() => {
     const change_start = () => {
-      const main_state_dispatch = dispatch["reflex___state____state"];
+      const main_state_dispatch = dispatch[state_name];
       if (main_state_dispatch !== undefined) {
         main_state_dispatch({ is_hydrated: false });
       }
diff --git a/reflex/app.py b/reflex/app.py
index 6f2d478c605..b391aa5eafe 100644
--- a/reflex/app.py
+++ b/reflex/app.py
@@ -1407,7 +1407,7 @@ async def modify_state(self, token: str) -> AsyncIterator[BaseState]:
         async with self.state_manager.modify_state(token) as state:
             # No other event handler can modify the state while in this context.
             yield state
-            delta = state.get_delta()
+            delta = state.get_delta(token=token)
             if delta:
                 # When the state is modified reset dirty status and emit the delta to the frontend.
                 state._clean()
diff --git a/reflex/app_mixins/middleware.py b/reflex/app_mixins/middleware.py
index caecdb01103..8c30c241b58 100644
--- a/reflex/app_mixins/middleware.py
+++ b/reflex/app_mixins/middleware.py
@@ -7,6 +7,7 @@
 
 from reflex.event import Event
 from reflex.middleware import HydrateMiddleware, Middleware
+from reflex.middleware.hydrate_middleware import PartialHyderateMiddleware
 from reflex.state import BaseState, StateUpdate
 
 from .mixin import AppMixin
@@ -21,6 +22,7 @@ class MiddlewareMixin(AppMixin):
 
     def _init_mixin(self):
         self._middlewares.append(HydrateMiddleware())
+        self._middlewares.append(PartialHyderateMiddleware())
 
     def add_middleware(self, middleware: Middleware, index: int | None = None):
         """Add middleware to the app.
diff --git a/reflex/compiler/utils.py b/reflex/compiler/utils.py
index 5e9947a937e..a5d386270ad 100644
--- a/reflex/compiler/utils.py
+++ b/reflex/compiler/utils.py
@@ -27,7 +27,7 @@
 )
 from reflex.components.component import Component, ComponentStyle, CustomComponent
 from reflex.istate.storage import Cookie, LocalStorage, SessionStorage
-from reflex.state import BaseState, _resolve_delta
+from reflex.state import BaseState, StateDelta, _resolve_delta
 from reflex.style import Style
 from reflex.utils import console, format, imports, path_ops
 from reflex.utils.exec import is_in_app_harness
@@ -187,7 +187,7 @@ def compile_state(state: Type[BaseState]) -> dict:
     Returns:
         A dictionary of the compiled state.
     """
-    initial_state = state(_reflex_internal_init=True).dict(initial=True)
+    initial_state = StateDelta(state(_reflex_internal_init=True).dict(initial=True))
     try:
         _ = asyncio.get_running_loop()
     except RuntimeError:
@@ -202,10 +202,10 @@ def compile_state(state: Type[BaseState]) -> dict:
                 console.warn(
                     f"Had to get initial state in a thread 🤮 {resolved_initial_state}",
                 )
-                return resolved_initial_state
+                return dict(**resolved_initial_state.data)
 
     # Normally the compile runs before any event loop starts, we asyncio.run is available for calling.
-    return asyncio.run(_resolve_delta(initial_state))
+    return dict(**asyncio.run(_resolve_delta(initial_state)).data)
 
 
 def _compile_client_storage_field(
diff --git a/reflex/config.py b/reflex/config.py
index ab86e004a09..bb33ba0da61 100644
--- a/reflex/config.py
+++ b/reflex/config.py
@@ -721,6 +721,9 @@ class EnvironmentVariables:
     # Used by flexgen to enumerate the pages.
     REFLEX_ADD_ALL_ROUTES_ENDPOINT: EnvVar[bool] = env_var(False)
 
+    # Use the JSON patch format for websocket messages.
+    REFLEX_USE_JSON_PATCH: EnvVar[bool] = env_var(True)
+
 
 environment = EnvironmentVariables()
 
diff --git a/reflex/constants/compiler.py b/reflex/constants/compiler.py
index 40134c15bba..93e25589ad0 100644
--- a/reflex/constants/compiler.py
+++ b/reflex/constants/compiler.py
@@ -55,6 +55,8 @@ class CompileVars(SimpleNamespace):
     EVENTS = "events"
     # The name of the initial hydrate event.
     HYDRATE = "hydrate"
+    # The name of the partial hydrate event.
+    PARTIAL_HYDRATE = "partial_hydrate"
     # The name of the is_hydrated variable.
     IS_HYDRATED = "is_hydrated"
     # The name of the function to add events to the queue.
diff --git a/reflex/constants/installer.py b/reflex/constants/installer.py
index f5fa2c3c7ef..8b33856043e 100644
--- a/reflex/constants/installer.py
+++ b/reflex/constants/installer.py
@@ -91,6 +91,7 @@ class Commands(SimpleNamespace):
         "react-dom": "19.0.0",
         "react-focus-lock": "2.13.6",
         "socket.io-client": "4.8.1",
+        "fast-json-patch": "3.1.1",
         "universal-cookie": "7.2.2",
     }
     DEV_DEPENDENCIES = {
diff --git a/reflex/event.py b/reflex/event.py
index 9d9a55b332a..902f64174d5 100644
--- a/reflex/event.py
+++ b/reflex/event.py
@@ -1199,6 +1199,18 @@ def get_hydrate_event(state: BaseState) -> str:
     return get_event(state, constants.CompileVars.HYDRATE)
 
 
+def get_partial_hydrate_event(state: BaseState) -> str:
+    """Get the name of the partial hydrate event for the state.
+
+    Args:
+        state: The state.
+
+    Returns:
+        The name of the partial hydrate event.
+    """
+    return get_event(state, constants.CompileVars.PARTIAL_HYDRATE)
+
+
 def call_event_handler(
     event_callback: EventHandler | EventSpec,
     event_spec: ArgsSpec | Sequence[ArgsSpec],
diff --git a/reflex/middleware/__init__.py b/reflex/middleware/__init__.py
index 8ba85b41a90..20c3e59747e 100644
--- a/reflex/middleware/__init__.py
+++ b/reflex/middleware/__init__.py
@@ -1,4 +1,5 @@
 """Reflex middleware."""
 
-from .hydrate_middleware import HydrateMiddleware
-from .middleware import Middleware
+from .hydrate_middleware import HydrateMiddleware as HydrateMiddleware
+from .hydrate_middleware import PartialHyderateMiddleware as PartialHydrateMiddleware
+from .middleware import Middleware as Middleware
diff --git a/reflex/middleware/hydrate_middleware.py b/reflex/middleware/hydrate_middleware.py
index ec18939dee3..656afdb7281 100644
--- a/reflex/middleware/hydrate_middleware.py
+++ b/reflex/middleware/hydrate_middleware.py
@@ -3,12 +3,12 @@
 from __future__ import annotations
 
 import dataclasses
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, ChainMap
 
 from reflex import constants
-from reflex.event import Event, get_hydrate_event
+from reflex.event import Event, get_hydrate_event, get_partial_hydrate_event
 from reflex.middleware.middleware import Middleware
-from reflex.state import BaseState, StateUpdate, _resolve_delta
+from reflex.state import BaseState, StateDelta, StateUpdate, _resolve_delta
 
 if TYPE_CHECKING:
     from reflex.app import App
@@ -42,9 +42,62 @@ async def preprocess(
         setattr(state, constants.CompileVars.IS_HYDRATED, False)
 
         # Get the initial state.
-        delta = await _resolve_delta(state.dict())
+        delta = await _resolve_delta(
+            StateDelta(
+                state.dict(),
+                client_token=state.router.session.client_token,
+                flush=True,
+            )
+        )
         # since a full dict was captured, clean any dirtiness
         state._clean()
 
         # Return the state update.
         return StateUpdate(delta=delta, events=[])
+
+
+@dataclasses.dataclass(init=True)
+class PartialHyderateMiddleware(Middleware):
+    """Middleware to handle partial app hydration."""
+
+    async def preprocess(
+        self, app: App, state: BaseState, event: Event
+    ) -> StateUpdate | None:
+        """Preprocess the event.
+
+        Args:
+            app: The app to apply the middleware to."
+            state: The client state.""
+            event: The event to preprocess.""
+
+        Returns:
+            An optional delta or list of state updates to return.""
+        """
+        # If this is not the partial hydrate event, return None
+        if event.name != get_partial_hydrate_event(state):
+            return None
+
+        substates_names = event.payload.get("states", [])
+        if not substates_names:
+            return None
+
+        substates = [
+            substate
+            for substate_name in substates_names
+            if (substate := state.get_substate(substate_name.split("."))) is not None
+        ]
+
+        delta = await _resolve_delta(
+            StateDelta(
+                ChainMap(*[substate.dict() for substate in substates]),
+                client_token=state.router.session.client_token,
+                flush=True,
+            )
+        )
+
+        # since a full dict was captured, clean any dirtiness
+        for substate in substates:
+            substate._clean()
+
+        # Return the state update.
+        return StateUpdate(delta=delta, events=[])
diff --git a/reflex/state.py b/reflex/state.py
index 9c6a8b3ad3c..b6a521e747f 100644
--- a/reflex/state.py
+++ b/reflex/state.py
@@ -27,6 +27,8 @@
     Callable,
     ClassVar,
     Dict,
+    Mapping,
+    NamedTuple,
     Optional,
     Sequence,
     Set,
@@ -41,6 +43,7 @@
 
 import pydantic.v1 as pydantic
 import wrapt
+from jsonpatch import make_patch
 from pydantic import BaseModel as BaseModelV2
 from pydantic.v1 import BaseModel as BaseModelV1
 from pydantic.v1 import validator
@@ -109,10 +112,137 @@
     from reflex.components.component import Component
 
 
-Delta = dict[str, Any]
 var = computed_var
 
 
+@dataclasses.dataclass
+class StateDelta:
+    """A dictionary representing the state delta."""
+
+    data: Mapping[str, Any] = dataclasses.field(default_factory=dict)
+    client_token: str | None = dataclasses.field(default=None)
+    flush: bool = dataclasses.field(default=False)
+
+    def __getitem__(self, key: str) -> Any:
+        """Get the item from the delta.
+
+        Args:
+            key: The key to get.
+
+        Returns:
+            The item from the delta.
+        """
+        return self.data[key]
+
+    def __iter__(self) -> Any:
+        """Iterate over the delta.
+
+        Returns:
+            The iterator over the delta.
+        """
+        return iter(self.data)
+
+    def __len__(self) -> int:
+        """Get the length of the delta.
+
+        Returns:
+            The length of the delta.
+        """
+        return len(self.data)
+
+    def __contains__(self, key: str) -> bool:
+        """Check if the delta contains the key.
+
+        Args:
+            key: The key to check.
+
+        Returns:
+            Whether the delta contains the key.
+        """
+        return key in self.data
+
+    def keys(self):
+        """Get the keys of the delta.
+
+        Returns:
+            The keys of the delta.
+        """
+        return self.data.keys()
+
+    def __reversed__(self):
+        """Reverse the delta.
+
+        Returns:
+            The reversed delta.
+        """
+        return reversed(dict(**self.data))
+
+    def values(self):
+        """Get the values of the delta.
+
+        Returns:
+            The values of the delta.
+        """
+        return self.data.values()
+
+    def items(self):
+        """Get the items of the delta.
+
+        Returns:
+            The items of the delta.
+        """
+        return self.data.items()
+
+
+class DeltaCache(NamedTuple):
+    """A named tuple representing the delta cache."""
+
+    hash: int
+    delta: dict[str, Any]
+
+
+LAST_DELTA_CACHE: dict[str, DeltaCache] = {}
+
+
+@serializer(to=dict)
+def serialize_state_delta(delta: StateDelta) -> dict[str, Any]:
+    """Serialize the state delta.
+
+    Args:
+        delta: The state delta to serialize.
+
+    Returns:
+        The serialized state delta.
+    """
+    if delta.client_token is not None and environment.REFLEX_USE_JSON_PATCH.get():
+        full_delta = {}
+        for state_name, new_state_value in delta.items():
+            json_str = format.json_dumps(new_state_value)
+            new_state_value = json.loads(json_str)
+            key = delta.client_token + state_name
+            cached = LAST_DELTA_CACHE.get(key)
+            hash_value = hash(json_str)
+            LAST_DELTA_CACHE[key] = DeltaCache(hash_value, new_state_value)
+            if cached is not None and not delta.flush:
+                patch = make_patch(cached.delta, new_state_value).patch
+                if not patch:
+                    continue
+                full_delta[state_name] = {
+                    "__patch": patch,
+                    "__previous_hash": cached.hash,
+                    "__hash": hash_value,
+                }
+            else:
+                full_delta[state_name] = {
+                    "__full": new_state_value,
+                    "__hash": hash_value,
+                }
+        return full_delta
+    return {
+        state_name: {"__full": state_value} for state_name, state_value in delta.items()
+    }
+
+
 if environment.REFLEX_PERF_MODE.get() != PerformanceMode.OFF:
     # If the state is this large, it's considered a performance issue.
     TOO_LARGE_SERIALIZED_STATE = environment.REFLEX_STATE_SIZE_LIMIT.get() * 1024
@@ -311,7 +441,7 @@ def get_var_for_field(cls: Type[BaseState], f: ModelField):
     )
 
 
-async def _resolve_delta(delta: Delta) -> Delta:
+async def _resolve_delta(delta: StateDelta) -> StateDelta:
     """Await all coroutines in the delta.
 
     Args:
@@ -1701,7 +1831,7 @@ async def _as_state_update(
 
         try:
             # Get the delta after processing the event.
-            delta = await _resolve_delta(state.get_delta())
+            delta = await _resolve_delta(state.get_delta(token=token))
             state._clean()
 
             return StateUpdate(
@@ -1910,9 +2040,12 @@ def _dirty_computed_vars(
             if include_backend or not self.computed_vars[cvar]._backend
         }
 
-    def get_delta(self) -> Delta:
+    def get_delta(self, *, token: str | None = None) -> StateDelta:
         """Get the delta for the state.
 
+        Args:
+            token: The client token.
+
         Returns:
             The delta for the state.
         """
@@ -1923,11 +2056,9 @@ def get_delta(self) -> Delta:
             name for name, cv in self.computed_vars.items() if not cv._backend
         }
 
-        # Return the dirty vars for this instance, any cached/dependent computed vars,
-        # and always dirty computed vars (cache=False)
-        delta_vars = self.dirty_vars.intersection(self.base_vars).union(
-            self.dirty_vars.intersection(frontend_computed_vars)
-        )
+        delta_vars = frontend_computed_vars.union(self.base_vars)
+        if not environment.REFLEX_USE_JSON_PATCH.get():
+            delta_vars = self.dirty_vars.intersection(delta_vars)
 
         subdelta: dict[str, Any] = {
             prop: self.get_value(prop)
@@ -1944,7 +2075,7 @@ def get_delta(self) -> Delta:
             delta.update(substates[substate].get_delta())
 
         # Return the delta.
-        return delta
+        return StateDelta(delta, client_token=token)
 
     def _mark_dirty(self):
         """Mark the substate and all parent states as dirty."""
@@ -2775,7 +2906,7 @@ class StateUpdate:
     """A state update sent to the frontend."""
 
     # The state delta.
-    delta: Delta = dataclasses.field(default_factory=dict)
+    delta: StateDelta = dataclasses.field(default_factory=StateDelta)
 
     # Events to be added to the event queue.
     events: list[Event] = dataclasses.field(default_factory=list)
diff --git a/tests/integration/test_background_task.py b/tests/integration/test_background_task.py
index 91a1b5ae15e..31ed03dea73 100644
--- a/tests/integration/test_background_task.py
+++ b/tests/integration/test_background_task.py
@@ -176,7 +176,7 @@ def index() -> rx.Component:
             rx.button("Reset", on_click=State.reset_counter, id="reset"),
         )
 
-    app = rx.App(_state=rx.State)
+    app = rx.App()
     app.add_page(index)
 
 
diff --git a/tests/integration/test_call_script.py b/tests/integration/test_call_script.py
index 8236bf8e736..f3ee5dae307 100644
--- a/tests/integration/test_call_script.py
+++ b/tests/integration/test_call_script.py
@@ -187,7 +187,7 @@ def reset_(self):
             yield rx.call_script("inline_counter = 0; external_counter = 0")
             self.reset()
 
-    app = rx.App(_state=rx.State)
+    app = rx.App()
     Path("assets/external.js").write_text(external_scripts)
 
     @app.add_page
diff --git a/tests/integration/test_client_storage.py b/tests/integration/test_client_storage.py
index 3618c779dc1..fcfd09ddb0b 100644
--- a/tests/integration/test_client_storage.py
+++ b/tests/integration/test_client_storage.py
@@ -127,7 +127,7 @@ def index():
             rx.box(ClientSideSubSubState.s1s, id="s1s"),
         )
 
-    app = rx.App(_state=rx.State)
+    app = rx.App()
     app.add_page(index)
     app.add_page(index, route="/foo")
 
diff --git a/tests/integration/test_component_state.py b/tests/integration/test_component_state.py
index 654dc7ce9de..4f8decd9be4 100644
--- a/tests/integration/test_component_state.py
+++ b/tests/integration/test_component_state.py
@@ -72,7 +72,7 @@ def increment(self):
             State=_Counter,
         )
 
-    app = rx.App(_state=rx.State)  # noqa: F841
+    app = rx.App()  # noqa: F841
 
     @rx.page()
     def index():
diff --git a/tests/integration/test_connection_banner.py b/tests/integration/test_connection_banner.py
index f7fd7365cc1..0ce33ed5746 100644
--- a/tests/integration/test_connection_banner.py
+++ b/tests/integration/test_connection_banner.py
@@ -39,7 +39,7 @@ def index():
             rx.button("Delay", id="delay", on_click=State.delay),
         )
 
-    app = rx.App(_state=rx.State)
+    app = rx.App()
     app.add_page(index)
 
 
diff --git a/tests/integration/test_deploy_url.py b/tests/integration/test_deploy_url.py
index 207f3760984..936f97a185b 100644
--- a/tests/integration/test_deploy_url.py
+++ b/tests/integration/test_deploy_url.py
@@ -26,7 +26,7 @@ def index():
             rx.button("GOTO SELF", on_click=State.goto_self, id="goto_self")
         )
 
-    app = rx.App(_state=rx.State)
+    app = rx.App()
     app.add_page(index)
 
 
diff --git a/tests/integration/test_dynamic_routes.py b/tests/integration/test_dynamic_routes.py
index 9cdb970ca80..0be90b6e236 100644
--- a/tests/integration/test_dynamic_routes.py
+++ b/tests/integration/test_dynamic_routes.py
@@ -138,7 +138,7 @@ def arg() -> rx.Component:
     def redirect_page():
         return rx.fragment(rx.text("redirecting..."))
 
-    app = rx.App(_state=rx.State)
+    app = rx.App()
     app.add_page(index, route="/page/[page_id]", on_load=DynamicState.on_load)
     app.add_page(index, route="/static/x", on_load=DynamicState.on_load)
     app.add_page(index)
diff --git a/tests/integration/test_event_actions.py b/tests/integration/test_event_actions.py
index aff3785ad7b..24d7469d9c6 100644
--- a/tests/integration/test_event_actions.py
+++ b/tests/integration/test_event_actions.py
@@ -59,128 +59,133 @@ def _get_custom_code(self) -> str | None:
         def get_event_triggers(self):
             return {"on_click": lambda: []}
 
-    def index():
-        return rx.vstack(
-            rx.input(
-                value=EventActionState.router.session.client_token,
-                is_read_only=True,
-                id="token",
-            ),
-            rx.button("No events", id="btn-no-events"),
-            rx.button(
-                "Stop Prop Only",
-                id="btn-stop-prop-only",
-                on_click=rx.stop_propagation,  # pyright: ignore [reportArgumentType]
-            ),
-            rx.button(
-                "Click event",
-                on_click=EventActionState.on_click("no_event_actions"),  # pyright: ignore [reportCallIssue]
-                id="btn-click-event",
-            ),
-            rx.button(
-                "Click stop propagation",
-                on_click=EventActionState.on_click("stop_propagation").stop_propagation,  # pyright: ignore [reportCallIssue]
-                id="btn-click-stop-propagation",
-            ),
-            rx.button(
-                "Click stop propagation2",
-                on_click=EventActionState.on_click2.stop_propagation,
-                id="btn-click-stop-propagation2",
-            ),
-            rx.button(
-                "Click event 2",
-                on_click=EventActionState.on_click2,
-                id="btn-click-event2",
-            ),
-            rx.link(
-                "Link",
-                href="#",
-                on_click=EventActionState.on_click("link_no_event_actions"),  # pyright: ignore [reportCallIssue]
-                id="link",
-            ),
-            rx.link(
-                "Link Stop Propagation",
-                href="#",
-                on_click=EventActionState.on_click(  # pyright: ignore [reportCallIssue]
-                    "link_stop_propagation"
-                ).stop_propagation,
-                id="link-stop-propagation",
-            ),
-            rx.link(
-                "Link Prevent Default Only",
-                href="/invalid",
-                on_click=rx.prevent_default,  # pyright: ignore [reportArgumentType]
-                id="link-prevent-default-only",
-            ),
-            rx.link(
-                "Link Prevent Default",
-                href="/invalid",
-                on_click=EventActionState.on_click(  # pyright: ignore [reportCallIssue]
-                    "link_prevent_default"
-                ).prevent_default,
-                id="link-prevent-default",
-            ),
-            rx.link(
-                "Link Both",
-                href="/invalid",
-                on_click=EventActionState.on_click(  # pyright: ignore [reportCallIssue]
-                    "link_both"
-                ).stop_propagation.prevent_default,
-                id="link-stop-propagation-prevent-default",
-            ),
-            EventFiringComponent.create(
-                id="custom-stop-propagation",
-                on_click=EventActionState.on_click(  # pyright: ignore [reportCallIssue]
-                    "custom-stop-propagation"
-                ).stop_propagation,
-            ),
-            EventFiringComponent.create(
-                id="custom-prevent-default",
-                on_click=EventActionState.on_click(  # pyright: ignore [reportCallIssue]
-                    "custom-prevent-default"
-                ).prevent_default,
-            ),
-            rx.button(
-                "Throttle",
-                id="btn-throttle",
-                on_click=lambda: EventActionState.on_click_throttle.throttle(  # pyright: ignore [reportFunctionMemberAccess]
-                    200
-                ).stop_propagation,
-            ),
-            rx.button(
-                "Debounce",
-                id="btn-debounce",
-                on_click=EventActionState.on_click_debounce.debounce(  # pyright: ignore [reportFunctionMemberAccess]
-                    200
-                ).stop_propagation,
-            ),
-            rx.list(  # pyright: ignore [reportAttributeAccessIssue]
-                rx.foreach(
-                    EventActionState.order,
-                    rx.list_item,
+    def index() -> rx.Component:
+        return rx.fragment(
+            rx.vstack(
+                rx.input(
+                    value=EventActionState.router.session.client_token,
+                    is_read_only=True,
+                    id="token",
                 ),
-            ),
-            on_click=EventActionState.on_click("outer"),  # pyright: ignore [reportCallIssue]
-        ), rx.form(
-            rx.dialog.root(
-                rx.dialog.trigger(
-                    rx.button("Open Dialog", type="button", id="btn-dialog"),
+                rx.button("No events", id="btn-no-events"),
+                rx.button(
+                    "Stop Prop Only",
+                    id="btn-stop-prop-only",
                     on_click=rx.stop_propagation,  # pyright: ignore [reportArgumentType]
                 ),
-                rx.dialog.content(
-                    rx.dialog.close(
-                        rx.form(
-                            rx.button("Submit", id="btn-submit"),
-                            on_submit=EventActionState.on_submit.stop_propagation,  # pyright: ignore [reportCallIssue]
+                rx.button(
+                    "Click event",
+                    on_click=EventActionState.on_click("no_event_actions"),  # pyright: ignore [reportCallIssue]
+                    id="btn-click-event",
+                ),
+                rx.button(
+                    "Click stop propagation",
+                    on_click=EventActionState.on_click(
+                        "stop_propagation"
+                    ).stop_propagation,  # pyright: ignore [reportCallIssue]
+                    id="btn-click-stop-propagation",
+                ),
+                rx.button(
+                    "Click stop propagation2",
+                    on_click=EventActionState.on_click2.stop_propagation,
+                    id="btn-click-stop-propagation2",
+                ),
+                rx.button(
+                    "Click event 2",
+                    on_click=EventActionState.on_click2,
+                    id="btn-click-event2",
+                ),
+                rx.link(
+                    "Link",
+                    href="#",
+                    on_click=EventActionState.on_click("link_no_event_actions"),  # pyright: ignore [reportCallIssue]
+                    id="link",
+                ),
+                rx.link(
+                    "Link Stop Propagation",
+                    href="#",
+                    on_click=EventActionState.on_click(  # pyright: ignore [reportCallIssue]
+                        "link_stop_propagation"
+                    ).stop_propagation,
+                    id="link-stop-propagation",
+                ),
+                rx.link(
+                    "Link Prevent Default Only",
+                    href="/invalid",
+                    on_click=rx.prevent_default,  # pyright: ignore [reportArgumentType]
+                    id="link-prevent-default-only",
+                ),
+                rx.link(
+                    "Link Prevent Default",
+                    href="/invalid",
+                    on_click=EventActionState.on_click(  # pyright: ignore [reportCallIssue]
+                        "link_prevent_default"
+                    ).prevent_default,
+                    id="link-prevent-default",
+                ),
+                rx.link(
+                    "Link Both",
+                    href="/invalid",
+                    on_click=EventActionState.on_click(  # pyright: ignore [reportCallIssue]
+                        "link_both"
+                    ).stop_propagation.prevent_default,
+                    id="link-stop-propagation-prevent-default",
+                ),
+                EventFiringComponent.create(
+                    id="custom-stop-propagation",
+                    on_click=EventActionState.on_click(  # pyright: ignore [reportCallIssue]
+                        "custom-stop-propagation"
+                    ).stop_propagation,
+                ),
+                EventFiringComponent.create(
+                    id="custom-prevent-default",
+                    on_click=EventActionState.on_click(  # pyright: ignore [reportCallIssue]
+                        "custom-prevent-default"
+                    ).prevent_default,
+                ),
+                rx.button(
+                    "Throttle",
+                    id="btn-throttle",
+                    on_click=lambda: EventActionState.on_click_throttle.throttle(  # pyright: ignore [reportFunctionMemberAccess]
+                        200
+                    ).stop_propagation,
+                ),
+                rx.button(
+                    "Debounce",
+                    id="btn-debounce",
+                    on_click=EventActionState.on_click_debounce.debounce(  # pyright: ignore [reportFunctionMemberAccess]
+                        200
+                    ).stop_propagation,
+                ),
+                rx.list(  # pyright: ignore [reportAttributeAccessIssue]
+                    rx.foreach(
+                        EventActionState.order,
+                        rx.list_item,
+                    ),
+                ),
+                on_click=EventActionState.on_click("outer"),  # pyright: ignore [reportCallIssue]
+            ),
+            rx.form(
+                rx.dialog.root(
+                    rx.dialog.trigger(
+                        rx.button("Open Dialog", type="button", id="btn-dialog"),
+                        on_click=rx.stop_propagation,  # pyright: ignore [reportArgumentType]
+                    ),
+                    rx.dialog.content(
+                        rx.dialog.close(
+                            rx.form(
+                                rx.button("Submit", id="btn-submit"),
+                                on_submit=EventActionState.on_submit.stop_propagation,  # pyright: ignore [reportCallIssue]
+                            ),
                         ),
                     ),
                 ),
+                on_submit=EventActionState.on_submit,  # pyright: ignore [reportCallIssue]
             ),
-            on_submit=EventActionState.on_submit,  # pyright: ignore [reportCallIssue]
         )
 
-    app = rx.App(_state=rx.State)
-    app.add_page(index)  # pyright: ignore [reportArgumentType]
+    app = rx.App()
+    app.add_page(index)
 
 
 @pytest.fixture(scope="module")
diff --git a/tests/integration/test_event_chain.py b/tests/integration/test_event_chain.py
index 755f64b20bd..dff099f538c 100644
--- a/tests/integration/test_event_chain.py
+++ b/tests/integration/test_event_chain.py
@@ -143,7 +143,7 @@ def click_yield_interim_value(self):
             time.sleep(0.5)
             self.interim_value = "final"
 
-    app = rx.App(_state=rx.State)
+    app = rx.App()
 
     token_input = rx.input(
         value=State.router.session.client_token, is_read_only=True, id="token"
diff --git a/tests/integration/test_exception_handlers.py b/tests/integration/test_exception_handlers.py
index 71858b8995b..1c15a6e72d8 100644
--- a/tests/integration/test_exception_handlers.py
+++ b/tests/integration/test_exception_handlers.py
@@ -39,7 +39,7 @@ def divide_by_number(self, number: int):
             """
             print(1 / number)
 
-    app = rx.App(_state=rx.State)
+    app = rx.App()
 
     @app.add_page
     def index():
diff --git a/tests/integration/test_extra_overlay_function.py b/tests/integration/test_extra_overlay_function.py
index 2e36057ca0b..bbaaa53f76c 100644
--- a/tests/integration/test_extra_overlay_function.py
+++ b/tests/integration/test_extra_overlay_function.py
@@ -25,7 +25,7 @@ def index():
             ),
         )
 
-    app = rx.App(_state=rx.State)
+    app = rx.App()
     app.add_page(index)
 
 
diff --git a/tests/integration/test_form_submit.py b/tests/integration/test_form_submit.py
index 69c55c05762..6f850607661 100644
--- a/tests/integration/test_form_submit.py
+++ b/tests/integration/test_form_submit.py
@@ -30,7 +30,7 @@ class FormState(rx.State):
         def form_submit(self, form_data: Dict):
             self.form_data = form_data
 
-    app = rx.App(_state=rx.State)
+    app = rx.App()
 
     @app.add_page
     def index():
@@ -90,7 +90,7 @@ class FormState(rx.State):
         def form_submit(self, form_data: Dict):
             self.form_data = form_data
 
-    app = rx.App(_state=rx.State)
+    app = rx.App()
 
     @app.add_page
     def index():
diff --git a/tests/integration/test_input.py b/tests/integration/test_input.py
index 5f2948feb7a..eccfbfe7afa 100644
--- a/tests/integration/test_input.py
+++ b/tests/integration/test_input.py
@@ -16,7 +16,7 @@ def FullyControlledInput():
     class State(rx.State):
         text: str = "initial"
 
-    app = rx.App(_state=rx.State)
+    app = rx.App()
 
     @app.add_page
     def index():
diff --git a/tests/integration/test_login_flow.py b/tests/integration/test_login_flow.py
index a1df015bf49..9bcb1aba7a9 100644
--- a/tests/integration/test_login_flow.py
+++ b/tests/integration/test_login_flow.py
@@ -45,7 +45,7 @@ def login():
             rx.button("Do it", on_click=State.login, id="doit"),
         )
 
-    app = rx.App(_state=rx.State)
+    app = rx.App()
     app.add_page(index)
     app.add_page(login)
 
diff --git a/tests/integration/test_server_side_event.py b/tests/integration/test_server_side_event.py
index 3050a4e363d..547a4ea6453 100644
--- a/tests/integration/test_server_side_event.py
+++ b/tests/integration/test_server_side_event.py
@@ -38,7 +38,7 @@ def set_value_return(self):
         def set_value_return_c(self):
             return rx.set_value("c", "")
 
-    app = rx.App(_state=rx.State)
+    app = rx.App()
 
     @app.add_page
     def index():
diff --git a/tests/integration/test_upload.py b/tests/integration/test_upload.py
index af359348c8b..a00015b64d7 100644
--- a/tests/integration/test_upload.py
+++ b/tests/integration/test_upload.py
@@ -166,7 +166,7 @@ def index():
             rx.text(UploadState.event_order.to_string(), id="event-order"),
         )
 
-    app = rx.App(_state=rx.State)
+    app = rx.App()
     app.add_page(index)
 
 
diff --git a/tests/integration/test_var_operations.py b/tests/integration/test_var_operations.py
index 14a4781d4ba..4bd6abd1e4a 100644
--- a/tests/integration/test_var_operations.py
+++ b/tests/integration/test_var_operations.py
@@ -51,7 +51,7 @@ class VarOperationState(rx.State):
         )
         obj: rx.Field[Object] = rx.field(Object())
 
-    app = rx.App(_state=rx.State)
+    app = rx.App()
 
     @rx.memo
     def memo_comp(list1: list[int], int_var1: int, id: str):
diff --git a/tests/integration/tests_playwright/test_datetime_operations.py b/tests/integration/tests_playwright/test_datetime_operations.py
index 2ac516d4acb..3e2c67d27cd 100644
--- a/tests/integration/tests_playwright/test_datetime_operations.py
+++ b/tests/integration/tests_playwright/test_datetime_operations.py
@@ -16,7 +16,7 @@ class DtOperationsState(rx.State):
         date2: datetime = datetime(2031, 1, 1)
         date3: datetime = datetime(2021, 1, 1)
 
-    app = rx.App(_state=DtOperationsState)
+    app = rx.App()
 
     @app.add_page
     def index():
diff --git a/tests/integration/tests_playwright/test_table.py b/tests/integration/tests_playwright/test_table.py
index 68c56a2b923..36604aa7397 100644
--- a/tests/integration/tests_playwright/test_table.py
+++ b/tests/integration/tests_playwright/test_table.py
@@ -20,7 +20,7 @@ def Table():
     """App using table component."""
     import reflex as rx
 
-    app = rx.App(_state=rx.State)
+    app = rx.App()
 
     @app.add_page
     def index():
diff --git a/tests/units/middleware/test_hydrate_middleware.py b/tests/units/middleware/test_hydrate_middleware.py
index 7b02f8515cf..fcd0bd6809c 100644
--- a/tests/units/middleware/test_hydrate_middleware.py
+++ b/tests/units/middleware/test_hydrate_middleware.py
@@ -46,5 +46,5 @@ async def test_preprocess_no_events(hydrate_middleware, event1, mocker):
         state=state,
     )
     assert isinstance(update, StateUpdate)
-    assert update.delta == state.dict()
+    assert update.delta.data == state.dict()
     assert not update.events
diff --git a/tests/units/test_app.py b/tests/units/test_app.py
index ee67dfa9d3b..7aec896e564 100644
--- a/tests/units/test_app.py
+++ b/tests/units/test_app.py
@@ -35,12 +35,14 @@
 from reflex.components.radix.themes.typography.text import Text
 from reflex.event import Event
 from reflex.middleware import HydrateMiddleware
+from reflex.middleware.hydrate_middleware import PartialHyderateMiddleware
 from reflex.model import Model
 from reflex.state import (
     BaseState,
     OnLoadInternalState,
     RouterData,
     State,
+    StateDelta,
     StateManagerDisk,
     StateManagerMemory,
     StateManagerRedis,
@@ -210,7 +212,7 @@ def test_default_app(app: App):
     Args:
         app: The app to test.
     """
-    assert app._middlewares == [HydrateMiddleware()]
+    assert app._middlewares == [HydrateMiddleware(), PartialHyderateMiddleware()]
     assert app.style == Style()
     assert app.admin_dash is None
 
@@ -479,7 +481,7 @@ async def test_dynamic_var_event(test_state: Type[ATestState], token: str):
             payload={"value": 50},
         )
     ):
-        assert result.delta == {test_state.get_name(): {"int_val": 50}}
+        assert result.delta.data == {test_state.get_name(): {"int_val": 50}}
 
 
 @pytest.mark.asyncio
@@ -593,7 +595,7 @@ async def test_list_mutation_detection__plain_list(
         ):
             # prefix keys in expected_delta with the state name
             expected_delta = {list_mutation_state.get_name(): expected_delta}
-            assert result.delta == expected_delta
+            assert result.delta.data == expected_delta
 
 
 @pytest.mark.asyncio
@@ -719,7 +721,7 @@ async def test_dict_mutation_detection__plain_list(
             # prefix keys in expected_delta with the state name
             expected_delta = {dict_mutation_state.get_name(): expected_delta}
 
-            assert result.delta == expected_delta
+            assert result.delta.data == expected_delta
 
 
 @pytest.mark.asyncio
@@ -730,7 +732,7 @@ async def test_dict_mutation_detection__plain_list(
             FileUploadState,
             {
                 FileUploadState.get_full_name(): {
-                    "img_list": ["image1.jpg", "image2.jpg"]
+                    "__full": {"img_list": ["image1.jpg", "image2.jpg"]}
                 }
             },
         ),
@@ -738,7 +740,7 @@ async def test_dict_mutation_detection__plain_list(
             ChildFileUploadState,
             {
                 ChildFileUploadState.get_full_name(): {
-                    "img_list": ["image1.jpg", "image2.jpg"]
+                    "__full": {"img_list": ["image1.jpg", "image2.jpg"]}
                 }
             },
         ),
@@ -746,7 +748,7 @@ async def test_dict_mutation_detection__plain_list(
             GrandChildFileUploadState,
             {
                 GrandChildFileUploadState.get_full_name(): {
-                    "img_list": ["image1.jpg", "image2.jpg"]
+                    "__full": {"img_list": ["image1.jpg", "image2.jpg"]}
                 }
             },
         ),
@@ -1050,14 +1052,17 @@ def _dynamic_state_event(name, val, **kwargs):
         update = await process_coro.__anext__()
         # route change (on_load_internal) triggers: [call on_load events, call set_is_hydrated(True)]
         assert update == StateUpdate(
-            delta={
-                state.get_name(): {
-                    arg_name: exp_val,
-                    f"comp_{arg_name}": exp_val,
-                    constants.CompileVars.IS_HYDRATED: False,
-                    "router": exp_router,
-                }
-            },
+            delta=StateDelta(
+                {
+                    state.get_name(): {
+                        arg_name: exp_val,
+                        f"comp_{arg_name}": exp_val,
+                        constants.CompileVars.IS_HYDRATED: False,
+                        "router": exp_router,
+                    }
+                },
+                client_token=token,
+            ),
             events=[
                 _dynamic_state_event(
                     name="on_load",
@@ -1093,11 +1098,14 @@ def _dynamic_state_event(name, val, **kwargs):
         )
         on_load_update = await process_coro.__anext__()
         assert on_load_update == StateUpdate(
-            delta={
-                state.get_name(): {
-                    "loaded": exp_index + 1,
+            delta=StateDelta(
+                {
+                    state.get_name(): {
+                        "loaded": exp_index + 1,
+                    },
                 },
-            },
+                client_token=token,
+            ),
             events=[],
         )
         # complete the processing
@@ -1114,11 +1122,14 @@ def _dynamic_state_event(name, val, **kwargs):
         )
         on_set_is_hydrated_update = await process_coro.__anext__()
         assert on_set_is_hydrated_update == StateUpdate(
-            delta={
-                state.get_name(): {
-                    "is_hydrated": True,
+            delta=StateDelta(
+                {
+                    state.get_name(): {
+                        "is_hydrated": True,
+                    },
                 },
-            },
+                client_token=token,
+            ),
             events=[],
         )
         # complete the processing
@@ -1135,11 +1146,14 @@ def _dynamic_state_event(name, val, **kwargs):
         )
         update = await process_coro.__anext__()
         assert update == StateUpdate(
-            delta={
-                state.get_name(): {
-                    "counter": exp_index + 1,
-                }
-            },
+            delta=StateDelta(
+                {
+                    state.get_name(): {
+                        "counter": exp_index + 1,
+                    }
+                },
+                client_token=token,
+            ),
             events=[],
         )
         # complete the processing
diff --git a/tests/units/test_state.py b/tests/units/test_state.py
index d9421620dbe..356a027cb1a 100644
--- a/tests/units/test_state.py
+++ b/tests/units/test_state.py
@@ -34,6 +34,7 @@
     OnLoadInternalState,
     RouterData,
     State,
+    StateDelta,
     StateManager,
     StateManagerDisk,
     StateManagerMemory,
@@ -784,7 +785,7 @@ async def test_process_event_simple(test_state):
         assert test_state.num1 == 69
 
         # The delta should contain the changes, including computed vars.
-        assert update.delta == {
+        assert update.delta.data == {
             TestState.get_full_name(): {"num1": 69, "sum": 72.14},
             GrandchildState3.get_full_name(): {"computed": ""},
         }
@@ -811,7 +812,7 @@ async def test_process_event_substate(test_state, child_state, grandchild_state)
     async for update in test_state._process(event):
         assert child_state.value == "HI"
         assert child_state.count == 24
-        assert update.delta == {
+        assert update.delta.data == {
             # TestState.get_full_name(): {"sum": 3.14, "upper": ""},
             ChildState.get_full_name(): {"value": "HI", "count": 24},
             GrandchildState3.get_full_name(): {"computed": ""},
@@ -827,7 +828,7 @@ async def test_process_event_substate(test_state, child_state, grandchild_state)
     )
     async for update in test_state._process(event):
         assert grandchild_state.value2 == "new"
-        assert update.delta == {
+        assert update.delta.data == {
             # TestState.get_full_name(): {"sum": 3.14, "upper": ""},
             GrandchildState.get_full_name(): {"value2": "new"},
             GrandchildState3.get_full_name(): {"computed": ""},
@@ -849,11 +850,11 @@ async def test_process_event_generator():
     async for update in gen:
         count += 1
         if count == 6:
-            assert update.delta == {}
+            assert update.delta.data == {}
             assert update.final
         else:
             assert gen_state.value == count
-            assert update.delta == {
+            assert update.delta.data == {
                 GenState.get_full_name(): {"value": count},
             }
             assert not update.final
@@ -1073,7 +1074,7 @@ def test_not_dirty_computed_var_from_var(
         interdependent_state: A state with varying Var dependencies.
     """
     interdependent_state.x = 5
-    assert interdependent_state.get_delta() == {
+    assert interdependent_state.get_delta().data == {
         interdependent_state.get_full_name(): {"x": 5},
     }
 
@@ -1088,7 +1089,7 @@ def test_dirty_computed_var_from_var(interdependent_state: InterdependentState)
         interdependent_state: A state with varying Var dependencies.
     """
     interdependent_state.v1 = 1
-    assert interdependent_state.get_delta() == {
+    assert interdependent_state.get_delta().data == {
         interdependent_state.get_full_name(): {"v1": 1, "v1x2": 2, "v1x2x2": 4},
     }
 
@@ -1104,7 +1105,7 @@ def test_dirty_computed_var_from_backend_var(
     # Accessing ._v3 returns the immutable var it represents instead of the actual computed var
     # assert InterdependentState._v3._backend is True
     interdependent_state._v2 = 2
-    assert interdependent_state.get_delta() == {
+    assert interdependent_state.get_delta().data == {
         interdependent_state.get_full_name(): {"v2x2": 4, "v3x2": 4},
     }
 
@@ -1275,23 +1276,23 @@ def comp_v(self) -> int:
 
     cs = ComputedState()
     assert cs.dirty_vars == set()
-    assert cs.get_delta() == {cs.get_name(): {"no_cache_v": 0, "dep_v": 0}}
+    assert cs.get_delta().data == {cs.get_name(): {"no_cache_v": 0, "dep_v": 0}}
     cs._clean()
     assert cs.dirty_vars == set()
-    assert cs.get_delta() == {cs.get_name(): {"no_cache_v": 0, "dep_v": 0}}
+    assert cs.get_delta().data == {cs.get_name(): {"no_cache_v": 0, "dep_v": 0}}
     cs._clean()
     assert cs.dirty_vars == set()
     cs.v = 1
     assert cs.dirty_vars == {"v", "comp_v", "dep_v", "no_cache_v"}
-    assert cs.get_delta() == {
+    assert cs.get_delta().data == {
         cs.get_name(): {"v": 1, "no_cache_v": 1, "dep_v": 1, "comp_v": 1}
     }
     cs._clean()
     assert cs.dirty_vars == set()
-    assert cs.get_delta() == {cs.get_name(): {"no_cache_v": 1, "dep_v": 1}}
+    assert cs.get_delta().data == {cs.get_name(): {"no_cache_v": 1, "dep_v": 1}}
     cs._clean()
     assert cs.dirty_vars == set()
-    assert cs.get_delta() == {cs.get_name(): {"no_cache_v": 1, "dep_v": 1}}
+    assert cs.get_delta().data == {cs.get_name(): {"no_cache_v": 1, "dep_v": 1}}
     cs._clean()
     assert cs.dirty_vars == set()
 
@@ -2017,14 +2018,17 @@ async def test_state_proxy(grandchild_state: GrandchildState, mock_app: rx.App):
     mcall = mock_app.event_namespace.emit.mock_calls[0]  # pyright: ignore [reportFunctionMemberAccess]
     assert mcall.args[0] == str(SocketEvent.EVENT)
     assert mcall.args[1] == StateUpdate(
-        delta={
-            grandchild_state.get_full_name(): {
-                "value2": "42",
-            },
-            GrandchildState3.get_full_name(): {
-                "computed": "",
+        delta=StateDelta(
+            {
+                grandchild_state.get_full_name(): {
+                    "value2": "42",
+                },
+                GrandchildState3.get_full_name(): {
+                    "computed": "",
+                },
             },
-        }
+            client_token="_" + grandchild_state.get_full_name(),
+        )
     )
     assert mcall.kwargs["to"] == grandchild_state.router.session.session_id
 
@@ -2179,18 +2183,21 @@ async def test_background_task_no_block(mock_app: rx.App, token: str):
     ):
         # other task returns delta
         assert update == StateUpdate(
-            delta={
-                BackgroundTaskState.get_full_name(): {
-                    "order": [
-                        "background_task:start",
-                        "other",
-                    ],
-                    "computed_order": [
-                        "background_task:start",
-                        "other",
-                    ],
-                }
-            }
+            delta=StateDelta(
+                {
+                    BackgroundTaskState.get_full_name(): {
+                        "order": [
+                            "background_task:start",
+                            "other",
+                        ],
+                        "computed_order": [
+                            "background_task:start",
+                            "other",
+                        ],
+                    }
+                },
+                client_token=token,
+            )
         )
 
     # Explicit wait for background tasks
@@ -2221,42 +2228,54 @@ async def test_background_task_no_block(mock_app: rx.App, token: str):
         is not None
     )
     assert first_ws_message == StateUpdate(
-        delta={
-            BackgroundTaskState.get_full_name(): {
-                "order": ["background_task:start"],
-                "computed_order": ["background_task:start"],
-            }
-        },
+        delta=StateDelta(
+            {
+                BackgroundTaskState.get_full_name(): {
+                    "order": ["background_task:start"],
+                    "computed_order": ["background_task:start"],
+                }
+            },
+            client_token=token + "_" + BackgroundTaskState.get_full_name(),
+        ),
         events=[],
         final=True,
     )
     for call in emit_mock.mock_calls[1:5]:  # pyright: ignore [reportFunctionMemberAccess]
         assert call.args[1] == StateUpdate(
-            delta={
-                BackgroundTaskState.get_full_name(): {
-                    "computed_order": ["background_task:start"],
-                }
-            },
+            delta=StateDelta(
+                {
+                    BackgroundTaskState.get_full_name(): {
+                        "computed_order": ["background_task:start"],
+                    }
+                },
+                client_token=token + "_" + BackgroundTaskState.get_full_name(),
+            ),
             events=[],
             final=True,
         )
     assert emit_mock.mock_calls[-2].args[1] == StateUpdate(  # pyright: ignore [reportFunctionMemberAccess]
-        delta={
-            BackgroundTaskState.get_full_name(): {
-                "order": exp_order,
-                "computed_order": exp_order,
-                "dict_list": {},
-            }
-        },
+        delta=StateDelta(
+            {
+                BackgroundTaskState.get_full_name(): {
+                    "order": exp_order,
+                    "computed_order": exp_order,
+                    "dict_list": {},
+                }
+            },
+            client_token=token + "_" + BackgroundTaskState.get_full_name(),
+        ),
         events=[],
         final=True,
     )
     assert emit_mock.mock_calls[-1].args[1] == StateUpdate(  # pyright: ignore [reportFunctionMemberAccess]
-        delta={
-            BackgroundTaskState.get_full_name(): {
-                "computed_order": exp_order,
+        delta=StateDelta(
+            {
+                BackgroundTaskState.get_full_name(): {
+                    "computed_order": exp_order,
+                },
             },
-        },
+            client_token=token,
+        ),
         events=[],
         final=True,
     )
@@ -2896,14 +2915,14 @@ async def test_preprocess(app_module_mock, token, test_state, expected, mocker):
         updates.append(update)
     assert len(updates) == 1
     assert updates[0].delta[State.get_name()].pop("router") is not None
-    assert updates[0].delta == exp_is_hydrated(state, False)
+    assert updates[0].delta.data == exp_is_hydrated(state, False)
 
     events = updates[0].events
     assert len(events) == 2
     async for update in state._process(events[0]):
-        assert update.delta == {test_state.get_full_name(): {"num": 1}}
+        assert update.delta.data == {test_state.get_full_name(): {"num": 1}}
     async for update in state._process(events[1]):
-        assert update.delta == exp_is_hydrated(state)
+        assert update.delta.data == exp_is_hydrated(state)
 
     if isinstance(app.state_manager, StateManagerRedis):
         await app.state_manager.close()
@@ -2944,16 +2963,16 @@ async def test_preprocess_multiple_load_events(app_module_mock, token, mocker):
         updates.append(update)
     assert len(updates) == 1
     assert updates[0].delta[State.get_name()].pop("router") is not None
-    assert updates[0].delta == exp_is_hydrated(state, False)
+    assert updates[0].delta.data == exp_is_hydrated(state, False)
 
     events = updates[0].events
     assert len(events) == 3
     async for update in state._process(events[0]):
-        assert update.delta == {OnLoadState.get_full_name(): {"num": 1}}
+        assert update.delta.data == {OnLoadState.get_full_name(): {"num": 1}}
     async for update in state._process(events[1]):
-        assert update.delta == {OnLoadState.get_full_name(): {"num": 2}}
+        assert update.delta.data == {OnLoadState.get_full_name(): {"num": 2}}
     async for update in state._process(events[2]):
-        assert update.delta == exp_is_hydrated(state)
+        assert update.delta.data == exp_is_hydrated(state)
 
     if isinstance(app.state_manager, StateManagerRedis):
         await app.state_manager.close()
@@ -3025,7 +3044,7 @@ async def test_get_state(mock_app: rx.App, token: str):
     )
     grandchild_state.value2 = "set_value"
 
-    assert test_state.get_delta() == {
+    assert test_state.get_delta().data == {
         GrandchildState.get_full_name(): {
             "value2": "set_value",
         },
@@ -3062,7 +3081,7 @@ async def test_get_state(mock_app: rx.App, token: str):
     child_state2 = new_test_state.get_substate((ChildState2.get_name(),))
     child_state2.value = "set_c2_value"
 
-    assert new_test_state.get_delta() == {
+    assert new_test_state.get_delta().data == {
         ChildState2.get_full_name(): {
             "value": "set_c2_value",
         },
@@ -3666,7 +3685,7 @@ class GetValueState(rx.State):
             "bar": "BAR",
         }
     }
-    assert state.get_delta() == {}
+    assert state.get_delta().data == {}
 
     state.bar = "foo"
 
@@ -3676,7 +3695,7 @@ class GetValueState(rx.State):
             "bar": "foo",
         }
     }
-    assert state.get_delta() == {
+    assert state.get_delta().data == {
         state.get_full_name(): {
             "bar": "foo",
         }
@@ -3799,7 +3818,7 @@ async def test_upcast_event_handler_arg(handler, payload):
     """
     state = UpcastState()
     async for update in state._process_event(handler, state, payload):
-        assert update.delta == {UpcastState.get_full_name(): {"passed": True}}
+        assert update.delta.data == {UpcastState.get_full_name(): {"passed": True}}
 
 
 @pytest.mark.asyncio
diff --git a/tests/units/test_state_tree.py b/tests/units/test_state_tree.py
index 70ef71cb864..b6069beca1b 100644
--- a/tests/units/test_state_tree.py
+++ b/tests/units/test_state_tree.py
@@ -374,7 +374,7 @@ async def test_get_state_tree(
     assert sorted(state.substates) == sorted(exp_root_substates)
 
     # Only computed vars should be returned
-    assert state.get_delta() == ALWAYS_COMPUTED_VARS
+    assert state.get_delta().data == ALWAYS_COMPUTED_VARS
 
     # All of TreeA, TreeD, and TreeE substates should be in the dict
     assert sorted(state.dict()) == sorted(exp_root_dict_keys)
diff --git a/uv.lock b/uv.lock
index b4ee97f8888..ef509699fdc 100644
--- a/uv.lock
+++ b/uv.lock
@@ -713,6 +713,27 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/62/a1/3d680cbfd5f4b8f15abc1d571870c5fc3e594bb582bc3b64ea099db13e56/jinja2-3.1.6-py3-none-any.whl", hash = "sha256:85ece4451f492d0c13c5dd7c13a64681a86afae63a5f347908daf103ce6d2f67", size = 134899 },
 ]
 
+[[package]]
+name = "jsonpatch"
+version = "1.33"
+source = { registry = "https://pypi.org/simple" }
+dependencies = [
+    { name = "jsonpointer" },
+]
+sdist = { url = "https://files.pythonhosted.org/packages/42/78/18813351fe5d63acad16aec57f94ec2b70a09e53ca98145589e185423873/jsonpatch-1.33.tar.gz", hash = "sha256:9fcd4009c41e6d12348b4a0ff2563ba56a2923a7dfee731d004e212e1ee5030c", size = 21699 }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/73/07/02e16ed01e04a374e644b575638ec7987ae846d25ad97bcc9945a3ee4b0e/jsonpatch-1.33-py2.py3-none-any.whl", hash = "sha256:0ae28c0cd062bbd8b8ecc26d7d164fbbea9652a1a3693f3b956c1eae5145dade", size = 12898 },
+]
+
+[[package]]
+name = "jsonpointer"
+version = "3.0.0"
+source = { registry = "https://pypi.org/simple" }
+sdist = { url = "https://files.pythonhosted.org/packages/6a/0a/eebeb1fa92507ea94016a2a790b93c2ae41a7e18778f85471dc54475ed25/jsonpointer-3.0.0.tar.gz", hash = "sha256:2b2d729f2091522d61c3b31f82e11870f60b68f43fbc705cb76bf4b832af59ef", size = 9114 }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/71/92/5e77f98553e9e75130c78900d000368476aed74276eb8ae8796f65f00918/jsonpointer-3.0.0-py2.py3-none-any.whl", hash = "sha256:13e088adc14fca8b6aa8177c044e12701e6ad4b28ff10e65f2267a90109c9942", size = 7595 },
+]
+
 [[package]]
 name = "keyring"
 version = "25.6.0"
@@ -1689,6 +1710,7 @@ dependencies = [
     { name = "gunicorn" },
     { name = "httpx" },
     { name = "jinja2" },
+    { name = "jsonpatch" },
     { name = "lazy-loader" },
     { name = "packaging" },
     { name = "platformdirs" },
@@ -1750,6 +1772,7 @@ requires-dist = [
     { name = "gunicorn", specifier = ">=20.1.0,<24.0" },
     { name = "httpx", specifier = ">=0.25.1,<1.0" },
     { name = "jinja2", specifier = ">=3.1.2,<4.0" },
+    { name = "jsonpatch", specifier = ">=1.33,<2.0" },
     { name = "lazy-loader", specifier = ">=0.4" },
     { name = "packaging", specifier = ">=23.1,<25.0" },
     { name = "platformdirs", specifier = ">=3.10.0,<5.0" },