diff --git a/news/6738.performance.md b/news/6738.performance.md
new file mode 100644
index 00000000000..2f719d22187
--- /dev/null
+++ b/news/6738.performance.md
@@ -0,0 +1 @@
+Skip deep element-wise type validation on state var hot paths: computed var return types are only checked on recompute (not on cache hits), and production mode no longer walks every element of assigned containers for the log-only type check.
diff --git a/news/6740.performance.md b/news/6740.performance.md
new file mode 100644
index 00000000000..3512eb02e06
--- /dev/null
+++ b/news/6740.performance.md
@@ -0,0 +1 @@
+Speed up MutableProxy: immutable elements retrieved through a proxy skip the dataclasses frame-walk check, and the proxy for each mutable state var is cached per instance instead of rebuilt on every attribute read.
diff --git a/news/6743.performance.md b/news/6743.performance.md
new file mode 100644
index 00000000000..9be8f6e8bbe
--- /dev/null
+++ b/news/6743.performance.md
@@ -0,0 +1 @@
+Dirty propagation now only walks newly-dirty vars per mutation and skips the computed var expiry scan for classes with no interval vars, roughly halving per-mutation overhead in states with computed vars.
diff --git a/packages/reflex-base/news/6738.performance.md b/packages/reflex-base/news/6738.performance.md
new file mode 100644
index 00000000000..2f719d22187
--- /dev/null
+++ b/packages/reflex-base/news/6738.performance.md
@@ -0,0 +1 @@
+Skip deep element-wise type validation on state var hot paths: computed var return types are only checked on recompute (not on cache hits), and production mode no longer walks every element of assigned containers for the log-only type check.
diff --git a/packages/reflex-base/news/6740.performance.md b/packages/reflex-base/news/6740.performance.md
new file mode 100644
index 00000000000..b4d796c4c7d
--- /dev/null
+++ b/packages/reflex-base/news/6740.performance.md
@@ -0,0 +1 @@
+Reserve the internal `_mutable_proxy_cache` state field name so user vars cannot collide with the per-instance proxy cache.
diff --git a/packages/reflex-base/news/6743.performance.md b/packages/reflex-base/news/6743.performance.md
new file mode 100644
index 00000000000..9be8f6e8bbe
--- /dev/null
+++ b/packages/reflex-base/news/6743.performance.md
@@ -0,0 +1 @@
+Dirty propagation now only walks newly-dirty vars per mutation and skips the computed var expiry scan for classes with no interval vars, roughly halving per-mutation overhead in states with computed vars.
diff --git a/packages/reflex-base/src/reflex_base/utils/types.py b/packages/reflex-base/src/reflex_base/utils/types.py
index 85c769523a0..1c0eb07a6df 100644
--- a/packages/reflex-base/src/reflex_base/utils/types.py
+++ b/packages/reflex-base/src/reflex_base/utils/types.py
@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import dataclasses
+import os
 import sys
 import types
 from collections.abc import Callable, Iterable, Mapping, Sequence
@@ -161,7 +162,15 @@ def __call__(
     dict: Dict,  # noqa: UP006
 }
 
-RESERVED_BACKEND_VAR_NAMES = {"_abc_impl", "_backend_vars", "_was_touched", "_mixin"}
+RESERVED_BACKEND_VAR_NAMES = {
+    "_abc_impl",
+    "_backend_vars",
+    "_was_touched",
+    "_mixin",
+    "_mutable_proxy_cache",
+    "_propagated_dirty_vars",
+    "_propagated_generation",
+}
 
 
 class Unset:
@@ -633,6 +642,37 @@ def does_obj_satisfy_typed_dict(
     return required_keys.issubset(frozenset(obj))
 
 
+@lru_cache
+def _validation_depth_for_mode(raw_mode: str) -> int:
+    """Get the validation depth for a raw REFLEX_ENV_MODE value.
+
+    Args:
+        raw_mode: The stripped environment variable value ("" if unset).
+
+    Returns:
+        The `nested` depth to pass to `_isinstance`.
+    """
+    return 0 if raw_mode == constants.Env.PROD.value else 1
+
+
+def _validation_depth() -> int:
+    """Get the container depth for hot-path state var type validation.
+
+    The result of these checks only gates a diagnostic log, so production
+    mode skips the per-element walk of large containers and only validates
+    the outer type. The environment is re-read on every call so in-process
+    mode changes take effect immediately.
+
+    Returns:
+        The `nested` depth to pass to `_isinstance`.
+    """
+    # Read the raw env var directly: interpreting it through
+    # environment.REFLEX_ENV_MODE.get() on this hot path would re-parse the
+    # enum on every state var assignment (and the import would be circular).
+    # Strip to match the canonical parser's whitespace tolerance.
+    return _validation_depth_for_mode(os.environ.get("REFLEX_ENV_MODE", "").strip())
+
+
 def _isinstance(
     obj: Any,
     cls: GenericType,
diff --git a/packages/reflex-base/src/reflex_base/vars/base.py b/packages/reflex-base/src/reflex_base/vars/base.py
index fa461cf5a0b..1853646261f 100644
--- a/packages/reflex-base/src/reflex_base/vars/base.py
+++ b/packages/reflex-base/src/reflex_base/vars/base.py
@@ -66,6 +66,7 @@
     GenericType,
     Self,
     _isinstance,
+    _validation_depth,
     get_origin,
     has_args,
     safe_issubclass,
@@ -2255,6 +2256,19 @@ def is_computed_var(obj: Any) -> TypeGuard[ComputedVar]:
     return isinstance(obj, FakeComputedVarBaseClass)
 
 
+# Incremented whenever a cached computed var is recomputed. State dirty
+# propagation uses this to know when an already-propagated dependency needs to
+# be re-propagated (a recompute re-materializes a cache that a later mutation
+# of its dependencies must invalidate again).
+_computed_var_recompute_generation: int = 0
+
+
+def _bump_computed_var_recompute_generation() -> None:
+    """Record that a cached computed var was recomputed."""
+    global _computed_var_recompute_generation
+    _computed_var_recompute_generation += 1
+
+
 @dataclasses.dataclass(
     eq=False,
     frozen=True,
@@ -2587,23 +2601,29 @@ def __get__(self, instance: BaseState | None, owner: type):
 
         if not self._cache:
             value = self.fget(instance)
-        else:
-            # handle caching
-            if not hasattr(instance, self._cache_attr) or self.needs_update(instance):
-                # Set cache attr on state instance.
-                setattr(instance, self._cache_attr, self.fget(instance))
-                # Ensure the computed var gets serialized to redis.
-                instance._was_touched = True
-                # Set the last updated timestamp on the state instance.
-                setattr(instance, self._last_updated_attr, datetime.datetime.now())
-            value = getattr(instance, self._cache_attr)
+            self._check_deprecated_return_type(instance, value)
+            return value
 
-        self._check_deprecated_return_type(instance, value)
+        # handle caching
+        if not hasattr(instance, self._cache_attr) or self.needs_update(instance):
+            # Set cache attr on state instance.
+            setattr(instance, self._cache_attr, self.fget(instance))
+            # Ensure the computed var gets serialized to redis.
+            instance._was_touched = True
+            # Set the last updated timestamp on the state instance.
+            setattr(instance, self._last_updated_attr, datetime.datetime.now())
+            _bump_computed_var_recompute_generation()
+            value = getattr(instance, self._cache_attr)
+            # Only validate the return type when the value was just computed.
+            self._check_deprecated_return_type(instance, value)
+            return value
 
-        return value
+        return getattr(instance, self._cache_attr)
 
     def _check_deprecated_return_type(self, instance: BaseState, value: Any) -> None:
-        if not _isinstance(value, self._var_type, nested=1, treat_var_as_type=False):
+        if not _isinstance(
+            value, self._var_type, nested=_validation_depth(), treat_var_as_type=False
+        ):
             console.error(
                 f"Computed var '{type(instance).__name__}.{self._name}' must return"
                 f" a value of type '{escape(str(self._var_type))}', got '{value!s}' of type {type(value)}."
@@ -2858,9 +2878,12 @@ async def _awaitable_result(instance: BaseState = instance) -> RETURN_TYPE:
                 instance._was_touched = True
                 # Set the last updated timestamp on the state instance.
                 setattr(instance, self._last_updated_attr, datetime.datetime.now())
-            value = getattr(instance, self._cache_attr)
-            self._check_deprecated_return_type(instance, value)
-            return value
+                _bump_computed_var_recompute_generation()
+                value = getattr(instance, self._cache_attr)
+                # Only validate the return type when the value was just computed.
+                self._check_deprecated_return_type(instance, value)
+                return value
+            return getattr(instance, self._cache_attr)
 
         return _awaitable_result()
 
diff --git a/reflex/istate/proxy.py b/reflex/istate/proxy.py
index ce2eef96834..769f697bbff 100644
--- a/reflex/istate/proxy.py
+++ b/reflex/istate/proxy.py
@@ -664,13 +664,14 @@ def _wrap_mutable(self, value: Any, path: tuple[_AccessSpec, ...]) -> Any:
         Returns:
             The wrapped value.
         """
-        # When called from dataclasses internal code, return the unwrapped value
-        if self._is_called_from_dataclasses_internal():
-            return value
         # If we already have a proxy, unwrap and rewrap to make sure the state
         # reference is up to date.
         if isinstance(value, MutableProxy):
             value = value.__wrapped__
+        # When called from dataclasses internal code, return the unwrapped value
+        if self._is_called_from_dataclasses_internal():
+            return value
+        # Recursively wrap mutable types.
         return globals()[self.__base_proxy__](
             wrapped=value,
             state=self._self_state,
diff --git a/reflex/state.py b/reflex/state.py
index bad54ff4e8e..bc60a06ca16 100644
--- a/reflex/state.py
+++ b/reflex/state.py
@@ -54,8 +54,9 @@
 )
 from reflex_base.utils.exceptions import ImmutableStateError as ImmutableStateError
 from reflex_base.utils.serializers import serializer
-from reflex_base.utils.types import _isinstance
+from reflex_base.utils.types import _isinstance, _validation_depth
 from reflex_base.vars import Field, VarData, field
+from reflex_base.vars import base as reflex_base_vars_base
 from reflex_base.vars.base import (
     ComputedVar,
     DynamicRouteVar,
@@ -369,6 +370,7 @@ def _is_user_descriptor(value: Any) -> bool:
     "_always_dirty_computed_vars",
     "_always_dirty_substates",
     "_potentially_dirty_states",
+    "_interval_computed_vars",
 })
 
 
@@ -408,6 +410,11 @@ class BaseState(EvenMoreBasicBaseState):
     # Set of states which might need to be recomputed if vars in this state change.
     _potentially_dirty_states: ClassVar[set[str]] = set()
 
+    # Names of computed vars with an update interval, refreshed whenever
+    # computed_vars changes. Empty for most classes, which lets dirty
+    # propagation skip the per-mutation expiry scan.
+    _interval_computed_vars: ClassVar[tuple[str, ...]] = ()
+
     # The parent state.
     parent_state: BaseState | None = field(default=None, is_var=False)
 
@@ -438,6 +445,18 @@ class BaseState(EvenMoreBasicBaseState):
     # Whether the state has ever been touched since instantiation.
     _was_touched: bool = field(default=False, is_var=False)
 
+    # Per-field cache of the MutableProxy wrapping each mutable var, so
+    # repeated reads don't rebuild the proxy. Never pickled.
+    _mutable_proxy_cache: builtins.dict[str, MutableProxy] = field(
+        default_factory=builtins.dict, is_var=False
+    )
+    # Dirty vars whose dependency closure was already propagated this cycle.
+    # Transient: cleared by _clean and never pickled.
+    _propagated_dirty_vars: set[str] = field(default_factory=set, is_var=False)
+
+    # Recompute generation the propagation frontier was built in.
+    _propagated_generation: int = field(default=0, is_var=False)
+
     # A special event handler for setting base vars.
     setvar: ClassVar[EventHandler]
 
@@ -717,9 +736,22 @@ def __init_subclass__(cls, mixin: bool = False, **kwargs):
         # Initialize per-class var dependency tracking.
         cls._var_dependencies = {}
         cls._init_var_dependency_dicts()
+        cls._refresh_interval_computed_vars()
 
         all_base_state_classes[cls.get_full_name()] = None
 
+    @classmethod
+    def _refresh_interval_computed_vars(cls) -> None:
+        """Recompute the names of computed vars that have an update interval.
+
+        Must be called whenever cls.computed_vars is modified.
+        """
+        cls._interval_computed_vars = tuple(
+            name
+            for name, cvar in cls.computed_vars.items()
+            if cvar._update_interval is not None
+        )
+
     @classmethod
     def _add_event_handler(
         cls,
@@ -824,6 +856,7 @@ def computed_var_func(state: Self):
 
         setattr(cls, unique_var_name, computed_var_func_arg)
         cls.computed_vars[unique_var_name] = computed_var_func_arg
+        cls._refresh_interval_computed_vars()
         cls.vars[unique_var_name] = computed_var_func_arg
         cls._update_substate_inherited_vars({unique_var_name: computed_var_func_arg})
         cls._always_dirty_computed_vars.add(unique_var_name)
@@ -1404,6 +1437,7 @@ def inner_func(self: BaseState) -> list[str]:
 
         # Update tracking dicts.
         cls.computed_vars.update(dynamic_vars)
+        cls._refresh_interval_computed_vars()
         cls.vars.update(dynamic_vars)
         cls._update_substate_inherited_vars(dynamic_vars)
 
@@ -1481,7 +1515,12 @@ def _get_attribute(self, name: str) -> Any:
             name in super().__getattribute__("base_vars") or name in backend_vars
         ):
             # track changes in mutable containers (list, dict, set, etc)
-            return MutableProxy(wrapped=value, state=self, field_name=name)
+            cache = super().__getattribute__("_mutable_proxy_cache")
+            proxy = cache.get(name)
+            if proxy is None or proxy.__wrapped__ is not value:
+                proxy = MutableProxy(wrapped=value, state=self, field_name=name)
+                cache[name] = proxy
+            return proxy
 
         return value
 
@@ -1516,6 +1555,8 @@ def __setattr__(self, name: str, value: Any):
 
         if name in self.backend_vars:
             self._backend_vars.__setitem__(name, value)
+            # Drop the proxy wrapping the replaced value.
+            self.__dict__["_mutable_proxy_cache"].pop(name, None)
             self.dirty_vars.add(name)
             self._mark_dirty()
             return
@@ -1538,7 +1579,9 @@ def __setattr__(self, name: str, value: Any):
 
         if (field := fields.get(name)) is not None and field.is_var:
             field_type = field.outer_type_
-            if not _isinstance(value, field_type, nested=1, treat_var_as_type=False):
+            if not _isinstance(
+                value, field_type, nested=_validation_depth(), treat_var_as_type=False
+            ):
                 console.error(
                     f"Expected field '{type(self).__name__}.{name}' to receive type '{escape(str(field_type))}',"
                     f" but got '{value}' of type '{type(value)}'."
@@ -1547,6 +1590,9 @@ def __setattr__(self, name: str, value: Any):
         # Set the attribute.
         object.__setattr__(self, name, value)
 
+        # Drop the proxy wrapping the replaced value.
+        self.__dict__["_mutable_proxy_cache"].pop(name, None)
+
         # Add the var to the dirty list.
         if name in self.base_vars:
             self.dirty_vars.add(name)
@@ -1801,14 +1847,31 @@ async def get_var_value(self, var: Var[VAR_TYPE]) -> VAR_TYPE:
 
     def _mark_dirty_computed_vars(self) -> None:
         """Mark ComputedVars that need to be recalculated based on dirty_vars."""
-        # Append expired computed vars to dirty_vars to trigger recalculation
-        self.dirty_vars.update(self._expired_computed_vars())
+        if self._interval_computed_vars:
+            # Append expired computed vars to dirty_vars to trigger recalculation
+            self.dirty_vars.update(self._expired_computed_vars())
         # Append always dirty computed vars to dirty_vars to trigger recalculation
         self.dirty_vars.update(self._always_dirty_computed_vars)
 
-        dirty_vars = self.dirty_vars
-        while dirty_vars:
-            calc_vars, dirty_vars = dirty_vars, set()
+        # Track which dirty vars already had their dependency closure
+        # propagated, so repeated mutations only process newly-dirty vars.
+        # Recomputing any cached var re-materializes a cache that a later
+        # mutation must invalidate again, so the frontier is only valid for
+        # the recompute generation it was built in.
+        # Go through __dict__ (not setattr) so this also works when self is a
+        # StateProxy: the proxy exposes the wrapped state's __dict__, while
+        # object.__setattr__ is rejected by wrapt's C ObjectProxy.
+        instance_dict = self.__dict__
+        propagated = instance_dict["_propagated_dirty_vars"]
+        current_gen = reflex_base_vars_base._computed_var_recompute_generation
+        if instance_dict["_propagated_generation"] != current_gen:
+            propagated.clear()
+            instance_dict["_propagated_generation"] = current_gen  # pyright: ignore[reportIndexIssue]
+
+        new_dirty = self.dirty_vars - propagated
+        while new_dirty:
+            propagated |= new_dirty
+            calc_vars, new_dirty = new_dirty, set()
             for state_name, cvar in self._dirty_computed_vars(from_vars=calc_vars):
                 if state_name == self.get_full_name():
                     defining_state = self
@@ -1821,7 +1884,8 @@ def _mark_dirty_computed_vars(self) -> None:
                 if actual_var is not None:
                     actual_var.mark_dirty(instance=defining_state)
                 if defining_state is self:
-                    dirty_vars.add(cvar)
+                    if cvar not in propagated:
+                        new_dirty.add(cvar)
                 else:
                     # mark dirty where this var is defined
                     defining_state._mark_dirty()
@@ -1832,10 +1896,11 @@ def _expired_computed_vars(self) -> set[str]:
         Returns:
             Set of computed vars to include in the delta.
         """
+        computed_vars = self.computed_vars
         return {
             cvar
-            for cvar, cvar_obj in self.computed_vars.items()
-            if cvar_obj.needs_update(instance=self)
+            for cvar in self._interval_computed_vars
+            if computed_vars[cvar].needs_update(instance=self)
         }
 
     def _dirty_computed_vars(
@@ -1955,6 +2020,8 @@ def _clean(self):
         # Clean this state.
         self.dirty_vars = set()
         self.dirty_substates = set()
+        # Discard the propagation frontier along with the dirty vars it tracked.
+        self.__dict__["_propagated_dirty_vars"].clear()
 
     def get_value(self, key: str) -> Any:
         """Get the value of a field (without proxying).
@@ -2072,6 +2139,11 @@ def __getstate__(self):
         state.pop("parent_state", None)
         state.pop("substates", None)
         state.pop("_was_touched", None)
+        # Proxies wrap live state references and are rebuilt on access.
+        state.pop("_mutable_proxy_cache", None)
+        # The propagation frontier is transient and rebuilt on demand.
+        state.pop("_propagated_dirty_vars", None)
+        state.pop("_propagated_generation", None)
         # Remove all inherited vars.
         for inherited_var_name in self.inherited_vars:
             state.pop(inherited_var_name, None)
@@ -2087,6 +2159,11 @@ def __setstate__(self, state: builtins.dict[str, Any]):
         """
         state["parent_state"] = None
         state["substates"] = {}
+        # The proxy cache is never pickled; recreate it on the restored instance.
+        state.setdefault("_mutable_proxy_cache", {})
+        # The propagation frontier is never pickled; recreate it on restore.
+        state.setdefault("_propagated_dirty_vars", set())
+        state.setdefault("_propagated_generation", 0)
         for key, value in state.items():
             object.__setattr__(self, key, value)
 
diff --git a/tests/units/istate/test_proxy.py b/tests/units/istate/test_proxy.py
index 43c750cc08e..df4d6a4a19b 100644
--- a/tests/units/istate/test_proxy.py
+++ b/tests/units/istate/test_proxy.py
@@ -713,3 +713,78 @@ async def test_mutable_proxy_custom_get_method_path_tracking(
     ) as state:
         assert isinstance(state, CustomGetState)
         assert state.registry.entries == {"a": [1, 2]}
+
+
+def test_mutable_proxy_cached_per_field():
+    """Repeated reads of a mutable var reuse the proxy until reassignment."""
+    state = ProxyTestState()
+    first = state.items
+    assert isinstance(first, MutableProxy)
+    assert state.items is first
+    # Reassignment invalidates the cached proxy.
+    state.items = [Item(2)]
+    second = state.items
+    assert isinstance(second, MutableProxy)
+    assert second is not first
+    assert second[0].id == 2
+    # In-place mutation keeps the same wrapped object, so the proxy is reused.
+    second.append(Item(3))
+    assert state.items is second
+    # Reassignment immediately evicts the cache entry, so no strong reference
+    # to the replaced value lingers until the next read.
+    state.items = [Item(4)]
+    assert "items" not in state.__dict__["_mutable_proxy_cache"]
+
+
+@pytest.mark.asyncio
+async def test_state_proxy_reassignment_evicts_cached_proxy(
+    attached_mock_event_context: EventContext,
+):
+    """Reassignment through a background-task StateProxy evicts the cached proxy.
+
+    StateProxy.__setattr__ delegates to setattr on the wrapped state, so the
+    eviction in BaseState.__setattr__ must also cover writes made inside an
+    `async with self` block.
+    """
+    state = ProxyTestState()
+    first = state.items
+    assert isinstance(first, MutableProxy)
+    assert state.__dict__["_mutable_proxy_cache"]["items"] is first
+
+    state_proxy = StateProxy(state)
+    # Simulate holding the lock inside `async with self`.
+    state_proxy._self_mutable = True
+    state_proxy.items = [Item(9)]
+    # The write reaches BaseState.__setattr__ on the wrapped state, evicting
+    # the proxy that wrapped the replaced list.
+    assert "items" not in state.__dict__["_mutable_proxy_cache"]
+    second = state.items
+    assert second is not first
+    assert second[0].id == 9
+
+
+def test_mutable_proxy_cache_not_serialized():
+    """The per-instance proxy cache never leaks into pickles or copies."""
+    state = ProxyTestState()
+    state.items.append(Item(1))  # populate the proxy cache
+    assert state.__dict__["_mutable_proxy_cache"]
+    assert "_mutable_proxy_cache" not in state.__getstate__()
+
+    restored = pickle.loads(pickle.dumps(state))
+    # The cache is recreated empty on the restored instance.
+    assert restored.__dict__["_mutable_proxy_cache"] == {}
+    restored_items = restored.items
+    assert isinstance(restored_items, MutableProxy)
+    # The restored proxy tracks the restored state, not the original.
+    assert restored_items._self_state is restored
+
+
+def test_mutable_proxy_iteration_yields_plain_immutables():
+    """Iterating a proxied container returns immutable elements unwrapped."""
+    state = ProxyTestState()
+    state.items = [Item(1), Item(2)]
+    numbers = [item.id for item in state.items]
+    assert numbers == [1, 2]
+    assert all(type(n) is int for n in numbers)
+    # Mutable elements remain wrapped so nested mutations mark the state dirty.
+    assert all(isinstance(item, MutableProxy) for item in state.items)
diff --git a/tests/units/reflex_base/utils/test_types.py b/tests/units/reflex_base/utils/test_types.py
index e4ab9dece45..b68eccf6850 100644
--- a/tests/units/reflex_base/utils/test_types.py
+++ b/tests/units/reflex_base/utils/test_types.py
@@ -1,6 +1,17 @@
 """Tests for reflex_base.utils.types."""
 
-from reflex_base.utils.types import ASGIApp, Message, Receive, Scope, Send
+import os
+
+from reflex_base import constants
+from reflex_base.environment import environment
+from reflex_base.utils.types import (
+    ASGIApp,
+    Message,
+    Receive,
+    Scope,
+    Send,
+    _validation_depth,
+)
 from typing_extensions import TypeAliasType
 
 
@@ -14,3 +25,23 @@ def test_asgi_aliases_keep_their_names():
     assert Receive.__name__ == "Receive"
     assert Send.__name__ == "Send"
     assert ASGIApp.__name__ == "ASGIApp"
+
+
+def test_validation_depth_by_env_mode():
+    """Hot-path validation walks containers in dev but stays shallow in prod.
+
+    In-process environment mode changes must take effect immediately, without
+    any cache invalidation by the caller.
+    """
+    initial = environment.REFLEX_ENV_MODE.getenv()
+    try:
+        environment.REFLEX_ENV_MODE.set(constants.Env.PROD)
+        assert _validation_depth() == 0
+        environment.REFLEX_ENV_MODE.set(constants.Env.DEV)
+        assert _validation_depth() == 1
+        # The canonical parser tolerates surrounding whitespace; the hot-path
+        # reader must agree with it.
+        os.environ["REFLEX_ENV_MODE"] = f" {constants.Env.PROD.value} "
+        assert _validation_depth() == 0
+    finally:
+        environment.REFLEX_ENV_MODE.set(initial)
diff --git a/tests/units/reflex_base/vars/test_base.py b/tests/units/reflex_base/vars/test_base.py
index b0ff880aa20..77f8a134e3d 100644
--- a/tests/units/reflex_base/vars/test_base.py
+++ b/tests/units/reflex_base/vars/test_base.py
@@ -6,6 +6,9 @@
 from reflex_base.utils.types import get_field_type
 from reflex_base.vars.base import EvenMoreBasicBaseState, field
 
+import reflex as rx
+from reflex.state import BaseState
+
 _MARKER_ATTR = "_marker"
 
 
@@ -87,3 +90,74 @@ class MyState(EvenMoreBasicBaseState):
 
     rebuilt = MyState.get_fields()["name"]
     assert rebuilt._check is check  # pyright: ignore[reportAttributeAccessIssue]
+
+
+def test_computed_var_return_type_checked_only_on_recompute(mocker):
+    """The return type of a cached computed var is only validated on recompute.
+
+    Args:
+        mocker: Pytest mocker object.
+    """
+
+    class ReturnTypeCheckState(BaseState):
+        v: int = 0
+
+        @rx.var
+        def wrong_typed(self) -> str:
+            return self.v  # pyright: ignore [reportReturnType]
+
+    state = ReturnTypeCheckState()
+    mock_error = mocker.patch("reflex_base.utils.console.error")
+    assert state.wrong_typed == 0
+    assert mock_error.call_count == 1
+    # Cache hits must not re-run the (potentially deep) type check.
+    assert state.wrong_typed == 0
+    assert mock_error.call_count == 1
+    # Invalidation triggers a recompute, which re-checks the return type.
+    state.v = 1
+    assert state.wrong_typed == 1
+    assert mock_error.call_count == 2
+
+
+def test_non_cached_computed_var_return_type_checked_every_access(mocker):
+    """A cache=False computed var recomputes, and is type-checked, on every access.
+
+    Args:
+        mocker: Pytest mocker object.
+    """
+
+    class NonCachedReturnTypeCheckState(BaseState):
+        v: int = 0
+
+        @rx.var(cache=False)
+        def wrong_typed(self) -> str:
+            return self.v  # pyright: ignore [reportReturnType]
+
+    state = NonCachedReturnTypeCheckState()
+    mock_error = mocker.patch("reflex_base.utils.console.error")
+    assert state.wrong_typed == 0
+    assert state.wrong_typed == 0
+    assert mock_error.call_count == 2
+
+
+async def test_async_computed_var_return_type_checked_only_on_recompute(mocker):
+    """The return type of a cached async computed var is only validated on recompute.
+
+    Args:
+        mocker: Pytest mocker object.
+    """
+
+    class AsyncReturnTypeCheckState(BaseState):
+        v: int = 0
+
+        @rx.var
+        async def wrong_typed(self) -> str:
+            return self.v  # pyright: ignore [reportReturnType]
+
+    state = AsyncReturnTypeCheckState()
+    mock_error = mocker.patch("reflex_base.utils.console.error")
+    assert await state.wrong_typed == 0
+    assert mock_error.call_count == 1
+    # Cache hits must not re-run the (potentially deep) type check.
+    assert await state.wrong_typed == 0
+    assert mock_error.call_count == 1
diff --git a/tests/units/test_state.py b/tests/units/test_state.py
index 05e06ee9f8c..130062f094b 100644
--- a/tests/units/test_state.py
+++ b/tests/units/test_state.py
@@ -1436,6 +1436,80 @@ def comp_v(self) -> int:
     assert comp_v_calls == 2
 
 
+def test_setattr_wrong_type_logs_error(mocker):
+    """Assigning a value of the wrong type to a base var logs an error in dev mode.
+
+    Args:
+        mocker: Pytest mocker object.
+    """
+
+    class WrongTypeState(BaseState):
+        n: int = 0
+
+    state = WrongTypeState()
+    mock_error = mocker.patch("reflex.utils.console.error")
+    setattr(state, "n", "not an int")  # noqa: B010
+    assert mock_error.call_count == 1
+
+
+def test_computed_var_recompute_after_mid_cycle_read():
+    """A dependency mutated again after a mid-cycle read still invalidates the cache."""
+
+    class FrontierState(BaseState):
+        v: int = 0
+
+        @rx.var
+        def doubled(self) -> int:
+            return self.v * 2
+
+    s = FrontierState()
+    s.v = 1
+    # Reading mid-cycle recomputes and re-caches the value.
+    assert s.doubled == 2
+    # Mutating the same dependency again must invalidate the fresh cache,
+    # even though dirty_vars already contained it.
+    s.v = 2
+    assert s.doubled == 4
+    assert s.get_delta()[s.get_full_name()]["doubled" + FIELD_MARKER] == 4
+
+
+def test_computed_var_recompute_after_mid_cycle_read_across_states():
+    """Cross-state dependency invalidation survives a mid-cycle recompute."""
+
+    class FrontierParentState(BaseState):
+        v: int = 0
+
+    class FrontierChildState(FrontierParentState):
+        @rx.var
+        def doubled(self) -> int:
+            return self.v * 2
+
+    parent = FrontierParentState()
+    child = parent.substates[FrontierChildState.get_name()]
+    assert isinstance(child, FrontierChildState)
+    parent.v = 1
+    assert child.doubled == 2
+    parent.v = 2
+    assert child.doubled == 4
+
+
+def test_interval_computed_vars_precomputed():
+    """Classes precompute which computed vars carry an update interval."""
+
+    class IntervalFreeState(BaseState):
+        @rx.var
+        def untimed(self) -> int:
+            return 2
+
+    class IntervalState(BaseState):
+        @rx.var(interval=15)
+        def timed(self) -> int:
+            return 1
+
+    assert IntervalFreeState._interval_computed_vars == ()
+    assert IntervalState._interval_computed_vars == ("timed",)
+
+
 def test_computed_var_cached_depends_on_non_cached():
     """Test that a cached var is recalculated if it depends on non-cached ComputedVar."""
 
@@ -3242,11 +3316,11 @@ class BaseFieldSetterState(BaseState):
     bfss.dirty_vars.clear()
     assert "c1" not in bfss.dirty_vars
 
-    # Assert identity of MutableProxy
+    # Repeated reads reuse the cached MutableProxy for the same field.
     mp = bfss.c1
     assert isinstance(mp, MutableProxy)
     mp3 = bfss.c1
-    assert mp is not mp3
+    assert mp is mp3
     # Since none of these set calls had values, the state should not be dirty
     assert not bfss.dirty_vars